RSocket & AsyncAPI

Primeiro, explica o método básico de uso do ponto de entrada /rsocket. O fluxo de geração de tipos de route/request/response e funções de serviço no AsyncAPI é a próxima forma opcional de integração do projeto.

Uso básico do adaptador RSocket.

Instale pacotes peer apenas em projetos que utilizam RSocket.

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

O uso básico não requer AsyncAPI generator ou WebFlux. Importe createRSocketApi de /rsocket e passe o endpoint WebSocket, e apenas se necessário, passe o JWT como segundo argumento. mono ou stream é retornado como um Observable RxJS.

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);
});
Opcional: Geração automática de serviço AsyncAPI para integração de projeto.

Fluxo quase idêntico ao REST

REST é swagger.jsonusado, e RSocket éasyncapi-rsocket.jsonusado. Leia o documento para gerar tipos TypeScript e funções de serviço a partir da rota, tipo de interação, esquema de requisição e resposta.

1. Primeiro: verifique o que é rsocketCommonService com o código completo.

rsocketCommonService.ts também não é um arquivo gerado automaticamente pela biblioteca. Ele envolve diretamente o createRSocketApi no projeto, centralizando a política de conexão/reuso de JWT, e faz com que o serviço RSocket gerado importe apenas callRSocketMono/callRSocketStream. Abaixo está o código completo desse arquivo. Ele lê a autorização do headerStore do serviço REST público, e se não houver, chama a API de token uma vez para preparar o JWT, e se o JWT mudar, fecha a conexão RSocket existente e cria uma nova.

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

Após a cópia, modifique o caminho de importação do headerStore, a localização do AsyncApiTypes, o caminho da API de token, NEXT_PUBLIC_RSOCKET_URL e o endpoint RSocket real de acordo com o projeto.

2. Funções do código do rsocketCommonService.

getRSocketUrl

No servidor, usa variáveis de ambiente e, no navegador, cria URLs ws ou wss com base no protocolo e host atuais.

fetchAuthorization

Quando não há JWT no headerStore, chama a API de token incluindo o cookie de sessão e lê os valores dos campos token, accessToken ou jwt.

getRsoketApi

Mesmo com várias solicitações simultâneas, compartilha a Promise de consulta do token e reutiliza a conexão existente para o mesmo JWT, criando uma nova conexão se o JWT mudar.

resetRsoketApi

Ao fazer logout ou remover informações de autenticação, fecha a conexão RSocket atual e inicializa o JWT e o cliente armazenados.

callRSocketMono / callRSocketStream

Verifica a solicitação e a resposta com o tipo de rota gerado e combina a Promise de preparação da conexão ao fluxo Observable RxJS para executar mono ou stream.

3. Criação de serviço 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. Chamada da função gerada.

a seguir é um exemplo comum de requestStream que recebe o progresso do tratamento de várias tarefas. A string de rota e o tipo de evento estão dentro do serviço gerado.

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

Escolha entre mono e stream

Se a interação AsyncAPI for requestResponsea função gerada serácallRSocketMonose, requestStreamse não, callRSocketStreamusará. Na tela, ambos são Observables, mas o stream pode ser chamado várias vezes. next.

© 2026 Byeolnaerim. Todos os direitos reservados.IntroduçãoPolítica de Privacidade