Java 类io.netty.channel.ChannelFutureListener 实例源码

public void checkProducerTransactionState(
    final Channel channel,
    final CheckTransactionStateRequestHeader requestHeader,
    final SelectMappedBufferResult selectMappedBufferResult) {
    RemotingCommand request =
        RemotingCommand.createRequestCommand(RequestCode.CHECK_TRANSACTION_STATE, requestHeader);
    request.markOnewayRPC();

    try {
        FileRegion fileRegion =
            new OneMessageTransfer(request.encodeHeader(selectMappedBufferResult.getSize()),
                selectMappedBufferResult);
        channel.writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                selectMappedBufferResult.release();
                if (!future.isSuccess()) {
                    log.error("invokeProducer failed,", future.cause());
                }
            }
        });
    } catch (Throwable e) {
        log.error("invokeProducer exception", e);
        selectMappedBufferResult.release();
    }
}

/**
 * Reopens the listening channel for this node. If the channel was already open, has no effect and
 * future completes immediately.
 *
 * @return future that completes when listening channel is reopened.
 */
private CompletableFuture<Void> rebind() {
  if (this.channel.get().isOpen()) {
    // already accepting...
    return CompletableFuture.completedFuture(null);
  }
  CompletableFuture<Void> future = new CompletableFuture<>();
  ChannelFuture bindFuture = bootstrap.bind(this.getAddress());
  bindFuture.addListener(
      (ChannelFutureListener)
          channelFuture -> {
            if (channelFuture.isSuccess()) {
              channelFuture.channel().attr(Server.HANDLER).set(this);
              logger.debug("Bound {} to {}", BoundNode.this, channelFuture.channel());
              future.complete(null);
              channel.set(channelFuture.channel());
            } else {
              // If failed, propagate it.
              future.completeExceptionally(
                  new BindNodeException(BoundNode.this, getAddress(), channelFuture.cause()));
            }
          });
  return future;
}

/**
 * 返回http信息
 * @param ctx
 * @param req
 * @param res
 */
private static void sendHttpResponse(
        ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.getStatus().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        HttpHeaders.setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.channel().writeAndFlush(res);
    if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

public void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
   if (res.getStatus().code() != 200) {
      ByteBuf f = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
      res.content().clear();
      res.content().writeBytes(f);
      f.release();
   }

   HttpHeaders.setContentLength(res, res.content().readableBytes());
   ChannelFuture f1;
   f1 = ctx.channel().writeAndFlush(res);

   if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
      f1.addListener(ChannelFutureListener.CLOSE);
   }
}

static void closeChannelGracefully(InetSocketAddress udpSource) {
    Channel socksChannel = removeSocksMapping(udpSource);
    Channel udpChannel = removeUdpMapping(udpSource);
    Channel tcpChannel = removeTcpMapping(udpSource);
    if (tcpChannel.isActive()) {
        tcpChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        log.info("\t          Proxy << Target \tDisconnect");
    }
    if (socksChannel.isActive()) {
        socksChannel.close();
        log.info("\tClient << Proxy           \tDisconnect");
    }
    if (udpChannel.isActive()) {
        udpChannel.close();
    }
}

/**
 * Write and flush channel context
 *
 * @param close whether close http connection
 */
private void writeFlush(boolean close) {

    if (flushed()) {

        return;
    }

    setDate();
    setPowerBy();
    setResponseTime();

    if (close) {

        channelCxt.writeAndFlush(httpResponse).addListener(ChannelFutureListener.CLOSE);
    } else {

        header(CONNECTION, "keep-alive");
        channelCxt.writeAndFlush(httpResponse);
    }

    flush();
}

private void sendOutOfMemory(OutOfMemoryException e, final ChannelHandlerContext ctx, int coordinationId){
  final UserException uex = UserException.memoryError(e)
      .message("Out of memory while receiving data.")
      .build(logger);

  final OutboundRpcMessage outMessage = new OutboundRpcMessage(
      RpcMode.RESPONSE_FAILURE,
      0,
      coordinationId,
      uex.getOrCreatePBError(false)
      );

  if (RpcConstants.EXTRA_DEBUGGING) {
    logger.debug("Adding message to outbound buffer. {}", outMessage);
  }

  ChannelFuture future = ctx.writeAndFlush(outMessage);
  // if we were unable to report back the failure make sure we close the channel otherwise we may cause the sender
  // to block undefinitely waiting for an ACK on this message
  future.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
}

@Override
protected void doStart() {
    try {
        ChannelFuture f = bootstrap.bind(address);
        f.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    notifyStarted();
                } else {
                    notifyFailed(future.cause());
                }
            }
        });
    } catch (Throwable t) {
        notifyFailed(t);
        Throwables.propagate(t);
    }
}

private void onGetFileChecksum(ChannelHandlerContext ctx) throws IOException {
  MD5MD5CRC32FileChecksum checksum = null;
  final String nnId = params.namenodeId();
  DFSClient dfsclient = newDfsClient(nnId, conf);
  try {
    checksum = dfsclient.getFileChecksum(path, Long.MAX_VALUE);
    dfsclient.close();
    dfsclient = null;
  } finally {
    IOUtils.cleanup(LOG, dfsclient);
  }
  final byte[] js = JsonUtil.toJsonString(checksum).getBytes(Charsets.UTF_8);
  DefaultFullHttpResponse resp =
    new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(js));

  resp.headers().set(CONTENT_TYPE, APPLICATION_JSON_UTF8);
  resp.headers().set(CONTENT_LENGTH, js.length);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
}

public static void main(String[] args) throws IOException, InterruptedException {
    Bootstrap b = new Bootstrap();
    b.group(new NioEventLoopGroup())
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                }
            });
    b.connect("localhost", 8090).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                future.channel().write(Unpooled.buffer().writeBytes("123".getBytes()));
                future.channel().flush();
                future.channel().close();
            }
        }
    });
}

public static int writeBack(Channel channel, boolean isSuccess, String resultStr, boolean isKeepAlive) {
    ByteBuf content = Unpooled.copiedBuffer(resultStr, Constants.DEFAULT_CHARSET);
    HttpResponseStatus status;
    if (isSuccess) {
        status = HttpResponseStatus.OK;
    } else {
        status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
    }
    FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, status, content);
    //logger.info("result str:{}", resultStr);
    res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
    HttpHeaders.setContentLength(res, content.readableBytes());
    try {
        ChannelFuture f = channel.writeAndFlush(res);
        if (isKeepAlive) {
            HttpHeaders.setKeepAlive(res, true);
        } else {
            HttpHeaders.setKeepAlive(res, false);//set keepalive closed
            f.addListener(ChannelFutureListener.CLOSE);
        }
    } catch (Exception e2) {
        logger.warn("Failed to send HTTP response to remote, cause by:", e2);
    }

    return content.readableBytes();
}

public static void sendMessage(AbstractBinaryEncoder encoder, Channel channel, Message msg,
        Proc3<Boolean, Throwable, Channel> listener) throws MessageCodecException {
    if (channel == null) {
        Procs.invoke(listener, false, new NullPointerException("channel"), null);
        return;
    }
    if (!channel.isWritable()) {
        Procs.invoke(listener, false, new IOException(" channel " + channel.remoteAddress() + " is unwritable"),
                channel);
        return;
    }
    ByteBuf buffer = Unpooled.buffer();
    encoder.beforeWriteBody(buffer, msg.getMessageId());
    msg.buffer(buffer);
    try {
        msg.encode();
    } catch (Exception e) {
        throw new MessageCodecException(e);
    }
    msg.buffer(null);
    encoder.afterWriteBody(buffer);
    flow(msg.getClass(), buffer);
    channel.writeAndFlush(buffer).addListener((ChannelFutureListener) arg0 -> {
        Procs.invoke(listener, arg0.isSuccess(), arg0.cause(), arg0.channel());
    });
}

private void onSocksSuccess(ChannelHandlerContext ctx, Socks5CommandRequest request) {
    Address serverAddr = new Address(request.dstAddr(), request.dstPort());
    createServerChannel(ctx, serverAddr).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                ConnectionInfo newConnectionInfo = new ConnectionInfo(
                        connectionInfo.getClientAddr(), serverAddr);
                ctx.writeAndFlush(new DefaultSocks5CommandResponse(
                        Socks5CommandStatus.SUCCESS,
                        request.dstAddrType(),
                        request.dstAddr(),
                        request.dstPort()));

                onServerConnected(ctx, newConnectionInfo, future.channel());
            } else {
                ctx.channel().writeAndFlush(new DefaultSocks5CommandResponse(
                        Socks5CommandStatus.FAILURE,
                        request.dstAddrType(),
                        request.dstAddr(),
                        request.dstPort()));
                ctx.close();
            }
        }
    });
}

@Override
public void channelRead(final ChannelHandlerContext context, Object msg) {
    if(client.getInbound().isActive()) {
        client.getInbound().writeAndFlush(msg).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) {
                if (future.isSuccess()) {
                    context.channel().read();
                } else {
                    future.channel().close();
                }
            }
        });
    } else {
        client.getOutbound().close();
    }
}

protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object object) throws Exception {
    System.out.println("handler中exportedServices：" + exportedServices.size());
    Request request = (Request)object;
    String serviceName = request.getInterfaceName();
    String methodName = request.getMethodName();
    Class<?>[] parameterTypes = request.getParameterTypes();
    Object[] arguments = request.getArgs();
    Class serviceClass = exportedServices.get(serviceName);

    Method method = serviceClass.getMethod(methodName,parameterTypes);
    Object result = method.invoke(serviceClass.newInstance(),arguments);

    Response response = new Response();
    response.setResult(result);

    channelHandlerContext.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

private void sendToWire(MessageRoundtrip messageRoundtrip) {

        if (messageRoundtrip != null && messageRoundtrip.getRetryTimes() == 0) {
            // TODO: retry logic || messageRoundtrip.hasToRetry()){

            Message msg = messageRoundtrip.getMsg();

            ethereumListener.onSendMessage(channel, msg);

            ctx.writeAndFlush(msg).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);

            if (msg.getAnswerMessage() != null) {
                messageRoundtrip.incRetryTimes();
                messageRoundtrip.saveTime();
            }
        }
    }

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if(msg instanceof  HttpContent){
        System.out.println(msg);
    }
    if (msg instanceof HttpRequest) {
        HttpRequest req = (HttpRequest) msg;
        String path = URI.create(req.getUri()).getPath();

        boolean keepAlive = HttpHeaders.isKeepAlive(req);
        FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(CONTENT));
        response.headers().set(CONTENT_TYPE, "text/plain");
        response.headers().set(CONTENT_LENGTH, response.content().readableBytes());

        if (!keepAlive) {
            ctx.write(response).addListener(ChannelFutureListener.CLOSE);
        } else {
            response.headers().set(CONNECTION, Values.KEEP_ALIVE);
            ctx.write(response);
        }
    }
}

@Override
public Future<Channel> getConnection(HostAndPort address)
{
    try {
        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .option(CONNECT_TIMEOUT_MILLIS, saturatedCast(connectTimeout.toMillis()))
                .handler(new ThriftClientInitializer(
                        messageFraming,
                        messageEncoding,
                        requestTimeout,
                        socksProxy,
                        sslContextSupplier));

        Promise<Channel> promise = group.next().newPromise();
        bootstrap.connect(new InetSocketAddress(address.getHost(), address.getPort()))
                .addListener((ChannelFutureListener) future -> notifyConnect(future, promise));
        return promise;
    }
    catch (Throwable e) {
        return group.next().newFailedFuture(new TTransportException(e));
    }
}

private static void sendHttpResponse(ChannelHandlerContext ctx,
        HttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.getStatus().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
                CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.writeAndFlush(res);
    if (!isKeepAlive(req) || res.getStatus().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

private static void sendHttpResponse(ChannelHandlerContext ctx,
        HttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.getStatus().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
                CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.writeAndFlush(res);
    if (!isKeepAlive(req) || res.getStatus().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

public Future startAsync() {
    bossGroup = new NioEventLoopGroup(1);
    workerGroup = new NioEventLoopGroup();
    bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new SocksServerInitializer())
             .childAttr(OPTION_ATTRIBUTE_KEY, option);

    return bootstrap.bind(option.getLocalHost(), option.getLocalPort()).addListener(
            new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (infoEnable) {
                        if (future.isSuccess()) {
                            logger.info("Listening on local port {}", option.getLocalPort());
                        } else {
                            logger.info("Shadowsocks client startup failed", future.cause());
                        }
                    }
                }
            });
}

private ReplyMsg doWriteAndSync(final Channel channel, final AskMsg request, final long timeout, final WriteFuture<BaseMsg> writeFuture) throws Exception {

        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                writeFuture.setWriteResult(future.isSuccess());
                writeFuture.setCause(future.cause());
                //失败移除
                if (!writeFuture.isWriteSuccess()) {
                    SyncWriteMap.syncKey.remove(writeFuture.requestId());
                }
            }
        });

        ReplyMsg response = (ReplyMsg)writeFuture.get(timeout, TimeUnit.MILLISECONDS);
        if (response == null) {
            if (writeFuture.isTimeout()) {
                throw new TimeoutException();
            } else {
                // write exception
                throw new Exception(writeFuture.cause());
            }
        }
        return response;
    }

public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
     throws RemotingTimeoutException, RemotingSendRequestException {
    try {
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) throws Exception {
                if (!f.isSuccess()) {
                    log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                }
            }
        });
    } catch (Exception e) {
        log.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
        throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
    }
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  LOG.debug("Error ", cause);
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
    throws Exception {
  // If not complete, try to negotiate
  if (!saslClient.isComplete()) {
    super.write(ctx, msg, promise);
  } else {
    ByteBuf in = (ByteBuf) msg;

    try {
      saslToken = saslClient.wrap(in.array(), in.readerIndex(), in.readableBytes());
    } catch (SaslException se) {
      try {
        saslClient.dispose();
      } catch (SaslException ignored) {
        LOG.debug("Ignoring SASL exception", ignored);
      }
      promise.setFailure(se);
    }
    if (saslToken != null) {
      ByteBuf out = ctx.channel().alloc().buffer(4 + saslToken.length);
      out.writeInt(saslToken.length);
      out.writeBytes(saslToken, 0, saslToken.length);

      ctx.write(out).addListener(new ChannelFutureListener() {
        @Override public void operationComplete(ChannelFuture future) throws Exception {
          if (!future.isSuccess()) {
            exceptionCaught(ctx, future.cause());
          }
        }
      });

      saslToken = null;
    }
  }
}

@Override
protected void channelRead0(ChannelHandlerContext aContext, FullHttpRequest aRequest) throws Exception {

    FullHttpResponse response = createResponse(aRequest);

    response.headers()
            .add("vetch", "1.0")
            .setInt(CONTENT_LENGTH, response.content().readableBytes());

    aContext.writeAndFlush(response)
            .addListener(ChannelFutureListener.CLOSE);
}

@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) {
  client.writeAndFlush(msg).addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) {
      if (future.isSuccess()) {
        ctx.channel().read();
      } else {
        LOG.debug("Proxy failed. Cause: ", future.cause());
        future.channel().close();
      }
    }
  });
}

public void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
   FullHttpResponse response = new DefaultFullHttpResponse(
         HttpVersion.HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
   response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain; charset=UTF-8");

   // Close the connection as soon as the error message is sent.
   ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    cause.printStackTrace();
    if (ctx.channel().isActive()) {
        ctx.writeAndFlush("ERR: " + cause.getClass().getSimpleName() + ": " + cause.getMessage() + '\n')
                .addListener(ChannelFutureListener.CLOSE);
    }
}

@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) {
    inboundChannel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> {
        if (future.isSuccess()) {
            ctx.channel().read();
        } else {
            future.channel().close();
        }
    });
}

@Override
public void putResponseBodyInString(String payload, int retCode, String encoding) {

    // this.response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(retCode),
    // Unpooled.copiedBuffer(payload, CharsetUtil.UTF_8));
    this.response.setStatus(HttpResponseStatus.valueOf(retCode));

    this.response.content().writeBytes(Unpooled.copiedBuffer(payload, Charset.forName(encoding)));

    ctx.write(response);

    // write end marker
    ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  log.error("[{}] Unexpected Exception: {}", sessionId, cause);
  try {
    if (cause.getCause() instanceof ReadTimeoutException) {
      ctx.write(PINGRESP).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
    } else {
      ctx.close();
    }
  } catch (Throwable t) {
    t.printStackTrace();
    ctx.close();
  }

}

static void closeChannelGracefully(InetSocketAddress udpSource) {
    Channel udpChannel = removeUdpMapping(udpSource);
    Channel tcpChannel = removeTcpMapping(udpSource);
    if (udpChannel.isActive()) {
        log.info("\t          Proxy << Target \tDisconnect");
        udpChannel.close();
    }
    if (tcpChannel.isActive()) {
        tcpChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        log.info("\tClient << Proxy           \tDisconnect");
    }
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    IdleStateEvent event = (IdleStateEvent) evt;
    if (event.state() == IdleState.WRITER_IDLE) {
        ctx.writeAndFlush("heartbeat").addListener((ChannelFutureListener) future -> {
            if (! future.isSuccess()) {
                future.channel().close();
            }
        });
    } else {
        ctx.fireUserEventTriggered(evt);
    }
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof IdleStateEvent) {
        ctx.writeAndFlush(heart.duplicate()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

public static void closeChannel(Channel channel) {
    final String addrRemote = parseRemoteAddr(channel);
    channel.close().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            logger.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote,
                    future.isSuccess());
        }
    });
}

private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
    // Generate an error page if response getStatus code is not OK (200).
    if (res.status().code() != 200) {
        ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
        res.content().writeBytes(buf);
        buf.release();
        HttpUtil.setContentLength(res, res.content().readableBytes());
    }

    // Send the response and close the connection if necessary.
    ChannelFuture f = ctx.channel().writeAndFlush(res);
    if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
        f.addListener(ChannelFutureListener.CLOSE);
    }
}

private void closeChannel(Channel channel) {
    final String addrRemote = RemotingHelper.parseChannelRemoteAddr(channel);
    channel.close().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            logger.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote, future.isSuccess());
        }
    });
}

/**
 * When file timestamp is the same as what the browser is sending up, send a
 * "304 Not Modified"
 *
 * @param ctx Context
 */
private void sendNotModified(ChannelHandlerContext ctx) {
    FullHttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_MODIFIED);
    setDateHeader(response);
    // Close the connection as soon as the error message is sent.
    ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void write(Object msg) {
    channel.writeAndFlush(msg).addListener(new ChannelFutureListener() {

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                future.cause().printStackTrace();
            }
        }
    });
}