@Test
public void addDecoderReplaysLastHttp() {
	ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
	EmbeddedChannel channel = new EmbeddedChannel();
	new HttpClientOperations(() -> channel, ConnectionObserver.emptyListener(),
			ClientCookieEncoder.STRICT, ClientCookieDecoder.STRICT)
			.addHandler(new JsonObjectDecoder());
	channel.writeInbound(new DefaultLastHttpContent(buf));

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

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

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

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

@Test
public void addNamedDecoderReplaysLastHttp() {
	ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
	EmbeddedChannel channel = new EmbeddedChannel();
	new HttpClientOperations(() -> channel, ConnectionObserver.emptyListener(),
			ClientCookieEncoder.STRICT, ClientCookieDecoder.STRICT)
			.addHandler("json", new JsonObjectDecoder());
	channel.writeInbound(new DefaultLastHttpContent(buf));

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

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

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

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

@Test
public void addEncoderReplaysLastHttp() {
	ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
	EmbeddedChannel channel = new EmbeddedChannel();
	new HttpClientOperations(() -> channel, ConnectionObserver.emptyListener(),
			ClientCookieEncoder.STRICT, ClientCookieDecoder.STRICT)
			.addHandler(new JsonObjectDecoder());
	channel.writeInbound(new DefaultLastHttpContent(buf));

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

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

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

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

@Test
public void addNamedEncoderReplaysLastHttp() {
	ByteBuf buf = Unpooled.copiedBuffer("{\"foo\":1}", CharsetUtil.UTF_8);
	EmbeddedChannel channel = new EmbeddedChannel();
	new HttpClientOperations(() -> channel, ConnectionObserver.emptyListener(),
			ClientCookieEncoder.STRICT, ClientCookieDecoder.STRICT)
			.addHandler("json", new JsonObjectDecoder());
	channel.writeInbound(new DefaultLastHttpContent(buf));

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

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

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

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

private static void echoJsonStreamDecoding() {

        TcpServer<Person, Person> transport = Netty4TcpServer.<Person, Person>create(
                0,
                new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addFirst(
                                new JsonObjectDecoder(),
                                new JacksonJsonCodec());
                    }
                });

        ReactorTcpServer.create(transport)
                .start(connection -> {
                    connection.log("input")
                            .observeComplete(v -> LOG.info("Connection input complete"))
                            .capacity(1)
                            .consume(person -> {
                                person = new Person(person.getLastName(), person.getFirstName());
                                Streams.wrap(connection.writeWith(Streams.just(person))).consume();
                            });
                    return Streams.never();
                });

    }
 

public <T> void post(TypeReference<T> typeReference, ResultCallback<T> resultCallback, InputStream body) {
    HttpRequestProvider requestProvider = httpPostRequestProvider(null);

    Channel channel = getChannel();

    JsonResponseCallbackHandler<T> jsonResponseHandler = new JsonResponseCallbackHandler<>(
            objectMapper,
            typeReference,
            resultCallback);

    HttpResponseHandler responseHandler = new HttpResponseHandler(requestProvider, resultCallback);

    channel.pipeline().addLast(new ChunkedWriteHandler());
    channel.pipeline().addLast(responseHandler);
    channel.pipeline().addLast(new JsonObjectDecoder(3 * 1024 * 1024));
    channel.pipeline().addLast(jsonResponseHandler);

    postChunkedStreamRequest(requestProvider, channel, body);
}
 

public <T> void get(TypeReference<T> typeReference, ResultCallback<T> resultCallback) {

        HttpRequestProvider requestProvider = httpGetRequestProvider();

        Channel channel = getChannel();

        JsonResponseCallbackHandler<T> jsonResponseHandler = new JsonResponseCallbackHandler<>(
                objectMapper,
                typeReference,
                resultCallback);

        HttpResponseHandler responseHandler = new HttpResponseHandler(requestProvider, resultCallback);

        channel.pipeline().addLast(responseHandler);
        channel.pipeline().addLast(new JsonObjectDecoder(3 * 1024 * 1024));
        channel.pipeline().addLast(jsonResponseHandler);

        sendRequest(requestProvider, channel);

        return;
    }
 

public <T> void post(final Object entity, TypeReference<T> typeReference, final ResultCallback<T> resultCallback) {

        HttpRequestProvider requestProvider = httpPostRequestProvider(entity);

        Channel channel = getChannel();

        JsonResponseCallbackHandler<T> jsonResponseHandler = new JsonResponseCallbackHandler<>(
                objectMapper,
                typeReference,
                resultCallback);

        HttpResponseHandler responseHandler = new HttpResponseHandler(requestProvider, resultCallback);

        channel.pipeline().addLast(responseHandler);
        channel.pipeline().addLast(new JsonObjectDecoder(3 * 1024 * 1024));
        channel.pipeline().addLast(jsonResponseHandler);

        sendRequest(requestProvider, channel);

        return;
    }