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

Odpowiedz Nowy wątek
2019-09-11 13:44
0

Cześć,

czy wie ktoś może jak w WebFluxie wysłać wiadomość do konkretnej grupy userów przy pomocy websocketów? Chodzi mi o np. aplikację typu czat z wieloma pokojami. Każdy użytkownik może wejść do wielu pokojów, ale wysłana wiadomość w danym pokoju trafia tylko do znajdujących się w nim użytkowników.
W Springu na servletach mogę użyć @EnableWebSocketMessageBroker, SimpMessageSendingOperations i np @MessageMapping("chat/{roomId}/sendMessage") w controllerze, ale WebFlux nie obsługuje stomp i nie wiem jak to rozwiązać.

Robił ktoś z Was coś podobnego?


Pozostało 580 znaków

2019-09-11 14:20
0

Patrzyłeś tutaj ?:
https://developer.okta.com/bl[...]tive-apis-with-spring-webflux


Bardzo lubie Singletony, dlatego robię po kilka instancji każdego.

Pozostało 580 znaków

2019-09-11 15:07
0

Tak, skleiłem z tego taki kod:


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

W ReactiveWebSocketHandler mam this.publisher = Flux.create(greetingsPublisher).share();. Kiedy dostaję message przez WebSocket, to pcham ją na kolejkę w GreetingsPublisher, z której pobieram przez sink.next(greeting) i emituję we Fluxa... W ten sposób wszyscy subskrybenci dostają dany message - każdy połączony przez WebSocket. Nie mam natomiast pojęcia jak zrobić, żeby wysłać message do konkretnej grupy użytkowników?


Pozostało 580 znaków

2019-09-11 18:05
0

Niedawno się tym bawiłem i zrobiłem czata p2p. Proponuję na początek wyrzucić WebFluksa i zrobić to na WebMVC. Jak zadziała i ogarniesz, co Spring konfiguruje pod spodem (jest sporo konwencji nazewniczych np. topiców, prawdopodobnie będziesz musiał zejść głębiej), powrót na reaktywny stack nie powinno być problemem. Ewentualnie wrzuć koda na githuba, chętnie pomogę.

edytowany 2x, ostatnio: Charles_Ray, 2019-09-11 18:10
Imo to tak zupełnie niepodobne stosy, że nijak się to nie mapuje. - jarekr000000 2019-09-11 18:21
Chodzi o to, żeby rozpoznać jak Spring ogarnia sobie topici Stompowe bez wprowadzania dodatkowej złożoności. Ale z drugiej strony może i masz racje :) - Charles_Ray 2019-09-11 19:18
Na servletach już to ogarnalem, nawet opisałem to na swoim blogu http://kojotdev.com/2019/09/m[...]ring-websocket-nuxt-vue-vuex/ ale nijak to się ma reaktywnego stacka. I szczerze mówiąc to tak, jestem o krok od porzucenia WebFluxa ;) - kkojot 2019-09-11 20:00
Ooook, czyli problem z samymi Fluksami - Charles_Ray 2019-09-11 20:04

Pozostało 580 znaków

2019-09-11 19:34
0

Chwilowo nie mam sensownej klawiatury więc napisze ogólnie. Zamiast subscribe wpisz flatMap. W jego srodku wyciagnij usera ( z websocketa lub wiadomosci). Wyciagnij pokoje. Potem wyciagnij z publishera wiadomosci, filteruj po pokojach i to zwroc jako wynik flatMap. (No i słowko return trzeba przeniesc).


Bardzo lubie Singletony, dlatego robię po kilka instancji każdego.
edytowany 1x, ostatnio: jarekr000000, 2019-09-11 19:34

Pozostało 580 znaków

2019-09-11 20:16
0

Tak jak @jarekr000000 pisze, musisz zwinąć wszystkie instrukcje w jeden pipeline. Nawet IntelliJ podświetla subscribe(): "Calling subscribe() in non-blocking scope". Postaram się wkleić kod.

Update. Kod, który wkleiłeś można zapisać jak poniżej. Nie ma jeszcze obsługi "pokojów" :)

 @Override
    public Mono<Void> handle(WebSocketSession webSocketSession) {
        return webSocketSession.send(
                webSocketSession.receive()
                .map(WebSocketMessage::getPayloadAsText)
                .map(greetingsService::greeting)
        );
    }
edytowany 3x, ostatnio: Charles_Ray, 2019-09-11 20:26

Pozostało 580 znaków

Odpowiedz
Liczba odpowiedzi na stronę

1 użytkowników online, w tym zalogowanych: 0, gości: 1, botów: 0