How to map all interaction models of RSocket in Spring Boot

Question

There are 4 interaction models provided in RSocket.

• fire and forget
• request and response
• request stream
• request channel
• (metadata push)

Spring(and Spring Boot) provides RSocket integration, it is easy to build a RSocket server with the existing messaging infrastructure to hide the original RSocket APIs.

   @MessageMapping("hello")
    public Mono<Void> hello(Greeting p) {
        log.info("received: {} at {}", p, Instant.now());
        return Mono.empty();
    }

    @MessageMapping("greet.{name}")
    public Mono<String> greet(@DestinationVariable String name, @Payload Greeting p) {
        log.info("received: {}, {} at {}", name, p, Instant.now());
        return Mono.just("Hello " + name + ", " + p.getMessage() + " at " + Instant.now());
    }

    @MessageMapping("greet-stream")
    public Flux<String> greetStream(@Payload Greeting p) {
        log.info("received:  {} at {}", p, Instant.now());
        return Flux.interval(Duration.ofSeconds(1))
                .map(i -> "Hello #" + i + "," + p.getMessage() + " at " + Instant.now());
    }

And in the client side, there is a RescoketRequester provided to shake hands with the server.

    @GetMapping("hello")
    Mono<Void> hello() {
        return this.requester.route("hello").data(new Greeting("Welcome to Rsocket")).send();
    }

    @GetMapping("name/{name}")
    Mono<String> greet(@PathVariable String name) {
        return this.requester.route("greet." + name).data(new Greeting("Welcome to Rsocket")).retrieveMono(String.class);
    }

    @GetMapping(value = "stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<String> greetStream() {
        return this.requester.route("greet-stream").data(new Greeting("Welcome to Rsocket"))
                .retrieveFlux(String.class)
                .doOnNext(msg -> log.info("received messages::" + msg));
    }

But how to use requestChannel and metadataPush model in Spring way(using messaging infrastructure)?

The sample codes is on Github. Update: added requestChannel sample.

Update: SETUP and METADATA_PUSH can be handled by @ConnectMapping. And Spring Security RSocket can secure SETUP and REQUEST.

Answer 1

Reference example

For a reference example, let's refer to the client-to-server integration tests and, in particular, to the ServerController class: spring-framework/RSocketClientToServerIntegrationTests.java (line 200) at 6d7bf8050fe710c5253e6032233021d5e025e1d5 · spring-projects/spring-framework · GitHub.

This commit has been mentioned in the release notes:

<…>

• RSocket support including response handling via annotated @MessageMapping methods and performing requests via RSocketRequester.

<…>

— Spring Framework 5.2.0.M1 available now.

Channel interaction model

The corresponding code part of the reference example:

@MessageMapping("echo-channel")
Flux<String> echoChannel(Flux<String> payloads) {
    return payloads.delayElements(Duration.ofMillis(10)).map(payload -> payload + " async");
}

Metadata push

It seems that, currently, it is not supported by the @MessageMapping annotation.