RSocket 與 AsyncAPI

首先說明直接使用 /rsocket 入口點的基本方法。AsyncAPI 中生成 route/request/response 類型和服務函數的流程是接下來的選擇性專案整合方式。

RSocket adapter 基本用法

僅在使用 RSocket 的項目中安裝對等包。

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

基本使用不需要 AsyncAPI generator 或 WebFlux。在 /rsocket 中導入 createRSocketApi，傳遞 WebSocket 端點，並在需要身份驗證的情況下將 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 的專案擁有 adapter。以下是該文件的完整代碼。從 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位置、token API路徑、NEXT_PUBLIC_RSOCKET_URL和實際RSocket端點。

2. rsocketCommonService 代碼結構的角色

getRSocketUrl

伺服器使用環境變數，瀏覽器根據當前協議和主機生成ws或wss URL。

fetchAuthorization

當headerStore中沒有JWT時，調用token API並包含會話cookie，從token、accessToken或jwt字段中讀取值。

getRsoketApi

即使同時有多個請求進來，也共享token查詢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. 調用生成的函數

以下是接收多個任務處理進度的一般請求流示例。路由字串和事件類型包含在生成的服務中。

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();

如果 AsyncAPI 互動是

則生成的函數是 requestResponse，callRSocketStream如果是 requestStream則使用 3. 調用生成的函數。在畫面上，兩者都是 Observable，但 stream 可以被多次 next調用。

© 2026 Byeolnaerim. 版權所有。介紹隱私政策