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

/**
 * This method will be called when events need to be published via this sink
 *
 * @param payload        payload of the event based on the supported event class exported by the extensions
 * @param dynamicOptions holds the dynamic options of this sink and Use this object to obtain dynamic options.
 * @throws ConnectionUnavailableException if end point is unavailable the ConnectionUnavailableException thrown
 *                                        such that the  system will take care retrying for connection
 */
@Override
public void publish(Object payload, DynamicOptions dynamicOptions) throws ConnectionUnavailableException {
    String headers = httpHeaderOption.getValue(dynamicOptions);
    String httpMethod = HttpConstants.EMPTY_STRING.equals(httpMethodOption.getValue(dynamicOptions)) ?
            HttpConstants.METHOD_DEFAULT : httpMethodOption.getValue(dynamicOptions);
    List<Header> headersList = HttpSinkUtil.getHeaders(headers);
    String contentType = HttpSinkUtil.getContentType(mapType, headersList);
    String messageBody = (String) payload;
    HTTPCarbonMessage cMessage = createHttpCarbonMessage(httpMethod);
    cMessage = generateCarbonMessage(headersList, contentType, httpMethod, cMessage);
    cMessage.addHttpContent(new DefaultLastHttpContent(Unpooled.wrappedBuffer(messageBody
            .getBytes(Charset.defaultCharset()))));
    clientConnector.send(cMessage);

}

public static void startDownTask(TaskInfo taskInfo, HttpRequest httpRequest,
    HttpResponse httpResponse, Channel clientChannel) {
  HttpHeaders httpHeaders = httpResponse.headers();
  HttpDownInfo httpDownInfo = new HttpDownInfo(taskInfo, httpRequest);
  HttpDownServer.DOWN_CONTENT.put(taskInfo.getId(), httpDownInfo);
  httpHeaders.clear();
  httpResponse.setStatus(HttpResponseStatus.OK);
  httpHeaders.set(HttpHeaderNames.CONTENT_TYPE, "text/html");
  String host = HttpDownServer.isDev() ? "localhost"
      : ((InetSocketAddress) clientChannel.localAddress()).getHostString();
  String js =
      "<script>window.top.location.href='http://" + host + ":" + HttpDownServer.VIEW_SERVER_PORT
          + "/#/tasks/new/" + httpDownInfo
          .getTaskInfo().getId()
          + "';</script>";
  HttpContent content = new DefaultLastHttpContent();
  content.content().writeBytes(js.getBytes());
  httpHeaders.set(HttpHeaderNames.CONTENT_LENGTH, js.getBytes().length);
  clientChannel.writeAndFlush(httpResponse);
  clientChannel.writeAndFlush(content);
  clientChannel.close();
}

@Test
public void addContentChunk_adds_last_chunk_trailing_headers() {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    LastHttpContent lastChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8));
    String headerKey = UUID.randomUUID().toString();
    List<String> headerVal = Arrays.asList(UUID.randomUUID().toString(), UUID.randomUUID().toString());
    lastChunk.trailingHeaders().add(headerKey, headerVal);

    // when
    requestInfo.addContentChunk(lastChunk);

    // then
    assertThat(requestInfo.trailingHeaders.names().size(), is(1));
    assertThat(requestInfo.trailingHeaders.getAll(headerKey), is(headerVal));
}

@Test
public void addContentChunk_adds_chunk_content_length_to_rawContentLengthInBytes() throws IOException {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    String chunk1String = UUID.randomUUID().toString();
    String lastChunkString = UUID.randomUUID().toString();
    byte[] chunk1Bytes = chunk1String.getBytes();
    byte[] lastChunkBytes = lastChunkString.getBytes();
    HttpContent chunk1 = new DefaultHttpContent(Unpooled.copiedBuffer(chunk1Bytes));
    HttpContent lastChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(lastChunkBytes));

    // when
    requestInfo.addContentChunk(chunk1);
    requestInfo.addContentChunk(lastChunk);

    // then
    assertThat(requestInfo.contentChunks.size(), is(2));
    assertThat(requestInfo.isCompleteRequestWithAllChunks(), is(true));
    assertThat(requestInfo.getRawContentLengthInBytes(), is(chunk1Bytes.length + lastChunkBytes.length));
}

@Test
public void addDecoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler(new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("JsonObjectDecoder$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void addNamedDecoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler("json", new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("json$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void addEncoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler(new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("JsonObjectDecoder$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void addNamedEncoderReplaysLastHttp() throws Exception {
    ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
    EmbeddedChannel channel = new EmbeddedChannel();
    HttpClientOperations ops = new HttpClientOperations(channel,
            (response, request) -> null, handler);

    ops.addHandler("json", new JsonObjectDecoder());
    channel.writeInbound(new DefaultLastHttpContent(buf));

    assertThat(channel.pipeline().names().iterator().next(), is("json$extractor"));

    Object content = channel.readInbound();
    assertThat(content, instanceOf(ByteBuf.class));
    ((ByteBuf) content).release();

    content = channel.readInbound();
    assertThat(content, instanceOf(LastHttpContent.class));
    ((LastHttpContent) content).release();

    assertThat(channel.readInbound(), nullValue());
}

@Test
public void testBuildContent()
    throws Exception {
  HttpRequest nettyRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_0, HttpMethod.GET, "www.google.com");
  RecordedHttpRequestBuilder recordedHttpRequestBuilder = new RecordedHttpRequestBuilder(nettyRequest);

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

  String lastStr = "Last chunk";
  HttpContent lastContent = new DefaultLastHttpContent(Unpooled.copiedBuffer(lastStr.getBytes(charset)));
  recordedHttpRequestBuilder.appendHttpContent(lastContent);

  RecordedHttpRequest recordedHttpRequest = recordedHttpRequestBuilder.build();
  Assert
      .assertEquals((str1 + str2 + lastStr).getBytes(charset), recordedHttpRequest.getHttpBody().getContent(charset));
}

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

private void handleThrowable(MicroservicesRegistryImpl currentMicroservicesRegistry, Throwable throwable,
                             Request request) {
    Optional<ExceptionMapper> exceptionMapper = currentMicroservicesRegistry.getExceptionMapper(throwable);
    if (exceptionMapper.isPresent()) {
        org.wso2.msf4j.Response msf4jResponse = new org.wso2.msf4j.Response(request);
        msf4jResponse.setEntity(exceptionMapper.get().toResponse(throwable));
        msf4jResponse.send();
    } else {
        log.warn("Unmapped exception", throwable);
        try {
            HTTPCarbonMessage response = HttpUtil.createTextResponse(
                    javax.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
                    "Exception occurred :" + throwable.getMessage());
            response.addHttpContent(new DefaultLastHttpContent());
            request.respond(response);
        } catch (ServerConnectorException e) {
            log.error("Error while sending the response.", e);
        }
    }
}

/**
 * Write the entity to the carbon message.
 */
@Override
public void writeData(HTTPCarbonMessage carbonMessage, Object entity, String mediaType, int chunkSize,
                      HTTPCarbonMessage responder) {
    mediaType = (mediaType != null) ? mediaType : MediaType.WILDCARD;
    ByteBuffer byteBuffer = BeanConverter.getConverter(mediaType).convertToMedia(entity);
    carbonMessage.addHttpContent(new DefaultLastHttpContent(Unpooled.wrappedBuffer(byteBuffer)));
    if (chunkSize == Response.NO_CHUNK) {
        carbonMessage.setHeader(Constants.HTTP_CONTENT_LENGTH, String.valueOf(byteBuffer.remaining()));
    } else {
        carbonMessage.setHeader(Constants.HTTP_TRANSFER_ENCODING, CHUNKED);
    }
    carbonMessage.setHeader(Constants.HTTP_CONTENT_TYPE, mediaType);
    try {
        responder.respond(carbonMessage);
    } catch (ServerConnectorException e) {
        throw new RuntimeException("Error while sending the response.", e);
    }
}

/**
 * Create a CarbonMessage for a specific status code.
 *
 * @param status HTTP status code
 * @param msg message text
 * @return CarbonMessage representing the status
 */
public static HTTPCarbonMessage createTextResponse(int status, String msg) {
    HTTPCarbonMessage response = new HTTPCarbonMessage(
            new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.valueOf(status)));
    response.setProperty(Constants.HTTP_STATUS_CODE, status);
    if (msg != null) {
        response.setHeader(HttpHeaders.CONTENT_LENGTH, String.valueOf(msg.length()));
        byte[] msgArray = null;
        try {
            msgArray = msg.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException("Failed to get the byte array from responseValue", e);
        }
        ByteBuffer byteBuffer = ByteBuffer.allocate(msgArray.length);
        byteBuffer.put(msgArray);
        byteBuffer.flip();
        response.addHttpContent(new DefaultLastHttpContent(Unpooled.wrappedBuffer(byteBuffer)));
    } else {
        response.setHeader(HttpHeaders.CONTENT_LENGTH, "0");
    }
    return response;
}

@Override
    public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
            throws Http2Exception {

        HTTPCarbonMessage cMsg = streamIdRequestMap.get(streamId);
        if (cMsg != null) {
            cMsg.addHttpContent(new DefaultLastHttpContent(data.retain()));
            if (endOfStream) {
                cMsg.setEndOfMsgAdded(true);
//                if (HTTPTransportContextHolder.getInstance().getHandlerExecutor() != null) {
//                    HTTPTransportContextHolder.getInstance().getHandlerExecutor().executeAtSourceRequestSending(cMsg);
//                }
            }
        }
        return data.readableBytes() + padding;
    }

@Override
public void onMessage(HTTPCarbonMessage httpRequestMessage) {
    executor.execute(() -> {
        HTTPCarbonMessage cMsg = new HTTPCarbonMessage(new DefaultHttpResponse(HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK));
        cMsg.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
        cMsg.setHeader(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
        cMsg.setHeader(HttpHeaders.Names.CONTENT_TYPE, Constants.TEXT_PLAIN);
        cMsg.setProperty(Constants.HTTP_STATUS_CODE, 200);
        try {
            httpRequestMessage.respond(cMsg);
        } catch (ServerConnectorException e) {
            logger.error("Error occurred during message notification: " + e.getMessage());
        }
        while (true) {
            HttpContent httpContent = httpRequestMessage.getHttpContent();
            cMsg.addHttpContent(httpContent);
            if (httpContent instanceof LastHttpContent) {
                cMsg.addHttpContent(new DefaultLastHttpContent());
                httpRequestMessage.release();
                break;
            }
        }
    });
}

private HttpContent nextChunk( ChannelHandlerContext context )
    throws Exception
{
    if( isLastChunk )
    {
        isLastChunkRead = true;
        LastHttpContent lastChunk = new DefaultLastHttpContent( EMPTY_BUFFER );
        lastChunk.trailingHeaders().add( X_WERVAL_CONTENT_LENGTH, contentLength );
        return lastChunk;
    }
    ByteBuf buffer = chunkedBody.readChunk( context );
    if( chunkedBody.isEndOfInput() )
    {
        isLastChunk = true;
    }
    contentLength += buffer.readableBytes();
    return new DefaultHttpContent( buffer );
}

@Test
public void shouldDecodeSuccessfulGetDesignDocumentResponse() throws Exception {
    String response = Resources.read("designdoc_success.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    GetDesignDocumentRequest requestMock = mock(GetDesignDocumentRequest.class);
    when(requestMock.name()).thenReturn("name");
    when(requestMock.development()).thenReturn(true);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    GetDesignDocumentResponse inbound = (GetDesignDocumentResponse) firedEvents.get(0);

    assertTrue(inbound.status().isSuccess());
    assertEquals("name", inbound.name());
    assertEquals(true, inbound.development());
    assertEquals(response, inbound.content().toString(CharsetUtil.UTF_8));
    ReferenceCountUtil.releaseLater(inbound);
}

@Test
@SuppressWarnings("unchecked")
public void shouldParseErrorWithEmptyRows() throws Exception {
    String response = Resources.read("error_empty_rows.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    ViewQueryRequest requestMock = mock(ViewQueryRequest.class);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    ViewQueryResponse inbound = (ViewQueryResponse) firedEvents.get(0);
    assertTrue(inbound.status().isSuccess());

    assertEquals(0, countAndRelease(inbound.rows()));

    String error = inbound.error().toBlocking().single();
    Map<String, Object> parsed = mapper.readValue(error, Map.class);
    assertEquals(1, parsed.size());
    assertNotNull(parsed.get("errors"));
}

@Test
@SuppressWarnings("unchecked")
public void shouldParseErrorAfterRows() throws Exception {
    String response = Resources.read("error_rows.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    ViewQueryRequest requestMock = mock(ViewQueryRequest.class);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    ViewQueryResponse inbound = (ViewQueryResponse) firedEvents.get(0);
    assertTrue(inbound.status().isSuccess());

    assertEquals(10, countAndRelease(inbound.rows()));

    String error = inbound.error().toBlocking().single();
    Map<String, Object> parsed = mapper.readValue(error, Map.class);
    assertEquals(1, parsed.size());
    assertNotNull(parsed.get("errors"));
}

@Test
public void shouldParseErrorWithDesignNotFound() throws Exception {
    String response = Resources.read("designdoc_notfound.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(404, "Object Not Found"));
    HttpContent responseChunk1 = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    ViewQueryRequest requestMock = mock(ViewQueryRequest.class);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    ViewQueryResponse inbound = (ViewQueryResponse) firedEvents.get(0);
    assertFalse(inbound.status().isSuccess());
    assertEquals(ResponseStatus.NOT_EXISTS, inbound.status());

    assertEquals(0, countAndRelease(inbound.rows()));

    String error = inbound.error().toBlocking().single();
    assertEquals("{\"errors\":[{\"error\":\"not_found\",\"reason\":\"Design document _design/designdoc not found\"}]}", error);

}

@Test
public void shouldDecodeSuccessBucketConfigResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultHttpContent(Unpooled.copiedBuffer("foo", CHARSET));
    HttpContent responseChunk2 = new DefaultLastHttpContent(Unpooled.copiedBuffer("bar", CHARSET));

    BucketConfigRequest requestMock = mock(BucketConfigRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1, responseChunk2);
    channel.readInbound();

    assertEquals(1, eventSink.responseEvents().size());
    BucketConfigResponse event = (BucketConfigResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.SUCCESS, event.status());
    assertEquals("foobar", event.config());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeNotFoundBucketConfigResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
        new HttpResponseStatus(404, "Object Not Found"));
    HttpContent responseChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer("Not found.", CharsetUtil.UTF_8));

    BucketConfigRequest requestMock = mock(BucketConfigRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);

    assertEquals(1, eventSink.responseEvents().size());
    BucketConfigResponse event = (BucketConfigResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.NOT_EXISTS, event.status());
    assertEquals("Not found.", event.config());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeFlushNotEnabledResponse() throws Exception {
    String content = "{\"_\":\"Flush is disabled for the bucket\"}";
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
        new HttpResponseStatus(400, "Bad Request"));
    HttpContent responseChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(content, CharsetUtil.UTF_8));

    FlushRequest requestMock = mock(FlushRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);

    assertEquals(1, eventSink.responseEvents().size());
    FlushResponse event = (FlushResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.INVALID_ARGUMENTS, event.status());
    assertEquals("{\"_\":\"Flush is disabled for the bucket\"}", event.content());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeListDesignDocumentsResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultHttpContent(Unpooled.copiedBuffer("foo", CharsetUtil.UTF_8));
    HttpContent responseChunk2 = new DefaultLastHttpContent(Unpooled.copiedBuffer("bar", CharsetUtil.UTF_8));

    GetDesignDocumentsRequest requestMock = mock(GetDesignDocumentsRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1, responseChunk2);

    assertEquals(1, eventSink.responseEvents().size());
    GetDesignDocumentsResponse event = (GetDesignDocumentsResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.SUCCESS, event.status());
    assertEquals("foobar", event.content());
    assertTrue(requestQueue.isEmpty());
}

/**
 * Tests for POST request that has no content.
 * @throws Exception
 */
@Test
public void zeroSizeContentTest() throws Exception {
  Channel channel = new MockChannel();
  NettyRequest nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel);
  HttpContent httpContent = new DefaultLastHttpContent();

  nettyRequest.addContent(httpContent);
  assertEquals("Reference count is not as expected", 2, httpContent.refCnt());

  ByteBufferAsyncWritableChannel writeChannel = new ByteBufferAsyncWritableChannel();
  ReadIntoCallback callback = new ReadIntoCallback();
  Future<Long> future = nettyRequest.readInto(writeChannel, callback);
  assertEquals("There should be no content", 0, writeChannel.getNextChunk().remaining());
  writeChannel.resolveOldestChunk(null);
  closeRequestAndValidate(nettyRequest, channel);
  writeChannel.close();
  assertEquals("Reference count of http content has changed", 1, httpContent.refCnt());
  callback.awaitCallback();
  if (callback.exception != null) {
    throw callback.exception;
  }
  long futureBytesRead = future.get();
  assertEquals("Total bytes read does not match (callback)", 0, callback.bytesRead);
  assertEquals("Total bytes read does not match (future)", 0, futureBytesRead);
}

/**
 * Splits the given {@code contentBytes} into {@code numChunks} chunks and stores them in {@code httpContents}.
 * @param contentBytes the content that needs to be split.
 * @param numChunks the number of chunks to split {@code contentBytes} into.
 * @param httpContents the {@link List<HttpContent>} that will contain all the content in parts.
 * @param useCopyForcingByteBuf if {@code true}, uses {@link CopyForcingByteBuf} instead of the default
 *                              {@link ByteBuf}.
 */
private void splitContent(byte[] contentBytes, int numChunks, List<HttpContent> httpContents,
    boolean useCopyForcingByteBuf) {
  int individualPartSize = contentBytes.length / numChunks;
  ByteBuf content;
  for (int addedContentCount = 0; addedContentCount < numChunks - 1; addedContentCount++) {
    if (useCopyForcingByteBuf) {
      content =
          CopyForcingByteBuf.wrappedBuffer(contentBytes, addedContentCount * individualPartSize, individualPartSize);
    } else {
      content = Unpooled.wrappedBuffer(contentBytes, addedContentCount * individualPartSize, individualPartSize);
    }
    httpContents.add(new DefaultHttpContent(content));
  }
  if (useCopyForcingByteBuf) {
    content =
        CopyForcingByteBuf.wrappedBuffer(contentBytes, (numChunks - 1) * individualPartSize, individualPartSize);
  } else {
    content = Unpooled.wrappedBuffer(contentBytes, (numChunks - 1) * individualPartSize, individualPartSize);
  }
  httpContents.add(new DefaultLastHttpContent(content));
}

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

@Override
public void afterResponse(Channel clientChannel, Channel proxyChannel, HttpContent httpContent,
    HttpProxyInterceptPipeline pipeline) throws Exception {
  if (isMatch) {
    try {
      contentBuf.writeBytes(httpContent.content());
      if (httpContent instanceof LastHttpContent) {
        ByteUtil.insertText(contentBuf,ByteUtil.findText(contentBuf,"<head>"),hookResponse());
        HttpContent hookHttpContent = new DefaultLastHttpContent();
        if (isGzip) { //转化成gzip编码
          byte[] temp = new byte[contentBuf.readableBytes()];
          contentBuf.readBytes(temp);
          ByteArrayOutputStream baos = new ByteArrayOutputStream();
          GZIPOutputStream outputStream = new GZIPOutputStream(baos);
          outputStream.write(temp);
          outputStream.finish();
          hookHttpContent.content().writeBytes(baos.toByteArray());
        } else {
          hookHttpContent.content().writeBytes(contentBuf);
        }
        pipeline.getDefault()
            .afterResponse(clientChannel, proxyChannel, hookHttpContent, pipeline);
      }
    } finally {
      ReferenceCountUtil.release(httpContent);
    }
  } else {
    pipeline.afterResponse(clientChannel, proxyChannel, httpContent);
  }
}

@Test
public void addContentChunk_and_getRawConent_and_getRawContentBytes_work_as_expected_for_last_chunk() throws IOException {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    String chunk1String = UUID.randomUUID().toString();
    String lastChunkString = UUID.randomUUID().toString();
    byte[] chunk1Bytes = chunk1String.getBytes();
    byte[] lastChunkBytes = lastChunkString.getBytes();
    HttpContent chunk1 = new DefaultHttpContent(Unpooled.copiedBuffer(chunk1Bytes));
    HttpContent lastChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(lastChunkBytes));
    assertThat(chunk1.refCnt(), is(1));
    assertThat(lastChunk.refCnt(), is(1));
    assertThat(requestInfo.getRawContentBytes(), nullValue());
    assertThat(requestInfo.getRawContent(), nullValue());

    // when
    requestInfo.addContentChunk(chunk1);
    requestInfo.addContentChunk(lastChunk);

    // then
    assertThat(chunk1.refCnt(), is(2));
    assertThat(lastChunk.refCnt(), is(2));
    assertThat(requestInfo.contentChunks.size(), is(2));
    assertThat(requestInfo.isCompleteRequestWithAllChunks(), is(true));
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    baos.write(chunk1Bytes);
    baos.write(lastChunkBytes);
    assertThat(requestInfo.getRawContentBytes(), is(baos.toByteArray()));
    String rawContentString = requestInfo.getRawContent();
    assertThat(requestInfo.getRawContent(), is(chunk1String + lastChunkString));
    assertThat(requestInfo.getRawContent() == rawContentString, is(true)); // Verify that the raw content string is cached the first time it's loaded and reused for subsequent calls
    assertThat(chunk1.refCnt(), is(1));
    assertThat(lastChunk.refCnt(), is(1));
}

@Test(expected = IllegalStateException.class)
public void addContentChunk_throws_IllegalStateException_if_requestInfo_trailingHeaders_is_already_populated_when_last_chunk_arrives() {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    LastHttpContent lastChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8));
    requestInfo.trailingHeaders.add("somekey", "someval");

    // expect
    requestInfo.addContentChunk(lastChunk);
    fail("Expected an IllegalStateException, but no exception was thrown");
}

@Test
public void addContentChunk_does_not_throw_IllegalStateException_if_requestInfo_trailingHeaders_is_already_populated_when_last_chunk_arrives_if_same_instance() {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    LastHttpContent lastChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8));
    lastChunk.trailingHeaders().add("somekey", "someval");
    requestInfo.trailingHeaders = lastChunk.trailingHeaders();

    // when
    requestInfo.addContentChunk(lastChunk);

    // then
    assertThat(requestInfo.trailingHeaders, is(lastChunk.trailingHeaders()));
}

@Test
public void addContentChunk_does_not_set_isCompleteRequestWithAllChunks_to_true_if_contentChunksWillBeReleasedExternally_is_true() {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    requestInfo.contentChunksWillBeReleasedExternally();
    LastHttpContent lastChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8));

    // when
    requestInfo.addContentChunk(lastChunk);

    // then
    Assertions.assertThat(requestInfo.isCompleteRequestWithAllChunks()).isFalse();
    Assertions.assertThat(requestInfo.contentChunks).isEmpty();
}

@Test
public void testDecodeChunkedRequest() throws Exception {
  HttpRequest req = newChunkedHttpRequest();
  decoder.channelRead(ctx, req);
  HttpContent chunk1 = new DefaultHttpContent(Unpooled.wrappedBuffer(new byte[123]));
  decoder.channelRead(ctx, chunk1);
  LastHttpContent chunk2 = new DefaultLastHttpContent(Unpooled.wrappedBuffer(new byte[200]));
  decoder.channelRead(ctx, chunk2);

  resultTester.assertValue(result -> result.length == 323);
}

@Test
public void testDecodeChunkedResponse() throws Exception {
  HttpResponse res = newChunkedHttpResponse();
  decoder.channelRead(ctx, res);
  HttpContent chunk1 = new DefaultHttpContent(Unpooled.wrappedBuffer(new byte[123]));
  decoder.channelRead(ctx, chunk1);
  LastHttpContent chunk2 = new DefaultLastHttpContent(Unpooled.wrappedBuffer(new byte[200]));
  decoder.channelRead(ctx, chunk2);

  resultTester.assertValue(result -> result.length == 323);
}

@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject)
    throws Exception {
  _channelMediator.readFromServerChannel(httpObject);
  if (httpObject instanceof DefaultLastHttpContent) {
    _channelMediator.writeToClientAndDisconnect(httpObject);
  } else {
    _channelMediator.writeToClient(httpObject);
  }
}

@SuppressWarnings("unchecked")
private void processEntity() {
    if (entity != null) {
        EntityWriter entityWriter = EntityWriterRegistry.getEntityWriter(entity.getClass());
        entityWriter.writeData(httpCarbonMessage, entity, mediaType, chunkSize, request.getHttpCarbonMessage());
    } else {
        ByteBuffer byteBuffer = ByteBuffer.allocate(0);
        httpCarbonMessage.addHttpContent(new DefaultLastHttpContent(Unpooled.wrappedBuffer(byteBuffer)));
        try {
            request.getHttpCarbonMessage().respond(httpCarbonMessage);
        } catch (ServerConnectorException e) {
            throw new RuntimeException("Error while sending the response.", e);
        }
    }
}

private void handleHandlerException(HandlerException e, Request request) {
    try {
        HTTPCarbonMessage failureResponse = e.getFailureResponse();
        failureResponse.addHttpContent(new DefaultLastHttpContent());
        request.respond(failureResponse);
    } catch (ServerConnectorException e1) {
        log.error("Error while sending the response.", e);
    }
}

@Override
public void close() {
    try {
        flush();
        httpCarbonMessage.addHttpContent(new DefaultLastHttpContent());
        super.close();
    } catch (IOException e) {
        LOG.error("Error while closing output stream but underlying resources are reset", e);
    } finally {
        byteBufferOutputStream = null;
        buffer = null;
    }
}

@SuppressWarnings("unchecked")
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    if (msg instanceof FullHttpMessage) {
        FullHttpMessage fullHttpMessage = (FullHttpMessage) msg;
        sourceReqCmsg = setupCarbonMessage(fullHttpMessage);
        notifyRequestListener(sourceReqCmsg, ctx);
        ByteBuf content = ((FullHttpMessage) msg).content();
        sourceReqCmsg.addHttpContent(new DefaultLastHttpContent(content));
        if (handlerExecutor != null) {
            handlerExecutor.executeAtSourceRequestSending(sourceReqCmsg);
        }

    } else if (msg instanceof HttpRequest) {
        HttpRequest httpRequest = (HttpRequest) msg;
        sourceReqCmsg = setupCarbonMessage(httpRequest);
        notifyRequestListener(sourceReqCmsg, ctx);
    } else {
        if (sourceReqCmsg != null) {
            if (msg instanceof HttpContent) {
                HttpContent httpContent = (HttpContent) msg;
                sourceReqCmsg.addHttpContent(httpContent);
                if (Util.isLastHttpContent(httpContent)) {
                    if (handlerExecutor != null) {
                        handlerExecutor.executeAtSourceRequestSending(sourceReqCmsg);
                    }
                }
            }
        }
    }
}

@Test
public void testProxyServer() {

    try {
        ByteBuffer byteBuffer = ByteBuffer.wrap(testValue.getBytes(Charset.forName("UTF-8")));
        HTTPCarbonMessage msg = new HTTPCarbonMessage(
                new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "https://localhost:8081"));
        msg.setHeader("Host", "localhost:8081");
        msg.addHttpContent(new DefaultLastHttpContent(Unpooled.wrappedBuffer(byteBuffer)));

        CountDownLatch latch = new CountDownLatch(1);
        HTTPConnectorListener listener = new HTTPConnectorListener(latch);
        HttpResponseFuture responseFuture = httpClientConnector.send(msg);
        responseFuture.setHttpConnectorListener(listener);

        latch.await(5, TimeUnit.SECONDS);

        HTTPCarbonMessage response = listener.getHttpResponseMessage();
        assertNotNull(response);
        String result = new BufferedReader(
                new InputStreamReader(new HttpMessageDataStreamer(response).getInputStream())).lines()
                .collect(Collectors.joining("\n"));
        assertEquals(testValue, result);
    } catch (Exception e) {
        TestUtil.handleException("Exception occurred while running testProxyServer", e);
    }
}