Getting Started With RSocket Part 1

In this blog, you will learn the basics of RSocket, a binary application protocol which supports Reactive Streams. After the introduction, you will learn how to use RSocket in combination with Spring Boot. Enjoy!

1. Introduction

RSocket is a binary protocol to be used on top of TCP or WebSockets. RSocket is a communication protocol which embraces the Reactive principles. This means that RSocket uses asynchronuous communication. It is also suited for push notifications. When using HTTP for example, there will be a need for polling in order to check whether new messages are available. This causes unnessary network load. RSocket provides a solution for this. There are 4 communication models which can be used:

• Request-Response (a stream of 1)
• Fire-and-Forget (no response)
• Request-Stream (a stream of many)
• Channel (bi-directional streams)

RSocket is situated at the OSI layer 5/6 and therefore at the Application Layer of the TCP/IP Model.

In the next sections, you will find examples for each communication model: the server side, client side and a unit test. The source code being used in this post is of course available at GitHub.

2. Reference Documentation

Before getting started, it is useful to know where some interesting documentation can be found. During writing this blog, it appeared that reference documentation, examples, etc. cannot easily be found. This list should give you a flying start when taking your first steps with RSocket.

All information about the protocol, the specification, implementations can be found at the official RSocket website.

The Spring Framework’s support for the RSocket protocol.

The Spring Boot reference section for the RSocket protocol.

Ben Wilcock has written some awesome blogs about the RSocket protocol. The complete list can be found at GitHub.

3. Request-Response Model

The Request-Response model will allow you to send one request and receive one response in return. First thing to do, is to set up a basic Spring Boot application. Navigate to the Spring Initializr website, add dependency RSocket and create the project which you can open in your favorite IDE. When checking the pom, you notice that the spring-boot-starter-rsocket dependency and the dependency reactor-test are added. The first one will enable RSocket support in your Spring Boot application, the second one is needed for testing purposes.

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>
...
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
</dependency>

The source code at GitHub is divided into two Maven modules, one for the server and one for the client.

In order to exchange information between client and server, you create a Notification data class which will be the item to transport via RSocket. The Notification class contains a Source, a Destination and some free Text. The toString implementation will be used for logging purposes.

public class Notification {
    private String source;
    private String destination;
    private String text;

    public Notification() {
        super();
    }

    public Notification(String source, String destination, String text) {
        this.source = source;
        this.destination = destination;
        this.text = text;
    }

    public String getSource() {
        return source;
    }

    public String getDestination() {
        return destination;
    }

    public String getText() {
        return text;
    }

    @Override
    public String toString() {
        return "Notification{" +
                "source='" + source + '\'' +
                ", destination='" + destination + '\'' +
                ", text='" + text + '\'' +
                '}';
    }
}

3.1 The Server Side

You create a RsocketServerController and annotate it with @Controller. In order to create your first RSocket Request-Response example, you just add a method requestResponse which takes a Notification, logs the received Notification and returns a new Notification where you swap the received source and destination and add a simple text to it. In order to make it a RSocket request, you need to annotate the method with @MessageMapping and give it a name, e.g. my-request-response.

@Controller
public class RsocketServerController {

    Logger logger = LoggerFactory.getLogger(RsocketServerController.class);

    @MessageMapping("my-request-response")
    public Notification requestResponse(Notification notification) {
        logger.info("Received notification for my-request-response: " + notification);
        return new Notification(notification.getDestination(), notification.getSource(), "In response to: " + notification.getText());
    }
}

In order to ensure that the RSocket server is started, you also need to add the port to the application.properties file:

spring.rsocket.server.port=7000

Start the server:

$ mvn spring-boot:run

In the logging you notice that the Netty Webserver has started. Netty is the reactive counterpart of a Jetty Webserver.

Netty RSocket started on port(s): 7000

3.2 The Client Side

The client side is a little bit more complex. You will again create a Spring Boot application which will send a Notification message to the server. Sending the message will be invoked by means of an http call. Therefore, you add the dependency spring-boot-starter-webflux to the client pom. Beware that you cannot use spring-boot-starter-web, you need to use the reactive webflux variant.

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

Ensure that you do not define the port in the application.properties, otherwise a RSocket server will be started and that is not what is needed for your client. When you do so, the following error will appear in your console.

2021-01-02 12:04:58.853 ERROR 19058 --- [           main] o.s.boot.SpringApplication               : Application run failed
org.springframework.context.ApplicationContextException: Failed to start bean 'rSocketServerBootstrap'; nested exception is reactor.netty.ChannelBindException: Failed to bind on [0.0.0.0:7000]
...
Caused by: reactor.netty.ChannelBindException: Failed to bind on [0.0.0.0:7000]
    Suppressed: java.lang.Exception: #block terminated with an error
...

You create a RsocketClientController and annotate it with @RestController. Next, you need to create a RSocketRequester instance in order to be able to connect to the RSocket server. In the requestResponse method, you create the Notification message (for ease of use, just copy the Notification class from the server module and make sure that it is also present on the client side) and with the rSocketRequester instance, you specify the route you want the message to be sent to (name equals the name as specified with the @MessageMapping annotation on the server side), the data you want to send and finally the response you expect. The response will be a Mono which means that you expect one response from the server and the response needs to be a Notification message. The message itself is returned to the caller.

@RestController
public class RsocketClientController {

    private static final String CLIENT = "Client";
    private static final String SERVER = "Server";

    private final RSocketRequester rSocketRequester;

    Logger logger = LoggerFactory.getLogger(RsocketClientController.class);

    public RsocketClientController(@Autowired RSocketRequester.Builder builder) {
        this.rSocketRequester = builder.tcp("localhost", 7000);
    }

    @GetMapping("/request-response")
    public Mono<Notification> requestResponse() {
        Notification notification = new Notification(CLIENT, SERVER, "Test the Request-Response interaction model");
        logger.info("Send notification for my-request-response: " + notification);
        return rSocketRequester
                .route("my-request-response")
                .data(notification)
                .retrieveMono(Notification.class);
    }
}

Start both the server and the client and invoke the URL:

$ curl http://localhost:8080/request-response
{"source":"Server","destination":"Client","text":"In response to: Test the Request-Response interaction model"}

As you can see, the server response is returned. In the logging of client and server, you can verify the sending and receiving messages.

Client:

Send notification for my-request-response: Notification{source='Client', destination='Server', text='Test the Request-Response interaction model'}

Server:

Received notification for my-request-response: Notification{source='Client', destination='Server', text='Test the Request-Response interaction model'}

3.3 The Test Side

Creating a test for the server code is quite similar as creating the client code. A RSocketRequester needs to be created in order to setup the connection. Sending a message is identical to the client code, only this time you put the response into a result variable of type Mono<Notification>. You can use this result variable in a StepVerifier in order to validate the response which is received. With a StepVerifier you can verify reactive responses in your unit test.

@SpringBootTest
class MyRsocketServerPlanetApplicationTests {

    private static final String CLIENT = "Client";
    private static final String SERVER = "Server";

    private static RSocketRequester rSocketRequester;

    @BeforeAll
    public static void setupOnce(@Autowired RSocketRequester.Builder builder, @Value("${spring.rsocket.server.port}") Integer port) {
        rSocketRequester = builder.tcp("localhost", port);
    }

    @Test
    void testRequestResponse() {
        // Send a request message
        Mono<Notification> result = rSocketRequester
                .route("my-request-response")
                .data(new Notification(CLIENT, SERVER, "Test the Request-Response interaction model"))
                .retrieveMono(Notification.class);

        // Verify that the response message contains the expected data
        StepVerifier
                .create(result)
                .consumeNextWith(notification -> {
                    assertThat(notification.getSource()).isEqualTo(SERVER);
                    assertThat(notification.getDestination()).isEqualTo(CLIENT);
                    assertThat(notification.getText()).isEqualTo("In response to: Test the Request-Response interaction model");
                })
                .verifyComplete();
    }
}

4. Conclusion

You have learnt the basics of the RSocket application protocol and explored how to create a server, client and unit test for the Request-Response communication model. In the next blog, you will learn how to create a server, client and unit test for the remaining three communication models.