Java 类org.jboss.netty.channel.DefaultFileRegion 实例源码

/**
 * @see org.apache.james.imap.encode.ImapResponseWriter#write(org.apache.james.imap.message.response.Literal)
 */
public void write(Literal literal) throws IOException {
    if (channel.isConnected()) {
        InputStream in = literal.getInputStream();
        if (in instanceof FileInputStream && channel.getFactory() instanceof NioServerSocketChannelFactory) {
            FileChannel fc = ((FileInputStream) in).getChannel();
               // Zero-copy is only possible if no SSL/TLS  and no COMPRESS is in place
            //
            // See JAMES-1305 and JAMES-1306
            ChannelPipeline cp = channel.getPipeline();
            if (zeroCopy && cp.get(SslHandler.class) == null && cp.get(ZlibEncoder.class) == null ) {
                channel.write(new DefaultFileRegion(fc, fc.position(), literal.size()));
            } else {
                channel.write(new ChunkedNioFile(fc, 8192));
            }
        } else {
            channel.write(new ChunkedStream(literal.getInputStream()));
        }
    }
}

public void release() {
    if (file instanceof DefaultFileRegion) {
        if (((DefaultFileRegion) file).releaseAfterTransfer()) {
            // Make sure the FileRegion resource are released otherwise it may cause a FD
            // leak or something similar
            file.releaseExternalResources();
        }
    }
}

protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch,
    String user, String mapId, int reduce, MapOutputInfo mapOutputInfo)
    throws IOException {
  final TezIndexRecord info = mapOutputInfo.indexRecord;
  final ShuffleHeader header =
    new ShuffleHeader(mapId, info.getPartLength(), info.getRawLength(), reduce);
  final DataOutputBuffer dob = new DataOutputBuffer();
  header.write(dob);
  ch.write(wrappedBuffer(dob.getData(), 0, dob.getLength()));
  final File spillfile =
      new File(mapOutputInfo.mapOutputFileName.toString());
  RandomAccessFile spill;
  try {
    spill = SecureIOUtils.openForRandomRead(spillfile, "r", user, null);
  } catch (FileNotFoundException e) {
    LOG.info(spillfile + " not found");
    return null;
  }
  ChannelFuture writeFuture;
  final DefaultFileRegion partition =
      new DefaultFileRegion(spill.getChannel(), info.getStartOffset(), info.getPartLength());
  writeFuture = ch.write(partition);
  writeFuture.addListener(new ChannelFutureListener() {
    // TODO error handling; distinguish IO/connection failures,
    //      attribute to appropriate spill output
    @Override
    public void operationComplete(ChannelFuture future) {
      partition.releaseExternalResources();
    }
  });
  return writeFuture;
}

@Override
public void messageReceived( ChannelHandlerContext ctx, MessageEvent e ) throws Exception
{
    HttpRequest request = (HttpRequest)e.getMessage( ) ;
    if( request.getMethod( ) != GET )
    {
        sendError( ctx, FORBIDDEN ) ;
        return ;
    }

    final String path = sanitizeUri( request.getUri( ) ) ;
    if( path == null )
    {
        sendError( ctx, FORBIDDEN ) ;
        return ;
    }

    File file = new File(path) ;
    if( file.isHidden( ) || !file.exists( ) )
    {
        sendError( ctx, NOT_FOUND ) ;
        return ;
    }

    RandomAccessFile raf ;
    try
    {
        raf = new RandomAccessFile( file, "r" ) ;
    }
    catch( FileNotFoundException fnfe )
    {
        sendError( ctx, NOT_FOUND ) ;
        return;
    }
    long fileLength = raf.length( ) ;
    HttpResponse response = new DefaultHttpResponse( HTTP_1_1, OK ) ;
    setContentLength( response, fileLength ) ;

    Channel ch = e.getChannel( ) ;

    //Escreve a linha inicial do cabe�alho
    ch.write( response ) ;

    // Escreve o conte�do
    ChannelFuture writeFuture ;
    if( ch.getPipeline( ).get( SslHandler.class ) != null )
    {
        writeFuture = ch.write( new ChunkedFile( raf, 0, fileLength, 8192 ) ) ;
    }
    else
    {
        final FileRegion region = new DefaultFileRegion( raf.getChannel( ), 0, fileLength ) ;
        writeFuture = ch.write( region ) ;
        writeFuture.addListener( new ChannelFutureProgressListener( )
        {
            @Override
            public void operationComplete( ChannelFuture arg0 ) throws Exception
            {
                region.releaseExternalResources( ) ;
            }

            @Override
            public void operationProgressed( ChannelFuture future, long amount, long current,  long total ) throws Exception
            {
                System.out.printf( "%s: %d / %d (+%d)%n", path, current, total, amount );
            }
        }) ;
    }

    // Decide se fecha a conex�o ou n�o!!
    if( !isKeepAlive( request ) )
    {
        writeFuture.addListener( ChannelFutureListener.CLOSE ) ;
    }
}

protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch,
    String jobId, String mapId, int reduce) throws IOException {
  LocalDirAllocator lDirAlloc = attributes.getLocalDirAllocator();
  FileSystem rfs = ((LocalFileSystem) attributes.getLocalFS()).getRaw();

  ShuffleServerMetrics shuffleMetrics = attributes.getShuffleServerMetrics();
  TaskTracker tracker = attributes.getTaskTracker();

  // Index file
  Path indexFileName = lDirAlloc.getLocalPathToRead(
      TaskTracker.getIntermediateOutputDir(jobId, mapId)
      + "/file.out.index", attributes.getJobConf());
  // Map-output file
  Path mapOutputFileName = lDirAlloc.getLocalPathToRead(
      TaskTracker.getIntermediateOutputDir(jobId, mapId)
      + "/file.out", attributes.getJobConf());

  /**
   * Read the index file to get the information about where
   * the map-output for the given reducer is available.
   */
  IndexRecord info = tracker.getIndexInformation(mapId, reduce,indexFileName);

  HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);

  //set the custom "from-map-task" http header to the map task from which
  //the map output data is being transferred
  response.setHeader(MRConstants.FROM_MAP_TASK, mapId);

  //set the custom "Raw-Map-Output-Length" http header to
  //the raw (decompressed) length
  response.setHeader(MRConstants.RAW_MAP_OUTPUT_LENGTH,
      Long.toString(info.rawLength));

  //set the custom "Map-Output-Length" http header to
  //the actual number of bytes being transferred
  response.setHeader(MRConstants.MAP_OUTPUT_LENGTH,
      Long.toString(info.partLength));

  //set the custom "for-reduce-task" http header to the reduce task number
  //for which this map output is being transferred
  response.setHeader(MRConstants.FOR_REDUCE_TASK, Integer.toString(reduce));

  ch.write(response);
  File spillfile = new File(mapOutputFileName.toString());
  RandomAccessFile spill;
  try {
    spill = new RandomAccessFile(spillfile, "r");
  } catch (FileNotFoundException e) {
    LOG.info(spillfile + " not found");
    return null;
  }
  final FileRegion partition = new DefaultFileRegion(
    spill.getChannel(), info.startOffset, info.partLength);
  ChannelFuture writeFuture = ch.write(partition);
  writeFuture.addListener(new ChanneFutureListenerMetrics(partition));
  shuffleMetrics.outputBytes(info.partLength); // optimistic
  LOG.info("Sending out " + info.partLength + " bytes for reduce: " +
           reduce + " from map: " + mapId + " given " +
           info.partLength + "/" + info.rawLength);
  return writeFuture;
}