Java 类org.apache.commons.io.input.TeeInputStream 实例源码

@Override
public ServletInputStream getInputStream() throws IOException {
    return new ServletInputStream() {
        @Override
        public boolean isFinished() {
            return false;
        }

        @Override
        public boolean isReady() {
            return false;
        }

        @Override
        public void setReadListener(ReadListener readListener) {

        }

        private TeeInputStream tee = new TeeInputStream(RequestWrapper.super.getInputStream(), bos);

        @Override
        public int read() throws IOException {
            return tee.read();
        }
    };
}

@Override
public ServletInputStream getInputStream() throws IOException {
    return new ServletInputStream() {
        @Override
        public boolean isFinished() {
            return false;
        }

        @Override
        public boolean isReady() {
            return false;
        }

        @Override
        public void setReadListener(ReadListener readListener) {

        }

        private TeeInputStream tee = new TeeInputStream(RequestWrapper.super.getInputStream(), bos, true);

        @Override
        public int read() throws IOException {
            return tee.read();
        }
    };
}

@Override
public void run() {
    logger.debug("Checking for a change");
    if (targetChanged()) {
        logger.debug("Target changed, checking if it's different than current flow.");
        try (FileInputStream configFile = new FileInputStream(configFilePath.toFile());
            ByteArrayOutputStream pipedOutputStream = new ByteArrayOutputStream();
            TeeInputStream teeInputStream = new TeeInputStream(configFile, pipedOutputStream)) {

            if (differentiator.isNew(teeInputStream)) {
                logger.debug("New change, notifying listener");
                // Fill the byteArrayOutputStream with the rest of the request data
                while (teeInputStream.available() != 0) {
                    teeInputStream.read();
                }

                ByteBuffer newConfig = ByteBuffer.wrap(pipedOutputStream.toByteArray());
                ByteBuffer readOnlyNewConfig = newConfig.asReadOnlyBuffer();

                configurationChangeNotifier.notifyListeners(readOnlyNewConfig);
                logger.debug("Listeners notified");
            }
        } catch (Exception e) {
            logger.error("Could not successfully notify listeners.", e);
        }
    }
}

private InputStream getConfigInputStream() throws IOException {
    MCRConfigurationInputStream configurationInputStream = MCRConfigurationInputStream
        .getMyCoRePropertiesInstance();
    File configFile = MCRConfigurationDir.getConfigFile("mycore.active.properties");
    if (configFile != null) {
        FileOutputStream fout = new FileOutputStream(configFile);
        return new TeeInputStream(configurationInputStream, fout, true);
    }
    return configurationInputStream;
}

public InputStream fetchInputStream(URL url, String prefix, String suffix) throws IOException {
    createCacheDir(prefix);
    File localDatasourceFile = urlToLocalFile(url, prefix, suffix);
    log.info("Fetching local file: {}", localDatasourceFile.getCanonicalPath());
    if (!localDatasourceFile.exists()){
        log.info("Local file not found: {} \nDownloading external resource: {}",
                                            localDatasourceFile.getCanonicalPath(), url.toString());
        URLConnection connection = url.openConnection();
        if (suffix.equals(".json")) connection.setRequestProperty("Accept", "application/json");
        return new TeeInputStream(connection.getInputStream(), new FileOutputStream(localDatasourceFile));
    } else {
        return new FileInputStream(localDatasourceFile);
    }
}

private InputStream getConfigInputStream() throws IOException {
    ConfigurationInputStream configurationInputStream = ConfigurationInputStream.getConfigPropertiesInstance();
    File configFile = ConfigurationDir.getConfigFile("config.active.properties");
    if (configFile != null) {
        FileOutputStream fout = new FileOutputStream(configFile);
        TeeInputStream tin = new TeeInputStream(configurationInputStream, fout, true);
        return tin;
    }
    return configurationInputStream;
}

/**
    * Creates an input stream that maps the remote file chunk
    * 
    * @param remoteSession
    *            the current file session in use
    * @param fileChunk
    *            the fileChunk to store the downlad in
    * @param remoteFile
    *            the remote file name with path
    * @return the input stream that maps the remote file
    * @throws UnknownHostException
    * @throws ConnectException
    * @throws RemoteException
    * @throws IOException
    * @throws InvalidLoginException
    * @throws FileNotFoundException
    */
   public InputStream downloadOneChunk(
    File fileChunk, String remoteFile, long chunkLength) throws UnknownHostException,
    ConnectException, RemoteException, IOException,
    InvalidLoginException, FileNotFoundException {
// debug("downloadFile Begin");

// Prepare the request parameters
List<SimpleNameValuePair> requestParams = new Vector<SimpleNameValuePair>();
requestParams.add(new SimpleNameValuePair(Parameter.ACTION,
    Action.DOWNLOAD_FILE_ACTION));
requestParams.add(new SimpleNameValuePair(Parameter.USERNAME,
    username));
requestParams.add(new SimpleNameValuePair(Parameter.TOKEN, authenticationToken));
requestParams
    .add(new SimpleNameValuePair(Parameter.FILENAME, remoteFile));
requestParams.add(new SimpleNameValuePair(Parameter.CHUNKLENGTH, ""
    + chunkLength));

InputStream in = httpTransfer.getInputStream(requestParams);

// If there is a non null FileChunk, save content of stream for reuse
// in case of download interruptions
if (fileChunk != null) {
    OutputStream out = new BufferedOutputStream(new FileOutputStream(
        fileChunk));
    TeeInputStream teeIn = new TeeInputStream(in, out, true);
    return teeIn;
} else {
    return in;
}
   }

private Blob pipeBlobAndReturn(String container, Blob blob) throws IOException {
    String name = blob.getMetadata().getName();
    logger.debug("piping {} from {} to {}", name, getDestStoreName(), getSourceStoreName());

    PipedInputStream pipeIn = new PipedInputStream();
    PipedOutputStream pipeOut = new PipedOutputStream(pipeIn);

    Payload blobPayload = blob.getPayload();
    MutableContentMetadata contentMetadata = blob.getMetadata().getContentMetadata();
    Blob retBlob = new BlobImpl(replaceSystemMetadata(blob.getMetadata()));
    retBlob.setPayload(pipeIn);
    retBlob.setAllHeaders(blob.getAllHeaders());
    TeeInputStream tee = new TeeInputStream(blobPayload.openStream(), pipeOut, true);
    retBlob.getMetadata().setContentMetadata(contentMetadata);

    app.executeBackgroundTask(() -> {
        try {
            logger.debug("copying {} to tee stream", name);
            return Utils.copyBlob(getDestination(), getSource(), container, blob, tee);
        } catch (RuntimeException e) {
            logger.error("copying " + name + " to tee stream failed", e);
            throw e;
        } finally {
            tee.close();
        }
    });
    return retBlob;
}

@Override
public ServletInputStream getInputStream() throws IOException {
    return new ServletInputStream() {
        private TeeInputStream tee = new TeeInputStream(RequestWrapper.super.getInputStream(), bos);

        @Override
        public int read() throws IOException {
            return tee.read();
        }

        @Override
        public boolean isFinished() {
            // Auto-generated method stub
            return false;
        }

        @Override
        public boolean isReady() {
            // Auto-generated method stub
            return false;
        }

        @Override
        public void setReadListener(ReadListener listener) {
            // Auto-generated method stub
        }
    };
}

@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response)
        throws IOException, ServletException {

    logRequest(request);

    baseRequest.setHandled(true);

    if (POST.equals(request.getMethod())) {
        int statusCode;
        String responseText;
        try (ByteArrayOutputStream pipedOutputStream = new ByteArrayOutputStream();
             TeeInputStream teeInputStream = new TeeInputStream(request.getInputStream(), pipedOutputStream)) {

            if (differentiator.isNew(teeInputStream)) {
                // Fill the pipedOutputStream with the rest of the request data
                while (teeInputStream.available() != 0) {
                    teeInputStream.read();
                }

                ByteBuffer newConfig = ByteBuffer.wrap(pipedOutputStream.toByteArray());
                ByteBuffer readOnlyNewConfig = newConfig.asReadOnlyBuffer();

                Collection<ListenerHandleResult> listenerHandleResults = configurationChangeNotifier.notifyListeners(readOnlyNewConfig);

                statusCode = 200;
                for (ListenerHandleResult result : listenerHandleResults) {
                    if (!result.succeeded()) {
                        statusCode = 500;
                        break;
                    }
                }
                responseText = getPostText(listenerHandleResults);
            } else {
                statusCode = 409;
                responseText = "Request received but instance is already running this config.";
            }

            writeOutput(response, responseText, statusCode);
        }
    } else if (GET.equals(request.getMethod())) {
        writeOutput(response, GET_TEXT, 200);
    } else {
        writeOutput(response, OTHER_TEXT, 404);
    }
}

protected void processRequest(HttpServletRequest req, HttpServletResponse res) throws Exception {

        String dest = req.getParameter(DEST_PARAM);
        String preload = req.getParameter(PRELOAD_PARAM);
        String overrideCacheKey= req.getParameter(CACHE_KEY);
        String fileType= req.getParameter(FILE_TYPE);

        if (! ServletFileUpload.isMultipartContent(req)) {
            sendReturnMsg(res, 400, "Is not a Multipart request. Request rejected.", "");
        }
        StopWatch.getInstance().start("Upload File");

        ServletFileUpload upload = new ServletFileUpload();
        FileItemIterator iter = upload.getItemIterator(req);
        while (iter.hasNext()) {
            FileItemStream item = iter.next();

            if (!item.isFormField()) {
                String fileName = item.getName();
                InputStream inStream = new BufferedInputStream(item.openStream(), IpacTableUtil.FILE_IO_BUFFER_SIZE);
                String ext = resolveExt(fileName);
                FileType fType = resolveType(fileType, ext, item.getContentType());
                File destDir = resolveDestDir(dest, fType);
                boolean doPreload = resolvePreload(preload, fType);

                File uf = File.createTempFile("upload_", ext, destDir);
                String rPathInfo = ServerContext.replaceWithPrefix(uf);

                UploadFileInfo fi= new UploadFileInfo(rPathInfo,uf,fileName,item.getContentType());
                String fileCacheKey= overrideCacheKey!=null ? overrideCacheKey : rPathInfo;
                UserCache.getInstance().put(new StringKey(fileCacheKey), fi);

                if (doPreload && fType == FileType.FITS) {
                    BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(uf), IpacTableUtil.FILE_IO_BUFFER_SIZE);
                    TeeInputStream tee = new TeeInputStream(inStream, bos);
                    try {
                        final Fits fits = new Fits(tee);
                        FitsRead[] frAry = FitsRead.createFitsReadArray(fits);
                        FitsCacher.addFitsReadToCache(uf, frAry);
                    } finally {
                        FileUtil.silentClose(bos);
                        FileUtil.silentClose(tee);
                    }
                } else {
                    FileUtil.writeToFile(inStream, uf);
                }
                sendReturnMsg(res, 200, null, fileCacheKey);
                Counters.getInstance().increment(Counters.Category.Upload, fi.getContentType());
                return;
            }
        }
        StopWatch.getInstance().printLog("Upload File");
    }

private InputStreamReader spawnSpy(InputStream in, PipedOutputStream pipeout) throws InterruptedException {
    return new InputStreamReader(new TeeInputStream(in,pipeout));
}