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

    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;

        storeRequestContent(httpContent);

        if (httpContent instanceof LastHttpContent) {
            LastHttpContent lastHttpContent = (LastHttpContent) httpContent;
            trailingHeaders = lastHttpContent .trailingHeaders();
        }
    }

    return null;
}
 

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

@Override
public void runBufferedBodyContentThroughFilter(ZuulFilter<?, ?> filter) {
    //Loop optimized for the common case: Most filters' processContentChunk() return
    // original chunk passed in as is without any processing
    for (int i=0; i < bodyChunks.size(); i++) {
        final HttpContent origChunk = bodyChunks.get(i);
        final HttpContent filteredChunk = filter.processContentChunk(this, origChunk);
        if ((filteredChunk != null) && (filteredChunk != origChunk)) {
            //filter actually did some processing, set the new chunk in and release the old chunk.
            bodyChunks.set(i, filteredChunk);
            final int refCnt = origChunk.refCnt();
            if (refCnt > 0) {
                origChunk.release(refCnt);
            }
        }
    }
}
 

protected Ack toAck(HttpContent content) {
  ByteBuf byteBuf = content.content() ;
  byte[] data = new byte[byteBuf.readableBytes()] ;
  byteBuf.readBytes(data) ;
  //byteBuf.release() ;
  Ack ack = null;
  try {
    ack = (Ack)IOUtil.deserialize(data);
  } catch (Exception e) {
    e.printStackTrace();
    ack = new Ack() ;
    ack.setStatus(Ack.Status.ERROR);
    ack.setMessage(e.getMessage());
  }
  return ack ;
}
 

@Test
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 = DefaultObjectMapper.readValueAsMap(error);
    assertEquals(1, parsed.size());
    assertNotNull(parsed.get("errors"));
}
 

@Test
public void releaseContentChunks_clear_on_chunk_list_but_does_not_release_chunks_if_contentChunksWillBeReleasedExternally_is_true() {
    // given
    RequestInfoImpl<?> requestInfo = RequestInfoImpl.dummyInstanceForUnknownRequests();
    requestInfo.contentChunksWillBeReleasedExternally();
    List<HttpContent> contentChunkList = Arrays.asList(mock(HttpContent.class), mock(HttpContent.class));
    requestInfo.contentChunks.addAll(contentChunkList);
    assertThat(requestInfo.contentChunks.size(), is(contentChunkList.size()));

    // when
    requestInfo.releaseContentChunks();

    // then
    for (HttpContent chunkMock : contentChunkList) {
        verify(chunkMock, never()).release();
    }
    assertThat(requestInfo.contentChunks.isEmpty(), is(true));
}
 

/**
 * Initialized the internals from a new chunk
 *
 * @param content
 *            the new received chunk
 * @throws ErrorDataDecoderException
 *             if there is a problem with the charset decoding or other
 *             errors
 */
@Override
public HttpPostMultipartRequestDecoder offer(HttpContent content) {
    checkDestroyed();

    // Maybe we should better not copy here for performance reasons but this will need
    // more care by the caller to release the content in a correct manner later
    // So maybe something to optimize on a later stage
    ByteBuf buf = content.content();
    if (undecodedChunk == null) {
        undecodedChunk = buf.copy();
    } else {
        undecodedChunk.writeBytes(buf);
    }
    if (content instanceof LastHttpContent) {
        isLastChunk = true;
    }
    parseBody();
    if (undecodedChunk != null && undecodedChunk.writerIndex() > discardThreshold) {
        undecodedChunk.discardReadBytes();
    }
    return this;
}
 

@Test
public void testDecodeDataAsClient() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(new Http2StreamFrameToHttpObjectCodec(false));
    ByteBuf hello = Unpooled.copiedBuffer("hello world", CharsetUtil.UTF_8);
    assertTrue(ch.writeInbound(new DefaultHttp2DataFrame(hello)));

    HttpContent content = ch.readInbound();
    try {
        assertThat(content.content().toString(CharsetUtil.UTF_8), is("hello world"));
        assertFalse(content instanceof LastHttpContent);
    } finally {
        content.release();
    }

    assertThat(ch.readInbound(), is(nullValue()));
    assertFalse(ch.finish());
}
 

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception
{
    State state = null;

    // Reset the state as each new outbound request goes out.
    if (msg instanceof HttpRequest) {
        state = createNewState(ctx.channel());
    }

    // Update the outbound body size with this chunk.
    if (msg instanceof HttpContent) {
        if (state == null) {
            state = getOrCreateCurrentState(ctx.channel());
        }
        state.outboundBodySize += ((HttpContent) msg).content().readableBytes();
    }

    super.write(ctx, msg, promise);
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
{
    State state = null;
    
    // Reset the state as each new inbound request comes in.
    if (msg instanceof HttpRequest) {
        state = createNewState(ctx.channel());
    }
    
    // Update the inbound body size with this chunk.
    if (msg instanceof HttpContent) {
        if (state == null) {
            state = getOrCreateCurrentState(ctx.channel());
        }
        state.inboundBodySize += ((HttpContent) msg).content().readableBytes();
    }

    super.channelRead(ctx, msg);
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
{
    try {
        CurrentPassport passport = passport(ctx);

        if (msg instanceof HttpResponse) {
            passport.add(PassportState.IN_RESP_HEADERS_RECEIVED);
        }

        if (msg instanceof LastHttpContent) {
            passport.add(PassportState.IN_RESP_LAST_CONTENT_RECEIVED);
        }
        else if (msg instanceof HttpContent) {
            passport.add(PassportState.IN_RESP_CONTENT_RECEIVED);
        }
    }
    finally {
        super.channelRead(ctx, msg);
    }
}
 

@Override
public HttpObject serverToProxyResponse(HttpObject httpObject) {
    if (httpObject instanceof HttpResponse) {
        httpResponse = (HttpResponse) httpObject;
        captureContentEncoding(httpResponse);
    }

    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;

        storeResponseContent(httpContent);

        if (httpContent instanceof LastHttpContent) {
            LastHttpContent lastContent = (LastHttpContent) httpContent;
            captureTrailingHeaders(lastContent);

            captureFullResponseContents();
        }
    }

    return super.serverToProxyResponse(httpObject);
}
 

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

    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;

        storeRequestContent(httpContent);

        if (httpContent instanceof LastHttpContent) {
            LastHttpContent lastHttpContent = (LastHttpContent) httpContent;
            trailingHeaders = lastHttpContent .trailingHeaders();
        }
    }

    return null;
}
 

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

    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;

        storeRequestContent(httpContent);

        if (httpContent instanceof LastHttpContent) {
            LastHttpContent lastHttpContent = (LastHttpContent) httpContent;
            trailingHeaders = lastHttpContent .trailingHeaders();
        }
    }

    return null;
}
 

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (null == stringBuilder) {
        stringBuilder = new StringBuilder();
    }

    if (msg instanceof io.netty.handler.codec.http.HttpRequest) {
        io.netty.handler.codec.http.HttpRequest request = (io.netty.handler.codec.http.HttpRequest)msg;
        if (null == httpRequest) {
            httpRequest = new NettyHttpRequest(ctx, request);
        }

    } else if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent)msg;
        ByteBuf byteBuf = content.content();
        byte []data = new byte[byteBuf.capacity()];
        byteBuf.readBytes(data);

        stringBuilder.append(new String(data));
    }

    super.channelRead(ctx, msg);
}
 

@Override
public void channelRead(
        final ChannelHandlerContext context,
        final Object message) {

    runIf(message, HttpResponse.class, httpResponse -> {
        this.response = new Response(REMOTE, httpResponse);
        this.responseStage = requestStage.process(response);
    });

    runIf(message, HttpContent.class, response::buffer);

    runIf(message, LastHttpContent.class, content ->
            sequence.set(1, throwingRunnable(responseStage::write)));

    context.fireChannelRead(message);
}
 

@Test
public void write_adds_to_finalContentLength_if_msg_is_HttpContent_and_finalContentLength_is_not_null() throws Exception {
    // given
    HttpContent msgMock = mock(HttpContent.class);
    ByteBuf contentMock = mock(ByteBuf.class);
    int contentBytes = (int)(Math.random() * 10000);

    doReturn(contentMock).when(msgMock).content();
    doReturn(contentBytes).when(contentMock).readableBytes();

    int initialFinalContentLengthValue = (int)(Math.random() * 10000);
    responseInfo.setFinalContentLength((long)initialFinalContentLengthValue);
    assertThat(responseInfo.getFinalContentLength()).isEqualTo(initialFinalContentLengthValue);

    // when
    handler.write(ctxMock, msgMock, promiseMock);

    // then
    assertThat(responseInfo.getFinalContentLength()).isEqualTo(initialFinalContentLengthValue + contentBytes);
}
 

/**
 * Initialized the internals from a new chunk
 *
 * @param content
 *            the new received chunk
 * @throws ErrorDataDecoderException
 *             if there is a problem with the charset decoding or other
 *             errors
 */
@Override
public HttpPostMultipartRequestDecoder offer(HttpContent content) {
    checkDestroyed();

    // Maybe we should better not copy here for performance reasons but this will need
    // more care by the caller to release the content in a correct manner later
    // So maybe something to optimize on a later stage
    ByteBuf buf = content.content();
    if (undecodedChunk == null) {
        undecodedChunk = buf.copy();
    } else {
        undecodedChunk.writeBytes(buf);
    }
    if (content instanceof LastHttpContent) {
        isLastChunk = true;
    }
    parseBody();
    if (undecodedChunk != null && undecodedChunk.writerIndex() > discardThreshold) {
        undecodedChunk.discardReadBytes();
    }
    return this;
}
 

/**
 *
 * @param factory the factory used to create InterfaceHttpData
 * @param request the request to decode
 * @param charset the charset to use as default
 * @throws NullPointerException      for request or charset or factory
 * @throws ErrorDataDecoderException if the default charset was wrong when
 *                                   decoding or other errors
 */
public HttpPostMultipartRequestDecoder(HttpDataFactory factory, HttpRequest request, Charset charset) {
	this.request = checkNotNull(request, "request");
	this.charset = checkNotNull(charset, "charset");
	this.factory = checkNotNull(factory, "factory");
	// Fill default values

	setMultipart(this.request.headers().get(HttpHeaderNames.CONTENT_TYPE));
	if (request instanceof HttpContent) {
		// Offer automatically if the given request is als type of HttpContent
		// See #1089
		offer((HttpContent) request);
	} else {
		undecodedChunk = buffer();
		parseBody();
	}
}
 

/**
 * Initialized the internals from a new chunk
 *
 * @param content the new received chunk
 * @throws ErrorDataDecoderException if there is a problem with the charset
 *                                   decoding or other errors
 */
@Override
public HttpPostMultipartRequestDecoder offer(HttpContent content) {
	checkDestroyed();

	// Maybe we should better not copy here for performance reasons but this will
	// need
	// more care by the caller to release the content in a correct manner later
	// So maybe something to optimize on a later stage
	ByteBuf buf = content.content();
	if (undecodedChunk == null) {
		undecodedChunk = buf.copy();
	} else {
		undecodedChunk.writeBytes(buf);
	}
	if (content instanceof LastHttpContent) {
		isLastChunk = true;
	}
	parseBody();
	if (undecodedChunk != null && undecodedChunk.writerIndex() > discardThreshold) {
		undecodedChunk.discardReadBytes();
	}
	return this;
}
 

/**
 * Send the curated request and verify that the response and satisfaction are expected.
 * @param httpRequest the request to be handled by {@link MockNettyMessageProcessor}
 * @param requestContent the content needed for POST request
 * @param expectedStatus the expected {@link HttpResponseStatus}
 * @param shouldBeSatisfied whether the request should be satisfied or not
 * @throws IOException
 */
private void sendRequestAndEvaluateResponsePerformance(HttpRequest httpRequest, String requestContent,
    HttpResponseStatus expectedStatus, boolean shouldBeSatisfied) throws IOException {
  EmbeddedChannel channel = createEmbeddedChannel();
  channel.pipeline().get(MockNettyMessageProcessor.class);
  channel.writeInbound(httpRequest);
  if (requestContent != null) {
    channel.writeInbound(createContent(requestContent, true));
  }
  HttpResponse response = channel.readOutbound();
  assertEquals("Unexpected response status", expectedStatus, response.status());
  if (requestContent != null) {
    HttpContent responseContent = channel.readOutbound();
    String returnedContent = RestTestUtils.getContentString(responseContent);
    responseContent.release();
    assertEquals("Content does not match with expected content", requestContent, returnedContent);
  }
  if (shouldBeSatisfied) {
    assertTrue(httpRequest.method() + " request should be satisfied", MockNettyRequest.mockTracker.isSatisfied());
  } else {
    assertFalse(httpRequest.method() + " request should be unsatisfied", MockNettyRequest.mockTracker.isSatisfied());
  }
}
 

@Override
public void close() {
  if (channelOpen.compareAndSet(true, false)) {
    setAutoRead(true);
    contentLock.lock();
    try {
      logger.trace("Closing NettyRequest {} with {} content chunks unread", getUri(), requestContents.size());
      // For non-POST we usually have one content chunk unread - this the LastHttpContent chunk. This is OK.
      HttpContent content = requestContents.poll();
      while (content != null) {
        ReferenceCountUtil.release(content);
        content = requestContents.poll();
      }
    } finally {
      contentLock.unlock();
      restRequestMetricsTracker.nioMetricsTracker.markRequestCompleted();
      if (digestCalculationTimeInMs >= 0) {
        nettyMetrics.digestCalculationTimeInMs.update(digestCalculationTimeInMs);
      }
      if (callbackWrapper != null) {
        callbackWrapper.invokeCallback(channelException);
      }
    }
  }
}
 

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

/**
 * Tests {@link NettyRequest#addContent(HttpContent)} and
 * {@link NettyRequest#readInto(AsyncWritableChannel, Callback)} by creating a {@link NettyRequest}, adding a few
 * pieces of content to it and then reading from it to match the stream with the added content.
 * <p/>
 * The read happens at different points of time w.r.t content addition (before, during, after).
 * @param digestAlgorithm the digest algorithm to use. Can be empty or {@code null} if digest checking is not
 *                        required.
 * @param useCopyForcingByteBuf if {@code true}, uses {@link CopyForcingByteBuf} instead of the default
 *                              {@link ByteBuf}.
 * @param method Http method
 * @throws Exception
 */
private void contentAddAndReadTest(String digestAlgorithm, boolean useCopyForcingByteBuf, HttpMethod method)
    throws Exception {
  // non composite content
  // start reading before addition of content
  List<HttpContent> httpContents = new ArrayList<>();
  ByteBuffer content = generateContent(httpContents, useCopyForcingByteBuf);
  doContentAddAndReadTest(digestAlgorithm, content, httpContents, 0, method);

  // start reading in the middle of content add
  httpContents.clear();
  content = generateContent(httpContents, useCopyForcingByteBuf);
  doContentAddAndReadTest(digestAlgorithm, content, httpContents, httpContents.size() / 2, method);

  // start reading after all content added
  httpContents.clear();
  content = generateContent(httpContents, useCopyForcingByteBuf);
  doContentAddAndReadTest(digestAlgorithm, content, httpContents, httpContents.size(), method);

  // composite content
  httpContents.clear();
  content = generateCompositeContent(httpContents);
  doContentAddAndReadTest(digestAlgorithm, content, httpContents, 0, method);
}
 

@Override
public HttpObject serverToProxyResponse(HttpObject httpObject) {
    // if a ServerResponseCaptureFilter is configured, delegate to it to collect the server's response. if it is not
    // configured, we still need to capture basic information (timings, HTTP status, etc.), just not content.
    if (responseCaptureFilter != null) {
        responseCaptureFilter.serverToProxyResponse(httpObject);
    }

    if (httpObject instanceof HttpResponse) {
        HttpResponse httpResponse = (HttpResponse) httpObject;

        captureResponse(httpResponse);
    }

    if (httpObject instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) httpObject;

        captureResponseSize(httpContent);
    }

    if (httpObject instanceof LastHttpContent) {
        if (dataToCapture.contains(CaptureType.RESPONSE_CONTENT)) {
            captureResponseContent(responseCaptureFilter.getHttpResponse(), responseCaptureFilter.getFullResponseContents());
        }

        harEntry.getResponse().setBodySize((long)(responseBodySize.get()));
    }

    this.harEntry.setTime(getTotalElapsedTime(this.harEntry.getTimings()));

    return super.serverToProxyResponse(httpObject);
}
 

@Override
public final void channelRead0(final ChannelHandlerContext ctx, HttpObject msg) throws Exception {
  if (msg instanceof HttpRequest) {
    HttpRequest req = (HttpRequest) msg;
    log.info("Received Request {}", req);
    route = proxyConfig.getRouteProvider(req);
    updater = route.handle(req, ctx);
  } else if (msg instanceof LastHttpContent) {
    updater.update((LastHttpContent) msg);
  } else if (msg instanceof HttpContent) {
    updater.update((HttpContent) msg);
  }
}
 

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpContent chunk)
  throws IOException {
  chunk.content().readBytes(out, chunk.content().readableBytes());
  if (chunk instanceof LastHttpContent) {
    response.headers().set(CONNECTION, CLOSE);
    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
    releaseDfsResources();
  }
}
 

@Override
public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) {
    if (msg instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) msg;

        System.err.println("STATUS: " + response.status());
        System.err.println("VERSION: " + response.protocolVersion());
        System.err.println();

        if (!response.headers().isEmpty()) {
            for (CharSequence name: response.headers().names()) {
                for (CharSequence value: response.headers().getAll(name)) {
                    System.err.println("HEADER: " + name + " = " + value);
                }
            }
            System.err.println();
        }

        if (HttpUtil.isTransferEncodingChunked(response)) {
            System.err.println("CHUNKED CONTENT {");
        } else {
            System.err.println("CONTENT {");
        }
    }
    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;

        System.err.print(content.content().toString(CharsetUtil.UTF_8));
        System.err.flush();

        if (content instanceof LastHttpContent) {
            System.err.println("} END OF CONTENT");
            ctx.close();
        }
    }
}
 

@Override
public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
    if (msg instanceof HttpResponse) {
        HttpResponse response = (HttpResponse) msg;

        System.out.println("STATUS: " + response.status());
        System.out.println("VERSION: " + response.protocolVersion());
        System.out.println();

        if (!response.headers().isEmpty()) {
            for (CharSequence name : response.headers().names()) {
                for (CharSequence value : response.headers().getAll(name)) {
                    System.out.println("HEADER: " + name + " = " + value);
                }
            }
            System.out.println();
        }

        if (HttpUtil.isTransferEncodingChunked(response)) {
            System.out.println("CHUNKED CONTENT {");
        } else {
            System.out.println("CONTENT {");
        }
    }
    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;

        System.out.print(content.content().toString(CharsetUtil.UTF_8));
        System.out.flush();

        if (content instanceof LastHttpContent) {
            System.out.println("} END OF CONTENT");
            queue.add(ctx.channel().newSucceededFuture());
        }
    }
}
 

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