RSocketApplication protocol providing Reactive Streams semantics

Spec
Java
JavaScript
C++
Kotlin

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)
  • channel (bi-directional streams)

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.

Detailed information can be found in these documents:

Following is a brief example of a server and client in Java:

Example Java Server:

RSocketFactory.receive()
    .frameDecoder(Frame::retain)
    .acceptor(new PingHandler())
    .transport(TcpServerTransport.create(7878))
    .start()
    .block()
    .onClose();

Example Java Client:

Mono<RSocket> client =
    RSocketFactory.connect()
        .frameDecoder(Frame::retain)
        .transport(TcpClientTransport.create(7878))
        .start();

PingClient pingClient = new PingClient(client);

Recorder recorder = pingClient.startTracker(Duration.ofSeconds(1));

int count = 1_000;

pingClient
    .startPingPong(count, recorder)
    .doOnTerminate(() -> System.out.println("Sent " + count + " messages."))
    .blockLast();