Как дождаться изменения переменных перед отправкой следующего контента, websocket по processor-netty

Question

что делать? ReadFlushMessage () будет ждать этого. сообщение для получения нового контента.

Не выходить из соединения через веб-сокет немедленно

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