RSocket & AsyncAPI

まず、/rsocketエントリーポイントを直接使用する基本的な方法について説明します。AsyncAPIでroute/request/responseタイプとサービス関数を生成する流れは、その後のオプションのプロジェクト統合方式です。

RSocketアダプターの基本的な使用法

RSocketを使用するプロジェクトでのみピアパッケージをインストールします。

bash
npm i rsocket-core rsocket-types rsocket-websocket-client

基本的な使用にはAsyncAPIジェネレーターやWebFluxは必要ありません。/rsocketからcreateRSocketApiをインポートし、WebSocketエンドポイントを渡し、認証が必要な場合のみ2番目の引数としてJWTを渡します。monoまたはstreamはRxJS Observableとして返されます。

ts
import { createRSocketApi } from "@byeolnaerim/typed-rx-http/rsocket";

const api = createRSocketApi("ws://localhost:7000/rsocket", jwt);

api.mono<User>("user.findById", { id: 1 }).subscribe((user) => {
  console.log(user);
});

api.stream<Notification>("notification.stream").subscribe((notification) => {
  console.log(notification);
});
オプション: AsyncAPIサービスの自動生成プロジェクト統合

RESTとほぼ同じ流れ

RESTは swagger.jsonを使用し、RSocketはasyncapi-rsocket.jsonを使用します。文書からルート、インタラクションタイプ、リクエストとレスポンススキーマを読み取り、TypeScriptタイプとサービス関数を生成します。

1. まず: rsocketCommonServiceが何であるかを全コードで確認

rsocketCommonService.tsもライブラリが自動生成するファイルではありません。createRSocketApiをプロジェクトで直接ラップして接続/JWT再利用ポリシーを一箇所にまとめ、生成されたRSocketサービスがcallRSocketMono/callRSocketStreamのみをインポートするようにするプロジェクト所有のアダプターです。以下はそのファイルの全コードです。REST共通サービスのheaderStoreからAuthorizationを読み込み、存在しない場合はtoken APIを一度呼び出してJWTを準備し、JWTが変更された場合は既存のRSocket接続を閉じて新しい接続を生成します。

rsocketCommonService.ts
ts
'use client';
import {
    createRSocketApi,
    type RSocketApi,
} from '@byeolnaerim/typed-rx-http/rsocket';

import { defer, from, Observable } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

import { headerStore } from '@ui/handler/service/rxjsHttpService';
import type {
    AsyncApiRSocketMonoRoute,
    AsyncApiRSocketOperations,
    AsyncApiRSocketRequest,
    AsyncApiRSocketResponse,
    AsyncApiRSocketStreamRoute,
} from './@types/auto/AsyncApiTypes';

const getRSocketUrl = () => {
    if (typeof window === 'undefined') {
        return (
            process.env.NEXT_PUBLIC_RSOCKET_URL ||
            'ws://localhost:8788/api/rsocket'
        );
    }

    return `${window.location.protocol === 'https:' ? 'wss:' : 'ws:'}//${
        window.location.host.startsWith('localhost') ||
        window.location.host.startsWith('127.0.0.1')
            ? 'localhost:8788'
            : window.location.host
    }/api/rsocket`;
};

const getJwtFromAuthorization = (
    authorization?: string
): string | undefined => {
    const jwt = authorization?.replace(/^Bearer\s+/i, '').trim();

    return jwt || undefined;
};

const fetchAuthorization = async (): Promise<string | undefined> => {
    const response = await fetch('/api/auth/token', {
        method: 'GET',
        credentials: 'include',
    });

    if (!response.ok) {
        headerStore.remove('Authorization');
        return undefined;
    }

    const json = (await response.json()) as {
        token?: string;
        accessToken?: string;
        jwt?: string;
    };
    const token = json.token ?? json.accessToken ?? json.jwt;
    const jwt = getJwtFromAuthorization(token);

    if (!jwt) {
        headerStore.remove('Authorization');
        return undefined;
    }

    return jwt;
};

let rsoketApi: RSocketApi<AsyncApiRSocketOperations> | null = null;
let currentJwt: string | undefined;
let pendingAuthorization: Promise<string | undefined> | null = null;

export const getRsoketApi = async (): Promise<
    RSocketApi<AsyncApiRSocketOperations>
> => {
    let authorization: string | undefined = headerStore.get()['Authorization'];
    let jwt = getJwtFromAuthorization(authorization);

    if (!jwt) {
        if (!pendingAuthorization) {
            pendingAuthorization = fetchAuthorization().finally(() => {
                pendingAuthorization = null;
            });
        }

        authorization = await pendingAuthorization;
        jwt = getJwtFromAuthorization(authorization);
    }

    if (!jwt) {
        resetRsoketApi();
        throw new Error('RSocket JWT is missing.');
    }

    if (!rsoketApi || currentJwt !== jwt) {
        rsoketApi?.close();

        currentJwt = jwt;
        rsoketApi = createRSocketApi<AsyncApiRSocketOperations>(
            getRSocketUrl(),
            jwt
        );
    }

    return rsoketApi;
};

export const resetRsoketApi = (): void => {
    rsoketApi?.close();
    rsoketApi = null;
    currentJwt = undefined;
};

export const callRSocketMono = <TRoute extends AsyncApiRSocketMonoRoute>(
    route: TRoute,
    body?: AsyncApiRSocketRequest<TRoute>
): Observable<AsyncApiRSocketResponse<TRoute>> => {
    return defer(() =>
        from(getRsoketApi()).pipe(mergeMap(api => api.mono(route, body)))
    );
};

export const callRSocketStream = <TRoute extends AsyncApiRSocketStreamRoute>(
    route: TRoute,
    body?: AsyncApiRSocketRequest<TRoute>
): Observable<AsyncApiRSocketResponse<TRoute>> => {
    return defer(() =>
        from(getRsoketApi()).pipe(mergeMap(api => api.stream(route, body)))
    );
};

コピー後、headerStoreインポートパス、AsyncApiTypesの位置、トークンAPIパス、NEXT_PUBLIC_RSOCKET_URLと実際のRSocketエンドポイントをプロジェクトに合わせて修正します。

2. rsocketCommonServiceコード構成別の役割

getRSocketUrl

サーバーでは環境変数を使用し、ブラウザでは現在のプロトコルとホストを基にwsまたはwss URLを作成します。

fetchAuthorization

headerStoreにJWTがない場合、セッションクッキーを含めてトークンAPIを呼び出し、トークン、accessTokenまたはjwtフィールドから値を読み取ります。

getRsoketApi

同時に複数のリクエストが来てもトークン取得Promiseを共有し、同じJWTには既存の接続を再利用し、JWTが変更されると接続を再作成します。

resetRsoketApi

ログアウトや認証情報の削除時に現在のRSocket接続を閉じ、保存されたJWTとクライアントを初期化します。

callRSocketMono / callRSocketStream

生成されたルートタイプでリクエストとレスポンスを検査し、接続準備PromiseをRxJS Observableフローに統合してmonoまたはstreamを実行します。

3. AsyncAPIサービス生成

generateAsyncApiRSocket.cjs
js
const {
  generateAsyncApiRSocketFromHttp,
} = require("@byeolnaerim/typed-rx-http/auto-node-script/asyncapi-rsocket");

generateAsyncApiRSocketFromHttp({
  hostname: "localhost",
  port: 8123,
  path: "/for-local/get-asyncapi-rsocket",
  asyncApiRSocketFile: "./asyncapi-rsocket.json",
  serviceDir: "./src/handler/service",
  typesDir: "./src/handler/service/@types",
  typesFile:
    "./src/handler/service/@types/AsyncApiTypes.ts",
  commonRSocketServiceFile:
    "./src/handler/service/rsocketCommonService.ts",
});

4. 生成された関数呼び出し

次は複数の作業の処理進行状況を受け取る一般的なrequestStreamの例です。ルート文字列とイベントタイプは生成されたサービス内に含まれています。

BatchJobPage.tsx
tsx
import { processBatchJobs } from "@handler/service/auto/BatchJobRSocketService";

const subscription = processBatchJobs({
  body: targets,
}).subscribe({
  next: (event) => {
    setProcessedCount(event.processedCount || 0);
    setTotalCount(event.totalCount || 0);
  },
  error: (error) => {
    setErrorMessage(error.message);
  },
  complete: () => {
    setIsSubmitting(false);
  },
});

return () => subscription.unsubscribe();

monoとstream選択

AsyncAPIインタラクションが requestResponseの場合、生成関数はcallRSocketMonoを、 requestStream場合は callRSocketStreamを使用します。画面では両方ともObservableですが、streamは複数回 nextが呼び出される可能性があります。

© 2026 Byeolnaerim. 無断転載を禁じます。紹介プライバシーポリシー