Java 类io.netty.handler.codec.json.JsonObjectDecoder 实例源码

@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 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<T>(typeReference,
            resultCallback);

    HttpResponseHandler responseHandler = new HttpResponseHandler(requestProvider, resultCallback);

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

    postChunkedStreamRequest(requestProvider, channel, body);
}

/**
 * Modifies the pipeline to handle HTTP requests
 * @param ctx The calling channel handler context
 * @param maxRequestSize The maximum request size in bytes
 */
private void switchToHttp(final ChannelHandlerContext ctx, final int maxRequestSize) {
    ChannelPipeline p = ctx.pipeline();         
    p.addLast("httpHandler", new HttpServerCodec());  // TODO: config ?
    p.addLast("decompressor", new HttpContentDecompressor());
    p.addLast("aggregator", new HttpObjectAggregator(maxRequestSize)); 
    p.addLast("jsonDecoder", new JsonObjectDecoder(maxRequestSize, false));
    p.addLast("handler", jsonRpcHandler);
    p.remove(this);
}

@Override
    protected void decode(final ChannelHandlerContext ctx, final HttpRequest msg, final List<Object> out) throws Exception {
        final String uri = msg.uri();
        log.info("-----------------------> URI [{}]", uri);
        if(uri.endsWith("/favicon.ico")) {
            final DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, favicon);
            resp.headers().set(HttpHeaders.CONTENT_TYPE, "image/x-icon");
            resp.headers().setInt(HttpHeaders.CONTENT_LENGTH, favSize);             
            ctx.writeAndFlush(resp);
            return;
        }           
        ReferenceCountUtil.retain(msg);
        final ChannelPipeline p = ctx.pipeline();

        final int index = uri.indexOf("/api/");
        final String endpoint = index==-1 ? "" : uri.substring(5);
        if(index != -1 && pureJsonEndPoints.contains(endpoint) ) {
            log.info("Switching to PureJSON handler");
            p.addLast(eventExecutorGroup, "httpToJson", httpToJson);
//          p.addLast("jsonLogger", loggingHandler);
            p.addLast("jsonDecoder", new JsonObjectDecoder(true));
//          p.addLast("jsonLogger", loggingHandler);
            p.addLast("traceHandler", traceHandler);
            p.remove(this);
            if(msg instanceof FullHttpMessage) {
                out.add(msg);
            }

        } else {
            log.info("Switching to Http Request Manager");
            out.add(msg);
            p.addLast(eventExecutorGroup, "requestManager", HttpRequestManager.getInstance());
            p.remove(this);         
        }
    }

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

/**
     * {@inheritDoc}
     * @see io.netty.channel.SimpleChannelInboundHandler#channelRead0(io.netty.channel.ChannelHandlerContext, java.lang.Object)
     */
    @Override
    protected void channelRead0(final ChannelHandlerContext ctx, final HttpRequest msg) throws Exception {
        try {
            final String uri = msg.uri();
            final Channel channel = ctx.channel();
            if(uri.endsWith("/favicon.ico")) {
                final DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, favicon);
                resp.headers().set(HttpHeaders.CONTENT_TYPE, "image/x-icon");
                resp.headers().setInt(HttpHeaders.CONTENT_LENGTH, favSize);             
                ctx.writeAndFlush(resp);
                return;
            } else if(uri.equals("/api/put") || uri.equals("/api/metadata")) {              
                final ChannelPipeline p = ctx.pipeline();
//              p.addLast(loggingHandler, jsonAdapter, new JsonObjectDecoder(true), traceHandler);
                p.addLast(jsonAdapter, new JsonObjectDecoder(true), traceHandler);
//              if(msg instanceof FullHttpRequest) {
//                  ByteBuf b = ((FullHttpRequest)msg).content().copy();
//                  ctx.fireChannelRead(b);
//              }
                ctx.fireChannelRead(msg);
                p.remove("requestManager");
                log.info("Switched to JSON Trace Processor");
                return;
            }
            final TSDBHttpRequest r = new TSDBHttpRequest(msg, ctx.channel(), ctx);
            final HttpRequestHandler handler = requestHandlers.get(r.getRoute());
            if(handler==null) {
                r.send404().addListener(new GenericFutureListener<Future<? super Void>>() {
                    public void operationComplete(Future<? super Void> f) throws Exception {
                        log.info("404 Not Found for {} Complete: success: {}", r.getRoute(), f.isSuccess());
                        if(!f.isSuccess()) {
                            log.error("Error sending 404", f.cause());
                        }
                    };
                });
                return;
            }
            handler.process(r);         
        } catch (Exception ex) {
            log.error("HttpRequest Routing Error", ex);
        }       
    }

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

        HttpRequestProvider requestProvider = httpGetRequestProvider();

        Channel channel = getChannel();

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

        HttpResponseHandler responseHandler = new HttpResponseHandler(requestProvider, resultCallback);

        channel.pipeline().addLast(responseHandler);
        channel.pipeline().addLast(new JsonObjectDecoder());
        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<T>(typeReference,
                resultCallback);

        HttpResponseHandler responseHandler = new HttpResponseHandler(requestProvider, resultCallback);

        channel.pipeline().addLast(responseHandler);
        channel.pipeline().addLast(new JsonObjectDecoder());
        channel.pipeline().addLast(jsonResponseHandler);

        sendRequest(requestProvider, channel);

        return;
    }