websocket - How to wait for variables to change before sending the next content , websocket by reactor-netty

Question

what to do? ReadFlushMessage () will wait for this. message to get new content.

Not quitting the websocket connection immediately

    private Set<String> message = new HashSet<>();

    private void writeMessage(String message) {
        this.message.add(message);
    }

    private String[] readFlushMessage() {
        String[] _message = (String[])this.message.toArray();
        this.message = new HashSet<>();
        return _message;
    }

    private Publisher<Void> websocketPublisherA(HttpServerRequest request, HttpServerResponse response, WebSocketServerHandle handleObject) {
        return response
            .header("content-type", "text/plain")
            .sendWebsocket((in, out) ->
                out.options(NettyPipeline.SendOptions::flushOnEach)
                    .sendString(
                        Flux.just(readFlushMessage())
                    )
            );
    }

Answer 1

尝试使用Reactor Core类型。

这是您要实现的目标的一个非常简单的示例：

    @Test
    public void test() {
        FluxProcessor<String, String> serverMsg =
                ReplayProcessor.<String>create();

        Flux.range(1, 20)
                .map(Object::toString)
                .subscribe(serverMsg::onNext);

        DisposableServer server =
                HttpServer.create()
                        .port(0)
                        .handle((req, resp) ->
                                resp.header("content-type", "text/plain")
                                    .sendWebsocket((in, out) ->
                                            out.options(NettyPipeline.SendOptions::flushOnEach)
                                                    .sendString(serverMsg)
                                    ))
                        .bindNow();

        HttpClient.create()
                .port(server.port())
                .websocket()
                .receive()
                .asString()
                .doOnNext(System.out::println)
                .blockLast();

        server.disposeNow();
    }