1

我有一个 Reactor-Netty HttpServer，我从 HttpClient 向其发送超过 1024 个字节的数据。

在客户端中，我在 Flux 中获取请求的不同部分，其想法是将数据连接成一个我最终可以解析的字符串。我认为这可以由 HttpObjectAggregator 完成，但可惜不是。

当客户端收到内容时，它以

DefaultHttpContent(数据: PooledSlicedByteBuf(ridx: 1024, widx: 1024, cap: 1024/1024, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1024, widx: 1024, cap: 1024)), decoderResult: 成功)

和

DefaultLastHttpContent(数据: PooledSlicedByteBuf(ridx: 98, widx: 98, cap: 98/98, unwrapped: PooledUnsafeDirectByteBuf(ridx: 98, widx: 98, cap: 1024)), decoderResult: 成功)

如果我尝试“提取”字节，我发现它们不可用。

服务器

    private void startServer() {

    HttpServer server = HttpServer.create(opts -> opts.listen(8092).option(ChannelOption.SO_RCVBUF, 1024 * 1024)
            .option(ChannelOption.SO_SNDBUF, 1024 * 1024).option(ChannelOption.SO_KEEPALIVE, true)
            .afterChannelInit(channelInit -> {
                ChannelPipeline pipeline = channelInit.pipeline();
                pipeline.addLast("aggregator", new HttpObjectAggregator(64 * 1024));
            }));

    Mono<? extends NettyContext> context = server.newRouter(routes -> {
        routes.post("/test", postHandler());
        routes.put("/test", postHandler());
    });

    context.subscribe().block(Duration.ofSeconds(30));

}

BiFunction<? super HttpServerRequest, ? super HttpServerResponse, ? extends Publisher<Void>> postHandler() {
    return (req, resp) -> {
        final String combinedContent = new String();
        req.requestHeaders().entries().forEach(entry -> {
            String key = entry.getKey();
            String value = entry.getValue();
            if (key.equalsIgnoreCase("content-length")) {
                contentLength = Integer.parseInt(value);
            }
            log.debug(String.format("header [%s=>%s]", key, value));
        });

        req.receiveContent().doOnNext(request -> {
            ByteBuf bb = ByteBufUtil.readBytes(request.content().alloc(), request.content(), request.content().readableBytes());
            log.debug("HEX DUMP : " +  ByteBufUtil.prettyHexDump(request.content()));
            combinedContent.concat(new String(ByteBufUtil.getBytes(bb)));
        }).subscribe((what) -> {
            log.debug("Combined Request = " + combinedContent.toString());
            log.debug("what = " + what);
        });


        return Mono.empty();

    };
}

客户

@Test
public void testClientMessage() {
    HttpClient httpClient;
    try {
        httpClient = webClient.connect("localhost", 8092, false);
        webClient.send(httpClient, "/test", getTestRequest());

    } catch (SSLException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

private String getResponse() {
    return "Well Done";
}

private String getTestRequest() {
    return "moreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytesmoreThan1024bytes";
}

知道如何获取数据吗？

数据记录到控制台，例如：

98B +------------------------------------------------ -+ | 0 1 2 3 4 5 6 7 8 9 abcdef | +--------+---------------------------------------- ---------+----------------+ |00000000| 54 68 61 6e 31 30 32 34 62 79 74 65 73 6d 6f 72 |Than1024bytesmor| |00000010| 65 54 68 61 6e 31 30 32 34 62 79 74 65 73 6d 6f |eThan1024bytesmo| |00000020| 72 65 54 68 61 6e 31 30 32 34 62 79 74 65 73 6d |reThan1024bytesm| |00000030| 6f 72 65 54 68 61 6e 31 30 32 34 62 79 74 65 73 |oreThan1024bytes| |00000040| 6d 6f 72 65 54 68 61 6e 31 30 32 34 62 79 74 65 |超过1024字节| |00000050| 73 6d 6f 72 65 54 68 61 6e 31 30 32 34 62 79 74 |超过1024字节| |00000060| 65 73 |es | +--------+---------------------------------------- ---------+----------------+ | +--------+---------------------------------------- ---------+----------------+ | +--------+---------------------------------------- ---------+----------------+

谢谢

4

0 回答 0