java - 在 Reactor Netty HttpClient/HttpServer 配置中，我无法在发送到服务器的帖子上检索请求对象

Question

Reactor-Netty HttpClient 正在向 HttpServer 发送请求 HttpClientRequest。在服务器端，我无法从 HttpServerRequest 中获取请求对象。我可以从 HttpServerRequest 中检索 FluxByteBuf，但不能检索 Flux 对象中的 ByteBuf 对象。通常，订阅 Flux 将允许我获取请求对象，但这在这里不起作用。客户端已成功收到服务器返回的响应。有谁知道为什么订阅 Flux/Mono 在 Reactor-Netty 客户端/服务器的服务器端不起作用？

客户端代码是:

public class Client {

    private static final Logger log = Logger.getLogger(Client.class.getSimpleName());

    public static void main(String[] args) throws InterruptedException {

        Consumer<byte[]> onSuccess = (byte[] response) -> {
            ElectionResponse electionResponse = SerializationUtils.deserialize(response);
            log.info("response in onSuccess: "+electionResponse);

        };
        Consumer<Throwable> onError = (Throwable ex) -> {
            ex.getMessage();
        };

        Runnable onCompletion = () -> {
            System.out.println("Message Completed");

        };
        ElectionRequest electionRequest = new ElectionRequest("aRequest");
        byte[] requestBytes = SerializationUtils.serialize(electionRequest);
        ByteBuf requestByteBuf = Unpooled.copiedBuffer(requestBytes);

        HttpClient.create()
                .tcpConfiguration(tcpClient -> tcpClient.host("10.0.0.19"))
                .port(61005)
                .post()
                .uri("/echo")
                .send(Mono.just(requestByteBuf))
                .responseContent()
                .aggregate()
                .asByteArray()
                .subscribe(onSuccess, onError, onCompletion);

        try {
            Thread.sleep(15000);
        } catch(InterruptedException ie) {
            ie.printStackTrace();
        }
    }
}

服务器代码是:

public class Server {

    private static final Logger log = Logger.getLogger(Server.class.getSimpleName());

    public static void main(String[] args) {

        ElectionResponse electionResponse = new ElectionResponse("aResponse");
        byte[] responseArray = SerializationUtils.serialize(electionResponse);

        Consumer<ByteBuf> onSuccess = (ByteBuf request) -> {
            System.out.println("onSuccess: Request received!");
        };
        Consumer<Throwable> onError = (Throwable ex) -> {
            ex.getMessage();
        };
        Runnable onCompletion = () -> {
            System.out.println("Message Completed");

        };

        DisposableServer server =
            HttpServer.create()
                      .host("10.0.0.19")
                      .port(61005)
                      .route(routes ->
                          routes
                            .post("/echo",
                                (request, response) -> {
                                    request.receive().retain().next().subscribe(onSuccess, onError, onCompletion);
                                    return response.send(Mono.just(Unpooled.copiedBuffer(responseArray).retain()));
                                                       }
                                    ))
                        .bindNow();

        server.onDispose()
                .block();
    }
}

环境和maven依赖如下:

Apache Maven 3.6.1
Maven home: /usr/share/maven
Java version: 11.0.6, vendor: Oracle Corporation, runtime: /home/linuxlp/opt/graalvm/graalvm-svm-linux-20.1.0-ea+28
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "5.3.0-51-generic", arch: "amd64", family: "unix"

    <dependencies>
        <dependency>
            <groupId>io.projectreactor.netty</groupId>
            <artifactId>reactor-netty</artifactId>
            <version>0.9.0.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>3.3.5.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.reactivestreams</groupId>
            <artifactId>reactive-streams</artifactId>
            <version>1.0.3</version>
        </dependency>

Answer 1

您对这样编写请求/响应有何看法

public class Server {

    private static final Logger log = Logger.getLogger(Server.class.getSimpleName());

    public static void main(String[] args) {

        Consumer<byte[]> onSuccess = (byte[] request) -> {
            System.out.println("onSuccess: Request received!");
        };
        Consumer<Throwable> onError = (Throwable ex) -> {
            ex.getMessage();
        };
        Runnable onCompletion = () -> {
            System.out.println("Message Completed");

        };

        DisposableServer server =
                HttpServer.create()
                        .host("10.0.0.19")
                        .port(61005)
                        .route(routes ->
                                routes.post("/echo",
                                        (request, response) ->
                                                response.send(request.receive()
                                                                     .aggregate()
                                                                     .asByteArray()
                                                                     .doOnNext(onSuccess)
                                                                     .doOnError(onError)
                                                                     .doOnTerminate(onCompletion)
                                                                     .flatMap(bytes -> {
                                                                         ElectionRequest electionRequest = (ElectionRequest) SerializationUtils.deserialize(bytes);
                                                                         ElectionResponse electionResponse = new ElectionResponse(electionRequest.getStr());
                                                                         return Mono.just(Unpooled.copiedBuffer(SerializationUtils.serialize(electionResponse)));
                                                                     }))))
                        .bindNow();

        server.onDispose()
                .block();
    }
}