RSocket Interface in Spring

1. Overview

In this tutorial, we’ll explore how to utilize RSocket in the Spring Framework 6.

Working with RSocket has become simpler with the introduction of declarative RSocket clients in version 6 of the Spring Framework. This feature eliminates the need for repetitive boilerplate code, allowing developers to use RSocket more efficiently and effectively.

2. Maven Dependency

We start by creating a Spring Boot project in our preferred IDE and add the spring-boot-starter-rsocket dependency to the pom.xml file:

<dependency> 
    <groupId>org.springframework.boot</groupId> 
    <artifactId>spring-boot-starter-rsocket</artifactId>
    <version>3.1.4</version>
</dependency> 

3. Creating RSocket Server

First, we’ll create a responder that uses a controller to manage incoming requests:

@MessageMapping("MyDestination")
public Mono<String> message(Mono<String> input) {
    return input.doOnNext(msg -> System.out.println("Request is:" + msg + ",Request!"))
      .map(msg -> msg + ",Response!");
}

Besides, we’ll add the following property to the application.properties file to enable the server to listen on port 7000 via MyDestination:

spring.rsocket.server.port=7000

4. Client Code

Now, we need to develop the client code. To keep things simple, we’ll create the client code in the same project but in a separate package. In reality, they must be in a unique project.

To proceed, let us create the client interface:

public interface MessageClient {

    @RSocketExchange("MyDestination")
    Mono<String> sendMessage(Mono<String> input);
}

When using our client interface, we use @RSocketExchange to show the RSocket endpoint. Basically, this just means we need some info to establish the endpoint path. We can do that at the interface level by assigning a shared path. It’s super easy and helps us to know which endpoint we want to use.

5. Testing

Every Spring Boot project includes a class annotated with @SpringBootApplication. This class runs when the project is loaded. Therefore, we can use this class and add some beans to test a scenario.

5.1. Create RSocketServiceProxyFactory Bean

First, we need to create a bean to generate an RSocketServiceProxyFactory.

This factory is responsible for creating proxy instances of the RSocket service interface. It handles the creation of these proxies and establishes the necessary connection to the RSocket server by specifying the host and port where the server will receive incoming connections:

@Bean
public RSocketServiceProxyFactory getRSocketServiceProxyFactory(RSocketRequester.Builder requestBuilder) {
    RSocketRequester requester = requestBuilder.tcp("localhost", 7000);
    return RSocketServiceProxyFactory.builder(requester).build();
}

5.2. Create Message Client

Then, we’ll create a Bean responsible for generating a client interface:

@Bean
public MessageClient getClient(RSocketServiceProxyFactory factory) {
    return factory.createClient(MessageClient.class);
}

5.3. Create Runner Bean

Finally, let’s create a runner bean that uses the MessageClient instance to send and receive messages from the server:

@Bean
public ApplicationRunner runRequestResponseModel(MessageClient client) {
    return args -> {
        client.sendMessage(Mono.just("Request-Response test "))
          .doOnNext(message -> {
              System.out.println("Response is :" + message);
          })
          .subscribe();
    };
}

5.4. Test Results

When we run our Spring Boot project through the command line, the following results are displayed:

>>c.b.r.responder.RSocketApplication : Started 
>>RSocketApplication in 1.127 seconds (process running for 1.398)
>>Request is:Request-Response test ,Request!
>>Response is :Request-Response test ,Response!

6. RSocket Interaction Models

RSocket is a binary protocol used to create fast and responsive distributed applications. It offers different communication patterns for exchanging data between servers and clients.

With these interaction models, developers can design systems that meet specific requirements for data flow, backlog, and application behavior.

RSocket has four main interaction models available. The main difference between these approaches is based on the cardinality of input and output.

6.1. Request-Response

In this approach, every request receives a single response. Therefore, we used a Mono request with a cardinality of one and received a Mono response with the same cardinality.

Until now, all our code in this article was based on a request-response model.

6.2. Request-Stream

When we subscribe to a newsletter, we receive a regular flow of updates from the server. When the client makes the initial request, the server sends a data stream in response.

The request can be either a Mono or a Void, but the response will always be a Flux:

@MessageMapping("Counter")
public Flux<String> Counter() {
    return Flux.range(1, 10)
      .map(i -> "Count is: " + i);
}

6.3. Fire-and-Forget

When we send a letter through the mail, we usually just drop it in the mailbox and don’t expect to receive a reply. Similarly, in the fire-and-forget context, the response can be either null or a single Mono:

@MessageMapping("Warning")
public Mono<Void> Warning(Mono<String> error) {
    error.doOnNext(e -> System.out.println("warning is :" + e))
      .subscribe();
    return Mono.empty();
}

6.4. Channel

Imagine a walkie-talkie that allows two-way communication where both parties can talk and listen simultaneously, just like having a conversation. This type of communication relies on sending and receiving data Flux:

@MessageMapping("channel")
public Flux<String> channel(Flux<String> input) {
    return input.doOnNext(i -> {
          System.out.println("Received message is : " + i);
      })
      .map(m -> m.toUpperCase())
      .doOnNext(r -> {
          System.out.println("RESPONSE IS :" + r);
      });
}

7. Conclusion

In this article, we explored the new declarative RSocket client feature in Spring 6. We also learned how to use it with the @RSocketExchange annotation.

Additionally, we saw in detail how to create and set up the service proxy so that we can easily and safely connect to a remote endpoint using the TCP protocol.