Make it work

RSocket examples – Java, JavaScript and Spring WebFlux

September 25, 2019 /
Overview

RSocket is a binary protocol that allows asynchronous message passing over a single connection. It uses Reactive Streams, provides symmetric API’s for server and client and supports TCP, WebSockets, and Aeron. It’s not widely used, but it’s worth taking a closer look, especially if you like reactive programming or build Spring WebFlux application. The Internet lacks RSocket examples and that’s the reason why I’m going to show you how to establish a client-server connection with Java, JavaScript and Spring WebFlux.
As always, remember you can see the full source code in this GitHub project.

Java server - Java client

Let’s start with a simple Java application to easy understand different interaction models provided by RSocket: fire-and-forget, request/response, request/stream, and channel. Let’s create a Maven Project with the following dependencies:

pom.xml
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5. <modelVersion>4.0.0</modelVersion>
  7. <groupId>com.kojotdev.blog.rsocket</groupId>
  8. <artifactId>rsocket-examples</artifactId>
  9. <version>1.0-SNAPSHOT</version>
  11. <dependencies>
  12. <dependency>
  13. <groupId>io.rsocket</groupId>
  14. <artifactId>rsocket-core</artifactId>
  15. <version>1.0.0-RC3</version>
  16. </dependency>
  17. <dependency>
  18. <groupId>io.rsocket</groupId>
  19. <artifactId>rsocket-transport-netty</artifactId>
  20. <version>1.0.0-RC3</version>
  21. </dependency>
  22. <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->
  23. <dependency>
  24. <groupId>com.fasterxml.jackson.core</groupId>
  25. <artifactId>jackson-databind</artifactId>
  26. <version>2.10.0.pr3</version>
  27. </dependency>
  29. <dependency>
  30. <groupId>ch.qos.logback</groupId>
  31. <artifactId>logback-classic</artifactId>
  32. <version>1.2.3</version>
  33. </dependency>
  34. <dependency>
  35. <groupId>ch.qos.logback</groupId>
  36. <artifactId>logback-core</artifactId>
  37. <version>1.2.3</version>
  38. </dependency>
  39. <dependency>
  40. <groupId>org.slf4j</groupId>
  41. <artifactId>slf4j-api</artifactId>
  42. <version>1.7.25</version>
  43. </dependency>
  44. </dependencies>
  46. <repositories>
  47. <repository>
  48. <id>spring-milestones</id>
  49. <name>Spring Milestones</name>
  50. <url>https://repo.spring.io/libs-milestone</url>
  51. <snapshots>
  52. <enabled>false</enabled>
  53. </snapshots>
  54. </repository>
  55. </repositories>
  57. <build>
  58. <plugins>
  59. <plugin>
  60. <groupId>org.apache.maven.plugins</groupId>
  61. <artifactId>maven-compiler-plugin</artifactId>
  62. <version>3.8.1</version>
  63. <configuration>
  64. <release>11</release>
  65. </configuration>
  66. </plugin>
  67. </plugins>
  68. </build>
  70. </project>
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.kojotdev.blog.rsocket</groupId>
    <artifactId>rsocket-examples</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>io.rsocket</groupId>
            <artifactId>rsocket-core</artifactId>
            <version>1.0.0-RC3</version>
        </dependency>
        <dependency>
            <groupId>io.rsocket</groupId>
            <artifactId>rsocket-transport-netty</artifactId>
            <version>1.0.0-RC3</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.10.0.pr3</version>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.3</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
            <version>1.2.3</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.25</version>
        </dependency>
    </dependencies>

    <repositories>
        <repository>
            <id>spring-milestones</id>
            <name>Spring Milestones</name>
            <url>https://repo.spring.io/libs-milestone</url>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <release>11</release>
                </configuration>
            </plugin>
        </plugins>
    </build>
    
</project>

We added RSocket dependencies, Jackson for mapping JSON objects, and a logback for logger. We also added a Spring Milestones repository because it’s needed with the current version. Otherwise, Maven will throw an exception Could not find artifact io.projectreactor:reactor-bom:pom:Dysprosium-M3 in central (https://repo.maven.apache.org/maven2)
OK, it’s time to add our domain Message class and so ugly MessageMapper 😉

Message.java
  1. import com.fasterxml.jackson.annotation.JsonCreator;
  2. import com.fasterxml.jackson.annotation.JsonProperty;
  4. public class Message {
  6. public final String message;
  8. @JsonCreator
  9. public Message(@JsonProperty("message") String message) {
  10. this.message = message;
  11. }
  12. }
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;

public class Message {

    public final String message;

    @JsonCreator
    public Message(@JsonProperty("message") String message) {
        this.message = message;
    }
}
MessageMapper.java
  1. import com.fasterxml.jackson.core.JsonProcessingException;
  2. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import java.io.IOException;
  6. public class MessageMapper {
  7. private static ObjectMapper jsonMapper = new ObjectMapper();
  9. public static String messageToJson(Message msg) {
  10. try {
  11. return jsonMapper.writeValueAsString(msg);
  12. } catch (JsonProcessingException e) {
  13. throw new IllegalStateException(e);
  14. }
  15. }
  17. public static Message jsonToMessage(String json) {
  18. try {
  19. return jsonMapper.readValue(json, Message.class);
  20. } catch (IOException e) {
  21. throw new IllegalStateException(e);
  22. }
  24. }
  25. }
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.IOException;

public class MessageMapper {
    private static ObjectMapper jsonMapper = new ObjectMapper();

    public static String messageToJson(Message msg) {
        try {
            return jsonMapper.writeValueAsString(msg);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException(e);
        }
    }

    public static Message jsonToMessage(String json) {
        try {
            return jsonMapper.readValue(json, Message.class);
        } catch (IOException e) {
            throw new IllegalStateException(e);
        }

    }
}

The Server class is listed below:

Server.java
  1. import ch.qos.logback.classic.Level;
  2. import ch.qos.logback.classic.Logger;
  3. import io.rsocket.AbstractRSocket;
  4. import io.rsocket.Payload;
  5. import io.rsocket.RSocketFactory;
  6. import io.rsocket.transport.netty.server.WebsocketServerTransport;
  7. import io.rsocket.util.DefaultPayload;
  8. import org.reactivestreams.Publisher;
  9. import org.slf4j.LoggerFactory;
  10. import reactor.core.publisher.Flux;
  11. import reactor.core.publisher.Mono;
  13. public final class Server {
  15. private static Logger log = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
  17. static {
  18. log.setLevel(Level.INFO);
  19. }
  21. public static void main(String[] args) {
  22. RSocketFactory.receive()
  23. .acceptor((setup, sendingSocket) -> Mono.just(new DefaultSimpleService()))
  24. .transport(WebsocketServerTransport.create(8801))
  25. .start()
  26. .block()
  27. .onClose()
  28. .block();
  29. }
  31. private static final class DefaultSimpleService extends AbstractRSocket {
  33. @Override
  34. public Mono<Void> fireAndForget(Payload payload) {
  35. log.info("got fireAndForget in Server");
  36. log.info(payload.getDataUtf8());
  37. return Mono.empty();
  38. }
  40. @Override
  41. public Mono<Payload> requestResponse(Payload payload) {
  42. log.info("got requestResponse in Server");
  43. log.info(payload.getDataUtf8());
  44. return Mono.just(payload.getDataUtf8())
  45. .map(payloadString -> MessageMapper.jsonToMessage(payloadString))
  46. .map(message -> message.message + " | requestReponse from Server #1")
  47. .map(responseText -> new Message(responseText))
  48. .map(responseMessage -> MessageMapper.messageToJson(responseMessage))
  49. .map(responseJson -> DefaultPayload.create(responseJson));
  50. }
  52. @Override
  53. public Flux<Payload> requestStream(Payload payload) {
  54. log.info("got requestStream in Server");
  55. log.info(payload.getDataUtf8());
  56. return Mono.just(payload.getDataUtf8())
  57. .map(payloadString -> MessageMapper.jsonToMessage(payloadString))
  58. .flatMapMany(msg -> Flux.range(0, 5)
  59. .map(count -> msg.message + " | requestStream from Server #" + count)
  60. .map(responseText -> new Message(responseText))
  61. .map(responseMessage -> MessageMapper.messageToJson(responseMessage)))
  62. .map(message -> DefaultPayload.create(message));
  63. }
  65. @Override
  66. public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
  67. log.info("got requestChannel in Server");
  68. return Flux.from(payloads)
  69. .map(payload -> payload.getDataUtf8())
  70. .map(payloadString -> {
  71. log.info(payloadString);
  72. return MessageMapper.jsonToMessage(payloadString);
  73. })
  74. .flatMap(msg -> Flux.range(0, 2)
  75. .map(count -> msg.message + " | requestChannel from Server #" + count)
  76. .map(responseText -> new Message(responseText))
  77. .map(responseMessage -> MessageMapper.messageToJson(responseMessage)))
  78. .map(message -> DefaultPayload.create(message));
  80. }
  81. }
  83. }
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.server.WebsocketServerTransport;
import io.rsocket.util.DefaultPayload;
import org.reactivestreams.Publisher;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public final class Server {

    private static Logger log = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);

    static {
        log.setLevel(Level.INFO);
    }

    public static void main(String[] args) {
        RSocketFactory.receive()
                .acceptor((setup, sendingSocket) -> Mono.just(new DefaultSimpleService()))
                .transport(WebsocketServerTransport.create(8801))
                .start()
                .block()
                .onClose()
                .block();
    }

    private static final class DefaultSimpleService extends AbstractRSocket {

        @Override
        public Mono<Void> fireAndForget(Payload payload) {
            log.info("got fireAndForget in Server");
            log.info(payload.getDataUtf8());
            return Mono.empty();
        }

        @Override
        public Mono<Payload> requestResponse(Payload payload) {
            log.info("got requestResponse in Server");
            log.info(payload.getDataUtf8());
            return Mono.just(payload.getDataUtf8())
                    .map(payloadString -> MessageMapper.jsonToMessage(payloadString))
                    .map(message -> message.message + " | requestReponse from Server #1")
                    .map(responseText -> new Message(responseText))
                    .map(responseMessage -> MessageMapper.messageToJson(responseMessage))
                    .map(responseJson -> DefaultPayload.create(responseJson));
        }

        @Override
        public Flux<Payload> requestStream(Payload payload) {
            log.info("got requestStream in Server");
            log.info(payload.getDataUtf8());
            return Mono.just(payload.getDataUtf8())
                    .map(payloadString -> MessageMapper.jsonToMessage(payloadString))
                    .flatMapMany(msg -> Flux.range(0, 5)
                            .map(count -> msg.message + " | requestStream from Server #" + count)
                            .map(responseText -> new Message(responseText))
                            .map(responseMessage -> MessageMapper.messageToJson(responseMessage)))
                    .map(message -> DefaultPayload.create(message));
        }

        @Override
        public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
            log.info("got requestChannel in Server");
            return Flux.from(payloads)
                    .map(payload -> payload.getDataUtf8())
                    .map(payloadString -> {
                        log.info(payloadString);
                        return MessageMapper.jsonToMessage(payloadString);
                    })
                    .flatMap(msg -> Flux.range(0, 2)
                            .map(count -> msg.message + " | requestChannel from Server #" + count)
                            .map(responseText -> new Message(responseText))
                            .map(responseMessage -> MessageMapper.messageToJson(responseMessage)))
                    .map(message -> DefaultPayload.create(message));

        }
    }

}

In the main static method, we start our RSocket server. It uses WebSocket at 8801 port and accepts interaction models implemented by the DefaultSimpleService class. All of these methods are very similar. They just get the payload, change its message and return it to the client. You can see the difference in the arguments and return types. 

    • fireAndForget just gets the payload and returns nothing (you don’t want to wait for the server response)
    • requestResponse gets the payload and returns a single response (like with the standard HTTP/REST application)
    • requestStream gets the payload and returns stream (no need to implement the SSE, it’s built-in)
    • channel – stream as a request and stream as a response (bi-directional streams)

The client class looks very similar to the server class:

Client.java
  1. import ch.qos.logback.classic.Level;
  2. import ch.qos.logback.classic.Logger;
  3. import io.rsocket.Payload;
  4. import io.rsocket.RSocket;
  5. import io.rsocket.RSocketFactory;
  6. import io.rsocket.transport.netty.client.WebsocketClientTransport;
  7. import io.rsocket.util.DefaultPayload;
  8. import org.slf4j.LoggerFactory;
  9. import reactor.core.publisher.Flux;
  11. public class Client {
  13. private static Logger log = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
  15. static {
  16. log.setLevel(Level.INFO);
  17. }
  19. public static void main(String[] args) {
  20. final Client client = new Client();
  21. final RSocket rSocket = client.connect();
  22. client.fireAndForget(rSocket);
  23. client.requestResponse(rSocket);
  24. client.requestStream(rSocket);
  25. client.requestChannel(rSocket);
  26. }
  28. private RSocket connect() {
  29. return RSocketFactory.connect()
  30. .transport(WebsocketClientTransport.create(8801))
  31. .start()
  32. .block();
  33. }
  35. private void fireAndForget(RSocket rSocket) {
  36. log.info("sending fire and forget from client");
  37. Flux.just(new Message("fire and forget JAVA client!"))
  38. .map(msg -> MessageMapper.messageToJson(msg))
  39. .map(json -> DefaultPayload.create(json))
  40. .flatMap(message -> rSocket.fireAndForget(message))
  41. .blockLast();
  42. }
  44. private void requestResponse(RSocket rSocket) {
  45. log.info("sending request-response from client");
  46. Flux.just(new Message("requestResponse from JAVA client!"))
  47. .map(msg -> MessageMapper.messageToJson(msg))
  48. .map(json -> DefaultPayload.create(json))
  49. .flatMap(message -> rSocket.requestResponse(message))
  50. .map(payload -> payload.getDataUtf8())
  51. .doOnNext(payloadString -> {
  52. log.info("got response in JAVA client");
  53. log.info(payloadString);
  54. })
  55. .blockLast();
  56. }
  58. private void requestStream(RSocket rSocket) {
  59. log.info("sending request-stream from client");
  60. Flux.just(new Message("requestStream from JAVA client!"))
  61. .map(msg -> MessageMapper.messageToJson(msg))
  62. .map(json -> DefaultPayload.create(json))
  63. .flatMap(message -> rSocket.requestStream(message))
  64. .map(payload -> payload.getDataUtf8())
  65. .doOnNext(payloadString -> log.info(payloadString))
  66. .blockLast();
  67. }
  69. private void requestChannel(RSocket rSocket) {
  70. log.info("sending request-channel from client");
  71. final Flux<Payload> requestPayload = Flux.range(0, 5)
  72. .map(count -> new Message("requestChannel from JAVA client! #" + count))
  73. .map(msg -> {
  74. log.info("sending message: {}", msg.message);
  75. return MessageMapper.messageToJson(msg);
  76. })
  77. .map(json -> DefaultPayload.create(json));
  79. rSocket
  80. .requestChannel(requestPayload)
  81. .map(payload -> payload.getDataUtf8())
  82. .doOnNext(payloadString -> log.info(payloadString))
  83. .blockLast();
  84. }
  85. }
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.WebsocketClientTransport;
import io.rsocket.util.DefaultPayload;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;

public class Client {

    private static Logger log = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);

    static {
        log.setLevel(Level.INFO);
    }

    public static void main(String[] args) {
        final Client client = new Client();
        final RSocket rSocket = client.connect();
        client.fireAndForget(rSocket);
        client.requestResponse(rSocket);
        client.requestStream(rSocket);
        client.requestChannel(rSocket);
    }

    private RSocket connect() {
        return RSocketFactory.connect()
                .transport(WebsocketClientTransport.create(8801))
                .start()
                .block();
    }

    private void fireAndForget(RSocket rSocket) {
        log.info("sending fire and forget from client");
        Flux.just(new Message("fire and forget JAVA client!"))
                .map(msg -> MessageMapper.messageToJson(msg))
                .map(json -> DefaultPayload.create(json))
                .flatMap(message -> rSocket.fireAndForget(message))
                .blockLast();
    }

    private void requestResponse(RSocket rSocket) {
        log.info("sending request-response from client");
        Flux.just(new Message("requestResponse from JAVA client!"))
                .map(msg -> MessageMapper.messageToJson(msg))
                .map(json -> DefaultPayload.create(json))
                .flatMap(message -> rSocket.requestResponse(message))
                .map(payload -> payload.getDataUtf8())
                .doOnNext(payloadString -> {
                    log.info("got response in JAVA client");
                    log.info(payloadString);
                })
                .blockLast();
    }

    private void requestStream(RSocket rSocket) {
        log.info("sending request-stream from client");
        Flux.just(new Message("requestStream from JAVA client!"))
                .map(msg -> MessageMapper.messageToJson(msg))
                .map(json -> DefaultPayload.create(json))
                .flatMap(message -> rSocket.requestStream(message))
                .map(payload -> payload.getDataUtf8())
                .doOnNext(payloadString -> log.info(payloadString))
                .blockLast();
    }

    private void requestChannel(RSocket rSocket) {
        log.info("sending request-channel from client");
        final Flux<Payload> requestPayload = Flux.range(0, 5)
                .map(count -> new Message("requestChannel from JAVA client! #" + count))
                .map(msg -> {
                    log.info("sending message: {}", msg.message);
                    return MessageMapper.messageToJson(msg);
                })
                .map(json -> DefaultPayload.create(json));

        rSocket
                .requestChannel(requestPayload)
                .map(payload -> payload.getDataUtf8())
                .doOnNext(payloadString -> log.info(payloadString))
                .blockLast();
    }
}

We still use the RSocketFactory, but this time with the connect() method instead of receive(). As you can see I use WebsocketServerTransport, but you can easily change it to the TCP (same with the port and host). After establishing the connection, we call the appropriate method and handle the response.

Java server - JavaScript client

With rsocket-js we can easily write our JavaScript client. As usual, I’ve created the Vue project, but I’m pretty sure you will understand the code even if you never used the Vue before. To make it work, we need to install the following packages:

    • npm i rsocket-core
    • npm i rsocket-websocket-client
    • npm i rsocket-flowable

Let’s connect with the backend:

Home.vue
  1. import { RSocketClient, JsonSerializers } from "rsocket-core";
  2. import RSocketWebSocketClient from "rsocket-websocket-client";
  3. ...
  4. connect() {
  5. console.log("connecting with RSocket...");
  6. const transport = new RSocketWebSocketClient({
  7. url: "ws://localhost:8801"
  8. });
  9. const client = new RSocketClient({
  10. // send/receive JSON objects instead of strings/buffers
  11. serializers: JsonSerializers,
  12. setup: {
  13. // ms btw sending keepalive to server
  14. keepAlive: 60000,
  15. // ms timeout if no keepalive response
  16. lifetime: 180000,
  17. // format of `data`
  18. dataMimeType: "application/json",
  19. // format of `metadata`
  20. metadataMimeType: "application/json"
  21. },
  22. transport
  23. });
  24. client.connect().subscribe({
  25. onComplete: socket => {
  26. this.socket = socket;
  27. },
  28. onError: error => {
  29. console.log("got connection error");
  30. console.error(error);
  31. },
  32. onSubscribe: cancel => {
  33. /* call cancel() to abort */
  34. }
  35. });
import { RSocketClient, JsonSerializers } from "rsocket-core";
import RSocketWebSocketClient from "rsocket-websocket-client";
...
connect() {
      console.log("connecting with RSocket...");
      const transport = new RSocketWebSocketClient({
        url: "ws://localhost:8801"
      });
      const client = new RSocketClient({
        // send/receive JSON objects instead of strings/buffers
        serializers: JsonSerializers,
        setup: {
          // ms btw sending keepalive to server
          keepAlive: 60000,
          // ms timeout if no keepalive response
          lifetime: 180000,
          // format of `data`
          dataMimeType: "application/json",
          // format of `metadata`
          metadataMimeType: "application/json"
        },
        transport
      });
      client.connect().subscribe({
        onComplete: socket => {
          this.socket = socket;
        },
        onError: error => {
          console.log("got connection error");
          console.error(error);
        },
        onSubscribe: cancel => {
          /* call cancel() to abort */
        }
      });

The signature and implementation of each method are presented below.

fireAndForget(payload: Payload): void

FireAndForget.vue
  1. if (this.socket) {
  2. const message = { message: "fire and forget from JavaScript!" };
  3. this.socket.fireAndForget({
  4. data: message,
  5. metadata: ""
  6. });
  7. this.sent.push(message);
  8. } else {
  9. console.log("not connected...");
  10. }
if (this.socket) {
        const message = { message: "fire and forget from JavaScript!" };
        this.socket.fireAndForget({
          data: message,
          metadata: ""
        });
        this.sent.push(message);
      } else {
        console.log("not connected...");
      }
requestResponse(payload: Payload): Single
RequestResponse.vue
  1. if (this.socket) {
  2. const message = { message: "requestResponse from JavaScript!" };
  3. this.socket
  4. .requestResponse({
  5. data: message,
  6. metadata: ""
  7. })
  8. .subscribe({
  9. onComplete: data => {
  10. console.log("got response with requestResponse");
  11. this.received.push(data.data);
  12. },
  13. onError: error => {
  14. console.log("got error with requestResponse");
  15. console.error(error);
  16. },
  17. onSubscribe: cancel => {
  18. this.sent.push(message);
  19. /* call cancel() to stop onComplete/onError */
  20. }
  21. });
  22. } else {
  23. console.log("not connected...");
  24. }
  if (this.socket) {
	const message = { message: "requestResponse from JavaScript!" };
	this.socket
	  .requestResponse({
		data: message,
		metadata: ""
	  })
	  .subscribe({
		onComplete: data => {
		  console.log("got response with requestResponse");
		  this.received.push(data.data);
		},
		onError: error => {
		  console.log("got error with requestResponse");
		  console.error(error);
		},
		onSubscribe: cancel => {
		  this.sent.push(message);
		  /* call cancel() to stop onComplete/onError */
		}
	  });
  } else {
	console.log("not connected...");
  }
requestStream(payload: Payload): Flowable
RequestStream.vue
  1. if (this.socket) {
  2. const message = { message: "requestStream from JavaScript!" };
  3. this.socket
  4. .requestStream({
  5. data: message,
  6. metadata: ""
  7. })
  8. .subscribe({
  9. onComplete: () => {
  10. console.log("requestStream done");
  11. this.received.push("requestStream done");
  12. },
  13. onError: error => {
  14. console.log("got error with requestStream");
  15. console.error(error);
  16. },
  17. onNext: value => {
  18. // console.log("got next value in requestStream..");
  19. this.received.push(value.data);
  20. },
  21. // Nothing happens until `request(n)` is called
  22. onSubscribe: sub => {
  23. console.log("subscribe request Stream!");
  24. sub.request(7);
  25. this.sent.push(message);
  26. }
  27. });
  28. } else {
  29. console.log("not connected...");
  30. }
  if (this.socket) {
	const message = { message: "requestStream from JavaScript!" };
	this.socket
	  .requestStream({
		data: message,
		metadata: ""
	  })
	  .subscribe({
		onComplete: () => {
		  console.log("requestStream done");
		  this.received.push("requestStream done");
		},
		onError: error => {
		  console.log("got error with requestStream");
		  console.error(error);
		},
		onNext: value => {
		  // console.log("got next value in requestStream..");
		  this.received.push(value.data);
		},
		// Nothing happens until `request(n)` is called
		onSubscribe: sub => {
		  console.log("subscribe request Stream!");
		  sub.request(7);
		  this.sent.push(message);
		}
	  });
  } else {
	console.log("not connected...");
  }
requestChannel(payload: Flowable): Flowable
RequestChannel.vue
  1. if (this.socket) {
  2. const flowablePayload = new Flowable(subscriber => {
  3. subscriber.onSubscribe({
  4. cancel: () => {},
  5. request: n => {
  6. for (let index = 0; index < n; index++) {
  7. const message = {
  8. message: "requestChannel from JavaScript! #" + index
  9. };
  10. subscriber.onNext(message);
  11. }
  12. subscriber.onComplete();
  13. }
  14. });
  15. });
  17. // test flowable payload
  18. // flowablePayload.subscribe({
  19. // onComplete: () => console.log("done"),
  20. // onError: error => console.error(error),
  21. // onNext: value => {
  22. // console.log("got onNext value ");
  23. // console.log(value);
  24. // },
  25. // // Nothing happens until `request(n)` is called
  26. // onSubscribe: sub => {
  27. // sub.request(5);
  28. // }
  29. // });
  31. this.socket
  32. .requestChannel({
  33. data: flowablePayload,
  34. metadata: ""
  35. })
  36. .subscribe({
  37. onComplete: () => {
  38. console.log("requestChannel done");
  39. this.received.push("requestChannel done");
  40. },
  41. onError: error => {
  42. console.log("got error with requestChannel");
  43. console.error(error);
  44. },
  45. onNext: value => {
  46. // console.log("got next value in requestChannel..");
  47. this.received.push(value.data);
  48. },
  49. // Nothing happens until `request(n)` is called
  50. onSubscribe: sub => {
  51. console.log("subscribe request Channel!");
  52. sub.request(7);
  53. this.sent.push(message);
  54. }
  55. });
  56. } else {
  57. console.log("not connected...");
  58. }
  if (this.socket) {
	const flowablePayload = new Flowable(subscriber => {
	  subscriber.onSubscribe({
		cancel: () => {},
		request: n => {
		  for (let index = 0; index < n; index++) {
			const message = {
			  message: "requestChannel from JavaScript! #" + index
			};
			subscriber.onNext(message);
		  }
		  subscriber.onComplete();
		}
	  });
	});

	// test flowable payload
	// flowablePayload.subscribe({
	//   onComplete: () => console.log("done"),
	//   onError: error => console.error(error),
	//   onNext: value => {
	//     console.log("got onNext value ");
	//     console.log(value);
	//   },
	//   // Nothing happens until `request(n)` is called
	//   onSubscribe: sub => {
	//     sub.request(5);
	//   }
	// });

	this.socket
	  .requestChannel({
		data: flowablePayload,
		metadata: ""
	  })
	  .subscribe({
		onComplete: () => {
		  console.log("requestChannel done");
		  this.received.push("requestChannel done");
		},
		onError: error => {
		  console.log("got error with requestChannel");
		  console.error(error);
		},
		onNext: value => {
		  // console.log("got next value in requestChannel..");
		  this.received.push(value.data);
		},
		// Nothing happens until `request(n)` is called
		onSubscribe: sub => {
		  console.log("subscribe request Channel!");
		  sub.request(7);
		  this.sent.push(message);
		}
	  });
  } else {
	console.log("not connected...");
  }

At this moment, the requestChannel method doesn’t work, because it is not yet implemented in the rsocket-js.

WebFlux server - WebFlux client

The latest version of Spring Boot (2.2.0 M6) supports RSocket. It means that you can very easily configure the server and client. I’m not going to implement all of the interaction models, they just depend on the method arguments and return type. Simple server that handles requestResponse model requires below configuration:

ServerController.java
  1. import com.kojotdev.blog.webfluxrsocketexample.domain.Message;
  2. import org.slf4j.Logger;
  3. import org.slf4j.LoggerFactory;
  4. import org.springframework.messaging.handler.annotation.MessageMapping;
  5. import org.springframework.stereotype.Controller;
  6. import reactor.core.publisher.Mono;
  8. @Controller
  9. public class ServerController {
  10. private static final Logger log = LoggerFactory.getLogger(ServerController.class);
  12. @MessageMapping("hello")
  13. public Mono<Message> helloServer(Message message) {
  14. log.info("got message in Server {}", message.message);
  15. return Mono.just(message)
  16. .map(msg -> new Message(msg.message + " | Server says hello!"));
  17. }
  18. }
import com.kojotdev.blog.webfluxrsocketexample.domain.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Mono;

@Controller
public class ServerController {
    private static final Logger log = LoggerFactory.getLogger(ServerController.class);

    @MessageMapping("hello")
    public Mono<Message> helloServer(Message message) {
        log.info("got message in Server {}", message.message);
        return Mono.just(message)
                .map(msg -> new Message(msg.message + " | Server says hello!"));
    }
}
application.properties
  1. #spring.rsocket.server.port=7000
  2. spring.rsocket.server.transport=websocket
  3. spring.rsocket.server.mapping-path=/rsocket
  4. #logging.level.root=DEBUG
#spring.rsocket.server.port=7000
spring.rsocket.server.transport=websocket
spring.rsocket.server.mapping-path=/rsocket
#logging.level.root=DEBUG
pom.xml
  1. <dependency>
  2. <groupId>org.springframework.boot</groupId>
  3. <artifactId>spring-boot-starter-rsocket</artifactId>
  4. </dependency>
  5. <dependency>
  6. <groupId>org.springframework.boot</groupId>
  7. <artifactId>spring-boot-starter-webflux</artifactId>
  8. </dependency>
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

The client is a separate Spring Webflux application.

ClientConfiguration.java
  1. import io.rsocket.transport.netty.client.WebsocketClientTransport;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.context.annotation.Configuration;
  4. import org.springframework.messaging.rsocket.RSocketRequester;
  5. import org.springframework.messaging.rsocket.RSocketStrategies;
  7. import java.net.URI;
  9. @Configuration
  10. public class ClientConfiguration {
  12. @Bean
  13. public RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) {
  14. final URI uri = URI.create("ws://localhost:8080/rsocket");
  15. return RSocketRequester
  16. .builder()
  17. .rsocketStrategies(rSocketStrategies)
  18. .connect(WebsocketClientTransport.create(uri))
  19. .block();
  20. }
  21. }
import io.rsocket.transport.netty.client.WebsocketClientTransport;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;

import java.net.URI;

@Configuration
public class ClientConfiguration {

    @Bean
    public RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) {
        final URI uri = URI.create("ws://localhost:8080/rsocket");
        return RSocketRequester
                .builder()
                .rsocketStrategies(rSocketStrategies)
                .connect(WebsocketClientTransport.create(uri))
                .block();
    }
}
MessageController.java
  1. import com.kojotdev.blog.webfluxrsocketexample.domain.Message;
  2. import org.slf4j.Logger;
  3. import org.slf4j.LoggerFactory;
  4. import org.springframework.messaging.rsocket.RSocketRequester;
  5. import org.springframework.web.bind.annotation.PostMapping;
  6. import org.springframework.web.bind.annotation.RequestBody;
  7. import org.springframework.web.bind.annotation.RestController;
  8. import reactor.core.publisher.Mono;
  10. @RestController
  11. public class MessageController {
  13. private static final Logger log = LoggerFactory.getLogger(MessageController.class);
  15. public final RSocketRequester rSocketRequester;
  17. public MessageController(RSocketRequester rSocketRequester) {
  18. this.rSocketRequester = rSocketRequester;
  19. }
  21. @PostMapping("/message")
  22. public Mono<String> message(@RequestBody Message message) {
  23. log.info("got message in the rest request {}", message.message);
  24. return rSocketRequester
  25. .route("hello")
  26. .data(message)
  27. .retrieveMono(Message.class)
  28. .map(serverMessage -> serverMessage.message)
  29. .doOnNext(serverMsg -> log.info("server response in the client: {}", serverMsg));
  30. }
  31. }
import com.kojotdev.blog.webfluxrsocketexample.domain.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

@RestController
public class MessageController {

    private static final Logger log = LoggerFactory.getLogger(MessageController.class);

    public final RSocketRequester rSocketRequester;

    public MessageController(RSocketRequester rSocketRequester) {
        this.rSocketRequester = rSocketRequester;
    }

    @PostMapping("/message")
    public Mono<String> message(@RequestBody Message message) {
        log.info("got message in the rest request {}", message.message);
        return rSocketRequester
                .route("hello")
                .data(message)
                .retrieveMono(Message.class)
                .map(serverMessage -> serverMessage.message)
                .doOnNext(serverMsg -> log.info("server response in the client: {}", serverMsg));
    }
}

We declared an RSocketRequester bean to make a connection with the server. With this bean, we can easily send and retrieve the messages. The flow of our application looks like this:

    • post the message with the REST endpoint “/message”
    • send it to the server with the injected RSocketRequester bean
    • retrieve the response 
    • log the response and return it with the REST endpoint

Spring WebFlux doesn’t support WebSocket with STOMP, so the RSocket sounds very promising for solving the message mapping problem. The documentation is still very modest about this, but it looks like everything goes in the right direction.

WebFlux server - JavaScript client

It would be very nice to push the messages with the JavaScript client and the Spring WebFlux server. Unfortunately, I couldn’t figure it out how to do this. The problem lies in the metadata and routing. I found this issue in the GitHub project and hope they’re going to solve this soon.

2 Comments

  • Michael

    November 26, 2020 at 17:43 Reply

    Hi,
    to use rsocket-js with webflux server, you need to configure the mapping.
    In your example, you use hello entrypoint on server: @MessageMapping(“hello”).
    The same thing should be done on client side:
    1. Change connection endpoint:
    const transport = new RSocketWebSocketClient({
    url: “ws://localhost:8801/rsocket”
    });

    2. change serializers & metadataMimeType in RSocketClient:

    const client = new RSocketClient({
    // send/receive JSON objects instead of strings/buffers
    serializers: {
    data: JsonSerializer,
    metadata: IdentitySerializer
    },
    setup: {
    // other properties….
    // format of `data`
    dataMimeType: ‘application/json’,
    // format of `metadata`
    metadataMimeType: ‘message/x.rsocket.routing.v0’,
    },
    transport
    });

    3. Finally, set the same MessageMapping for a method:

    this.socket
    .requestResponse({
    data: message,
    metadata: String.fromCharCode(‘hello’.length) + ‘hello’
    })

    That’s all, now your web client can connect to webflux server.

Leave a Reply

Your email address will not be published.