Reactor-netty TCPClient无法接收响应

Question

我正在尝试在最新的reactor-netty版本上做一些项目前的经验，这个版本缺乏文档；我使用的是0.8.0.M3版本。

我已经用这个tcp服务器开发了一个简单的spring引导应用程序，它可以正确启动，而且似乎可以正常工作：

@PostConstruct
public void startServer() throws InterruptedException {
    TcpServer.create().
              host("localhost").
              port(1235).
              handle((in, out) -> {

                    Flux<String> fluxString = in.receive().asString().log().
                        map(text -> {
                            return "Hi server have received "+text;});  
                        return out.sendString(fluxString).then();

              } 
              ).
              wiretap().bindNow();
}

如果我尝试使用客户端进行测试，交互似乎是正确的，但我无法收到任何响应：

int counter = 10;
    CountDownLatch latch = new CountDownLatch(counter);
    Flux<String> input = Flux.range(0, counter).map(i->""+i);

    TcpClient.create().
      host("localhost").
      port(1235).
      handle((in, out) -> {

          in.receive().subscribe(receiv -> {System.out.println(receiv);latch.countDown();});
                return out.sendString(input).neverComplete();

      } 
      ).
      wiretap().connectNow();
    System.out.println("waiting closure");
    boolean result = latch.await(5, TimeUnit.SECONDS);

看一下窃听日志，似乎客户端将每个int作为字符串单独发送，而服务器只接收一个聚合字符串"0123456789“，并且只发送一个响应。客户端没有接收到任何东西，锁存器也没有减去1，而是保持为10 (我预计至少会收到一个聚合响应)。

谁能解释一下客户端出了什么问题，以及服务器如何分别接收每个单独的整数？

Thx G

Answer 1

您可能需要解决一些问题。我想说这对于学习来说有点复杂。

对于服务器：

    TcpServer.create()
            .host("localhost")
            .port(1235)
            .doOnConnection(c ->
                    //The origin input are 0,1,2,3,4,5,6,7,8,9.
                    //So you need a decoder split every 1 byte as a ByteBuf.
                    c.addHandler(
                            "1ByteStringDecoder",
                            new ByteToMessageDecoder() {
                                @Override
                                protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
                                    out.add(in.readBytes(1));
                                }
                            }
                    )
            )
            .handle((in, out) -> {
                        Flux<String> fluxString = in.receive()
                                .asString()
                                .log()
                                .map(text -> {
                                    return "Hi server have received " + text;
                                });
                        //Since the output is quite small, you need flush it
                        return out.options(o -> o.flushOnEach())
                                .sendString(fluxString)
                                .neverComplete();

                    }
            )
            .wiretap()
            .bindNow();

对于客户端：

    int counter = 10;
    CountDownLatch latch = new CountDownLatch(counter);
    startServer();
    Flux<String> input = Flux.range(0, counter)
            .map(i -> "" + i);

    TcpClient.create()
            .host("localhost")
            .port(1235)
            .doOnConnected(c ->
                    c.addHandler(
                            //The covert input are "Hi server have received " + (0,1,2,3,4,5,6,7,8,9).
                            //So you need a decoder split every 25 byte as a ByteBuf.
                            "25ByteStringDecoder",
                            new ByteToMessageDecoder() {
                                @Override
                                protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
                                    out.add(in.readBytes(25));
                                }
                            }
                    )
            )
            .handle((in, out) -> {
                        in.receive()
                                .asString()//You need convert ByteBuf to String.
                                .subscribe(receiv -> {
                                    System.out.println(receiv);
                                    latch.countDown();
                                });

                        out.options(o -> o.flushOnEach())
                                .sendString(input)
                                .then()
                                .subscribe(); //You need to ask your client to send the data by subscribe
                        return Mono.never();

                    }
            )
            .wiretap()
            .connectNow();
    System.out.println("waiting closure");
    boolean result = latch.await(5, TimeUnit.SECONDS);