Vert.xでリクエストチェーンを作る

javaでtcp proxy的なものを作りたくて今更ながらVert.xを触ってみた。
プロキシする前に（認証認可処理など）何かしらの処理を挟みたく、どうするか迷ったので解決策をメモする。

Vert.xとは

Vert.xは非同期I/Oをベースにした高性能なリアクティブフレームワーク。
複数のプログラミング言語に対応し、スケーラブルな分散アプリケーション構築が可能。
クラスタリングや高可用性もサポートし、マイクロサービスやリアルタイムWebアプリに最適で、多様なエコシステムで拡張性も高く、分散システムにしている（と言われている）。

Vert.xでリクエストチェーン

Vert.xではFuture/Promiseを使った非同期処理を書くことができる。
Futureをチェイニングすることが可能なので、以下のようにリクエストチェーンをかける。

import io.vertx.core.Vertx;
import io.vertx.core.Promise;
import io.vertx.core.Future;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;

public class RequestChainExample {
    
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();

        // NetClient の作成
        NetClient client = vertx.createNetClient();

        // Step 1: リクエスト先のサーバーに接続
        connectToServer(client, "server-host", 8080)
            // Step 2: 1個目の処理実行
            .compose(socket -> firstRequest(socket))
            // Step 3: 2個目の処理実行
            .compose(socket -> secondRequest(socket, "Hello Server!"))
            // 全ての操作が成功したら
            .onSuccess(response -> {
                //　成功した場合の処理
                System.out.println(response);
            })
            // エラーがあればキャッチ
            .onFailure(err -> {
                System.out.println("Failed: " + err.getMessage());
            });
    }

    // サーバーへの接続 (Promise)
    private static Future<NetSocket> connectToServer(NetClient client, String host, int port) {
        Promise<NetSocket> promise = Promise.promise();
        client.connect(port, host, res -> {
            if (res.succeeded()) {
                System.out.println("Connected to Server");
                promise.complete(res.result());
            } else {
                promise.fail(res.cause());
            }
        });
        return promise.future();
    }

    // 1個目の処理 (Promise)
    private static Future<NetSocket> firstRequest(NetSocket socket) {
        Promise<NetSocket> promise = Promise.promise();
        String message = "送信内容";
        socket.write(Buffer.buffer(message), writeRes -> {
            if (writeRes.succeeded()) {
                promise.complete(socket);
            } else {
                promise.fail(writeRes.cause());
            }
        });
        return promise.future();
    }

    // 2個目の処理 (Promise)
    private static Future<String> secondRequest(NetSocket socket, String message) {
        Promise<String> promise = Promise.promise();

        socket.write(Buffer.buffer(message), writeRes -> {
            if (writeRes.succeeded()) {
                socket.handler(buffer -> {
                    String response = buffer.toString();
                    promise.complete(response); // レスポンスを次に渡す
                });
            } else {
                promise.fail(writeRes.cause());
            }
        });
        return promise.future();
    }
}

ポイントは以下

• Promise と compose の使用
  • 各操作（接続、１つ目の処理、２つ目の処理）を Promise でラップし、それぞれの処理を compose を使ってシーケンシャルにチェーンしている
  • それぞれの操作が完了すると、次のステップに自動的に進む
• エラーハンドリング
  • 各 Future の操作に対して onFailure を設定し、問題が発生した場合に適切にエラーを処理する

便利！