java - 反应式 WebSocketClient 数据发布

Question

我正在尝试使用从 websocket 读取数据，ReactorNettyWebSocketClient但我无法连接到它的 api。问题是我收到的所有数据都在 lambda 样式 websockethandler 的内部可用（#1），但我想在 client.execute(..) 上的 .subscribe 之后将它们提供给订阅者（#2）

WebSocketClient client = new ReactorNettyWebSocketClient();
        client.execute(
                URI.create(URL),
                session -> session.send(
                        Mono.just(session.textMessage(pairRqStr)))
                        .thenMany(session.receive()
                                .map(WebSocketMessage::getPayloadAsText)
                                .map(this::toResp)
                                .onErrorContinue((throwable, o) -> throwable.getMessage())
                        )
                        .log() // #1
                        .then()
        )
                .log()
                .subscribe(System.out::println); // #2

对此有点迷茫和陌生，所以，请指导我。

Answer 1

这就是我最后所做的，但仍然不太喜欢它：

    private Flux<WebSocketMessage> requestData(String req) {
        WebSocketClient client = new ReactorNettyWebSocketClient();
        return ConnectableFlux.create(sub -> {
            client.execute(
                    URI.create(URL),
                    session -> session.send(
                            Mono.just(session.textMessage(req)))
                            .thenMany(session.receive().doOnNext(sub::next))
                            .then()
            )
                    .log()
                    .subscribe();
        });
    }

Answer 2

要发送或接收消息，您必须先连接到该频道。这就是为什么client.execute返回Mono<Void>，这意味着它不返回任何数据，它只是表示handskahe的完成或失败。如果它会返回例如。Flux<WebSocketMessage>，你怎么知道握手是否成功完成？

如果你想访问 lambda 之外的通道，你不能通过实现处理程序方法来做到这一点：

        Consumer<WebSocketMessage> printingConsumer = webSocketMessage -> System.out.println(webSocketMessage.getPayloadAsText());

        client.execute(URI.create(URL), session -> handle(session, printingConsumer));
    }


    public Mono<Void> handle(WebSocketSession session, Consumer<WebSocketMessage> consumer) {
        return session.receive()
                .doOnNext(consumer::accept)
                .then();
    }