发送另一个味精后如何接收单声道味精答案

【问题标题】：How to receive a mono msg after sending another msg发送另一个味精后如何接收单声道味精
【发布时间】：2019-09-05 08:20:27
【问题描述】：

如何发送 buf 然后接收 msg

方法

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

我正在尝试通过从连接出站发送消息并从入站接收消息然后返回消息 Mono 来实现此方法。但我只能在 then(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.get(key) != null 处获得带有while 循环中断的Mono.fromSupply()。

它会阻塞线程。

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

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

【问题讨论】：

标签： java project-reactor reactor-netty

【解决方案1】：

您没有具体说明您的期望。看下面的例子，它发送一些数据，然后接收服务器返回的内容。

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

【讨论】：

谢谢。我正在尝试通过从连接出站发送一条消息并从入站接收一条消息然后返回一条消息单声道来实现我的方法。但我只能以then(Publisher<Void>) 方法接收消息。好像不能返回一个数据单声道。

【解决方案2】：

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

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

当入站接收到对象响应时，只需执行sink.success()

【讨论】：

【解决方案3】：

您应该结合使用 NettyOutbound::then 来监听写入完成和 Mono::then 在写入后读取您的 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));

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

【讨论】：