spring - 迁移现有的 Spring Websocket 处理程序以使用 rsocket

Question

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

@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 协议(protocol)？

Answer 1

Spring WebFlux 中的 RSocket Controller 看起来更像是一个 RestController 而不是 WebSocketHandler。所以上面的例子很简单:

@Controller
public class RSocketController {

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

这等同于 requestResponse 方法。

如果这个答案不能让您满意，请描述更多您想要实现的目标。

编辑

如果要向所有客户端广播消息，他们需要订阅同一个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() 的消息发送到何处，它都将被广播到所有客户端。