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

/**
 * {@inheritDoc}
 * @see org.jboss.netty.channel.Channel#write(java.lang.Object)
 */
@Override
public ChannelFuture write(Object message) {
    if(message!=null) {
        if(message instanceof FileRegion) {
            try {
                Pipe pipe = Pipe.open();
                FileRegion fr = (FileRegion)message;

                long bytesToRead = fr.getCount();
                fr.transferTo(pipe.sink(), 0L);
                byte[] content = new byte[(int)bytesToRead];
                pipe.source().read(ByteBuffer.wrap(content));
                channelWrites.add(content);
            } catch (Exception ex) {
                log.error("Failed to read content from pipe", ex);
                channelWrites.add(ex);
            }
        } else {
            channelWrites.add(message);
        }
        log.info("Received Channel Write [{}]  type:[{}]", message, message.getClass().getName());
    }

    return Channels.succeededFuture(this);
}

SendBuffer acquire(Object message) {
    if (message instanceof ChannelBuffer) {
        return acquire((ChannelBuffer) message);
    }
    if (message instanceof FileRegion) {
        return acquire((FileRegion) message);
    }

    throw new IllegalArgumentException(
            "unsupported message type: " + message.getClass());
}

@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;
}

private SendBuffer acquire(FileRegion src) {
    if (src.getCount() == 0) {
        return EMPTY_BUFFER;
    }
    return new FileSendBuffer(src);
}

FileSendBuffer(FileRegion file) {
    this.file = file;
}

/**
 * Constructor.
 *
 * @param partition Release resources on this when done
 */
private ChanneFutureListenerMetrics(FileRegion partition) {
  this.partition = partition;
}

/**
 * Constructor.
 *
 * @param partition Release resources on this when done
 */
private ChanneFutureListenerMetrics(FileRegion partition) {
  this.partition = partition;
}