netty - PooledUnsafeDirectByteBuf を持つ HttpContent の読み取り

Question

HttpClient から 1024 バイトを超えるデータを送信する Reactor-Netty HttpServer があります。

クライアントでは、データを最終的に解析できる文字列に連結するという考えで、リクエストのさまざまな部分を Flux で取得します。私はこれが HttpObjectAggregator によって行われた可能性があるという意見でしたが、残念ながら違います。

クライアントがコンテンツを受け取ると、次のようになります。

DefaultHttpContent(データ: PooledSlicedByteBuf(ridx: 1024、widx: 1024、キャップ: 1024/1024、アンラップ: PooledUnsafeDirectByteBuf(ridx: 1024、widx: 1024、キャップ: 1024))、decoderResult: 成功)

と

DefaultLastHttpContent(データ: PooledSlicedByteBuf(ridx: 98、widx: 98、キャップ: 98/98、アンラップ: PooledUnsafeDirectByteBuf(ridx: 98、widx: 98、キャップ: 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 |moreThan1024byte| |00000050| 73 6d 6f 72 65 54 68 61 6e 31 30 32 34 62 79 74 |smoreThan1024byt| |00000060| 65 73 |es | +--------+-------------------------------------------- ------+----------------+ | | +--------+-------------------------------------------- ------+----------------+ | | +--------+-------------------------------------------- ------+----------------+

ありがとう