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

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