迁移现有 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 协议？

Answer 1

Spring WebFlux 中的 RSocket 控制器看起来更像 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() 将消息发送到何处，它都会广播给所有客户端。