迁移现有的Spring Websocket处理程序以使用rsocket [英] Migrating existing Spring Websocket handler to use rsocket

问题描述

假设我有一个用于聊天消息的简单Websocket处理程序:

Suppose I have this simple Websocket handler for chat messages:

@Override
public Mono<Void> handle(WebSocketSession webSocketSession) {
    webSocketSession
            .receive()
            .map(webSocketMessage -> webSocketMessage.getPayloadAsText())
            .map(textMessage -> textMessageToGreeting(textMessage))
            .doOnNext(greeting-> greetingPublisher.push(greeting))
            .subscribe();
    final Flux<WebSocketMessage> message = publisher
            .map(greet -> processGreeting(webSocketSession, greet));
    return webSocketSession.send(message);
}

这里一般需要做什么，因为它将使用 rsocket 协议?

What is needed to be done here in general as such it will use the rsocket protocol?

推荐答案

Spring WebFlux中的RSocket控制器比WebSocketHandler看起来更像RestController.所以上面的例子很简单:

RSocket controller in the Spring WebFlux looks more like a RestController than WebSocketHandler. So the example above is simple like that:

@Controller
public class RSocketController {

    @MessageMapping("say.hello")
    public Mono<String> saHello(String name) {
        return Mono.just("server says hello " + name);
    }
}

，这等效于requestResponse方法.

如果此答案不能使您满意，请描述更多您想要实现的目标.

If this answer doesn't satisfy you, please describe more what you want to achieve.

编辑

如果要向所有客户端广播消息，则它们需要订阅相同的Flux.

If you want to broadcast messages to all clients, they need to subscribe to the same Flux.

public class GreetingPublisher {

    final FluxProcessor processor;
    final FluxSink sink;

    public GreetingPublisher() {
        this.processor = DirectProcessor.<String>create().serialize();
        this.sink = processor.sink();
    }

    public void addGreetings(String greeting) {
        this.sink.next(greeting);
    }

    public Flux<String> greetings() {
        return processor;
    }
}

@Controller
public class GreetingController{

    final GreetingPublisher greetingPublisher = new GreetingPublisher();

    @MessageMapping("greetings.add")
    public void addGreetings(String name) {
        greetingPublisher.addGreetings("Hello, " + name);
    }

    @MessageMapping("greetings")
    public Flux<String> sayHello() {
        return greetingPublisher.greetings();
    }
}

您的客户端必须使用requestStream方法调用greetings端点.无论您使用greetingPublisher.addGreetings()消息发送到哪里，它都将广播到所有客户端.

Your clients have to call the greetings endpoint with the requestStream method. Wherever you send the message with the greetingPublisher.addGreetings() it's going to be broadcasted to all clients.

这篇关于迁移现有的Spring Websocket处理程序以使用rsocket的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！