Java 类io.netty.handler.codec.http.DefaultFullHttpResponse 实例源码

/**
 * 接受http信息
 * @param ctx
 * @param req
 */
private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
    // Handle a bad request.
    if (!req.getDecoderResult().isSuccess()) {
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
        return;
    }

    // Handshake
    WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
            getWebSocketLocation(req), null, true);
    handshaker = wsFactory.newHandshaker(req);
    if (handshaker == null) {
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    } else {
        handshaker.handshake(ctx.channel(), req);
    }
}

private boolean writeResponse(HttpObject currentObj, ChannelHandlerContext ctx) {
    // Decide whether to close the connection or not.
    boolean keepAlive = HttpHeaders.isKeepAlive(request);

    // Build the response object.
    FullHttpResponse response = new DefaultFullHttpResponse(
        HTTP_1_1, currentObj.getDecoderResult().isSuccess() ? OK : BAD_REQUEST,
        Unpooled.copiedBuffer(buf.toString(), CharsetUtil.UTF_8));

    response.headers().set(CONTENT_TYPE, "application/json");

    if (keepAlive) {
        // Add 'Content-Length' header only for a keep-alive connection.
        response.headers().set(CONTENT_LENGTH, response.content().readableBytes());
        // Add keep alive header as per:
        // - http://www.w3.org/Protocols/HTTP/1.1/draft-ietf-http-v11-spec-01.html#Connection
        response.headers().set(CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
    }

    // Write the response.
    ctx.write(response);

    return keepAlive;
}

@Override
public FullHttpResponse get(ChannelHandlerContext channelHandlerContext, QueryDecoder queryDecoder, PathProvider path, HttpRequest httpRequest) throws Exception
{
    CloudNet.getLogger().debug("HTTP Request from " + channelHandlerContext.channel().remoteAddress());

    StringBuilder stringBuilder = new StringBuilder();

    try (InputStream inputStream = WebsiteDocumentation.class.getClassLoader().getResourceAsStream("files/api-doc.txt");
         BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)))
    {
        String input;
        while ((input = bufferedReader.readLine()) != null)
        {
            stringBuilder.append(input).append(System.lineSeparator());
        }
    }

    String output = stringBuilder.substring(0);
    ByteBuf byteBuf = Unpooled.wrappedBuffer(output.getBytes(StandardCharsets.UTF_8));
    FullHttpResponse fullHttpResponse = new DefaultFullHttpResponse(httpRequest.getProtocolVersion(), HttpResponseStatus.OK, byteBuf);
    fullHttpResponse.headers().set("Content-Type", "text/plain");
    return fullHttpResponse;
}

/**
 * Returns a full HTTP response with the specified status, content type, and custom headers.
 *
 * <p>Headers should be specified as a map of strings. For example, to allow CORS, add the
 * following key and value: "access-control-allow-origin", "http://foo.example"
 *
 * <p>If content type or content length are passed in as custom headers, they will be ignored.
 * Instead, content type will be as specified by the parameter contentType and content length will
 * be the length of the parameter contentLength.
 */
public static FullHttpResponse newResponse(
    HttpResponseStatus status,
    ByteBuf payload,
    ContentType contentType,
    Map<String, String> customHeaders) {
  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, payload);

  if (customHeaders != null) {
    for (Map.Entry<String, String> entry : customHeaders.entrySet()) {
      response.headers().set(entry.getKey(), entry.getValue());
    }
  }

  response.headers().set(CONTENT_TYPE, contentType.value);
  response.headers().setInt(CONTENT_LENGTH, payload.readableBytes());

  return response;
}

@Test
public void shouldHandlerRequestAndResponse() {
    inboundChannel.pipeline().addLast(handler);

    DefaultFullHttpRequest req = new DefaultFullHttpRequest(
            HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
    inboundChannel.write(req);

    assertEquals(1, inboundChannel.outboundMessages().size());

    Object outboundReq = inboundChannel.outboundMessages().poll();
    assertTrue(outboundReq instanceof ByteBuf);
    assertEquals("GET / HTTP/1.1\r\n\r\n", new String(readBytes((ByteBuf) outboundReq)));

    DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    assertFalse(inboundChannel.writeInbound(resp));

    assertEquals(1, outboundChannel.outboundMessages().size());
    assertEquals(resp, outboundChannel.outboundMessages().poll());

    resp.release();
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req)
        throws Exception {
    if (!req.getDecoderResult().isSuccess()) {
        logger.debug("invalid http request");
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                BAD_REQUEST));
        return;
    }

    if (req.getUri().equalsIgnoreCase(this.websocketUri)) {
        logger.debug("it is websocket request");
        ctx.fireChannelRead(req.retain());
        return;
    }

    HttpTransport transport = getTransport(req);
    if (transport == null) {
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                BAD_REQUEST));
    } else {
        transport.handleRequest(ctx, req);
    }
}

@Override
public void handleRequest(ChannelHandlerContext ctx, FullHttpRequest req)
        throws Exception {
    if (req.getUri().contains("/jsonp/connect")) {
        handleConnect(ctx, req);
    } else if (req.getUri().contains("/jsonp/subscribe")) {
        handleSubscrible(ctx, req);
    } else if (req.getUri().contains("/jsonp/waiting")) {
        handleWaitingMsg(ctx, req);
    } else if (req.getUri().contains("/jsonp/unsubscrible")) {
        handleUnsubscrible(ctx, req);
    } else if (req.getUri().contains("/jsonp/publish")) {
        handlePublish(ctx, req);
    } else if (req.getUri().contains("/jsonp/disconnect")) {
        handleDisconnect(ctx, req);
    } else { // invalid request
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                BAD_REQUEST));
    }
}

private void handleUnsubscrible(ChannelHandlerContext ctx, HttpRequest req) {
    if (!HttpSessionStore.checkJSessionId(req)) {
        sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                UNAUTHORIZED));
        return;
    }

    String topic = HttpSessionStore.getParameter(req, "topic");
    String sessionId = HttpSessionStore.getClientSessionId(req);

    HttpChannelEntity httpChannelEntity = (HttpChannelEntity) MemoryMetaPool
            .getChannelEntryByClientId(sessionId);

    MemoryMetaPool.unregisterTopic(httpChannelEntity, topic);

    Set<String> topicSet = MemoryMetaPool
            .getTopicsByChannelEntry(httpChannelEntity);

    Map<String, Object> map = new HashMap<String, Object>(2);
    map.put("status", true);
    map.put("topics", topicSet);
    String result = gson.toJson(map);

    logger.debug("unregister topic = " + topic + " and output = " + result);
    sendFullHttpOKResponse(ctx, req, result);
}

@Override
protected ChannelHandler newNonSslHandler(ChannelHandlerContext context) {
    return new ChannelInboundHandlerAdapter() {

        private HttpResponseEncoder encoder = new HttpResponseEncoder();

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            LOG.trace("Received non-SSL request, returning redirect");
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.MOVED_PERMANENTLY, Unpooled.EMPTY_BUFFER);
            response.headers().set(Names.LOCATION, redirectAddress);
            LOG.trace(Constants.LOG_RETURNING_RESPONSE, response);
            encoder.write(ctx, response, ctx.voidPromise());
            ctx.flush();
        }
    };
}

@Test
public void testBasicAuthenticationFailure() throws Exception {
    Configuration config = TestConfiguration.createMinimalConfigurationForTest();

    BasicAuthLogin auth = new BasicAuthLogin();
    auth.setUsername("test");
    auth.setPassword("test2");
    DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/login");
    request.content().writeBytes(JsonSerializer.getObjectMapper().writeValueAsBytes(auth));

    TestHttpQueryDecoder decoder = new TestHttpQueryDecoder(config);
    decoder.decode(null, request, results);
    Assert.assertEquals(1, results.size());
    Object result = results.iterator().next();
    Assert.assertEquals(BasicAuthLoginRequest.class, result.getClass());

    BasicAuthLoginRequestHandler handler = new BasicAuthLoginRequestHandler(config);
    CaptureChannelHandlerContext ctx = new CaptureChannelHandlerContext();
    handler.channelRead(ctx, result);
    Assert.assertNotNull(ctx.msg);
    Assert.assertTrue(ctx.msg instanceof DefaultFullHttpResponse);
    DefaultFullHttpResponse response = (DefaultFullHttpResponse) ctx.msg;
    Assert.assertEquals(HttpResponseStatus.UNAUTHORIZED, response.getStatus());
    Assert.assertTrue(response.headers().contains(Names.CONTENT_TYPE));
    Assert.assertEquals(Constants.JSON_TYPE, response.headers().get(Names.CONTENT_TYPE));
}

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();
}

private void sendTextResource(String prepend, String name, String mimeType, FullHttpRequest req, ChannelHandlerContext ctx) throws IOException {
    BufferedReader reader = new BufferedReader(new InputStreamReader((this.getResourceAsStream(name))));
    // TODO: read only once and buffer
    String line;
    StringBuffer buffer = new StringBuffer();
    if (prepend != null) buffer.append(prepend);
    while ((line = reader.readLine()) != null) {
        buffer.append(line);
        buffer.append('\n');
    }
    ByteBuf content = Unpooled.copiedBuffer(buffer, java.nio.charset.Charset.forName("UTF-8"));

    FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, content);

    res.headers().set(HttpHeaderNames.CONTENT_TYPE, mimeType);
    HttpUtil.setContentLength(res, content.readableBytes());

    sendHttpResponse(ctx, req, res);
}

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

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
        throws Exception {
  Exception e = cause instanceof Exception ? (Exception) cause : new
    Exception(cause);
  final String output = JsonUtil.toJsonString(e);
  ByteBuf content = Unpooled.wrappedBuffer(output.getBytes(Charsets.UTF_8));
  final DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
          HTTP_1_1, INTERNAL_SERVER_ERROR, content);

  resp.headers().set(CONTENT_TYPE, APPLICATION_JSON_UTF8);
  if (e instanceof IllegalArgumentException) {
    resp.setStatus(BAD_REQUEST);
  } else if (e instanceof FileNotFoundException) {
    resp.setStatus(NOT_FOUND);
  }

  resp.headers().set(CONTENT_LENGTH, resp.content().readableBytes());
  resp.headers().set(CONNECTION, CLOSE);
  ctx.write(resp).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof HttpRequest) {
        HttpRequest req = (HttpRequest) msg;

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

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

protected HttpResponse handleNonProxyRequest(FullHttpRequest req) {
    String uri = req.getUri();
    if ("/version".equals(uri)) {
        if (HttpMethod.GET.equals(req.getMethod())) {
            JsonObject jsonObj = new JsonObject();
            jsonObj.addProperty("name", m_appConfig.getAppName());
            jsonObj.addProperty("version", m_appConfig.getAppVersion());
            byte[] content = jsonObj.toString().getBytes(CharsetUtil.UTF_8);
            DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
                    HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                    Unpooled.copiedBuffer(content));
            HttpHeaders.setKeepAlive(resp, false);
            HttpHeaders.setHeader(resp, HttpHeaders.Names.CONTENT_TYPE,
                    "application/json");
            HttpHeaders.setContentLength(resp, content.length);
            return resp;
        }
    }

    return RESPONSE_404;
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
        throws Exception {
  Exception e = cause instanceof Exception ? (Exception) cause : new
    Exception(cause);
  final String output = JsonUtil.toJsonString(e);
  ByteBuf content = Unpooled.wrappedBuffer(output.getBytes(Charsets.UTF_8));
  final DefaultFullHttpResponse resp = new DefaultFullHttpResponse(
          HTTP_1_1, INTERNAL_SERVER_ERROR, content);

  resp.headers().set(CONTENT_TYPE, APPLICATION_JSON_UTF8);
  if (e instanceof IllegalArgumentException) {
    resp.setStatus(BAD_REQUEST);
  } else if (e instanceof FileNotFoundException) {
    resp.setStatus(NOT_FOUND);
  } else if (e instanceof IOException) {
    resp.setStatus(FORBIDDEN);
  }
  resp.headers().set(CONTENT_LENGTH, resp.content().readableBytes());
  resp.headers().set(CONNECTION, CLOSE);
  ctx.write(resp).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    logger.error("Exception caught: " + cause);

    HttpResponseStatus status = (cause instanceof BadRequestException) ? HttpResponseStatus.BAD_REQUEST :
             HttpResponseStatus.INTERNAL_SERVER_ERROR;

    StringWriter stringWriter = new StringWriter();
    PrintWriter printWriter = new PrintWriter(stringWriter);
    cause.printStackTrace(printWriter);
    String content = stringWriter.toString();

    FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1,
            status, Unpooled.copiedBuffer(content, CharsetUtil.UTF_8));

    response.headers().set(CONTENT_TYPE, "application/json; charset=UTF-8");
    response.headers().set(CONTENT_LENGTH,
            response.content().readableBytes());

    ctx.writeAndFlush(response);
    ctx.close();
}

public boolean sendResponseMessage(String requestId, String statusCode, String body) {
    ChannelHandlerContext ctx = sessionMap.get(requestId);

    if(ctx == null) return false;

    removeSession(requestId);

    try {
        DefaultFullHttpResponse response = this.makeHttpResponse(HttpResponseStatus.valueOf(Integer.parseInt(statusCode)), body != null ? body.getBytes() : null);
        response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);

        HttpServerHandler.sendHttpMessage(response, ctx.channel()).
                                addListener(ChannelFutureListener.CLOSE).
                                addListener(new FilnalEventListener(ctx, true));
    } catch (Exception e) {
        log.debug("Handled exception", e);

        sendError(ctx);
    }

    return true;

}

@Override
protected void channelRead0(ChannelHandlerContext ctx, SuggestRequest msg) throws Exception {
    byte[] buf = null;
    try {
        buf = JsonUtil.getObjectMapper().writeValueAsBytes(dataStore.suggest(msg));
    } catch (TimelyException e) {
        LOG.error(e.getMessage(), e);
        this.sendHttpError(ctx, e);
        return;
    }
    FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
            Unpooled.copiedBuffer(buf));
    response.headers().set(Names.CONTENT_TYPE, Constants.JSON_TYPE);
    response.headers().set(Names.CONTENT_LENGTH, response.content().readableBytes());
    sendResponse(ctx, response);
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, SearchLookupRequest msg) throws Exception {
    byte[] buf = null;
    try {
        buf = JsonUtil.getObjectMapper().writeValueAsBytes(dataStore.lookup(msg));
    } catch (TimelyException e) {
        LOG.error(e.getMessage(), e);
        this.sendHttpError(ctx, e);
        return;
    }
    FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
            Unpooled.copiedBuffer(buf));
    response.headers().set(Names.CONTENT_TYPE, Constants.JSON_TYPE);
    response.headers().set(Names.CONTENT_LENGTH, response.content().readableBytes());
    sendResponse(ctx, response);
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("netty4-http:http://0.0.0.0:{{port}}/foo")
                .to("mock:input")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                                                                            NettyConverter.toByteBuffer("Bye World".getBytes()));
                        response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, 9);

                        exchange.getOut().setBody(response);
                    }
                });
        }
    };
}

public static FullHttpResponse generateNoMatchResponse(RecordedHttpRequest recordedHttpRequest) {
  StringBuilder bodyTextBuilder = new StringBuilder();
  bodyTextBuilder.append("No Matching Request\n").append("Incoming Request Method: ")
      .append(recordedHttpRequest.getMethod()).append("\n").append("Incoming Request URI: ")
      .append(recordedHttpRequest.getUri()).append("\n").append("Incoming Request Headers: ")
      .append(recordedHttpRequest.getHeaders()).append("\n");
  RecordedHttpBody incomingBody = recordedHttpRequest.getHttpBody();
  if (incomingBody != null) {
    if (incomingBody instanceof RecordedEncodedHttpBody) {
      incomingBody = ((RecordedEncodedHttpBody) incomingBody).getDecodedBody();
    }
    if (incomingBody instanceof RecordedStringHttpBody) {
      bodyTextBuilder.append("Incoming Request Body: ").append(((RecordedStringHttpBody) incomingBody).getContent());
    } else {
      bodyTextBuilder.append("Incoming Request Body: (binary content)");
    }
  }
  ByteBuf badRequestBody = Unpooled.wrappedBuffer(bodyTextBuilder.toString().getBytes(Charset.forName("UTF-8")));
  return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST, badRequestBody);
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {

    ByteBuf buf = msg.content();
    byte[] bytes = new byte[buf.readableBytes()];
    buf.getBytes(0, bytes);

    YarRequest yarRequest = YarProtocol.buildRequest(bytes);
    YarResponse yarResponse = process(yarRequest);

    FullHttpResponse response =
            new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(YarProtocol
                    .toProtocolBytes(yarResponse)));
    response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "application/x-www-form-urlencoded");
    response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, response.content().readableBytes());
    if (HttpHeaders.isKeepAlive(msg)) {
        response.headers().set(HttpHeaders.Names.CONNECTION, Values.KEEP_ALIVE);
    }
    ctx.write(response);
    ctx.flush();
    ctx.close();
}

@Test
public void testBuild()
    throws IOException {
  HttpResponse httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_0, HttpResponseStatus.GATEWAY_TIMEOUT);
  RecordedHttpResponseBuilder recordedHttpResponseBuilder = new RecordedHttpResponseBuilder(httpResponse);

  String charset = "UTF-8";
  String str1 = "Hello world";
  HttpContent httpContent1 = new DefaultHttpContent(Unpooled.copiedBuffer(str1.getBytes(charset)));
  recordedHttpResponseBuilder.appendHttpContent(httpContent1);
  String str2 = "second content";
  HttpContent httpContent2 = new DefaultHttpContent(Unpooled.copiedBuffer(str2.getBytes(charset)));
  recordedHttpResponseBuilder.appendHttpContent(httpContent2);

  String lastStr = "Last chunk";
  HttpContent lastContent = new DefaultLastHttpContent(Unpooled.copiedBuffer(lastStr.getBytes(charset)));
  recordedHttpResponseBuilder.appendHttpContent(lastContent);
  RecordedHttpResponse recordedHttpResponse = recordedHttpResponseBuilder.build();
  Assert.assertEquals(recordedHttpResponse.getStatus(), HttpResponseStatus.GATEWAY_TIMEOUT.code());
  Assert.assertEquals((str1 + str2 + lastStr).getBytes(charset),
      recordedHttpResponse.getHttpBody().getContent(charset));
}

@Override
public DefaultFullHttpResponse handler(HttpContext context, DefaultFullHttpResponse response) {
    if (accessLogAppender != null) {
        String accessOnce = new AccessLogGenerator()
                .setIp(context.getRemoteAddress())
                .setTime(new Date(context.getCreationTime()))
                .setHttpMethod(context.getRequestMethod().name())
                .setURL(context.getUri())
                .setHttpCode(response.getStatus().hashCode())
                .setConsume(System.currentTimeMillis() - context.getCreationTime())
                .setTransitionSize(response.content().readableBytes())
                .setReqeustID(context.getRequestId())
                .makeAccess();

        accessLogAppender.writeLine(accessOnce);
    }

    return super.handler(context, response);
}

@Override
public boolean sendHttpResponse(OneM2mResponse resMessage) {
    ChannelHandlerContext ctx = sessionMap.get(resMessage.getRequestIdentifier());
    DefaultFullHttpResponse response = null;

    if(ctx == null) return false;

    removeSession(resMessage.getRequestIdentifier());

    try {
        response = HttpResponseCodec.encode(resMessage, httpVersion);
        response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);

        HttpServerHandler.sendHttpMessage(response, ctx.channel()).
                                addListener(ChannelFutureListener.CLOSE).
                                addListener(new FilnalEventListener(ctx, true));
    } catch (Exception e) {
        log.debug("Handled exception", e);

        sendError(ctx);
    }

    return true;
}

private void writeResponse(final Channel channel, final HttpResponseStatus statusCode) {
    // Convert the response content to a ChannelBuffer.
    final ByteBuf buf = copiedBuffer(responseContent.toString(), CharsetUtil.UTF_8);
    responseContent.setLength(0);
    // Decide whether to close the connection or not.
    final boolean close = HttpHeaders.Values.CLOSE.equalsIgnoreCase(request.headers().get(CONNECTION)) ||
            request.getProtocolVersion().equals(HttpVersion.HTTP_1_0) && !HttpHeaders.Values.KEEP_ALIVE.equalsIgnoreCase(request.headers().get(CONNECTION));
    // Build the response object.
    final FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, statusCode, buf);
    response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
    if (!close) {
        // There's no need to add 'Content-Length' header if this is the last response.
        response.headers().set(CONTENT_LENGTH, buf.readableBytes());
    }
    // Write the response.
    final ChannelFuture future = channel.writeAndFlush(response);
    // Close the connection after the write operation is done if necessary.
    if (close) {
        future.addListener(ChannelFutureListener.CLOSE);
    }
}

private static void sendError(final ChannelHandlerContext ctx, final HttpResponseStatus status,
                              final String message, final Optional<Throwable> t) {
    if (t.isPresent())
        logger.warn(String.format("Invalid request - responding with %s and %s", status, message), t.get());
    else
        logger.warn(String.format("Invalid request - responding with %s and %s", status, message));

    errorMeter.mark();
    final ObjectNode node = mapper.createObjectNode();
    node.put("message", message);
    final FullHttpResponse response = new DefaultFullHttpResponse(
            HTTP_1_1, status, Unpooled.copiedBuffer(node.toString(), CharsetUtil.UTF_8));
    response.headers().set(CONTENT_TYPE, "application/json");

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

@Override
public HttpResponse clientToProxyRequest(HttpObject httpObject) {
    if (httpObject instanceof HttpRequest) {
        HttpRequest httpRequest = (HttpRequest) httpObject;

        String url = getFullUrl(httpRequest);

        for (BlacklistEntry entry : blacklistedUrls) {
            if (HttpMethod.CONNECT.equals(httpRequest.getMethod()) && entry.getHttpMethodPattern() == null) {
                // do not allow CONNECTs to be blacklisted unless a method pattern is explicitly specified
                continue;
            }

            if (entry.matches(url, httpRequest.getMethod().name())) {
                HttpResponseStatus status = HttpResponseStatus.valueOf(entry.getStatusCode());
                HttpResponse resp = new DefaultFullHttpResponse(httpRequest.getProtocolVersion(), status);
                HttpHeaders.setContentLength(resp, 0L);

                return resp;
            }
        }
    }

    return null;
}

@Override
protected void onOutboundError(Throwable err) {

    if (!channel().isActive()) {
        super.onOutboundError(err);
        return;
    }

    discreteRemoteClose(err);
    if (markSentHeaders()) {
        log.error("Error starting response. Replying error status", err);

        HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                HttpResponseStatus.INTERNAL_SERVER_ERROR);
        response.headers()
                .setInt(HttpHeaderNames.CONTENT_LENGTH, 0)
                .set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
        channel().writeAndFlush(response)
                 .addListener(ChannelFutureListener.CLOSE);
        return;
    }

    markSentBody();
    channel().writeAndFlush(EMPTY_BUFFER)
             .addListener(ChannelFutureListener.CLOSE);
}

@Test
    public void prematureCancel() throws Exception {
        DirectProcessor<Void> signal = DirectProcessor.create();
        NettyContext x = TcpServer.create("localhost", 0)
                                  .newHandler((in, out) -> {
                                        signal.onComplete();
                                        return out.context(c -> c.addHandlerFirst(
                                                new HttpResponseEncoder()))
                                                  .sendObject(Mono.delay(Duration
                                                          .ofSeconds(2))
                                                          .map(t ->
                                                          new DefaultFullHttpResponse(
                                                                  HttpVersion.HTTP_1_1,
                                                                  HttpResponseStatus
                                                                          .PROCESSING)))
                                                .neverComplete();
                                  })
                                  .block(Duration.ofSeconds(30));

        StepVerifier.create(createHttpClientForContext(x)
                                      .get("/")
                                      .timeout(signal)
        )
                    .verifyError(TimeoutException.class);
//      Thread.sleep(1000000);
    }

private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {

        // Handle a bad request.
        if (!req.decoderResult().isSuccess()) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
            return;
        }

        // Allow only GET methods.
        if (req.method() != GET) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
            return;
        }

        // Handshake
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                getWebSocketLocation(req), null, false);
        handshaker = wsFactory.newHandshaker(req);

        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req)throws Exception {
    FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, FOUND);

    HttpHeaders headers = req.headers();

    IConfig cfg = Config.getInstance();

    StringBuilder sb = new StringBuilder();

    if (cfg.isEncrypted()) {
        sb.append(StringCache.HTTPS);
    } else {
        sb.append(StringCache.HTTP);
    }

    //finish up the url.
    sb.append(headers.get(HttpHeaderNames.HOST)).append(StringCache.COLON).append(cfg.getPort()).append(req.uri());

    //apply the redirect url
    response.headers().set(HttpHeaderNames.LOCATION, sb.toString());

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

private void writeResponse(Channel channel, Response response, HttpRequest httpRequest) {
    ByteBuf buf = Unpooled.copiedBuffer(JsonCodec.encodeResponse(response), CharsetUtil.UTF_8);
    FullHttpResponse httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);

    String contentType = "text/html; charset=UTF-8";
    httpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, contentType);

    boolean close = httpRequest.headers().contains(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE, true)
            || httpRequest.protocolVersion().equals(HttpVersion.HTTP_1_0)
            || !httpRequest.headers().contains(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE, true);

    if (!close) {
        httpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, buf.readableBytes());
    }

    ChannelFuture future = channel.writeAndFlush(response);
    future.addListener(ChannelFutureListener.CLOSE);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof HttpRequest) {
        HttpRequest req = (HttpRequest) msg;
        if (HttpUtil.is100ContinueExpected(req)) {
            ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
        }

        ///http://127.0.0.1:8080/aa/bb
        System.out.println();
        System.out.println(req.method());//     GET
        System.out.println(req.uri());   //     /aa/bb

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

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

@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof HttpRequest) {
        HttpRequest req = (HttpRequest) msg;

        if (is100ContinueExpected(req)) {
            ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
        }
        boolean keepAlive = isKeepAlive(req);

        ByteBuf content = Unpooled.copiedBuffer("Hello World " + new Date(), CharsetUtil.UTF_8);

        FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, content);
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
        response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());

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

private void writeAuthenticationRequired(String realm) {
    String body = "<!DOCTYPE HTML \"-//IETF//DTD HTML 2.0//EN\">\n"
            + "<html><head>\n"
            + "<title>407 Proxy Authentication Required</title>\n"
            + "</head><body>\n"
            + "<h1>Proxy Authentication Required</h1>\n"
            + "<p>This server could not verify that you\n"
            + "are authorized to access the document\n"
            + "requested.  Either you supplied the wrong\n"
            + "credentials (e.g., bad password), or your\n"
            + "browser doesn't understand how to supply\n"
            + "the credentials required.</p>\n" + "</body></html>\n";
    DefaultFullHttpResponse response = responseFor(HttpVersion.HTTP_1_1,
            HttpResponseStatus.PROXY_AUTHENTICATION_REQUIRED, body);
    HttpHeaders.setDate(response, new Date());
    response.headers().set("Proxy-Authenticate",
            "Basic realm=\"" + (realm == null ? "Restricted Files" : realm) + "\"");
    write(response);
}

/**
 * 写入数据到客户端
 *
 * @param messageResult
 */
public void write(final MessageResult messageResult) {
    String json = messageResult.toJson();
    ByteBuf content = Unpooled.copiedBuffer(json, CharsetUtil.UTF_8);
    FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, content);
    res.headers().set(CONTENT_TYPE, "text/html; charset=UTF-8");
    HttpHeaders.setContentLength(res, content.readableBytes());

    // Send the response
    ChannelFuture f = this.channel.writeAndFlush(res);
}

public void sendRedirect(ChannelHandlerContext ctx, String newUri) {
   FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.FOUND);
   response.headers().set(HttpHeaders.Names.LOCATION, newUri);

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