src/java/org/apache/cassandra/streaming/FileStreamTask.java
/**
     * Stream file by it's sections specified by this.header
     * @throws IOException on any I/O error
     */
Stream file by it's sections specified by this.header
@throws IOException on any I/O error
Stream file by it ' s sections specified by this . header @throws IOException on any I / O error
private void stream() throws IOException
    {
        ByteBuffer HeaderBuffer = MessagingService.instance().constructStreamHeader(header, false, Gossiper.instance.getVersion(to));
        output.write(ByteBufferUtil.getArray(HeaderBuffer));
        if (header.file == null)
            return;
        RandomAccessReader file = (header.file.sstable.compression) 
                                ? CompressedRandomAccessReader.open(header.file.getFilename(), header.file.sstable.getCompressionMetadata(), true)
                                : RandomAccessReader.open(new File(header.file.getFilename()), true);
        compressedoutput = new LZFOutputStream(output);
        MessagingService.instance().incrementActiveStreamsOutbound();
        try
        {
            for (Pair<Long, Long> section : header.file.sections)
            {
                file.seek(section.left);
                long length = section.right - section.left;
                long bytesTransferred = 0;
                while (bytesTransferred < length)
                {
                    long lastWrite = write(file, length, bytesTransferred);
                    bytesTransferred += lastWrite;
                    header.file.progress += lastWrite;
                }
                compressedoutput.flush();
                if (logger.isDebugEnabled())
                    logger.debug("Bytes transferred " + bytesTransferred + "/" + header.file.size);
            }
            receiveReply();
        }
        finally
        {
            MessagingService.instance().decrementActiveStreamsOutbound();
            FileUtils.closeQuietly(file);
        }
    }
private void stream ( ) throws IOException { ByteBuffer HeaderBuffer = MessagingService . instance ( ) . constructStreamHeader ( header , false , Gossiper . instance . getVersion ( to ) ) ; output . write ( ByteBufferUtil . getArray ( HeaderBuffer ) ) ; if ( header . file == null ) return ; RandomAccessReader file = ( header . file . sstable . compression ) ? CompressedRandomAccessReader . open ( header . file . getFilename ( ) , header . file . sstable . getCompressionMetadata ( ) , true ) : RandomAccessReader . open ( new File ( header . file . getFilename ( ) ) , true ) ; compressedoutput = new LZFOutputStream ( output ) ; MessagingService . instance ( ) . incrementActiveStreamsOutbound ( ) ; try { for ( Pair < Long , Long > section : header . file . sections ) { file . seek ( section . left ) ; long length = section . right - section . left ; long bytesTransferred = 0 ; while ( bytesTransferred < length ) { long lastWrite = write ( file , length , bytesTransferred ) ; bytesTransferred += lastWrite ; header . file . progress += lastWrite ; } compressedoutput . flush ( ) ; if ( logger . isDebugEnabled ( ) ) logger . debug ( "Bytes transferred " + bytesTransferred + "/" + header . file . size ) ; } receiveReply ( ) ; } finally { MessagingService . instance ( ) . decrementActiveStreamsOutbound ( ) ; FileUtils . closeQuietly ( file ) ; } }
Stream file by it's sections specified by this.header
 @throws IOException on any I/O error
Stream file by it ' s sections specified by this . header @throws IOException on any I / O error
// write header (this should not be compressed for compatibility with other messages) // TODO just use a raw RandomAccessFile since we're managing our own buffer here // try to skip kernel page cache if possible // setting up data compression stream // stream each of the required sections of the file // seek to the beginning of the section // length of the section to stream // tracks write progress // store streaming progress // make sure that current section is send // receive reply confirmation // no matter what happens close file