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

@Override
public void beforeRequest(Channel clientChannel, HttpContent httpContent,
    HttpProxyInterceptPipeline pipeline) throws Exception {
  if (content != null) {
    ByteBuf temp = httpContent.content().slice();
    content.writeBytes(temp);
    if (httpContent instanceof LastHttpContent) {
      try {
        byte[] contentBts = new byte[content.readableBytes()];
        content.readBytes(contentBts);
        ((HttpRequestInfo) pipeline.getHttpRequest()).setContent(contentBts);
      } finally {
        ReferenceCountUtil.release(content);
        content = null; //状态回归
      }
    }
  }
  pipeline.beforeRequest(clientChannel, httpContent);
}

@Override
protected List<Entry<String, String>> getTrailingHeaders(Object obj) {
    List chunkes = (List) obj;
    Map<String, String> trailingHeaders = new LinkedHashMap<>();
    for (Object chunk : chunkes) {
        if (chunk instanceof LastHttpContent) {
            Iterator<Entry<String, String>> it = ((LastHttpContent) chunk)
                    .trailingHeaders().iterator();
            while (it.hasNext()) {
                Entry<String, String> next = it.next();
                trailingHeaders.put(next.getKey(), next.getValue());
            }
        }
    }
    return new ArrayList<>(trailingHeaders.entrySet());
}

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {

    HttpProcessingState state = ChannelAttributes.getHttpProcessingStateForChannel(ctx).get();

    allowCompressionForThisRequest = false;

    if (state != null) {
        // We only want to allow compression if the endpoint being hit is *not* a ProxyRouterEndpoint, the response is full, and the response size
        // is greater than the threshold
        boolean isFull = msg instanceof HttpResponse && msg instanceof LastHttpContent;
        boolean endpointAllowed = endpointAllowsCompression(state.getEndpointForExecution());
        boolean responseInfoAllowed =
            state.getResponseInfo() == null || !state.getResponseInfo().isPreventCompressedOutput();
        if (isFull && endpointAllowed && responseInfoAllowed
            && ((LastHttpContent) msg).content().readableBytes() > responseSizeThresholdBytes) {
            allowCompressionForThisRequest = true;
        }
    }

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

@Override
public PipelineContinuationBehavior doChannelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof LastHttpContent) {
        HttpProcessingState state = ChannelAttributes.getHttpProcessingStateForChannel(ctx).get();
        Endpoint<?> endpoint = state.getEndpointForExecution();
        RequestInfo reqInfo = state.getRequestInfo();
        // Don't bother trying to deserialize until we have an endpoint and the request content has fully arrived
        if (endpoint != null && reqInfo.isCompleteRequestWithAllChunks()) {
            // Setup the content deserializer if desired
            TypeReference<?> contentTypeRef = endpoint.requestContentType();
            if (contentTypeRef != null) {
                // A non-null TypeReference is available, so deserialization is possible. Retrieve the appropriate
                //      deserializer and setup the RequestInfo so that it can lazily deserialize when requested.
                ObjectMapper deserializer = endpoint.customRequestContentDeserializer(reqInfo);
                if (deserializer == null)
                    deserializer = defaultRequestContentDeserializer;

                //noinspection unchecked
                reqInfo.setupContentDeserializer(deserializer, contentTypeRef);
            }
        }
    }

    return PipelineContinuationBehavior.CONTINUE;
}

@Test
public void StreamingChannel_doStreamChunk_works_as_expected_when_last_chunk_already_sent_downstream_and_incoming_chunk_is_empty_last_chunk() {
    // given
    streamingChannelSpy.downstreamLastChunkSentHolder.heldObject = true;

    LastHttpContent contentChunkMock = mock(LastHttpContent.class);
    ByteBuf contentByteBufMock = mock(ByteBuf.class);
    doReturn(contentByteBufMock).when(contentChunkMock).content();
    doReturn(0).when(contentByteBufMock).readableBytes();

    ChannelFuture successFutureMock = mock(ChannelFuture.class);
    doReturn(successFutureMock).when(channelMock).newSucceededFuture();

    // when
    ChannelFuture result = streamingChannelSpy.doStreamChunk(contentChunkMock);

    // then
    verify(channelMock, never()).writeAndFlush(any(Object.class));
    verify(contentChunkMock).release();
    verify(channelMock).newSucceededFuture();
    assertThat(result).isSameAs(successFutureMock);
}

@Before
public void beforeMethod() {
    channelMock = mock(Channel.class);
    ctxMock = mock(ChannelHandlerContext.class);
    stateAttributeMock = mock(Attribute.class);
    state = new HttpProcessingState();

    doReturn(channelMock).when(ctxMock).channel();
    doReturn(stateAttributeMock).when(channelMock).attr(ChannelAttributes.HTTP_PROCESSING_STATE_ATTRIBUTE_KEY);
    doReturn(state).when(stateAttributeMock).get();

    firstChunkMsgMock = mock(HttpRequest.class);
    lastChunkMsgMock = mock(LastHttpContent.class);

    filter1Mock = mock(RequestAndResponseFilter.class);
    filter2Mock = mock(RequestAndResponseFilter.class);
    filtersList = Arrays.asList(filter1Mock, filter2Mock);

    handlerSpy = spy(new RequestFilterHandler(filtersList));

    requestInfoMock = mock(RequestInfo.class);

    state.setRequestInfo(requestInfoMock);
}

@Test
public void argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo_only_returns_true_for_HttpRequest_or_LastHttpContent() {
    // given
    Object httpRequestMsg = mock(HttpRequest.class);
    Object lastHttpContentMsg = mock(LastHttpContent.class);
    Object httpMessageMsg = mock(HttpMessage.class);

    // expect
    assertThat(handlerSpy.argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo(
        DO_CHANNEL_READ, ctxMock, httpRequestMsg, null)
    ).isTrue();
    assertThat(handlerSpy.argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo(
        DO_CHANNEL_READ, ctxMock, lastHttpContentMsg, null)
    ).isTrue();
    assertThat(handlerSpy.argsAreEligibleForLinkingAndUnlinkingDistributedTracingInfo(
        DO_CHANNEL_READ, ctxMock, httpMessageMsg, null)
    ).isFalse();
}

@Before
public void beforeMethod() {
    stateMock = mock(HttpProcessingState.class);
    ctxMock = mock(ChannelHandlerContext.class);
    channelMock = mock(Channel.class);
    pipelineMock = mock(ChannelPipeline.class);
    stateAttrMock = mock(Attribute.class);
    proxyRouterProcessingStateAttrMock = mock(Attribute.class);
    metricsListenerMock = mock(MetricsListener.class);
    msgMockFirstChunkOnly = mock(HttpRequest.class);
    msgMockFullRequest = mock(FullHttpRequest.class);
    msgMockLastChunkOnly = mock(LastHttpContent.class);
    idleChannelTimeoutHandlerMock = mock(IdleChannelTimeoutHandler.class);

    doReturn(channelMock).when(ctxMock).channel();
    doReturn(pipelineMock).when(ctxMock).pipeline();
    doReturn(idleChannelTimeoutHandlerMock).when(pipelineMock).get(IDLE_CHANNEL_TIMEOUT_HANDLER_NAME);
    doReturn(stateAttrMock).when(channelMock).attr(ChannelAttributes.HTTP_PROCESSING_STATE_ATTRIBUTE_KEY);
    doReturn(stateMock).when(stateAttrMock).get();
    doReturn(proxyRouterProcessingStateAttrMock).when(channelMock).attr(ChannelAttributes.PROXY_ROUTER_PROCESSING_STATE_ATTRIBUTE_KEY);

    handler = new RequestStateCleanerHandler(metricsListenerMock, incompleteHttpCallTimeoutMillis);
}

@Before
public void beforeMethod() {
    msg = mock(LastHttpContent.class);
    stateMock = mock(HttpProcessingState.class);
    ctxMock = mock(ChannelHandlerContext.class);
    channelMock = mock(Channel.class);
    stateAttrMock = mock(Attribute.class);
    endpointMock = mock(Endpoint.class);
    requestInfoSpy = spy((RequestInfo<String>) RequestInfoImpl.dummyInstanceForUnknownRequests());
    String rawContentString = UUID.randomUUID().toString();
    Whitebox.setInternalState(requestInfoSpy, "rawContent", rawContentString);
    Whitebox.setInternalState(requestInfoSpy, "rawContentBytes", rawContentString.getBytes());
    defaultHandlerDeserializerMock = mock(ObjectMapper.class);

    doReturn(channelMock).when(ctxMock).channel();
    doReturn(stateAttrMock).when(channelMock).attr(ChannelAttributes.HTTP_PROCESSING_STATE_ATTRIBUTE_KEY);
    doReturn(stateMock).when(stateAttrMock).get();
    doReturn(endpointMock).when(stateMock).getEndpointForExecution();
    doReturn(requestInfoSpy).when(stateMock).getRequestInfo();
    doReturn(contentTypeRef).when(endpointMock).requestContentType();
    doReturn(endpointMatcher).when(endpointMock).requestMatcher();

    handler = new RequestContentDeserializerHandler(defaultHandlerDeserializerMock);
}

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

@Override
public void write(Channel channel, NettyResponseFuture<?> future) throws IOException {
    final InputStream is = inputStream;

    if (future.isStreamWasAlreadyConsumed()) {
        if (is.markSupported())
            is.reset();
        else {
            LOGGER.warn("Stream has already been consumed and cannot be reset");
            return;
        }
    } else {
        future.setStreamWasAlreadyConsumed(true);
    }

    channel.write(new ChunkedStream(is), channel.newProgressivePromise()).addListener(
            new ProgressListener(future.getAsyncHandler(), future, false, getContentLength()) {
                public void operationComplete(ChannelProgressiveFuture cf) {
                    closeSilently(is);
                    super.operationComplete(cf);
                }
            });
    channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
}

private void handleChunk(HttpContent chunk,//
        final Channel channel,//
        final NettyResponseFuture<?> future,//
        AsyncHandler<?> handler) throws IOException, Exception {

    boolean interrupt = false;
    boolean last = chunk instanceof LastHttpContent;

    // Netty 4: the last chunk is not empty
    if (last) {
        LastHttpContent lastChunk = (LastHttpContent) chunk;
        HttpHeaders trailingHeaders = lastChunk.trailingHeaders();
        if (!trailingHeaders.isEmpty()) {
            interrupt = handler.onHeadersReceived(new HttpResponseHeaders(trailingHeaders, true)) != State.CONTINUE;
        }
    }

    ByteBuf buf = chunk.content();
    if (!interrupt && !(handler instanceof StreamedAsyncHandler) && (buf.readableBytes() > 0 || last)) {
        HttpResponseBodyPart part = config.getResponseBodyPartFactory().newResponseBodyPart(buf, last);
        interrupt = updateBodyAndInterrupt(future, handler, part);
    }

    if (interrupt || last)
        finishUpdate(future, channel, !last);
}

@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 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
protected void onOutboundComplete() {
    if (isWebsocket() || isInboundCancelled()) {
        return;
    }
    if (markSentHeaderAndBody()) {
        if (log.isDebugEnabled()) {
            log.debug("No sendHeaders() called before complete, sending " + "zero-length header");
        }
        channel().writeAndFlush(newFullEmptyBodyMessage());
    }
    else if (markSentBody()) {
        channel().writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
    }
    channel().read();
}

@Override
protected void onInboundNext(ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof HttpContent) {
        if (msg != LastHttpContent.EMPTY_LAST_CONTENT) {
            super.onInboundNext(ctx, msg);
        }
        if (msg instanceof LastHttpContent) {
            onInboundComplete();
            if (isOutboundDone()) {
                onHandlerTerminate();
            }
            else {
                //force auto read to enable more accurate close selection now inbound is done
                channel().config()
                         .setAutoRead(true);
            }
        }
    }
    else {
        super.onInboundNext(ctx, msg);
    }
}

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

private void handleUploadMessage(HttpMessage httpMsg, Message uploadMessage) throws IOException{
    if (httpMsg instanceof HttpContent) { 
           HttpContent chunk = (HttpContent) httpMsg;
           decoder.offer(chunk); 
           try {
               while (decoder.hasNext()) {
                   InterfaceHttpData data = decoder.next();
                   if (data != null) {
                       try { 
                        handleUploadFile(data, uploadMessage);
                       } finally {
                           data.release();
                       }
                   }
               }
           } catch (EndOfDataDecoderException e1) { 
            //ignore
           }

           if (chunk instanceof LastHttpContent) {  
            resetUpload();
           }
       }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg)
        throws Exception {
    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;
        RandomAccessFile output = null;
        FileChannel oc = null;
        try {
            output = new RandomAccessFile(file, "rw");
            oc = output.getChannel();
            oc.position(oc.size());
            ByteBuf buffer = content.content();
            for (int i = 0, len = buffer.nioBufferCount(); i < len; i++) {
                oc.write(buffer.nioBuffers()[i]);
            }
        } finally {
            IOUtils.closeQuietly(oc);
            IOUtils.closeQuietly(output);
        }
        if (content instanceof LastHttpContent) {
            ctx.close();
        }
    }
}

@Override
public void handleReadFromClient(ChannelMediator channelMediator, HttpObject httpObject) {
  if (channelMediator == null) {
    throw new IllegalStateException("HRFC: ChannelMediator can't be null");
  }

  try {
    if (httpObject instanceof HttpRequest) {
      HttpRequest httpRequest = (HttpRequest) httpObject;
      _clientRequestBuilder.interpretHttpRequest(httpRequest);
      _clientRequestBuilder.addHeaders(httpRequest);
    }

    if (httpObject instanceof HttpContent) {
      _clientRequestBuilder.appendHttpContent((HttpContent) httpObject);
    }

    if (httpObject instanceof LastHttpContent) {
      HttpResponse httpResponse = playBack();
      channelMediator.writeToClientAndDisconnect(httpResponse);
    }
  } catch (IOException e) {
    throw new RuntimeException("HRFC: Failed to replay HttpContent", e);
  }
}

@Override
public void handleReadFromServer(HttpObject httpObject) {
  if (httpObject instanceof HttpResponse) {
    _serverResponseBuilder = new RecordedHttpResponseBuilder((HttpResponse) httpObject);
  }

  try {
    if (httpObject instanceof HttpContent) {
      _serverResponseBuilder.appendHttpContent((HttpContent) httpObject);
    }

    if (httpObject instanceof LastHttpContent) {
      _sceneAccessLayer.record(_clientRequestBuilder.build(), _serverResponseBuilder.build());
    }
  } catch (IOException e) {
    throw new RuntimeException("HRFS: Failed to record HttpContent", e);
  }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) {
    try {
        if (msg instanceof HttpRequest) {
            initFileChannel();
        } else if (msg instanceof HttpContent) {
            if (fileChnl == null) {
                initFileChannel();
            }
            ByteBuf byteBuf = ((HttpContent) msg).content();
            writeBytesToFile(byteBuf);
        } else if (msg instanceof LastHttpContent) {
            if (fileChnl != null && outStream != null) {
                fileChnl.close();
                outStream.close();
            }
            ctx.close();
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) {
    try {
        if (msg instanceof HttpRequest) {
            initFileChannel();
        } else if (msg instanceof HttpContent) {
            if (fileChnl == null) {
                initFileChannel();
            }
            ByteBuf byteBuf = ((HttpContent) msg).content();
            writeBytesToFile(byteBuf);
        } else if (msg instanceof LastHttpContent) {
            if (fileChnl != null && outStream != null) {
                fileChnl.close();
                outStream.close();
            }
            ctx.close();
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof HttpRequest) {
        httpRequest = (HttpRequest) msg;
        byteBuf = Unpooled.buffer();
    }

    if (msg instanceof HttpContent) {
        HttpContent httpContent = (HttpContent) msg;
        ByteBuf buf = httpContent.content();
        byteBuf.writeBytes(buf);
        buf.release();

        if (httpContent instanceof LastHttpContent) {
            try {
                ctx.fireChannelRead(new Object[] { httpRequest, byteBuf });
            } finally {
                httpRequest = null;
                byteBuf = null;
            }
        }
    }
}

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

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

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws
        Exception {
    try {
        if (msg instanceof LastHttpContent) {
            LastHttpContent res = (LastHttpContent) msg;

            DataTask task = queue.removeFirst();

            String json = res.content().toString(CharsetUtil.UTF_8);

            task.promise.setSuccess(mapper.readValue(json, task.clazz));

            waitingForResponse = false;
        }

        workQueue();
    } finally {
        ReferenceCountUtil.release(msg);
    }
}

@Override
protected void encode(
        ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {

    if (upgraded) {
        out.add(ReferenceCountUtil.retain(msg));
        return;
    }

    if (msg instanceof HttpRequest && !done) {
        queue.offer(((HttpRequest) msg).method());
    }

    super.encode(ctx, msg, out);

    if (failOnMissingResponse && !done) {
        // check if the request is chunked if so do not increment
        if (msg instanceof LastHttpContent) {
            // increment as its the last chunk
            requestResponseCounter.incrementAndGet();
        }
    }
}

@Override
public int read() throws IOException {
    httpCarbonMessage.setAlreadyRead(true); // TODO: No need to set this again and again
    if ((httpContent instanceof LastHttpContent) && chunkFinished) {
        return -1;
    } else if (chunkFinished) {
        httpContent = httpCarbonMessage.getHttpContent();
        byteBuffer = httpContent.content().nioBuffer();
        count = 0;
        limit = byteBuffer.limit();
        if (limit == 0) {
            return -1;
        }
        chunkFinished = false;
    }
    count++;
    if (count == limit) {
        int value = byteBuffer.get() & 0xff;
        chunkFinished = true;
        byteBuffer = null;
        httpContent.release();

        return value;
    }
    return byteBuffer.get() & 0xff;
}

public HttpContent getHttpContent() {
    try {
        if (!isConsumed.get() || !alreadyRead.get()) {
            HttpContent httpContent = httpContentQueue.poll(soTimeOut, TimeUnit.SECONDS);

            if (httpContent instanceof LastHttpContent) {
                isConsumed.set(true);
                alreadyRead.set(false);
                httpContentQueue.clear();
            }

            return httpContent;
        }
    } catch (InterruptedException e) {
        LOG.error("Error while retrieving http content from queue.", e);
    }
    return null;
}

public void waitAndReleaseAllEntities() {
    if (!isConsumed.get() && !alreadyRead.get()) {
        boolean isEndOfMessageProcessed = false;
        while (!isEndOfMessageProcessed) {
            try {
                HttpContent httpContent = httpContentQueue.poll(soTimeOut, TimeUnit.SECONDS);
                // This check is to make sure we add the last http content after getClone and avoid adding
                // empty content to bytebuf list again and again
                if (httpContent instanceof EmptyLastHttpContent) {
                    break;
                }

                if (httpContent instanceof LastHttpContent) {
                    isEndOfMessageProcessed = true;
                    isConsumed.set(true);
                }
                httpContent.release();
            } catch (InterruptedException e) {
                LOG.error("Error while getting full message body", e);
            }
        }
    }
}

/**
 * Check whether, redirect request is written to the backend when a redirect response is received.
 *
 * @throws URISyntaxException
 * @throws IOException
 */
@Test
public void unitTestForRedirectHandler() throws URISyntaxException, IOException {
    EmbeddedChannel embeddedChannel = new EmbeddedChannel();
    embeddedChannel.pipeline().addLast(new HttpResponseDecoder());
    embeddedChannel.pipeline().addLast(new HttpRequestEncoder());
    embeddedChannel.pipeline().addLast(new RedirectHandler(null, false, 5, false));
    HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.TEMPORARY_REDIRECT,
            Unpooled.EMPTY_BUFFER);
    response.headers().set(HttpHeaders.Names.LOCATION, FINAL_DESTINATION);
    embeddedChannel.attr(Constants.ORIGINAL_REQUEST)
            .set(createHttpRequest(Constants.HTTP_GET_METHOD, FINAL_DESTINATION));
    embeddedChannel.writeInbound(response);
    embeddedChannel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
    assertNotNull(embeddedChannel.readOutbound());
}

/**
 * When the maximum redirect count reached, channel should not do any more redirects.
 *
 * @throws URISyntaxException
 * @throws IOException
 */
@Test
public void unitTestForRedirectLoop() throws URISyntaxException, IOException {
    EmbeddedChannel embeddedChannel = new EmbeddedChannel();
    embeddedChannel.pipeline().addLast(new HttpResponseDecoder());
    embeddedChannel.pipeline().addLast(new HttpRequestEncoder());
    embeddedChannel.pipeline()
            .addLast(Constants.IDLE_STATE_HANDLER, new IdleStateHandler(50000, 50000, 0, TimeUnit.MILLISECONDS));
    embeddedChannel.pipeline().addLast(new RedirectHandler(null, false, 5, false, null, false));
    HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.TEMPORARY_REDIRECT,
            Unpooled.EMPTY_BUFFER);
    response.headers().set(HttpHeaders.Names.LOCATION, FINAL_DESTINATION);
    embeddedChannel.attr(Constants.ORIGINAL_REQUEST)
            .set(createHttpRequest(Constants.HTTP_GET_METHOD, FINAL_DESTINATION));
    embeddedChannel.attr(Constants.RESPONSE_FUTURE_OF_ORIGINAL_CHANNEL).set(new HttpResponseFutureImpl());
    TargetChannel targetChannel = new TargetChannel(null, null);
    targetChannel.setChannel(embeddedChannel);
    embeddedChannel.attr(Constants.TARGET_CHANNEL_REFERENCE).set(targetChannel);
    embeddedChannel.attr(Constants.REDIRECT_COUNT).set(5);
    embeddedChannel.writeInbound(response);
    embeddedChannel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
    assertNull(embeddedChannel.readOutbound());
}

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

@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg)
        throws Exception {
    if (msg instanceof HttpContent) {
        HttpContent content = (HttpContent) msg;
        RandomAccessFile output = null;
        FileChannel oc = null;
        try {
            output = new RandomAccessFile(file, "rw");
            oc = output.getChannel();
            oc.position(oc.size());
            ByteBuf buffer = content.content();
            for (int i = 0, len = buffer.nioBufferCount(); i < len; i++) {
                oc.write(buffer.nioBuffers()[i]);
            }
        } finally {
            IOUtils.closeQuietly(oc);
            IOUtils.closeQuietly(output);
        }
        if (content instanceof LastHttpContent) {
            ctx.close();
        }
    }
}

/**
 * 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
 */
public HttpPostRequestDecoder offer(HttpContent content) throws ErrorDataDecoderException {
    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;
}

/**
* The methods that reads server's response. 
* 
* It prints the server response for the sent HTTP request.
*
* @param ctx The channel handler context
* @param msg The received HTTP response
* 
*/
  @Override
  public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
    if (msg instanceof HttpResponse) {
          HttpResponse response = (HttpResponse) msg;

          logger.debug("----Header----");
          logger.debug(response.toString());
          logger.debug("----End of header----");
          if (HttpHeaders.isTransferEncodingChunked(response)) {
              logger.debug("----Chunked content----");
          } else {
              logger.debug("----Content----");
          }
      }
      if (msg instanceof HttpContent) {
          HttpContent content = (HttpContent) msg;

          logger.debug(content.content().toString(CharsetUtil.UTF_8));
          if (content instanceof LastHttpContent) {
              logger.debug("----End of content----");
              ctx.close();
          }
      }
  }