WebFlux, WebSocket i wysyłanie wiadomości do konkretnego usera/userów

import io.vavr.control.Try;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.FluxSink;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;
@Component
public class GreetingsPublisher implements Consumer<FluxSink<String>> {
    private static final Logger log = LoggerFactory.getLogger(GreetingsPublisher.class);
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private final Executor executor = Executors.newSingleThreadExecutor();
    public boolean push(String greeting) {
        return queue.offer(greeting);
    }
    @Override
    public void accept(FluxSink<String> sink) {
        this.executor.execute(() -> {
            while (true) {
                Try.of(() -> {
                    final String greeting = queue.take();
                    return sink.next(greeting);
                })
                        .onFailure(ex -> log.error("Could not take greeting from queue", ex));
            }
        });
    }

@Component
public class ReactiveWebSocketHandler implements WebSocketHandler {
    private final GreetingsService greetingsService = new GreetingsService();
    private final GreetingsPublisher greetingsPublisher;
    private final Flux<String> publisher;
    public ReactiveWebSocketHandler(GreetingsPublisher greetingsPublisher) {
        this.greetingsPublisher = greetingsPublisher;
        this.publisher = Flux.create(greetingsPublisher).share();
    }
    @Override
    public Mono<Void> handle(WebSocketSession webSocketSession) {
        webSocketSession
                .receive()
                .map(webSocketMessage -> webSocketMessage.getPayloadAsText())
                .map(helloMessage -> greetingsService.greeting(helloMessage))
                .doOnNext(greetings -> greetingsPublisher.push(greetings))
                .subscribe();
        final Flux<WebSocketMessage> message = publisher
                .map(greetings -> webSocketSession.textMessage(greetings));
        return webSocketSession.send(message);
    }
}

 @Override
    public Mono<Void> handle(WebSocketSession webSocketSession) {
        return webSocketSession.send(
                webSocketSession.receive()
                .map(WebSocketMessage::getPayloadAsText)
                .map(greetingsService::greeting)
        );
    }

        return webSocketSession.send(
                webSocketSession.receive()
                        .map(webSocketMessage -> webSocketMessage.getPayloadAsText())
                        .map(helloMessage -> greetingsService.greeting(helloMessage))
                        .map(greetings -> greetingsPublisher.push(greetings))
//                        .doOnNext(greetings -> greetingsPublisher.push(greetings))
                        .flatMap(greetings -> publisher
                                .map(publisherGreetings -> webSocketSession.textMessage(publisherGreetings)))
        );

    @Override
    public Mono<Void> handle(WebSocketSession webSocketSession) {
        /* share greetings across websocket sessions */
        webSocketSession
                .receive()
                .map(webSocketMessage -> webSocketMessage.getPayloadAsText())
                .map(message -> new SimpleTeamMessageResolver(webSocketSession, message))
                .map(resolver -> resolver.process())
                .map(message -> teamMessagePublisher.push(message))
                .subscribe();

        return webSocketSession.send(
                SimpleTeamMessageResolver
                        .filterTeamMessage(webSocketSession, publisher)
                        .map(message -> webSocketSession.textMessage(message)));
    }

public class SimpleTeamMessageResolver {

    private final WebSocketSession webSocketSession;
    private final String message;

    public SimpleTeamMessageResolver(WebSocketSession webSocketSession, String message) {
        this.webSocketSession = webSocketSession;
        this.message = message;
    }

    public static Flux<String> filterTeamMessage(WebSocketSession session, Flux<String> messages) {
        final String teamColor = teamColorFromSession(session);
        if (teamColor.equals("UNKNOWN")) return messages;
        final Flux<String> filtered = messages
                .filter(message -> message.startsWith(teamColor));
        return filtered;
    }

    private static String teamColorFromSession(WebSocketSession session) {
        return Option.of(session
                .getAttributes()
                .get("team"))
                .map(color -> color.toString())
                .getOrElse("UNKNOWN");
    }

    public String process() {
        if (message.startsWith("join:")) {
            return handleJoin();
        } else {
            return handleMessage();
        }
    }

    private String handleJoin() {
        final String teamColor = teamColorFromMessage();
        webSocketSession
                .getAttributes()
                .put("team", teamColor);
        return "New guy joined the " + teamColor + " team.";
    }

    private String handleMessage() {
        final String teamColor = teamColorFromSession(webSocketSession);
        return teamColor + ": " + message;
    }

    public String teamColorFromMessage() {
        return Option.of(message.split(":"))
                .map(strings -> strings.length > 1 ? strings[1] : "BLUE")
                .map(color -> color.toUpperCase().trim().equals("RED") ? "RED" : "BLUE")
                .getOrElse("BLUE");
    }
}