agave-systems/systems-core/src/main/java/org/iplantc/service/transfer/URLCopy.java
/**
     * Performs the transfer by streaming data from one system to another through
     * the local server.
     * <p>
     * If the destPath is a directory, then the srcPath will be copied to a file with
     * the exact same name within the destPath directory. This keeps the behavior
     * identical to that of the {@link RemoteDataClient#put(String, String)} method
     * used in the {@link #relayTransfer(String, String, TransferTask)} method.
     * </p>
     *
     * @param srcPath agave source path of the file on the remote system
     * @param destPath the destination agave path of the transfer on the remote system
     * @param listener the listener to track the transfer info
     * @throws RemoteDataException
     * @throws IOException
     * @throws TransferException
     * @throws ClosedByInterruptException
     */
Performs the transfer by streaming data from one system to another through
the local server.

If the destPath is a directory, then the srcPath will be copied to a file with
the exact same name within the destPath directory. This keeps the behavior
identical to that of the  RemoteDataClient#put(String, String) method
used in the  #relayTransfer(String, String, TransferTask) method.


Performs the transfer by streaming data from one system to another through the local server . If the destPath is a directory then the srcPath will be copied to a file with the exact same name within the destPath directory . This keeps the behavior identical to that of the RemoteDataClient#put ( String String ) method used in the #relayTransfer ( String String TransferTask ) method .
protected void streamingTransfer(String srcPath, String destPath, RemoteTransferListener listener)
            throws RemoteDataException, IOException, TransferException, ClosedByInterruptException {
        RemoteInputStream<?> in = null;
        InputStream bis = null;
        RemoteOutputStream<?> out = null;
        OutputStream bos = null;
        long bytesSoFar = 0;
        try {
            log.debug(String.format(
                    "Beginning streaming transfer for task %s. %s to %s . Protocol: %s => %s",
                    listener.getTransferTask().getUuid(),
                    listener.getTransferTask().getSource(),
                    listener.getTransferTask().getDest(),
                    getProtocolForClass(sourceClient.getClass()),
                    getProtocolForClass(destClient.getClass())));
            long totalSize = sourceClient.length(srcPath);
            try {
                in = getInputStream(sourceClient, srcPath);
            } catch (Exception e) {
                log.error("Unable to get input stream for " + sourceClient.getUsername() +
                        " on host " + sourceClient.getHost() + " for path " + srcPath + ".");
                throw e;
            }
            if (in.isBuffered()) bis = in;
            else bis = new BufferedInputStream(in);
            checkCancelled(listener);
            try {
                out = getOutputStream(destClient, destPath);
            } catch (Exception e) {
                log.error("Unable to get output stream for " + destClient.getUsername() +
                        " on host " + destClient.getHost() + " for path " + destPath + ".");
                throw e;
            }
            if (out.isBuffered()) bos = out;
            else bos = new BufferedOutputStream(out);
            checkCancelled(listener);
            int length = 0;
            long callbackTime = System.currentTimeMillis();
            int bufferSize = Math.min(sourceClient.getMaxBufferSize(), destClient.getMaxBufferSize());
            byte[] b = new byte[bufferSize];
            listener.started(totalSize, srcPath);
            while ((length = bis.read(b, 0, bufferSize)) != -1) {
                bytesSoFar += length;
                bos.write(b, 0, length);
                if (System.currentTimeMillis() > (callbackTime + 10000)) {
                    checkCancelled(listener);
                    callbackTime = System.currentTimeMillis();
                    listener.progressed(bytesSoFar);
                }
            }
            listener.progressed(bytesSoFar);
            listener.completed();
            log.debug(String.format(
                    "Completed streaming transfer for task %s. %s to %s . Protocol: %s => %s",
                    listener.getTransferTask().getUuid(),
                    listener.getTransferTask().getSource(),
                    listener.getTransferTask().getDest(),
                    getProtocolForClass(sourceClient.getClass()),
                    getProtocolForClass(destClient.getClass())));
        } catch (ClosedByInterruptException e) {
            log.debug(String.format(
                    "Aborted streaming transfer for task %s. %s to %s . Protocol: %s => %s",
                    listener.getTransferTask().getUuid(),
                    listener.getTransferTask().getSource(),
                    listener.getTransferTask().getDest(),
                    getProtocolForClass(destClient.getClass()),
                    getProtocolForClass(destClient.getClass())), e);
            log.info("Transfer task " + listener.getTransferTask().getUuid() + " killed by worker shutdown.");
            setKilled(true);
            listener.progressed(bytesSoFar);
            listener.cancel();
            Thread.currentThread().interrupt();
            throw e;
        } catch (RemoteDataException | IOException e) {
            log.debug(String.format(
                    "Failed streaming transfer for task %s. %s to %s . Protocol: %s => %s",
                    listener.getTransferTask().getUuid(),
                    listener.getTransferTask().getSource(),
                    listener.getTransferTask().getDest(),
                    getProtocolForClass(sourceClient.getClass()),
                    getProtocolForClass(destClient.getClass())), e);
            listener.failed();
            throw e;
        } catch (Throwable e) {
            log.debug(String.format(
                    "Failed streaming transfer for task %s. %s to %s . Protocol: %s => %s",
                    listener.getTransferTask().getUuid(),
                    listener.getTransferTask().getSource(),
                    listener.getTransferTask().getDest(),
                    getProtocolForClass(sourceClient.getClass()),
                    getProtocolForClass(destClient.getClass())), e);
            listener.failed();
            throw new RemoteDataException("Transfer task %s failed.", e);
        } finally {
            if (listener != null && listener.getTransferTask() != null) {
                log.info(String.format(
                        "Total of %s bytes transferred in task %s . Protocol %s => %s",
                        listener.getTransferTask().getBytesTransferred(),
                        listener.getTransferTask().getUuid(),
                        getProtocolForClass(sourceClient.getClass()),
                        getProtocolForClass(destClient.getClass())));
            }
            try { if (bis != null) bis.close(); bis = null; } catch (Throwable ignored) {}
            try { if (bos != null) bos.close(); bos = null; } catch (Throwable ignored) {}
            try { if (in != null) in.close(); in = null; } catch (Throwable ignored) {}
            try { if (out != null) out.close(); out = null; } catch (Throwable ignored) {}
        }
    }
protected void streamingTransfer ( String srcPath , String destPath , RemoteTransferListener listener ) throws RemoteDataException , IOException , TransferException , ClosedByInterruptException { RemoteInputStream < ? > in = null ; InputStream bis = null ; RemoteOutputStream < ? > out = null ; OutputStream bos = null ; long bytesSoFar = 0 ; try { log . debug ( String . format ( "Beginning streaming transfer for task %s. %s to %s . Protocol: %s => %s" , listener . getTransferTask ( ) . getUuid ( ) , listener . getTransferTask ( ) . getSource ( ) , listener . getTransferTask ( ) . getDest ( ) , getProtocolForClass ( sourceClient . getClass ( ) ) , getProtocolForClass ( destClient . getClass ( ) ) ) ) ; long totalSize = sourceClient . length ( srcPath ) ; try { in = getInputStream ( sourceClient , srcPath ) ; } catch ( Exception e ) { log . error ( "Unable to get input stream for " + sourceClient . getUsername ( ) + " on host " + sourceClient . getHost ( ) + " for path " + srcPath + "." ) ; throw e ; } if ( in . isBuffered ( ) ) bis = in ; else bis = new BufferedInputStream ( in ) ; checkCancelled ( listener ) ; try { out = getOutputStream ( destClient , destPath ) ; } catch ( Exception e ) { log . error ( "Unable to get output stream for " + destClient . getUsername ( ) + " on host " + destClient . getHost ( ) + " for path " + destPath + "." ) ; throw e ; } if ( out . isBuffered ( ) ) bos = out ; else bos = new BufferedOutputStream ( out ) ; checkCancelled ( listener ) ; int length = 0 ; long callbackTime = System . currentTimeMillis ( ) ; int bufferSize = Math . min ( sourceClient . getMaxBufferSize ( ) , destClient . getMaxBufferSize ( ) ) ; byte [ ] b = new byte [ bufferSize ] ; listener . started ( totalSize , srcPath ) ; while ( ( length = bis . read ( b , 0 , bufferSize ) ) != - 1 ) { bytesSoFar += length ; bos . write ( b , 0 , length ) ; if ( System . currentTimeMillis ( ) > ( callbackTime + 10000 ) ) { checkCancelled ( listener ) ; callbackTime = System . currentTimeMillis ( ) ; listener . progressed ( bytesSoFar ) ; } } listener . progressed ( bytesSoFar ) ; listener . completed ( ) ; log . debug ( String . format ( "Completed streaming transfer for task %s. %s to %s . Protocol: %s => %s" , listener . getTransferTask ( ) . getUuid ( ) , listener . getTransferTask ( ) . getSource ( ) , listener . getTransferTask ( ) . getDest ( ) , getProtocolForClass ( sourceClient . getClass ( ) ) , getProtocolForClass ( destClient . getClass ( ) ) ) ) ; } catch ( ClosedByInterruptException e ) { log . debug ( String . format ( "Aborted streaming transfer for task %s. %s to %s . Protocol: %s => %s" , listener . getTransferTask ( ) . getUuid ( ) , listener . getTransferTask ( ) . getSource ( ) , listener . getTransferTask ( ) . getDest ( ) , getProtocolForClass ( destClient . getClass ( ) ) , getProtocolForClass ( destClient . getClass ( ) ) ) , e ) ; log . info ( "Transfer task " + listener . getTransferTask ( ) . getUuid ( ) + " killed by worker shutdown." ) ; setKilled ( true ) ; listener . progressed ( bytesSoFar ) ; listener . cancel ( ) ; Thread . currentThread ( ) . interrupt ( ) ; throw e ; } catch ( RemoteDataException | IOException e ) { log . debug ( String . format ( "Failed streaming transfer for task %s. %s to %s . Protocol: %s => %s" , listener . getTransferTask ( ) . getUuid ( ) , listener . getTransferTask ( ) . getSource ( ) , listener . getTransferTask ( ) . getDest ( ) , getProtocolForClass ( sourceClient . getClass ( ) ) , getProtocolForClass ( destClient . getClass ( ) ) ) , e ) ; listener . failed ( ) ; throw e ; } catch ( Throwable e ) { log . debug ( String . format ( "Failed streaming transfer for task %s. %s to %s . Protocol: %s => %s" , listener . getTransferTask ( ) . getUuid ( ) , listener . getTransferTask ( ) . getSource ( ) , listener . getTransferTask ( ) . getDest ( ) , getProtocolForClass ( sourceClient . getClass ( ) ) , getProtocolForClass ( destClient . getClass ( ) ) ) , e ) ; listener . failed ( ) ; throw new RemoteDataException ( "Transfer task %s failed." , e ) ; } finally { if ( listener != null && listener . getTransferTask ( ) != null ) { log . info ( String . format ( "Total of %s bytes transferred in task %s . Protocol %s => %s" , listener . getTransferTask ( ) . getBytesTransferred ( ) , listener . getTransferTask ( ) . getUuid ( ) , getProtocolForClass ( sourceClient . getClass ( ) ) , getProtocolForClass ( destClient . getClass ( ) ) ) ) ; } try { if ( bis != null ) bis . close ( ) ; bis = null ; } catch ( Throwable ignored ) { } try { if ( bos != null ) bos . close ( ) ; bos = null ; } catch ( Throwable ignored ) { } try { if ( in != null ) in . close ( ) ; in = null ; } catch ( Throwable ignored ) { } try { if ( out != null ) out . close ( ) ; out = null ; } catch ( Throwable ignored ) { } } }
Performs the transfer by streaming data from one system to another through
 the local server.
Performs the transfer by streaming data from one system to another through the local server .
// The "b" in the variable names means "buffered". // Buffer the input stream only if it's not already buffered. // Buffer the output stream only if it's not already buffered. // update the progress every 15 seconds buffer cycle. This reduced the impact // from the observing process while keeping the update interval at a // rate the user can somewhat trust // check to see if this transfer has been cancelled due to outside // intervention such as updating the TransferTask record or the parent // thread interrupting this one and setting the AbstractTransferTask.cancelled // field to true // update with the final transferred blocks and wrap the transfer. // transfer failed due to connectivity issue // stuff happens, what are you going to do.