如何使用 Spring WebFlux 和 WebSockets 确保 Reactive Stream 完成

Question

我在 Kotlin 中为 Spring WebFlux 编写了一个测试客户端和服务器。客户端向服务器发送一个数字（例如 4）并返回那么多数字（例如 0、1、2 和 3）。这是服务器实现：

class NumbersWebSocketHandler : WebSocketHandler {
    override fun handle(session: WebSocketSession): Mono<Void> {
        var index = 0
        var count = 1
        val publisher = Flux.generate<Int> { sink ->
            if (index < count) {
                sink.next(index)
                index++
            } else {
                sink.complete()
            }
        }.map(Int::toString)
            .map(session::textMessage)
            .delayElements(Duration.ofMillis(500))

        return session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .doOnNext {
                println("About to send $it numbers")
                count = it.toInt()
            }
            .then()
            .and(session.send(publisher))
            .then()
    }
}

这里是客户端：

fun main(args: Array<String>) {
    val uri = URI("ws://localhost:8080/numbers")
    val client = ReactorNettyWebSocketClient()

    println("How many numbers would you like?")
    val input = Flux.just(readLine())

    client.execute(uri) { session ->
        session.send(input.map(session::textMessage))
            .then(
                session.receive()
                    .map(WebSocketMessage::getPayloadAsText)
                    .map { it.toInt() }
                    .reduce { a,b ->
                        println("Reduce called with $a and $b")
                        a + b
                    }
                    .doOnNext(::println)
                    .then()
            )
            .then()
    }.block()
}

客户端成功接收数字并调用reduce，如下：

reduce 用 0 和 1 调用

reduce 用 1 和 2 调用

Reduce 用 3 和 3 调用

但是，对 doOnNext 的调用从未到达 - 大概是因为客户端不知道最后一个项目已发送。我的问题是我需要在客户端或服务器上添加什么代码才能打印总数？

更新：在服务器端关闭会话无济于事。我试过了：

.delayElements(Duration.ofMillis(500))
.doOnComplete { session.close() }

还有：

.delayElements(Duration.ofMillis(500))
.doFinally { session.close() }

但是两者都不会对客户端的行为产生任何影响。在调用“发送”之后尝试显式关闭会话也没有：

.and(session.send(publisher))
.then()
.and { session.close() }
.then()

Answer 1

从WebSocketHandler 返回的Mono<Void> 指示处理何时完成，这反过来指示WebSocket 连接应保持打开多长时间。问题是双方返回的Mono<Void> 永远不会完成。

客户端使用reduce 等待来自服务器端的输入结束，但服务器使用receive() + doOnNext + then() 等待永远接收消息。所以客户端和服务器都互相等待。在客户端添加take有助于打破僵局：

client.execute(uri, session -> session
        .send(input.map(session::textMessage))
        .then(session.receive()
                .map(WebSocketMessage::getPayloadAsText)
                .map(Integer::valueOf)
                .take(3)
                .reduce((a,b) -> {
                    logger.debug("Reduce called with " + a + " and " + b);
                    return a + b;
                })
                .doOnNext(logger::debug)
        )
        .then()
).block();

第二个问题是服务器组成不正确。发送与通过.and 接收并行触发。相反，发送流应该依赖于首先接收计数：

session.receive()
        .map(WebSocketMessage::getPayloadAsText)
        .flatMap(s -> {
            logger.debug("About to send " + s + " numbers");
            count.set(Integer.parseInt(s));
            return session.send(publisher);
        })
        .then()

Answer 2

从marble diagram for reduce 可以看出，它期望上游发布者发送一个onComplete 事件来自己发出一个事件。我不确定，但我认为只有当连接正常终止时才会发出。这就是为什么doOnNext 永远不会被执行的原因。

我认为你应该使用subscribe() 而不是reduce，并定义你自己的Subscriber 实例来保持状态。

编辑：您不能在这里使用subscribe 方法。如果连接在服务器上关闭，它应该像这样工作：

.map(Int::toString)
        .map(session::textMessage)
        .delayElements(Duration.ofMillis(500))
        .then(session::close)