Spring Fest 2019 の「RSocket 徹底入門〜Spring 5.2 の目玉機能である RSocket 対応とは〜」で RSocket に興味を持って実験してみた結果をメモっとく。

本当にこの記事は実験メモなので、コードは参考程度にしてもらって、実用するときはきちんと調べてから利用してほしい。

背景

Give REST a Rest with RSocket（邦訳：RSocket で REST に安息 (Rest) を）のような記事を読んで、REST による通信だけだと実際厳しいよなと同意することが最近多いので、RSocket を試してみようじゃないかという気分になった。サーバー側を CQRS で構成したときに、クライアント側と REST で通信するのがイケてない気がしているというのもある。

概要

RSocket サーバーは Spring Boot で実現している。RSocket クライアントは CLI クライアント (rsocket-cli / rsc) と rsocket-js によるブラウザークライアントで実現している。RScoket の Interaction Model としては Request Response と Request Stream を試してみた。また、Routing Metadata Extension を使用している。ペイロードは JSON 形式で表現している。

Spring Boot サーバー間の通信は試していないので、その実験をしたい人は Spring Tips: RSocket Messaging in Spring Boot 2.2 などを参考にするといいと思う。この記事でもサーバー側の実装の参考にしている。また、RScoket の Interaction Model にある Request Channel と Fire and Forget も試していない。

Spring Boot で RSocket サーバーを実装する

Spring Boot プロジェクトを作成する。WebFlux、RSocket、Lombok を依存関係として追加する。

$ curl https://start.spring.io/starter.tgz \
    -d type=gradle-project \
    -d baseDir=rsocket \
    -d dependencies=webflux,rsocket,lombok \
| tar xz

application.properties に RSocket のプロトコルとパスの設定を追加する。プロトコルは WebSocket を使用する。プロトコルのデフォルトは TCP なのだが、ブラウザーから RSocket を利用するときは WebSocket に設定する。パスは /rsocket にする。これらの設定を入れておくと HTTP と WebSocket を同じポート番号で利用できる。ここでは明示的にポートを指定していないので8080番になる。

rsocket/src/main/resources/application.properties

spring.rsocket.server.transport=websocket
spring.rsocket.server.mapping-path=/rsocket

DemoApplication.java に GreetingController クラス、GreetingRequest クラス、GreetingResponse クラスを追加する。

rsocket/src/main/java/DemoApplication.java

package com.example.demo;

import lombok.Data;
import lombok.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.Instant;
import java.util.stream.Stream;

@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

}

@Controller
class GreetingController {
    @MessageMapping("greet")
    GreetingResponse greet(GreetingRequest request) {
        return new GreetingResponse("Hi " + request.getName());
    }

    @MessageMapping("greet-stream")
    Flux<GreetingResponse> greetStream(GreetingRequest request) {
        return Flux
                .fromStream(Stream.generate(() -> new GreetingResponse("Hi " + request.getName())))
                .delayElements(Duration.ofSeconds(1));
    }
}

@Data
class GreetingRequest {
    String name;
}

@Value
class GreetingResponse {
    String greeting;
    Instant timestamp = Instant.now();
}

GreetingController クラスは、RSocket 通信を処理するクラス。@MessageMapping アノテーションが付与されたメソッドが Routing Metadata のタグに応じて RSocket 通信を処理する。

greet メソッドは Request Response の通信を処理する。リクエストボディの JSON を GreetingRequest クラスのインスタンスにマッピングして、その name フィールドの値を用いて GreetingResponse クラスのインスタンスを作成して返している。

greetStream メソッドは Request Stream の通信を処理する。リクエストボディの JSON を GreetingRequest クラスのインスタンスにマッピングして、その name フィールドの値を用いて GreetingResponse クラスのインスタンスを1秒ごとに生成するストリームを返している。

GreetingRequest クラスは、リクエストボディの JSON をマッピングするためのクラス。@Data アノテーションの代わりに @Value アノテーションを付与したいときは、lombok.config に lombok.anyConstructor.addConstructorProperties = true を設定すること。

GreetingResponse クラスは、レスポンスボディの JSON をマッピングするためのクラス。レスポンスごとの差異がわかりやすいようにタイムスタンプを生成している。

RSocket 通信を CLI クライアントで検証する

RSocket の CLI クライアントを使用してサーバーの動作を検証してみる。

rsocket-cli コマンドで検証する

RSocket の公式リポジトリにある rsocket-cli コマンドをインストールする。

$ brew install yschimke/tap/rsocket-cli

Request Response 通信の動作を検証する。

$ rsocket-cli --request ws://localhost:8080/rsocket \
    --metadataFormat "message/x.rsocket.routing.v0" \
    --metadata $(printf "\x05greet")                \
    --dataFormat "application/json"                 \
    --input '{ "name": "takuya" }' --debug

request オプションで Request Response 通信を指定している。WebSocket プロトコルを指定して localhost の8080番ポートに接続する。metadataFormat オプションで Routing Metadata Extension の MIME タイプを指定している。metadata オプションで Routing Metadata Extension のタグ長（固定長：1バイト）とタグ（可変長）を指定している。タグが greet の5バイトなので、データの先頭に \x05 と記載して、printf コマンドでバイナリとして正しい値になるように変換している。dataFormat オプションでペイロードの MIME タイプを指定している。input オプションで JSON 文字列を指定している。debug オプションでリクエストとレスポンスのデバッグ情報を出力するようにしている。

Request Stream 通信の動作を検証する。

$ rsocket-cli --stream ws://localhost:8080/rsocket  \
    --metadataFormat "message/x.rsocket.routing.v0" \
    --metadata $(printf "\x0cgreet-stream")         \
    --dataFormat "application/json"                 \
    --input '{ "name": "takuya" }' --debug

Request Response 通信とほとんど同じだが、いくつかのオプションが異なっている。stream オプションで Request Stream 通信を指定している。metadata オプションで指定するタグが greet-stream の12バイトなので、データの先頭に \x0c と記載して、printf コマンドでバイナリとして正しい値になるように変換している。

rsc コマンドで検証する

rsocket-cli コマンドより使い勝手のいい rsc コマンドをインストールする。

$ curl -L -o /usr/local/bin/rsc https://github.com/making/rsc/releases/download/0.4.2/rsc-osx-x86_64
$ chmod +x /usr/local/bin/rsc

Request Response 通信の動作を検証する。

$ rsc --request ws://localhost:8080/rsocket \
    --route greet                           \
    --data '{ "name": "takuya" }' --debug

request オプションで Request Response 通信を指定している。WebSocket プロトコルを指定して localhost の8080番ポートに接続する。route オプションで Routing Metadata Extension のタグを指定している。data オプションで JSON 文字列を指定している。debug オプションでリクエストとレスポンスのデバッグ情報を出力するようにしている。

Request Stream 通信の動作を検証する。

$ rsc --stream ws://localhost:8080/rsocket \
    --route greet-stream                   \
    --data '{ "name": "takuya" }' --debug

Request Response 通信とほとんど同じだが、いくつかのオプションが異なっている。stream オプションで Request Stream 通信を指定している。route オプションで指定するタグを greet-stream にしている。

参考までに rsc コマンドの debug オプションなしの実行結果を掲載しておく。

$ rsc --request ws://localhost:8080/rsocket \
>     --route greet                         \
>     --data '{ "name": "takuya" }'
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:27:54.254838Z"}
$ rsc --stream ws://localhost:8080/rsocket \
>     --route greet-stream                 \
>     --data '{ "name": "takuya" }'
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:02.159310Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:03.168169Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:04.174156Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:05.178946Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:06.180451Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:07.208541Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:08.214503Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:09.219728Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:10.228279Z"}
{"greeting":"Hi takuya","timestamp":"2020-01-13T09:28:11.230968Z"}

RSocket 通信を rsocket-js によるブラウザークライアントで検証する

RSocket の JavaScript 実装である rsocket-js を用いたブラウザークライアントを実装してサーバーとの動作を検証してみる。

Yarn をインストールする

パッケージマネージャーの Yarn をインストールする。アプリケーションバンドラー Parcel 2 の資料に Yarn の手順しかなかったので、今回は Yarn を使用してみた。

$ brew install yarn

Yarn で新規プロジェクトを作成する

Yarn で rsocket-js を試すための新規プロジェクトを作成する。

$ mkdir rsocket-client
$ cd rsocket-client
$ yarn init -y

Parcel 2 は記事公開時点ではまだアルファ版なので、@next をパッケージ名に付与して追加する。

$ yarn add --dev parcel@next

rsocket-js ライブラリからブラウザーで動作させるための rsocket-websocket-client を追加する。

$ yarn add rsocket-websocket-client

Request Response 通信の動作を検証するブラウザークライアントを実装する

エントリーファイルになる request.html を作成する。

rsocket-client/request.html

<!DOCTYPE html>
<title>RSocket: Request Response</title>
<script src="request.js"></script>
<pre id="output"></pre>

RSocket の Request Response 通信を実行する request.js を作成する。とりあえず、async / await を使用せず、わりと昔ながらの方法で書いてみる。

rsocket-client/request.js

import {RSocketClient} from "rsocket-core";
import RSocketWebSocketClient from "rsocket-websocket-client";

window.addEventListener("load", function() {
    var client = new RSocketClient({
        setup: {
            metadataMimeType: "message/x.rsocket.routing.v0",
            dataMimeType: "application/json"
        },
        transport: new RSocketWebSocketClient({
            url: "ws://localhost:8080/rsocket"
        })
    });

    client
        .connect()
        .subscribe({
            onComplete: function(rsocket) {
                rsocket
                    .requestResponse({
                        metadata: "\x05greet",
                        data: '{ "name": "takuya" }'
                    })
                    .subscribe({
                        onComplete: function(response) {
                            document.getElementById("output").innerHTML += response.data + "\n";
                        }
                    });
            }
        });
});

これを async / await を使用して、わりと最近の方法で書き直してみる。こちらの方がわかりやすい。

rsocket-client/request.js

import {RSocketClient} from "rsocket-core";
import RSocketWebSocketClient from "rsocket-websocket-client";

window.addEventListener("load", async () => {
    const client = new RSocketClient({
        setup: {
            metadataMimeType: "message/x.rsocket.routing.v0",
            dataMimeType: "application/json"
        },
        transport: new RSocketWebSocketClient({
            url: "ws://localhost:8080/rsocket"
        })
    });

    const rsocket = await client.connect();

    const response = await rsocket.requestResponse({
        metadata: "\x05greet",
        data: '{ "name": "takuya" }'
    });

    document.getElementById("output").innerHTML += response.data + "\n";
});

これを Parcel で実行する。

$ yarn parcel request.html

参考までに Safari での実行結果のスクリーンショットを掲載しておく。

Request Stream 通信の動作を検証するブラウザークライアントを実装する

エントリーファイルになる stream.html を作成する。

rsocket-client/stream.html

<!DOCTYPE html>
<title>RSocket: Request Stream</title>
<script src="stream.js"></script>
<pre id="output"></pre>

RSocket の Request Stream 通信を実行する stream.js を作成する。

rsocket-client/stream.js

import {RSocketClient, MAX_STREAM_ID} from "rsocket-core";
import RSocketWebSocketClient from "rsocket-websocket-client";

window.addEventListener("load", async () => {
    const client = new RSocketClient({
        setup: {
            metadataMimeType: "message/x.rsocket.routing.v0",
            dataMimeType: "application/json"
        },
        transport: new RSocketWebSocketClient({
            url: "ws://localhost:8080/rsocket"
        })
    });

    const rsocket = await client.connect();

    rsocket
        .requestStream({
            data: '{ "name": "takuya" }',
            metadata: "\x0cgreet-stream"
        })
        .subscribe({
            onSubscribe(subscription) {
                subscription.request(MAX_STREAM_ID);
            },
            onNext(payload) {
                document.getElementById("output").innerHTML += payload.data + "\n";
            }
        });
});