RSocket y AsyncAPI

Primero, explica el método básico de uso directo del punto de entrada /rsocket. El flujo para generar tipos de ruta/petición/respuesta y funciones de servicio en AsyncAPI es el siguiente método de integración de proyecto opcional.

Uso básico del adaptador RSocket.

Solo se instalan paquetes peer en proyectos que utilizan RSocket.

bash

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

El uso básico no requiere AsyncAPI generator ni WebFlux. Importa createRSocketApi desde /rsocket y pasa el endpoint de WebSocket, y solo si es necesario, pasa el JWT como segundo argumento. mono o stream se devuelve como un Observable de RxJS.

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

La generación automática de AsyncAPI es opcional.

El flujo de AsyncAPI a continuación solo se utiliza si deseas generar automáticamente la cadena de ruta y los tipos de petición/respuesta para conectar con el wrapper RSocket público del proyecto. No es un requisito básico para el uso de createRSocketApi.

Opcional: Integración de proyecto para la generación automática de servicios AsyncAPI.

Flujo casi idéntico a REST

REST es swagger.jsonutiliza, y RSocket utilizaasyncapi-rsocket.jsonSe leen las rutas, tipos de interacción, y esquemas de solicitud y respuesta del documento para generar tipos de TypeScript y funciones de servicio.

webflux-fe-dev-assistant es opcional

En un endpoint funcional de WebFlux, se puede crear documentación AsyncAPI con webflux-fe-dev-assistant. typed-rx-http lee la documentación generada, por lo que, incluso si se creó con otra herramienta, se necesita la estructura AsyncAPI requerida y x-rsocket la información es la misma, incluso si se creó con otra herramienta.

1. Primero: verifica qué es rsocketCommonService con el código completo.

rsocketCommonService.ts tampoco es un archivo generado automáticamente por la biblioteca. Envuelve createRSocketApi directamente en el proyecto para centralizar la conexión/política de reutilización de JWT, y hace que el servicio RSocket generado solo importe callRSocketMono/callRSocketStream. A continuación se muestra el código completo de ese archivo. Lee el Authorization del headerStore del servicio REST público, y si no está presente, llama una vez a la API de token para preparar el JWT, y si el JWT cambia, cierra la conexión RSocket existente y crea una nueva conexión.

rsocketCommonService.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)))
    );
};

Después de copiar, modifica la ruta de importación de headerStore, la ubicación de AsyncApiTypes, la ruta de la API de token, NEXT_PUBLIC_RSOCKET_URL y el endpoint RSocket real según el proyecto.

2. Roles según la estructura del código de rsocketCommonService.

getRSocketUrl

En el servidor, se utilizan variables de entorno y en el navegador se crea una URL ws o wss basada en el protocolo y host actuales.

fetchAuthorization

Cuando no hay JWT en headerStore, se llama a la API de token incluyendo la cookie de sesión y se leen los valores de los campos token, accessToken o jwt.

getRsoketApi

Comparte la promesa de consulta de token incluso si llegan múltiples solicitudes al mismo tiempo, reutiliza la conexión existente para el mismo JWT y crea una nueva conexión si el JWT cambia.

resetRsoketApi

Al cerrar sesión o eliminar credenciales, cierra la conexión RSocket actual y reinicia el JWT y el cliente almacenados.

callRSocketMono / callRSocketStream

Verifica la solicitud y respuesta con el tipo de ruta generado, y combina la promesa de preparación de conexión en el flujo Observable de RxJS para ejecutar mono o stream.

3. Creación de servicio AsyncAPI.

generateAsyncApiRSocket.cjs

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",
});

Significado de commonRSocketServiceFile

Esta opción no generará rsocketCommonService.ts que se creó anteriormente. Especifica la ruta real del archivo de propiedad del proyecto que importará callRSocketMono y callRSocketStream.

4. Llamada a la función generada.

A continuación se presenta un ejemplo común de requestStream que recibe el progreso del manejo de múltiples tareas. La cadena de ruta y el tipo de evento están incluidos en el servicio generado.

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

Si la interacción de AsyncAPI es

la función generada es requestResponsela,callRSocketStreamsi es requestStreamse utiliza. 3. Llamar a la función generadaEn la pantalla, ambos son Observable, pero el stream puede ser llamado varias veces. nextpuede ser llamado.