RSocket & AsyncAPI

First, explains the basic method of directly using the /rsocket entry point. The flow of generating route/request/response types and service functions in AsyncAPI is the next optional project integration method.

Basic usage of the RSocket adapter

Install the peer packages only in projects that use RSocket.

bash

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

Basic usage does not require AsyncAPI generator or WebFlux. Import createRSocketApi from /rsocket and pass the WebSocket endpoint, and only pass JWT as the second argument if authentication is required. Mono or stream is returned as an RxJS Observable.

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 auto generation is optional.

The following AsyncAPI flow is only used when you want to automatically generate route strings and request/response types to connect to the project-wide RSocket wrapper. It is not a basic usage condition of createRSocketApi itself.

Optional: AsyncAPI service auto generation project integration.

Almost the same flow as REST.

REST uses swagger.jsonwhile RSocket usesasyncapi-rsocket.jsonRead the route, interaction type, request, and response schema from the documentation to generate TypeScript types and service functions.

webflux-fe-dev-assistant is optional.

In WebFlux functional endpoints, you can create AsyncAPI documentation with webflux-fe-dev-assistant. Since typed-rx-http reads the generated documentation, it can work with the necessary AsyncAPI structure even if created with other tools. x-rsocket information match.

1. First: Check what rsocketCommonService is with the full code.

rsocketCommonService.ts is also not a file automatically generated by the library. It wraps createRSocketApi directly in the project to centralize connection/JWT reuse policies, allowing the generated RSocket service to only import callRSocketMono/callRSocketStream. Below is the full code of that file. It reads Authorization from the headerStore of the REST common service, calls the token API once to prepare the JWT if it is missing, and closes the existing RSocket connection and creates a new one if the JWT changes.

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

After copying, modify the headerStore import path, AsyncApiTypes location, token API path, NEXT_PUBLIC_RSOCKET_URL, and the actual RSocket endpoint to fit your project.

2. Role of the rsocketCommonService code structure.

getRSocketUrl

On the server, environment variables are used, and in the browser, it creates ws or wss URLs based on the current protocol and host.

fetchAuthorization

When there is no JWT in headerStore, it calls the token API including the session cookie and reads values from the token, accessToken, or jwt fields.

getRsoketApi

Even with multiple requests coming in simultaneously, it shares the token retrieval Promise, reuses existing connections for the same JWT, and creates a new connection if the JWT changes.

resetRsoketApi

On logout or when removing authentication information, it closes the current RSocket connection and resets the stored JWT and client.

callRSocketMono / callRSocketStream

Checks request and response with the generated route type, and merges the connection preparation Promise into the RxJS Observable flow to execute mono or stream.

3. AsyncAPI service generation.

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

Meaning of commonRSocketServiceFile.

This option does not generate the previously created rsocketCommonService.ts. It specifies the actual path of the project-owned file that the generated service will import callRSocketMono and callRSocketStream.

4. Calling the generated function.

The following is a typical requestStream example that receives the progress of multiple task processes. The route string and event type are included in the generated service.

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

Choose between mono and stream.

If the AsyncAPI interaction is requestResponsethe created function usescallRSocketMono. requestStreamIf it is callRSocketStreamit uses. nextOn the screen, both are Observable, but the stream can be called multiple times.