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

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

@Override
public void onMessage(HTTPCarbonMessage httpMessage) {
    executor.execute(() -> {
        try {
            InputStream inputStream = new HttpMessageDataStreamer(httpMessage).getInputStream();
            String response = new String(ByteStreams.toByteArray(inputStream), Charset.defaultCharset());
            String alteredContent = "Altered " + response + " content";

            HTTPCarbonMessage newMsg = httpMessage.cloneCarbonMessageWithOutData();
            newMsg.addHttpContent(new DefaultHttpContent(
                    Unpooled.wrappedBuffer(alteredContent.getBytes(Charset.defaultCharset()))));
            newMsg.setEndOfMsgAdded(true);

            httpMessage.respond(newMsg);
        } catch (IOException | ServerConnectorException e) {
            logger.error("Error occurred during message processing ", e);
        }

    });
}

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

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    Class<?> recievedMsgClass = msg.getClass();

    if (HttpServerResponse.class.isAssignableFrom(recievedMsgClass)) {
        @SuppressWarnings("rawtypes")
        HttpServerResponse rxResponse = (HttpServerResponse) msg;
        if (keepAlive && !rxResponse.getHeaders().contains(HttpHeaders.Names.CONTENT_LENGTH)) {
            // If there is no content length & it is a keep alive connection. We need to specify the transfer
            // encoding as chunked as we always send data in multiple HttpContent.
            // On the other hand, if someone wants to not have chunked encoding, adding content-length will work
            // as expected.
            rxResponse.getHeaders().add(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
        }
        super.write(ctx, rxResponse.getNettyResponse(), promise);
    } else if (ByteBuf.class.isAssignableFrom(recievedMsgClass)) {
        HttpContent content = new DefaultHttpContent((ByteBuf) msg);
        super.write(ctx, content, promise);
    } else {
        super.write(ctx, msg, promise); // pass through, since we do not understand this message.
    }

}

/**
 * Tests that {@link NettyRequest#close()} leaves any added {@link HttpContent} the way it was before it was added.
 * (i.e no reference count changes).
 * @throws RestServiceException
 */
@Test
public void closeTest() throws RestServiceException {
  Channel channel = new MockChannel();
  NettyRequest nettyRequest = createNettyRequest(HttpMethod.POST, "/", null, channel);
  Queue<HttpContent> httpContents = new LinkedBlockingQueue<HttpContent>();
  for (int i = 0; i < 5; i++) {
    ByteBuffer content = ByteBuffer.wrap(TestUtils.getRandomBytes(1024));
    HttpContent httpContent = new DefaultHttpContent(Unpooled.wrappedBuffer(content));
    nettyRequest.addContent(httpContent);
    httpContents.add(httpContent);
  }
  closeRequestAndValidate(nettyRequest, channel);
  while (httpContents.peek() != null) {
    assertEquals("Reference count of http content has changed", 1, httpContents.poll().refCnt());
  }
}

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

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

@Override
public void subscribe(Subscriber<? super HttpContent> subscriber) {
    publisher.subscribe(new Subscriber<ByteBuffer>() {
        @Override
        public void onSubscribe(Subscription subscription) {
            subscriber.onSubscribe(subscription);
        }

        @Override
        public void onNext(ByteBuffer byteBuffer) {
            ByteBuf buffer = channel.alloc().buffer(byteBuffer.remaining());
            buffer.writeBytes(byteBuffer);
            HttpContent content = new DefaultHttpContent(buffer);
            subscriber.onNext(content);
        }

        @Override
        public void onError(Throwable t) {
            subscriber.onError(t);
        }

        @Override
        public void onComplete() {
            subscriber.onComplete();
        }
    });
}

@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
public void addContentChunk_does_not_add_chunk_to_contentChunks_list_if_contentChunksWillBeReleasedExternally_is_true() {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.isCompleteRequestWithAllChunks = false;
    requestInfo.contentChunksWillBeReleasedExternally();
    HttpContent chunk = new DefaultHttpContent(Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8));

    // when
    requestInfo.addContentChunk(chunk);

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

@Test
public void convertContentChunksToRawString_and_convertContentChunksToRawBytes_works_with_EmptyByteBuf_chunks() throws IOException {
    // given
    Charset contentCharset = CharsetUtil.UTF_8;
    String chunk1Content = UUID.randomUUID().toString();
    String chunk2Content = UUID.randomUUID().toString();
    byte[] chunk1Bytes = chunk1Content.getBytes(contentCharset);
    byte[] chunk2Bytes = chunk2Content.getBytes(contentCharset);
    ByteBuf chunk1ByteBuf = Unpooled.copiedBuffer(chunk1Bytes);
    ByteBuf chunk2ByteBuf = Unpooled.copiedBuffer(chunk2Bytes);
    Collection<HttpContent> chunkCollection = Arrays.asList(
            new DefaultHttpContent(chunk1ByteBuf),
            new DefaultHttpContent(new EmptyByteBuf(ByteBufAllocator.DEFAULT)),
            new DefaultHttpContent(chunk2ByteBuf),
            new DefaultHttpContent(new EmptyByteBuf(ByteBufAllocator.DEFAULT))
    );

    // when
    String resultString = HttpUtils.convertContentChunksToRawString(contentCharset, chunkCollection);
    byte[] resultBytes = HttpUtils.convertContentChunksToRawBytes(chunkCollection);

    // then
    String expectedResultString = chunk1Content + chunk2Content;
    assertThat(resultString, is(expectedResultString));
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    baos.write(chunk1Bytes);
    baos.write(chunk2Bytes);
    assertThat(resultBytes, is(baos.toByteArray()));
}

@Test
public void convertContentChunksToRawBytes_returns_null_if_total_bytes_is_zero() {
    // given
    Collection<HttpContent> chunkCollection = Arrays.asList(new DefaultHttpContent(new EmptyByteBuf(ByteBufAllocator.DEFAULT)),
            new DefaultHttpContent(new EmptyByteBuf(ByteBufAllocator.DEFAULT)));

    // when
    byte[] resultBytes = HttpUtils.convertContentChunksToRawBytes(chunkCollection);

    // then
    assertThat(resultBytes, nullValue());
}

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

private ChannelFuture doWriteSplitData(
        ChannelHandlerContext ctx, int id, HttpData data, boolean endStream) {

    try {
        int offset = data.offset();
        int remaining = data.length();
        ChannelFuture lastFuture;
        for (;;) {
            // Ensure an HttpContent does not exceed the maximum length of a cleartext TLS record.
            final int chunkSize = Math.min(MAX_TLS_DATA_LENGTH, remaining);
            lastFuture = write(ctx, id, new DefaultHttpContent(dataChunk(data, offset, chunkSize)), false);
            remaining -= chunkSize;
            if (remaining == 0) {
                break;
            }
            offset += chunkSize;
        }

        if (endStream) {
            lastFuture = write(ctx, id, LastHttpContent.EMPTY_LAST_CONTENT, true);
        }

        ctx.flush();
        return lastFuture;
    } finally {
        ReferenceCountUtil.safeRelease(data);
    }
}

@Override
public void write(int b) throws IOException {
    if (buffer == null) {
        buffer = BufferFactory.getInstance().getBuffer();
    }
    if (buffer.hasRemaining()) {
        buffer.put((byte) b);
    } else {
        buffer.flip();
        httpCarbonMessage.addHttpContent(new DefaultHttpContent(Unpooled.wrappedBuffer(buffer)));
        buffer = BufferFactory.getInstance().getBuffer();
        buffer.put((byte) b);
    }
}

@Override
public void flush() throws IOException {
    if (buffer != null && buffer.position() > 0) {
        buffer.flip();
        httpCarbonMessage.addHttpContent(new DefaultHttpContent(Unpooled.wrappedBuffer(buffer)));
        buffer = BufferFactory.getInstance().getBuffer();
    }
}

HttpContent buildContent(ChannelHandlerContext ctx, StreamingData data) {
  if (data.endOfStream()) {
    LastHttpContent last = new DefaultLastHttpContent(data.content());
    if (data.trailingHeaders() != null) {
      last.trailingHeaders().add(data.trailingHeaders().http1Headers(true, true));
    }
    // setChannelRequest(ctx, null);
    return last;
  } else {
    return new DefaultHttpContent(data.content());
  }
}

HttpContent buildContent(ChannelHandlerContext ctx, StreamingData data) {
  if (data.endOfStream()) {
    LastHttpContent last = new DefaultLastHttpContent(data.content());
    if (data.trailingHeaders() != null) {
      last.trailingHeaders().add(data.trailingHeaders().http1Headers(true, false));
    }
    setChannelRequest(ctx, null);
    return last;
  } else {
    return new DefaultHttpContent(data.content());
  }
}

@Override
public @Nullable HttpContent readChunk(ByteBufAllocator allocator) throws Exception {
    if (hasSentTerminatingChunk) {
        return null;
    }
    ByteBuf nextChunk = readNextChunk();
    if (nextChunk != null) {
        return new DefaultHttpContent(nextChunk);
    }
    // chunked transfer encoding must be terminated by a final chunk of length zero
    hasSentTerminatingChunk = true;
    return LastHttpContent.EMPTY_LAST_CONTENT;
}

@Override
public void flush() throws IOException {
    if (buffer.readableBytes() != 0) {
        ctx.writeAndFlush(new DefaultHttpContent(buffer.copy()));
        buffer.clear();
    }
    super.flush();
}

@Override
public HttpContent marshallResponsePart(final Object res,
                                        final HttpResponseStatus status,
                                        final boolean rawStream) throws IOException {
  final String content = rawStream ?
    mapper.writeValueAsString(res) + "<br/>\n" :
    ChunkHeader.ELEMENT_HEADER + mapper.writeValueAsString(res) + "\n";

  final ByteBuf buf = Unpooled.copiedBuffer(content, CharsetUtil.UTF_8);
  return new DefaultHttpContent(buf);
}

/**
 * @param fallbackContentType
 * Only used if Content-Type header has not been set.
 * If not given and Content-Type header is not set, it is set to
 * "application/xml" if text param is Node or NodeSeq, otherwise it is
 * set to "text/plain".
 */
public ChannelFuture respondText(Object text, String fallbackContentType) throws Exception {
    if (doneResponding) {
        throwDoubleResponseError(text);
    }

    final String respondedText = text.toString();

    if (!nonChunkedResponseOrFirstChunkSent && !response.headers().contains(CONTENT_TYPE)) {
        // Set content type
        if (fallbackContentType != null) {
            // https://developers.google.com/speed/docs/best-practices/rendering#SpecifyCharsetEarly
            final String withCharset =
                    fallbackContentType.toLowerCase().contains("charset") ?
                            fallbackContentType :
                            fallbackContentType + "; charset=" + server.charset();

            response.headers().set(CONTENT_TYPE, withCharset);
        } else {
            response.headers().set(CONTENT_TYPE, "text/plain; charset=" + server.charset());
        }
    }

    ByteBuf buf = Unpooled.copiedBuffer(respondedText, server.charset());
    if (HttpHeaders.isTransferEncodingChunked(response)) {
        respondHeadersOnlyForFirstChunk();
        return channel.writeAndFlush(new DefaultHttpContent(buf));
    } else {
        // Pitfall: Content-Length is number of bytes, not characters
        response.headers().set(CONTENT_LENGTH, buf.readableBytes());
        response.content().writeBytes(buf);
        return respond();
    }
}

/**
 * If Content-Type header is not set, it is set to "application/octet-stream".
 *
 * @param byteBuf Will be released
 */
public ChannelFuture respondBinary(ByteBuf byteBuf) throws Exception {
    if (HttpUtil.isTransferEncodingChunked(response)) {
        respondHeadersOnlyForFirstChunk();
        return channel.writeAndFlush(new DefaultHttpContent(byteBuf));
    } else {
        if (!response.headers().contains(CONTENT_TYPE))
            response.headers().set(CONTENT_TYPE, "application/octet-stream");

        response.headers().set(CONTENT_LENGTH, byteBuf.readableBytes());
        response.content().writeBytes(byteBuf);
        return respond();
    }
}

/**
 * Create a new handler emitting SSE events if a request is made on <code>urlMapping</code> URL.
 * @param urlMapping the path the handler will look for beginning to emit SSE events,
 *                   otherwise ignore HTTP messages
 * @param interval   interval between publication of events
 * @throws UnsupportedEncodingException
 */
public HystrixMetricsStreamHandler(String urlMapping, long interval) throws UnsupportedEncodingException {
    super();
    this.urlMapping = urlMapping;
    this.interval = interval;
    PING = new DefaultHttpContent(Unpooled.unreleasableBuffer(Unpooled.wrappedBuffer("ping: \n".getBytes("UTF-8"))));
    BEGIN_DATA = Unpooled.unreleasableBuffer(Unpooled.wrappedBuffer("data: ".getBytes("UTF-8")));
    END_DATA = Unpooled.unreleasableBuffer(Unpooled.wrappedBuffer("\n\n".getBytes("UTF-8")));
}

/**
 * Creates {@link HttpContent} wrapping the {@code content}.
 * @param content the content to wrap.
 * @param isLast {@code true} if this is the last piece of content. {@code false} otherwise.
 * @return a {@link HttpContent} wrapping the {@code content}.
 */
private HttpContent createContent(String content, boolean isLast) {
  ByteBuf buf = Unpooled.copiedBuffer(content.getBytes());
  if (isLast) {
    return new DefaultLastHttpContent(buf);
  } else {
    return new DefaultHttpContent(buf);
  }
}

@Override
public void onNext(ByteBuffer t) {
    ByteBuf buffer = Unpooled.wrappedBuffer(t.array());
    HttpContent content = new DefaultHttpContent(buffer);
    subscriber.onNext(content);
}

@Test
public void httpPipelining() throws Exception {

    AtomicInteger i = new AtomicInteger();

    NettyContext server = HttpServer.create(0)
                               .newHandler((req, resp) -> resp.header(HttpHeaderNames.CONTENT_LENGTH, "1")
                                                              .sendString(Mono.just(i.incrementAndGet())
                                                                              .flatMap(d -> Mono.delay(
                                                                                      Duration.ofSeconds(
                                                                                              4 - d))
                                                                                             .map(x -> d + "\n"))))
                               .block(Duration.ofSeconds(30));

    DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
            HttpMethod.GET,
            "/plaintext");

    CountDownLatch latch = new CountDownLatch(6);

    NettyContext client = TcpClient.create(server.address()
                                                 .getPort())
                                   .newHandler((in, out) -> {
                                           in.context()
                                             .addHandlerFirst(new HttpClientCodec());

                                           in.receiveObject()
                                             .ofType(DefaultHttpContent.class)
                                             .as(ByteBufFlux::fromInbound)
                                             .asString()
                                             .log()
                                             .map(Integer::parseInt)
                                             .subscribe(d -> {
                                                     for (int x = 0; x < d; x++) {
                                                         latch.countDown();
                                                     }
                                             });

                                           return out.sendObject(Flux.just(request.retain(),
                                                                           request.retain(),
                                                                           request.retain()))
                                                     .neverComplete();
                                   })
                                   .block(Duration.ofSeconds(30));

    Assert.assertTrue(latch.await(45, TimeUnit.SECONDS));

    server.dispose();
    client.dispose();
}

@Test
public void httpAndJsonDecoders() {

    EmbeddedChannel channel = new EmbeddedChannel();
    NettyContext testContext = () -> channel;

    ChannelHandler handler = new JsonObjectDecoder(true);
    testContext.addHandlerLast("foo", handler);

    HttpOperations.autoAddHttpExtractor(testContext, "foo", handler);

    String json1 = "[{\"some\": 1} , {\"valu";
    String json2 = "e\": true, \"test\": 1}]";

    Object[] content = new Object[3];
    content[0] = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    content[1] =
            new DefaultHttpContent(Unpooled.copiedBuffer(json1, CharsetUtil.UTF_8));
    content[2] = new DefaultLastHttpContent(Unpooled.copiedBuffer(json2,
            CharsetUtil.UTF_8));

    channel.writeInbound(content);

    Object t = channel.readInbound();
    assertThat(t, instanceOf(HttpResponse.class));
    assertThat(t, not(instanceOf(HttpContent.class)));

    t = channel.readInbound();
    assertThat(t, instanceOf(ByteBuf.class));
    assertThat(((ByteBuf) t).toString(CharsetUtil.UTF_8), is("{\"some\": 1}"));
    ((ByteBuf) t).release();

    t = channel.readInbound();
    assertThat(t, instanceOf(ByteBuf.class));
    assertThat(((ByteBuf) t).toString(CharsetUtil.UTF_8),
            is("{\"value\": true, \"test\": 1}"));
    ((ByteBuf) t).release();

    t = channel.readInbound();
    assertThat(t, is(LastHttpContent.EMPTY_LAST_CONTENT));
    ((LastHttpContent) t).release();

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

/** Builds HttpContent blobs from ByteBuffers. */
private static HttpContent toHttpContent(ByteBuffer buffer) {
  return new DefaultHttpContent(Unpooled.wrappedBuffer(buffer));
}

private static void testBinaryStreamUpload(boolean withSpace) throws Exception {
    // Boundary starts here with '=' to check against issue https://github.com/netty/netty/issues/3004
    final String boundary = "=dLV9Wyq26L_-JQxk6ferf-RT153LhOO";
    final String contentTypeValue;
    if (withSpace) {
        contentTypeValue = "multipart/form-data; boundary=" + boundary;
    } else {
        contentTypeValue = "multipart/form-data;boundary=" + boundary;
    }
    final DefaultHttpRequest req = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST,
            "http://localhost");

    req.setDecoderResult(DecoderResult.SUCCESS);
    req.headers().add(HttpHeaders.Names.CONTENT_TYPE, contentTypeValue);
    req.headers().add(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);

    // Force to use memory-based data.
    final DefaultHttpDataFactory inMemoryFactory = new DefaultHttpDataFactory(false);

    for (String data : Arrays.asList("", "\r", "\r\r", "\r\r\r")) {
        final String body =
                "--" + boundary + "\r\n" +
                "Content-Disposition: form-data; name=\"file\"; filename=\"tmp-0.txt\"\r\n" +
                "Content-Type: image/gif\r\n" +
                "\r\n" +
                data + "\r\n" +
                "--" + boundary + "--\r\n";

        // Create decoder instance to test.
        final HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(inMemoryFactory, req);

        decoder.offer(releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer(body, CharsetUtil.UTF_8))));
        decoder.offer(releaseLater(new DefaultHttpContent(Unpooled.EMPTY_BUFFER)));

        // Validate it's enough chunks to decode upload.
        assertTrue(decoder.hasNext());

        // Decode binary upload.
        MemoryFileUpload upload = (MemoryFileUpload) decoder.next();

        // Validate data has been parsed correctly as it was passed into request.
        assertEquals("Invalid decoded data [data=" + data.replaceAll("\r", "\\\\r") + ", upload=" + upload + ']',
            data, upload.getString(CharsetUtil.UTF_8));
        upload.release();
        decoder.destroy();
    }
}

@Test
public void testNoZeroOut() throws Exception {
    final String boundary = "E832jQp_Rq2ErFmAduHSR8YlMSm0FCY";

    final DefaultHttpDataFactory aMemFactory = new DefaultHttpDataFactory(false);

    DefaultHttpRequest aRequest = new DefaultHttpRequest(HttpVersion.HTTP_1_1,
                                                         HttpMethod.POST,
                                                         "http://localhost");
    aRequest.headers().set(HttpHeaders.Names.CONTENT_TYPE,
                           "multipart/form-data; boundary=" + boundary);
    aRequest.headers().set(HttpHeaders.Names.TRANSFER_ENCODING,
                           HttpHeaders.Values.CHUNKED);

    HttpPostRequestDecoder aDecoder = new HttpPostRequestDecoder(aMemFactory, aRequest);

    final String aData = "some data would be here. the data should be long enough that it " +
                         "will be longer than the original buffer length of 256 bytes in " +
                         "the HttpPostRequestDecoder in order to trigger the issue. Some more " +
                         "data just to be on the safe side.";

    final String body =
            "--" + boundary + "\r\n" +
            "Content-Disposition: form-data; name=\"root\"\r\n" +
            "Content-Type: text/plain\r\n" +
            "\r\n" +
            aData +
            "\r\n" +
            "--" + boundary + "--\r\n";

    byte[] aBytes = body.getBytes();

    int split = 125;

    ByteBufAllocator aAlloc = new UnpooledByteBufAllocator(true);
    ByteBuf aSmallBuf = aAlloc.heapBuffer(split, split);
    ByteBuf aLargeBuf = aAlloc.heapBuffer(aBytes.length - split, aBytes.length - split);

    aSmallBuf.writeBytes(aBytes, 0, split);
    aLargeBuf.writeBytes(aBytes, split, aBytes.length - split);

    aDecoder.offer(releaseLater(new DefaultHttpContent(aSmallBuf)));
    aDecoder.offer(releaseLater(new DefaultHttpContent(aLargeBuf)));

    aDecoder.offer(LastHttpContent.EMPTY_LAST_CONTENT);

    assertTrue("Should have a piece of data", aDecoder.hasNext());

    InterfaceHttpData aDecodedData = aDecoder.next();
    assertEquals(InterfaceHttpData.HttpDataType.Attribute, aDecodedData.getHttpDataType());

    Attribute aAttr = (Attribute) aDecodedData;
    assertEquals(aData, aAttr.getValue());

    aDecodedData.release();
    aDecoder.destroy();
}

@Test
public void testChunkCorrect() throws Exception {
   String payload = "town=794649819&town=784444184&town=794649672&town=794657800&town=" +
            "794655734&town=794649377&town=794652136&town=789936338&town=789948986&town=" +
            "789949643&town=786358677&town=794655880&town=786398977&town=789901165&town=" +
            "789913325&town=789903418&town=789903579&town=794645251&town=794694126&town=" +
            "794694831&town=794655274&town=789913656&town=794653956&town=794665634&town=" +
            "789936598&town=789904658&town=789899210&town=799696252&town=794657521&town=" +
            "789904837&town=789961286&town=789958704&town=789948839&town=789933899&town=" +
            "793060398&town=794659180&town=794659365&town=799724096&town=794696332&town=" +
            "789953438&town=786398499&town=794693372&town=789935439&town=794658041&town=" +
            "789917595&town=794655427&town=791930372&town=794652891&town=794656365&town=" +
            "789960339&town=794645586&town=794657688&town=794697211&town=789937427&town=" +
            "789902813&town=789941130&town=794696907&town=789904328&town=789955151&town=" +
            "789911570&town=794655074&town=789939531&town=789935242&town=789903835&town=" +
            "789953800&town=794649962&town=789939841&town=789934819&town=789959672&town=" +
            "794659043&town=794657035&town=794658938&town=794651746&town=794653732&town=" +
            "794653881&town=786397909&town=794695736&town=799724044&town=794695926&town=" +
            "789912270&town=794649030&town=794657946&town=794655370&town=794659660&town=" +
            "794694617&town=799149862&town=789953234&town=789900476&town=794654995&town=" +
            "794671126&town=789908868&town=794652942&town=789955605&town=789901934&town=" +
            "789950015&town=789937922&town=789962576&town=786360170&town=789954264&town=" +
            "789911738&town=789955416&town=799724187&town=789911879&town=794657462&town=" +
            "789912561&town=789913167&town=794655195&town=789938266&town=789952099&town=" +
            "794657160&town=789949414&town=794691293&town=794698153&town=789935636&town=" +
            "789956374&town=789934635&town=789935475&town=789935085&town=794651425&town=" +
            "794654936&town=794655680&town=789908669&town=794652031&town=789951298&town=" +
            "789938382&town=794651503&town=794653330&town=817675037&town=789951623&town=" +
            "789958999&town=789961555&town=794694050&town=794650241&town=794656286&town=" +
            "794692081&town=794660090&town=794665227&town=794665136&town=794669931";
    DefaultHttpRequest defaultHttpRequest =
                new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/");

    HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(defaultHttpRequest);

    int firstChunk = 10;
    int middleChunk = 1024;

    HttpContent part1 = new DefaultHttpContent(Unpooled.wrappedBuffer(
            payload.substring(0, firstChunk).getBytes()));
    HttpContent part2 = new DefaultHttpContent(Unpooled.wrappedBuffer(
            payload.substring(firstChunk, firstChunk + middleChunk).getBytes()));
    HttpContent part3 = new DefaultHttpContent(Unpooled.wrappedBuffer(
            payload.substring(firstChunk + middleChunk, firstChunk + middleChunk * 2).getBytes()));
    HttpContent part4 = new DefaultHttpContent(Unpooled.wrappedBuffer(
            payload.substring(firstChunk + middleChunk * 2).getBytes()));

    decoder.offer(part1);
    decoder.offer(part2);
    decoder.offer(part3);
    decoder.offer(part4);
}

public void addMessageBody(ByteBuffer msgBody) {
    isConsumed.set(false);
    httpContentQueue.add(new DefaultHttpContent(Unpooled.copiedBuffer(msgBody)));
}

@Override
public void write(ByteBuffer byteBuffer) {
    ByteBuf bbuf = Unpooled.copiedBuffer(byteBuffer);
    DefaultHttpContent httpContent = new DefaultHttpContent(bbuf);
    this.channelHandlerContext.write(httpContent);
}