反应式WebSocketClient数据发布

要发送或接收消息,必须首先连接到通道。这就是为什么 client.execute 退货 Mono<Void> 也就是说,它不返回任何数据,只是表示手持电话的完成或失败。 如果它会回来的话。 Flux<WebSocketMessage> ,您如何知道手持设备是否成功完成?

如果要访问lambda外部的通道,则无法通过实现handler方法来进行访问:

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

这是我最后做的,但还是不太喜欢:

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