Petros Stergioulas

Posted on Jun 10, 2019 • Edited on Jun 8, 2020

Getting started with RSocket in Spring Boot

#tutorial #java #spring #rsocket

Introduction

RSocket is a binary protocol for use on byte stream transports such as TCP, WebSockets, and Aeron.

It enables the following symmetric interaction models via async message passing over a single connection:

request/response (stream of 1)
request/stream (finite stream of many)
fire-and-forget (no response)
event subscription (infinite stream of many)

It supports session resumption, to allow resuming long-lived streams across different transport connections. This is particularly useful for mobile⬄server communication when network connections drop, switch, and reconnect frequently.

In our tutorial, we will implement RSocket using the Java programming language.

Why Springboot

Although I could simply implement RSocket with a simple Java application, I chose Springboot because it's a huge project on the JVM languages ecosystem. Springboot doesn't have a stable version for RSocket yet, but that shouldn't stop us experimenting with it.

Structure

Our project will consist of two sub-projects. The Consumer, who will do the requests and the Producer, who will provide the Consumer with data.

Getting Started

If you want to easily get started with a Springboot project I recommend always using the Spring Initializr.

Configuring Gradle

First of all we have to configure our build.gradle and include the rsocket-starter-dependency from Springboot:

	dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-rsocket'
	....
	}

view raw build.gradle hosted with ❤ by GitHub

For the Consumer we should include also the reactive-web-starter dependency, because we want to somehow display the data that we receive from the Producer.

	dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-rsocket'
	implementation 'org.springframework.boot:spring-boot-starter-webflux'
	....
	}

view raw build.gradle hosted with ❤ by GitHub

Configuration

Now let's get our hands dirty and write some code 😉.

Producer

First we have to configure in which port will our Producer receive new connections:

	spring.rsocket.server.port=7000
	spring.main.lazy-initialization=true

view raw application.properties hosted with ❤ by GitHub

The spring.rsocket.server.port=7000 configures the RSocket to start listening on port 7000 and the spring.main.lazy-initialization=true is to make sure that Springboot will initialize its beans lazily.

Consumer

Now, here, we configure the RSocket client, by creating two Beans of type RSocket and RSocketRequester. The RSocket bean is used to create the connection to the Producer. The RSocketRequester bean is like our WebClient (if we are using Reactive) or the RestTemplate (if we are using MVC), he will do the requests to Producer.

	@Bean
	@SneakyThrows
	RSocket rSocket() {
	return RSocketFactory.connect()
	.dataMimeType(MimeTypeUtils.APPLICATION_JSON_VALUE)
	.frameDecoder(PayloadDecoder.ZERO_COPY)
	.transport(TcpClientTransport.create(new InetSocketAddress("127.0.0.1", 7000)))
	.start()
	.block();
	}

	@Bean
	RSocketRequester requester(RSocketStrategies strategies) {
	return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, strategies);
	}

view raw RSocketConfig.java hosted with ❤ by GitHub

The Actual Code

Our two applications (the Producer and the Consumer) will use the following domain objects for their communication:

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	class GreetingsRequest {
	private String name;
	}

view raw GreetingsRequest.java hosted with ❤ by GitHub

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	class GreetingsResponse {
	private String greeting;
	}

view raw GreetingsResponse.java hosted with ❤ by GitHub

Simple Communication

Let's start with an easy example, a simple Request with a single Response.

Producer

	@MessageMapping("greet")
	Mono<GreetingsResponse> greet(GreetingsRequest request) {
	return Mono.just(new GreetingsResponse("Hello " + request.getName() + " @ " + Instant.now()));
	}

view raw GreetingsController.java hosted with ❤ by GitHub

If you ever used WebSockets with Springboot the @MessageMapping annotation should be familiar.
Here, with the @MessageMapping("greet") we specify the route "greet", which receives a GreetingsRequest and returns a single object GreetingsResponse as a response.

Consumer

	@RequiredArgsConstructor
	@RestController
	class GreetingsRestController {

	private final RSocketRequester requester;

	@GetMapping("/greet/{name}")
	public Publisher<GreetingsResponse> greet(@PathVariable String name) {
	return requester
	.route("greet")
	.data(new GreetingsRequest(name))
	.retrieveMono(GreetingsResponse.class);
	}
	}

view raw GreetingsRestController.java hosted with ❤ by GitHub

Here we see a simple @RestController which injects the RSocketRequester (the one we configured earlier) and uses him to do the Request. As you see, we tell the requester to do a Request to the "greet" route (which we configured earlier in the Producer).

Communication with Stream

Delivering a stream of data with RSocket comes natural and with not that much of an effort.

Producer

	@MessageMapping("greet-stream")
	Flux<GreetingsResponse> greetStream(GreetingsRequest request) {
	return Flux.fromStream(Stream.generate(
	() -> new GreetingsResponse("Hello " + request.getName() + " @ " + Instant.now())
	)).delayElements(Duration.ofSeconds(1));
	}

view raw GreetingsController.java hosted with ❤ by GitHub

As you see in the gist above, we configured our @MessageMapping as before, but now we return a Flux (don't know what Mono and Flux are? Check this StackOverflow post).

We also use the .delayElements() method to simulate a delay between our responses.

Consumer

	@GetMapping(value = "/greet-stream/{name}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	public Publisher<GreetingsResponse> greetStream(@PathVariable String name) {
	return requester
	.route("greet-stream")
	.data(new GreetingsRequest(name))
	.retrieveFlux(GreetingsResponse.class);
	}

view raw GreetingsRestController.java hosted with ❤ by GitHub

Not much have changed from our other request to the Producer, clearly the .route("greet-stream"). But if you see carefully, now our @GetMapping produces a text/event-stream and not the default application/json.

Let's check the result!