Although the implementation gives broader possibilities in writing asynchronous, low-latency, and high-throughput communication using Reactor API, it left much work with infrastructure configurations for developers. Fortunately, Spring teams valued that project and started experimenting with adopting such a great solution to the Spring Ecosystem with a simplified programming model over annotations.

One of the experiments was called Spring Cloud Sockets and was aimed at providing a familiar (in comparison to Spring Web) programming model for declaring annotations:

@SpringBootApplication                                             // (1)
@EnableReactiveSockets                                             // (1.1)
public static class TestApplication {                              //

   @RequestManyMapping(                                            // (2)
      value = "/stream1",                                          // (2.1)
      mimeType = "application/json"                                // (2.2)
   )                                                               //
   public Flux<String> stream1(@Payload String a) {                // (2.3)
      return Flux.just(a)                                          //
                 .mergeWith(                                       //
                    Flux.interval(Duration.ofMillis(100))          //
                        .map(i -> "1. Stream Message: [" + i + "]")//
                 );                                                //
   }                                                               //
 
   @RequestManyMapping(                                            // (2)
      value = "/stream2",                                          // (2.1)
      mimeType = "application/json"                                // (2.2)
   )                                                               //
   public Flux<String> stream2(@Payload String b) {                // (2.3)
      return Flux.just(b)                                          //
                 .mergeWith(                                       //
                    Flux.interval(Duration.ofMillis(500))          //
                        .map(i -> "2. Stream Message: [" + i + "]")//
                 );                                                //
   }                                                               //
}

The numeration in the code is explained here:

1. This is the @SpringBootApplication definition. Here, at point (1.1), we define the @EnableReactiveSockets annotation, which provides the required configurations and enables RSocket in the application.
2. This is the handler method declaration. Here, we use the @RequestManyMapping annotation to specify that the current method works in the request stream interaction model. One noticeable feature offered by the  Spring Cloud Sockets module is that it provides out-of-the-box mapping (routing) and allows the definition of the handler mapping's path (2.1) and mime-type for incoming messages (2.2). Finally, there is an additional @Payload annotation (2.3), which suggests that the given parameter is the incoming request's payload.

Here, we have the familiar Spring Boot application, which, with the support of Spring Cloud Socket, enables additional features of the RSocket-Java library. In turn, Spring Cloud Sockets provide simplification for interactions with a server from the client's perspective as well:

public interface TestClient {                                      // (1)
   @RequestManyMapping(                                            // (2)
      value = "/stream1",                                          //
      mimeType = "application/json"                                //
   )                                                               //
   Flux<String> receiveStream1(String a);                          //

   @RequestManyMapping(                                            // (2)
      value = "/stream1",                                          //
      mimeType = "application/json"                                //
   )                                                               //
   Flux<String> receiveStream2(String b);                          //
}

Here, all we have to do is declare an RSocket Client using Spring Cloud Sockets when providing an interface (1). In order to enable the RSocket client, we have to define identical annotations, as in the server example on top of client's methods, and define the corresponding handler path.

As a result, the interface can easily be transformed into a Proxy at runtime using the ReactiveSocketClient factory, as in the following example:

ReactiveSocketClient client = new ReactiveSocketClient(rSocket);
TestClient clientProxy = client.create(TestClient.class);

Flux.merge(
       clientProxy.receiveStream1("a"),
       clientProxy.receiveStream2("b")
    )
    .log()
    .subscribe();

In the preceding example, we created a client (please note that, in this example, we have to provide RSocket client's instance manually). For demonstration purposes, we merged two streams and endeavored to .log() the results.