RSocket 与 AsyncAPI

首先解释直接使用 /rsocket 入口点的基本方法。AsyncAPI 中生成 route/request/response 类型和服务函数的流程是接下来的可选项目集成方式.

RSocket 适配器基本用法.

仅在使用 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 的项目所有适配器。以下是该文件的完整代码。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. 调用生成的函数.

以下是接收多个任务处理进度的一般 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();

如果 AsyncAPI 交互是

则生成的函数是 requestStream，callRSocketStream如果是 next则使用 3. 调用生成的函数。在界面上，两者都是 Observable，但 stream 可以被多次调用。 下一个requestResponse

© 2026 Byeolnaerim. 保留所有权利.介绍隐私政策