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

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

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

private void writeResourceReport(Channel channel) {
  ByteBuf content = Unpooled.buffer();
  Writer writer = new OutputStreamWriter(new ByteBufOutputStream(content), CharsetUtil.UTF_8);
  try {
    reportAdapter.toJson(resourceReport.get(), writer);
    writer.close();
  } catch (IOException e) {
    LOG.error("error writing resource report", e);
    writeAndClose(channel, new DefaultFullHttpResponse(
      HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR,
      Unpooled.copiedBuffer(e.getMessage(), StandardCharsets.UTF_8)));
    return;
  }

  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
  HttpUtil.setContentLength(response, content.readableBytes());
  response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=UTF-8");
  channel.writeAndFlush(response);
}

@Override
public NettyOutbound send(Publisher<? extends ByteBuf> source) {
    if (method() == HttpMethod.GET || method() == HttpMethod.HEAD) {
        ByteBufAllocator alloc = channel().alloc();
        return then(Flux.from(source)
            .doOnNext(ByteBuf::retain)
            .collect(alloc::buffer, ByteBuf::writeBytes)
            .flatMapMany(agg -> {
                if (!hasSentHeaders() && !HttpUtil.isTransferEncodingChunked(
                        outboundHttpMessage()) && !HttpUtil.isContentLengthSet(
                        outboundHttpMessage())) {
                    outboundHttpMessage().headers()
                                         .setInt(HttpHeaderNames.CONTENT_LENGTH,
                                                 agg.readableBytes());
                }
                return send(Mono.just(agg)).then();
            }));
    }
    return super.send(source);
}

@Override
public Mono<Void> then() {
    if (markSentHeaders()) {
        if (HttpUtil.isContentLengthSet(outboundHttpMessage())) {
            outboundHttpMessage().headers()
                                 .remove(HttpHeaderNames.TRANSFER_ENCODING);
        }

        if (!HttpUtil.isTransferEncodingChunked(outboundHttpMessage())
                && !HttpUtil.isContentLengthSet(outboundHttpMessage())) {
            markPersistent(false);
        }

        return FutureMono.deferFuture(() -> channel().writeAndFlush(outboundHttpMessage()));
    }
    else {
        return Mono.empty();
    }
}

@Override
public final NettyOutbound sendFile(Path file, long position, long count) {
    Objects.requireNonNull(file);

    if (hasSentHeaders()) {
        return super.sendFile(file, position, count);
    }

    if (!HttpUtil.isTransferEncodingChunked(outboundHttpMessage()) && !HttpUtil.isContentLengthSet(
            outboundHttpMessage()) && count < Integer.MAX_VALUE) {
        outboundHttpMessage().headers()
                             .setInt(HttpHeaderNames.CONTENT_LENGTH, (int) count);
    }
    else if (!HttpUtil.isContentLengthSet(outboundHttpMessage())) {
        outboundHttpMessage().headers()
                             .remove(HttpHeaderNames.CONTENT_LENGTH)
                             .remove(HttpHeaderNames.TRANSFER_ENCODING);
        HttpUtil.setTransferEncodingChunked(outboundHttpMessage(), true);
    }

    return super.sendFile(file, position, count);
}

/**
 * Send http response.
 *
 * @param ctx the ctx
 * @param req the req
 * @param res the 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.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);
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
    if (HttpUtil.is100ContinueExpected(req)) {
        ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
    }
    boolean keepAlive = HttpUtil.isKeepAlive(req);

    ByteBuf content = ctx.alloc().buffer();
    content.writeBytes(HelloWorldHttp2Handler.RESPONSE_BYTES.duplicate());
    ByteBufUtil.writeAscii(content, " - via " + req.protocolVersion() + " (" + establishApproach + ")");

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

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

@Override
protected void sendResponse(final ChannelHandlerContext ctx, String streamId, int latency,
        final FullHttpResponse response, final FullHttpRequest request) {
    HttpUtil.setContentLength(response, response.content().readableBytes());
    ctx.executor().schedule(new Runnable() {
        @Override
        public void run() {
            if (isKeepAlive(request)) {
                response.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
                ctx.writeAndFlush(response);
            } else {
                ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
            }
        }
    }, latency, TimeUnit.MILLISECONDS);
}

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

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

@Override
public void handle(ChannelHandlerContext ctx, FullHttpRequest request, String path) throws Exception
{
    if (HttpUtil.is100ContinueExpected(request))
        ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
    boolean keepAlive = HttpUtil.isKeepAlive(request);
    byte[] data = plainTxt ? this.handlePlain() : this.handleWithHTML(ctx, request);
    FullHttpResponse response;
    if (data == null)
        response = new DefaultFullHttpResponse(HTTP_1_1, INTERNAL_SERVER_ERROR);
    else
        response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(data));
    response.headers().set(CONTENT_TYPE, plainTxt ? "text/plain" : "text/html");
    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);
    }
    ctx.flush();
}

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

public static void beginHTTPResponse(ChannelHandlerContext ctx, FullHttpRequest request, long lastModified,
        String path, long fileLength) {
    HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
    HttpUtil.setContentLength(response, fileLength);
    setContentTypeHeader(response, path);

    response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);

    setDateAndCacheHeaders(response, lastModified);
    if (HttpUtil.isKeepAlive(request)) {
        response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    }

    // Write the initial line and the header.
    ctx.write(response);
}

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

/**
 * Handle keep alive as long as there's the request contains
 * 'connection:Keep-Alive' header, no matter what the client is 1.0 or 1.1:
 * http://sockjs.github.com/sockjs-protocol/sockjs-protocol-0.3.3.html#section-157
 */
public static void if_keepAliveRequest_then_resumeReading_else_closeOnComplete(
        final HttpRequest request, final Channel channel, final ChannelFuture channelFuture
) {
    // TODO:
    // Add Connection: Close, or Keep-Alive?
    // res.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
    if (HttpUtil.isKeepAlive(request)) {
        channelFuture.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) {
                resumeReading(channel);
            }
        });
    } else {
        channelFuture.addListener(ChannelFutureListener.CLOSE);
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
  checkState(userPromise != null, "response before request");
  if (msg instanceof HttpResponse) {
    HttpResponse response = (HttpResponse) msg;
    keepAlive = HttpUtil.isKeepAlive((HttpResponse) msg);
    if (!response.status().equals(HttpResponseStatus.OK)) {
      failAndReset(
          new HttpException(
              response.status(), "Download failed with Status: " + response.status(), null),
          ctx);
    }
  } else if (msg instanceof HttpContent) {
    ByteBuf content = ((HttpContent) msg).content();
    content.readBytes(out, content.readableBytes());
    if (msg instanceof LastHttpContent) {
      succeedAndReset(ctx);
    }
  } else {
    failAndReset(
        new IllegalArgumentException(
            "Unsupported message type: " + StringUtil.simpleClassName(msg)),
        ctx);
  }
}

@SuppressWarnings("FutureReturnValueIgnored")
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse response)
    throws Exception {
  try {
    checkState(userPromise != null, "response before request");
    if (!response.status().equals(HttpResponseStatus.OK)
        && !response.status().equals(HttpResponseStatus.ACCEPTED)
        && !response.status().equals(HttpResponseStatus.CREATED)
        && !response.status().equals(HttpResponseStatus.NO_CONTENT)) {
      // Supporting more than OK status to be compatible with nginx webdav.
      failAndResetUserPromise(
          new HttpException(
              response.status(), "Download failed with " + "Status: " + response.status(), null));
    } else {
      succeedAndResetUserPromise();
    }
  } finally {
    if (!HttpUtil.isKeepAlive(response)) {
      ctx.close();
    }
  }
}

protected void writeResponse(ChannelHandlerContext ctx,
                             FullHttpRequest req,
                             FullHttpResponse res,
                             HttpResponseStatus status) {
  setDateHeader(req, res, status);
  if (!HttpUtil.isContentLengthSet(res)) {
    HttpUtil.setContentLength(res, res.content().readableBytes());
  }

  boolean keepAlive = HttpUtil.isKeepAlive(req);
  if (keepAlive) {
    res.headers().set(CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    ctx.write(res);
  } else {
    ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE);
  }
}

private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
  if (HttpUtil.is100ContinueExpected(req)) {
    ctx.write(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE));
  }

  if (webSocketPath.equals(req.uri())) {
    WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
        getWebSocketLocation(req), null, true, MAX_FRAME_PAYLOAD_LENGTH
    );

    handshaker = wsFactory.newHandshaker(req);
    if (handshaker == null) {
      WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());

    } else {
      handshaker.handshake(ctx.channel(), req);
    }
    return;
  }

  requestProcessor.handleRequest(ctx, req);
}

/**
 * Tests multipart POST and verifies it via GET operations.
 * @throws Exception
 */
@Test
public void multipartPostGetHeadTest() throws Exception {
  Account refAccount = ACCOUNT_SERVICE.createAndAddRandomAccount();
  Container refContainer = refAccount.getContainerById(Container.DEFAULT_PUBLIC_CONTAINER_ID);
  doPostGetHeadDeleteTest(0, refAccount, refContainer, refAccount.getName(), !refContainer.isCacheable(),
      refAccount.getName(), refContainer.getName(), true);
  doPostGetHeadDeleteTest(FRONTEND_CONFIG.frontendChunkedGetResponseThresholdInBytes * 3, refAccount, refContainer,
      refAccount.getName(), !refContainer.isCacheable(), refAccount.getName(), refContainer.getName(), true);

  // failure case
  // size of content being POSTed is higher than what is allowed via multipart/form-data
  long maxAllowedSizeBytes = new NettyConfig(FRONTEND_VERIFIABLE_PROPS).nettyMultipartPostMaxSizeBytes;
  ByteBuffer content = ByteBuffer.wrap(TestUtils.getRandomBytes((int) maxAllowedSizeBytes + 1));
  HttpHeaders headers = new DefaultHttpHeaders();
  setAmbryHeadersForPut(headers, 7200, !refContainer.isCacheable(), refAccount.getName(), "application/octet-stream",
      null, refAccount.getName(), refContainer.getName());
  HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.POST, "/", headers);
  HttpPostRequestEncoder encoder = createEncoder(httpRequest, content, ByteBuffer.allocate(0));
  ResponseParts responseParts = nettyClient.sendRequest(encoder.finalizeRequest(), encoder, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, response.status());
  assertTrue("No Date header", response.headers().getTimeMillis(HttpHeaderNames.DATE, -1) != -1);
  assertFalse("Channel should not be active", HttpUtil.isKeepAlive(response));
}

/**
 * Tests {@link RestUtils.SubResource#Replicas} requests
 * <p/>
 * For each {@link PartitionId} in the {@link ClusterMap}, a {@link BlobId} is created. The replica list returned from
 * server is checked for equality against a locally obtained replica list.
 * @throws Exception
 */
@Test
public void getReplicasTest() throws Exception {
  List<? extends PartitionId> partitionIds = CLUSTER_MAP.getWritablePartitionIds();
  for (PartitionId partitionId : partitionIds) {
    String originalReplicaStr = partitionId.getReplicaIds().toString().replace(", ", ",");
    BlobId blobId = new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
        ClusterMapUtils.UNKNOWN_DATACENTER_ID, Account.UNKNOWN_ACCOUNT_ID, Container.UNKNOWN_CONTAINER_ID,
        partitionId, false);
    FullHttpRequest httpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET,
        blobId.getID() + "/" + RestUtils.SubResource.Replicas, Unpooled.buffer(0));
    ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
    HttpResponse response = getHttpResponse(responseParts);
    assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
    ByteBuffer content = getContent(responseParts.queue, HttpUtil.getContentLength(response));
    JSONObject responseJson = new JSONObject(new String(content.array()));
    String returnedReplicasStr = responseJson.getString(GetReplicasHandler.REPLICAS_KEY).replace("\"", "");
    assertEquals("Replica IDs returned for the BlobId do no match with the replicas IDs of partition",
        originalReplicaStr, returnedReplicasStr);
  }
}

/**
 * Tests for handling of {@link HttpMethod#OPTIONS}.
 * @throws Exception
 */
@Test
public void optionsTest() throws Exception {
  FullHttpRequest httpRequest = buildRequest(HttpMethod.OPTIONS, "", null, null);
  ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
  assertTrue("No Date header", response.headers().getTimeMillis(HttpHeaderNames.DATE, -1) != -1);
  assertEquals("Content-Length is not 0", 0, HttpUtil.getContentLength(response));
  assertEquals("Unexpected value for " + HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS,
      FRONTEND_CONFIG.frontendOptionsAllowMethods,
      response.headers().get(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS));
  assertEquals("Unexpected value for " + HttpHeaderNames.ACCESS_CONTROL_MAX_AGE,
      FRONTEND_CONFIG.frontendOptionsValiditySeconds,
      Long.parseLong(response.headers().get(HttpHeaderNames.ACCESS_CONTROL_MAX_AGE)));
}

/**
 * Method to easily create a request.
 * @param httpMethod the {@link HttpMethod} desired.
 * @param uri string representation of the desired URI.
 * @param headers any associated headers as a {@link HttpHeaders} object. Can be null.
 * @param content the content that accompanies the request. Can be null.
 * @return A {@link FullHttpRequest} object that defines the request required by the input.
 */
private FullHttpRequest buildRequest(HttpMethod httpMethod, String uri, HttpHeaders headers, ByteBuffer content) {
  ByteBuf contentBuf;
  if (content != null) {
    contentBuf = Unpooled.wrappedBuffer(content);
  } else {
    contentBuf = Unpooled.buffer(0);
  }
  FullHttpRequest httpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, httpMethod, uri, contentBuf);
  if (headers != null) {
    httpRequest.headers().set(headers);
  }
  if (HttpMethod.POST.equals(httpMethod) && !HttpUtil.isContentLengthSet(httpRequest)) {
    HttpUtil.setTransferEncodingChunked(httpRequest, true);
  }
  return httpRequest;
}

/**
 * Gets the user metadata of the blob with blob ID {@code blobId} and verifies them against what is expected.
 * @param blobId the blob ID of the blob to HEAD.
 * @param getOption the options to use while getting the blob.
 * @param expectedHeaders the expected headers in the response.
 * @param usermetadata if non-null, this is expected to come as the body.
 * @throws ExecutionException
 * @throws InterruptedException
 */
private void getUserMetadataAndVerify(String blobId, GetOption getOption, HttpHeaders expectedHeaders,
    byte[] usermetadata) throws ExecutionException, InterruptedException {
  HttpHeaders headers = new DefaultHttpHeaders();
  if (getOption != null) {
    headers.add(RestUtils.Headers.GET_OPTION, getOption.toString());
  }
  FullHttpRequest httpRequest =
      buildRequest(HttpMethod.GET, blobId + "/" + RestUtils.SubResource.UserMetadata, headers, null);
  ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
  checkCommonGetHeadHeaders(response.headers());
  verifyUserMetadata(expectedHeaders, response, usermetadata, responseParts.queue);
  assertTrue("Channel should be active", HttpUtil.isKeepAlive(response));
}

/**
 * Gets the blob info of the blob with blob ID {@code blobId} and verifies them against what is expected.
 * @param blobId the blob ID of the blob to HEAD.
 * @param getOption the options to use while getting the blob.
 * @param expectedHeaders the expected headers in the response.
 * @param isPrivate {@code true} if the blob is expected to be private
 * @param accountName the expected account name in the response.
 * @param containerName the expected container name in response.
 * @param usermetadata if non-null, this is expected to come as the body.
 * @throws ExecutionException
 * @throws InterruptedException
 */
private void getBlobInfoAndVerify(String blobId, GetOption getOption, HttpHeaders expectedHeaders, boolean isPrivate,
    String accountName, String containerName, byte[] usermetadata) throws ExecutionException, InterruptedException {
  HttpHeaders headers = new DefaultHttpHeaders();
  if (getOption != null) {
    headers.add(RestUtils.Headers.GET_OPTION, getOption.toString());
  }
  FullHttpRequest httpRequest =
      buildRequest(HttpMethod.GET, blobId + "/" + RestUtils.SubResource.BlobInfo, headers, null);
  ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
  checkCommonGetHeadHeaders(response.headers());
  verifyBlobProperties(expectedHeaders, isPrivate, response);
  verifyAccountAndContainerHeaders(accountName, containerName, response);
  verifyUserMetadata(expectedHeaders, response, usermetadata, responseParts.queue);
  assertTrue("Channel should be active", HttpUtil.isKeepAlive(response));
}

@Override
public void operationComplete(ChannelFuture future) throws Exception {
  long writeFinishTime = System.currentTimeMillis();
  long channelWriteTime = writeFinishTime - responseWriteStartTime;
  if (future.isSuccess()) {
    completeRequest(!HttpUtil.isKeepAlive(finalResponseMetadata));
  } else {
    handleChannelWriteFailure(future.cause(), true);
  }
  long responseAfterWriteProcessingTime = System.currentTimeMillis() - writeFinishTime;
  nettyMetrics.channelWriteTimeInMs.update(channelWriteTime);
  nettyMetrics.responseMetadataAfterWriteProcessingTimeInMs.update(responseAfterWriteProcessingTime);
  if (request != null) {
    request.getMetricsTracker().nioMetricsTracker.addToResponseProcessingTime(
        channelWriteTime + responseAfterWriteProcessingTime);
  }
}

/**
 * Tests setting of different available {@link ResponseStatus} codes and sees that they are recognized and converted
 * in {@link NettyResponseChannel}.
 * <p/>
 * If this test fails, a case for conversion probably needs to be added in {@link NettyResponseChannel}.
 */
@Test
public void setStatusTest() {
  // ask for every status to be set
  for (ResponseStatus expectedResponseStatus : ResponseStatus.values()) {
    HttpRequest request = createRequestWithHeaders(HttpMethod.GET, TestingUri.SetStatus.toString());
    request.headers().set(MockNettyMessageProcessor.STATUS_HEADER_NAME, expectedResponseStatus);
    HttpUtil.setKeepAlive(request, false);
    EmbeddedChannel channel = createEmbeddedChannel();
    channel.writeInbound(request);

    // pull but discard response
    channel.readOutbound();
    assertFalse("Channel not closed on the server", channel.isActive());
  }
  // check if all the ResponseStatus codes were recognized.
  String metricName = MetricRegistry.name(NettyResponseChannel.class, "UnknownResponseStatusCount");
  long metricCount = MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(metricName).getCount();
  assertEquals("Some of the ResponseStatus codes were not recognized", 0, metricCount);
}

/**
 * Tests that the underlying network channel is closed when {@link NettyResponseChannel#close()} is called.
 */
@Test
public void closeTest() {
  // request is keep-alive by default.
  HttpRequest request = createRequestWithHeaders(HttpMethod.GET, TestingUri.Close.toString());
  EmbeddedChannel channel = createEmbeddedChannel();
  channel.writeInbound(request);

  HttpResponse response = (HttpResponse) channel.readOutbound();
  assertEquals("Unexpected response status", HttpResponseStatus.INTERNAL_SERVER_ERROR, response.status());
  assertFalse("Inconsistent value for Connection header", HttpUtil.isKeepAlive(response));
  // drain the channel of content.
  while (channel.readOutbound() != null) {
  }
  assertFalse("Channel should be closed", channel.isOpen());
}

/**
 * Asks the server to write more data than the set Content-Length and checks behavior.
 * @param chunkCount the number of chunks of {@link MockNettyMessageProcessor#CHUNK} to use to set Content-Length.
 * @throws Exception
 */
private void doWriteMoreThanContentLengthTest(int chunkCount) throws Exception {
  EmbeddedChannel channel = createEmbeddedChannel();
  MockNettyMessageProcessor processor = channel.pipeline().get(MockNettyMessageProcessor.class);
  HttpHeaders httpHeaders = new DefaultHttpHeaders();
  httpHeaders.set(MockNettyMessageProcessor.CHUNK_COUNT_HEADER_NAME, chunkCount);
  HttpRequest httpRequest =
      RestTestUtils.createRequest(HttpMethod.POST, TestingUri.WriteMoreThanContentLength.toString(), httpHeaders);
  HttpUtil.setKeepAlive(httpRequest, true);
  channel.writeInbound(httpRequest);

  try {
    verifyCallbacks(processor);
    fail("One of the callbacks should have failed because the data written was more than Content-Length");
  } catch (IllegalStateException e) {
    // expected. Nothing to do.
  }

  // It doesn't matter what the response is - because it may either fail or succeed depending on certain race
  // conditions. What matters is that the programming error is caught appropriately by NettyResponseChannel and it
  // makes a callback with the right exception.
  while (channel.readOutbound() != null) {
  }
  channel.close();
}

@Override
public void channelRead0(ChannelHandlerContext ctx, HttpObject in) {
  // Make sure that we increase refCnt because we are going to process it async. The other end has to release
  // after processing.
  responseParts.offer(ReferenceCountUtil.retain(in));
  if (in instanceof HttpResponse && in.decoderResult().isSuccess()) {
    isKeepAlive = HttpUtil.isKeepAlive((HttpResponse) in);
  } else if (in.decoderResult().isFailure()) {
    Throwable cause = in.decoderResult().cause();
    if (cause instanceof Exception) {
      exception = (Exception) cause;
    } else {
      exception =
          new Exception("Encountered Throwable when trying to decode response. Message: " + cause.getMessage());
    }
    invokeFutureAndCallback("CommunicationHandler::channelRead0 - decoder failure");
  }
  if (in instanceof LastHttpContent) {
    if (isKeepAlive) {
      invokeFutureAndCallback("CommunicationHandler::channelRead0 - last content");
    } else {
      // if not, the future will be invoked when the channel is closed.
      ctx.close();
    }
  }
}

/**
 * Does a test to see that request handling results in expected entries in public access log
 * @param httpMethod the {@link HttpMethod} for the request.
 * @param uri Uri to be used during the request
 * @param testErrorCase true if error case has to be tested, false otherwise
 * @param useSSL {@code true} to test SSL logging.
 * @throws Exception
 */
private void doRequestHandleTest(HttpMethod httpMethod, String uri, boolean testErrorCase, boolean useSSL)
    throws Exception {
  EmbeddedChannel channel = createChannel(useSSL);
  List<HttpHeaders> httpHeadersList = getHeadersList();
  for (HttpHeaders headers : httpHeadersList) {
    HttpRequest request = RestTestUtils.createRequest(httpMethod, uri, headers);
    HttpUtil.setKeepAlive(request, true);
    sendRequestCheckResponse(channel, request, uri, headers, testErrorCase, false, useSSL);
    if (!testErrorCase) {
      Assert.assertTrue("Channel should not be closed ", channel.isOpen());
    } else {
      Assert.assertFalse("Channel should have been closed ", channel.isOpen());
      channel = createChannel(useSSL);
    }
  }
  channel.close();
}

/**
 * Does a test to see that a health check request results in expected response from the health check handler
 * @param httpMethod the {@link HttpMethod} for the request.
 * @param keepAlive true if keep alive has to be set in the request, false otherwise
 * @throws IOException
 */
private void testHealthCheckRequest(HttpMethod httpMethod, boolean isServiceUp, boolean keepAlive)
    throws IOException {
  EmbeddedChannel channel = createChannel();
  for (int i = 0; i < 2; i++) {
    if (isServiceUp) {
      restServerState.markServiceUp();
    }
    HttpRequest request = RestTestUtils.createRequest(httpMethod, healthCheckUri, null);
    HttpUtil.setKeepAlive(request, keepAlive);
    FullHttpResponse response = sendRequestAndGetResponse(channel, request);
    HttpResponseStatus httpResponseStatus =
        (isServiceUp) ? HttpResponseStatus.OK : HttpResponseStatus.SERVICE_UNAVAILABLE;
    assertEquals("Unexpected response status", httpResponseStatus, response.status());
    String expectedStr = (isServiceUp) ? goodStr : badStr;
    assertEquals("Unexpected content", expectedStr, RestTestUtils.getContentString(response));
    restServerState.markServiceDown();
    if (keepAlive && isServiceUp) {
      Assert.assertTrue("Channel should not be closed ", channel.isOpen());
    } else {
      Assert.assertFalse("Channel should have been closed by now ", channel.isOpen());
      channel = createChannel();
    }
  }
  channel.close();
}

/**
 * Sends the provided {@code httpRequest} and verifies that the response is an echo of the {@code restMethod}.
 * @param channel the {@link EmbeddedChannel} to send the request over.
 * @param httpMethod the {@link HttpMethod} for the request.
 * @param restMethod the equivalent {@link RestMethod} for {@code httpMethod}. Used to check for correctness of
 *                   response.
 * @param isKeepAlive if the request needs to be keep-alive.
 * @throws IOException
 */
private void sendRequestCheckResponse(EmbeddedChannel channel, HttpMethod httpMethod, RestMethod restMethod,
    boolean isKeepAlive) throws IOException {
  long requestId = REQUEST_ID_GENERATOR.getAndIncrement();
  String uri = MockBlobStorageService.ECHO_REST_METHOD + requestId;
  HttpRequest httpRequest = RestTestUtils.createRequest(httpMethod, uri, null);
  HttpUtil.setKeepAlive(httpRequest, isKeepAlive);
  channel.writeInbound(httpRequest);
  channel.writeInbound(new DefaultLastHttpContent());
  HttpResponse response = (HttpResponse) channel.readOutbound();
  assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
  // MockBlobStorageService echoes the RestMethod + request id.
  String expectedResponse = restMethod.toString() + requestId;
  assertEquals("Unexpected content", expectedResponse,
      RestTestUtils.getContentString((HttpContent) channel.readOutbound()));
  assertTrue("End marker was expected", channel.readOutbound() instanceof LastHttpContent);
}

/**
 * Does the post test by sending the request and content to {@link NettyMessageProcessor} through an
 * {@link EmbeddedChannel} and returns the data stored in the {@link InMemoryRouter} as a result of the post.
 * @param postRequest the POST request as a {@link HttpRequest}.
 * @param contentToSend the content to be sent as a part of the POST.
 * @return the data stored in the {@link InMemoryRouter} as a result of the POST.
 * @throws InterruptedException
 */
private ByteBuffer doPostTest(HttpRequest postRequest, List<ByteBuffer> contentToSend) throws InterruptedException {
  EmbeddedChannel channel = createChannel();

  // POST
  notificationSystem.reset();
  postRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream");
  HttpUtil.setKeepAlive(postRequest, false);
  channel.writeInbound(postRequest);
  if (contentToSend != null) {
    for (ByteBuffer content : contentToSend) {
      channel.writeInbound(new DefaultHttpContent(Unpooled.wrappedBuffer(content)));
    }
    channel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
  }
  if (!notificationSystem.operationCompleted.await(100, TimeUnit.MILLISECONDS)) {
    fail("Post did not succeed after 100ms. There is an error or timeout needs to increase");
  }
  assertNotNull("Blob id operated on cannot be null", notificationSystem.blobIdOperatedOn);
  return router.getActiveBlobs().get(notificationSystem.blobIdOperatedOn).getBlob();
}

public static HttpResponse createHttpResponse(final String origin, ByteBuf content, boolean json) {
  FullHttpResponse res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
  if (json) {
    res.headers().add(HttpHeaderNames.CONTENT_TYPE, "text/javascript; charset=UTF-8");
  } else {
    res.headers().add(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
  }
  res.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
  if (origin != null) {
    res.headers().add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, origin);
    res.headers().add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_CREDENTIALS, "true");
  }
  HttpUtil.setContentLength(res, content.readableBytes());

  return res;
}

/**
 * 响应HTTP的请求
 *
 * @param ctx     ChannelHandlerContext
 * @param req     FullHttpRequest
 * @param jsonStr String
 */
private void sendResponse(ChannelHandlerContext ctx, FullHttpRequest req, String jsonStr) {
    boolean keepAlive = HttpUtil.isKeepAlive(req);
    byte[] jsonByteByte = jsonStr.getBytes();
    FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(jsonByteByte));
    response.headers().set(CONTENT_TYPE, APPLICATION_JSON);
    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);
    }
}

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

@Override
protected void encode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out)
    throws Exception {
  if (msg instanceof HttpResponse) {
    HttpResponse res = (HttpResponse) msg;

    skipCompression = false;

    // if an "content-encoding: identity" header was set, we do not compress
    if (skipCompression = res.headers().containsValue(
        HttpHeaderNames.CONTENT_ENCODING,
        HttpHeaderValues.IDENTITY,
        true)) {
      // remove header as one should not send Identity as content encoding
      res.headers().remove(HttpHeaderNames.CONTENT_ENCODING);
    } else {
      CharSequence mimeType = HttpUtil.getMimeType(res);
      // skip compression if the media type is not compressible by the server
      skipCompression = mimeType != null && !isCompressable(MediaType.parse(mimeType.toString()));

      // skip compression if the content length is less than expected by the server
      int contentLength = res.headers().getInt(HttpHeaderNames.CONTENT_LENGTH, 0);
      skipCompression = contentLength > 0 && contentLength < compressionContentLength;
    }
  }

  super.encode(ctx, msg, out);
}