public void retrieve ( DataSource source ) { try { localControlChannel . resetReplyCount ( ) ; if ( session . transferMode != GridFTPSession . MODE_EBLOCK ) { // // No EBLOCK // EBlockParallelTransferContext context = ( EBlockParallelTransferContext ) createTransferContext ( ) ; context . setEodsTotal ( 0 ) ; logger . debug ( "starting outgoing transfer without mode E" ) ; if ( session . serverMode == Session . SERVER_PASSIVE ) { transferThreadManager . passiveConnect ( source , context , serverSocket ) ; } else { transferThreadManager . startTransfer ( source , context , 1 , ManagedSocketBox . NON_REUSABLE ) ; } return ; } else if ( session . serverMode == Session . SERVER_ACTIVE ) { // // EBLOCK, no striping // // data channels will share this transfer context EBlockParallelTransferContext context = ( EBlockParallelTransferContext ) createTransferContext ( ) ; int total = gSession . parallel ; //we should send as many EODS as there are parallel streams context . setEodsTotal ( total ) ; int free = socketPool . countFree ( ) ; int willReuseConnections = ( total > free ) ? free : total ; int willCloseConnections = ( free > total ) ? free - total : 0 ; int needNewConnections = ( total > free ) ? total - free : 0 ; logger . debug ( "will reuse " + willReuseConnections + " connections, start " + needNewConnections + " new ones, and close " + willCloseConnections ) ; if ( needNewConnections > 0 ) { transferThreadManager . activeConnect ( this . remoteServerAddress , needNewConnections ) ; } if ( willCloseConnections > 0 ) { transferThreadManager . activeClose ( context , willCloseConnections ) ; } transferThreadManager . startTransfer ( source , context , willReuseConnections + needNewConnections , ManagedSocketBox . REUSABLE ) ; } else if ( session . serverMode == GridFTPSession . SERVER_EACT ) { // // EBLOCK, striping // if ( stripeRetrContextManager == null ) { throw new IllegalStateException ( ) ; } int stripes = stripeRetrContextManager . getStripes ( ) ; for ( int stripe = 0 ; stripe < stripes ; stripe ++ ) { EBlockParallelTransferContext context = stripeRetrContextManager . getStripeContext ( stripe ) ; context . setEodsTotal ( gSession . parallel ) ; transferThreadManager . startTransfer ( source , context , gSession . parallel , ManagedSocketBox . REUSABLE ) ; } } else { // // EBLOCK and local server not active // throw new DataChannelException ( DataChannelException . BAD_SERVER_MODE ) ; } } catch ( Exception e ) { exceptionToControlChannel ( e , "ocurred during retrieve()" ) ; } }
Retrieve the data from the data source and write to the data channel . This method does not block . If operation fails exception might be thrown via local control channel .