Reactor网络--如何用延迟通量发送

Question

在Reactor Netty中，当通过out.send（publisher)向TCP通道发送数据时，任何发布服务器都可以工作。但是，如果不是简单的即时flux而是使用带有延迟元素的更复杂的flux，那么它就会停止正常工作。例如，如果我们使用这个hello world TCP echo服务器，它将按照预期工作：

import reactor.core.publisher.Flux;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;

import java.time.Duration;

public class Reactor1 {
    public static void main(String[] args) throws Exception {
        DisposableServer server = TcpServer.create()
            .port(3344)
            .handle((in, out) -> in
                .receive()
                .asString()
                .flatMap(s ->
                    out.sendString(Flux.just(s.toUpperCase()))
                ))
            .bind()
            .block();
        server.channel().closeFuture().sync();
    }
}

但是，如果我们将out.sendstring更改为

out.sendString(Flux.just(s.toUpperCase()).delayElements(Duration.ofSeconds(1)))

那么我们将期望对于每一个接收到的项目，都将产生一个延迟一秒的输出。

但是，html" target="_blank">服务器的行为方式是，如果它在间隔期间接收到多个项目，它将只针对第一个项目产生输出。例如，下面我们在第一秒内键入aa和bb，但只有aa作为输出（一秒后）：

$ nc localhost 3344
aa
bb
AA <after one second>

然后，如果我们稍后输入附加行，我们得到输出（一秒钟后），但来自前面的输入：

cc
BB <after one second>

如何使send()在延迟的flux情况下按预期工作？

Answer 1

尝试使用concatmap。这是可行的：

DisposableServer server = TcpServer.create()
        .port(3344)
        .handle((in, out) -> in
                .receive()
                .asString()
                .concatMap(s ->
                        out.sendString(Flux.just(s.toUpperCase())
                                           .delayElements(Duration.ofSeconds(1)))
                ))
            .bind()
            .block();
server.channel().closeFuture().sync();

延迟来话流量

DisposableServer server = TcpServer.create()
        .port(3344)
        .handle((in, out) -> in
                .receive()
                .asString()
                .timestamp()
                .delayElements(Duration.ofSeconds(1))
                .concatMap(tuple2 ->
                        out.sendString(
                                Flux.just(tuple2.getT2().toUpperCase() +
                                        " " +
                                        (System.currentTimeMillis() - tuple2.getT1())
                                ))
                ))
        .bind()
        .block();

Answer 2

我认为您不应该为out.sendString(...)重新创建publisher，这是可行的：

DisposableServer server = TcpServer.create()
        .port(3344)
        .handle((in, out) -> out
                .options(NettyPipeline.SendOptions::flushOnEach)
                .sendString(in.receive()
                        .asString()
                        .map(String::toUpperCase)
                        .delayElements(Duration.ofSeconds(1))))
        .bind()
        .block();
server.channel().closeFuture().sync();