RSocket & AsyncAPI

Pertama, menjelaskan cara dasar menggunakan titik masuk /rsocket secara langsung. Alur untuk menghasilkan tipe route/request/response dan fungsi layanan di AsyncAPI adalah metode integrasi proyek opsional berikutnya.

Cara dasar penggunaan adapter RSocket.

Hanya proyek yang menggunakan RSocket yang perlu menginstal paket peer.

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

Penggunaan dasar tidak memerlukan AsyncAPI generator atau WebFlux. Impor createRSocketApi dari /rsocket dan berikan endpoint WebSocket, serta hanya berikan JWT sebagai argumen kedua jika diperlukan. mono atau stream akan dikembalikan sebagai 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);
});
Opsional: Integrasi proyek pembuatan layanan AsyncAPI otomatis.

Alur yang hampir sama dengan REST

REST menggunakan swagger.jsonsedangkan RSocket menggunakanasyncapi-rsocket.jsonMembaca route, interaction type, request dan response schema dari dokumen untuk menghasilkan tipe TypeScript dan fungsi service.

1. Pertama: Periksa apa itu rsocketCommonService dengan kode lengkap.

rsocketCommonService.ts juga bukan file yang dihasilkan secara otomatis oleh pustaka. Menghubungkan createRSocketApi secara langsung di proyek dan menempatkan kebijakan penggunaan kembali JWT di satu tempat, sehingga layanan RSocket yang dihasilkan hanya mengimpor callRSocketMono/callRSocketStream. Berikut adalah kode lengkap dari file tersebut. Membaca Authorization dari headerStore layanan publik REST, dan jika tidak ada, memanggil token API sekali untuk menyiapkan JWT, dan jika JWT berubah, menutup koneksi RSocket yang ada dan membuat koneksi baru.

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

Setelah menyalin, sesuaikan jalur impor headerStore, lokasi AsyncApiTypes, jalur token API, NEXT_PUBLIC_RSOCKET_URL, dan endpoint RSocket yang sebenarnya sesuai dengan proyek.

2. Peran berdasarkan struktur kode rsocketCommonService.

getRSocketUrl

Di server, menggunakan variabel lingkungan dan di browser, membuat URL ws atau wss berdasarkan protokol dan host saat ini.

fetchAuthorization

Memanggil token API dengan menyertakan cookie sesi saat tidak ada JWT di headerStore dan membaca nilai dari field token, accessToken, atau jwt.

getRsoketApi

Membagikan Promise pencarian token meskipun ada beberapa permintaan yang masuk secara bersamaan, menggunakan kembali koneksi yang ada untuk JWT yang sama, dan membuat koneksi baru jika JWT berubah.

resetRsoketApi

Saat logout atau menghapus informasi otentikasi, menutup koneksi RSocket saat ini dan menginisialisasi JWT dan klien yang disimpan.

callRSocketMono / callRSocketStream

Memeriksa request dan response dengan tipe route yang dihasilkan, dan menggabungkan Promise persiapan koneksi ke dalam aliran Observable RxJS untuk menjalankan mono atau stream.

3. Pembuatan layanan 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. Memanggil fungsi yang dihasilkan.

Berikut adalah contoh umum requestStream yang menerima kemajuan pemrosesan dari beberapa pekerjaan. String route dan tipe event ada di dalam service yang dihasilkan.

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

Jika interaction AsyncAPI adalah

maka fungsi yang dihasilkan adalah requestStreamjika,callRSocketStreammaka nextakan menggunakan. 3. Memanggil fungsi yang dihasilkanDi layar, keduanya adalah Observable tetapi stream dapat dipanggil beberapa kali. Streaming & UploadrequestResponse

© 2026 Byeolnaerim. Semua hak dilindungi.PengenalanKebijakan Penanganan Data Pribadi