如何在发送另一个消息后接收mono消息

Question

如何发送buf然后接收msg

方法

Mono<ByteBuf> send(ByteBuf buf){
    // how to send the buf then receive a msg
}

我正在尝试通过从连接出站发送msg并从入站接收msg然后返回消息Mono来实现此方法。但我只能在that（Publisher）方法中接收消息。它似乎无法返回数据Mono

我试过这个。

// the connecttion has been initialized before entering this method.

        Mono.just(buf)
                .doOnNext(data -> connection.outbound().sendObject(data).then().subscribe())
                .then(connection
                        .inbound()
                        .receiveObject()
                        .single()
                        .map(RpcDataPackage.class::cast)
                        .map(RpcDataPackage::getData)
                        .map(data -> {
                            try {
                                return resCodec.decode(data);
                            } catch (IOException e) {
                                throw new RpcRequestException(e);
                            }
                        })
                );

但它会一直阻塞，直到连接超时

我尝试了另一个代码。我添加了一个handle方法，并将响应放到map中。然后我可以得到单声道。fromSupply（），在映射处有一个while循环中断。获取（键）！=空。

它会阻塞线程。

                .handle(((nettyInbound, nettyOutbound) -> nettyInbound
                        .receiveObject()
                        .map(RpcDataPackage.class::cast)
                        .doOnNext(pkg -> {
                            String responseKey = "a key"

                            responseMap.put(responseKey, pkg);
                        })
                        .then()))

Answer 1

您应该结合使用NettyOutbound::然后监听写入完成和Mono::然后在写入后读取NettyInboud。

  Mono<String> resposeMono = TcpClient.create()
            .connect()
            .flatMap(connection -> connection.outbound().sendString(Mono.just("Hello!"))
                    .then()
                    .then(connection.inbound().receive().aggregate().asString())
                    .doOnTerminate(connection::dispose));

这将写“你好！”对于输出，将输入中的所有字节作为字符串读取，然后处理连接。

Answer 2

我阅读了Mono javadoc并找到了MonoSink。

Mono.create(monoSink -> {
  // some call
})

当入站接收到对象响应时，只需执行sink。success（）

Answer 3

您没有指定您的期望。参见下面的示例，它发送一些数据，然后接收服务器返回的数据。

    @Test
    public void test() {
        Connection connection =
                TcpClient.create()
                         .wiretap(true)
                         .host("example.com")
                         .port(80)
                         .connect()
                         .block();

        assertNotNull(connection);

        connection.outbound()
                  .sendObject(Unpooled.wrappedBuffer("test".getBytes()))
                  .then(connection.inbound()
                                  .receiveObject()
                                  .last()
                                  .doOnNext(System.out::println)
                                  .then())
                  .then()
                  .block();
    }