快速开始

适用于TypeScript的基于RxJS的类型安全HTTP客户端 + （可选）Next.js/RSocket适配器。

核心特性

通过注入从Swagger/OpenAPI / AsyncAPI派生的模式/自定义契约等获得的OpenAPI风格的Paths类型（习惯上称为paths），确保路由（请求）类型的安全性。
所有API返回RxJS Observable
核心是框架独立的（不依赖于Next.js）
Next.js专用功能分离到@byeolnaerim/typed-rx-http/next入口点 → 仅在导入/next时需要Next.js
RSocket专用功能分离到@byeolnaerim/typed-rx-http/rsocket入口点 → 仅在导入/rsocket时需要RSocket包

安装

npm

bash

npm i @byeolnaerim/typed-rx-http rxjs

入口点

Core（框架独立）

import {
  createHttpClient,
  createHeaderStore,
  createCsrCache,
  createSessionAuth,
  HttpResponseError,
  isHttpResponseError,
  type ServiceArguments,
  type CacheForService,
} from "@byeolnaerim/typed-rx-http";

Next.js适配器（可选）

import {
  callApiSsrCache,
  redirectToUnauthorizedOnServer401,
} from "@byeolnaerim/typed-rx-http/next";

在不使用Next.js的项目中，请勿导入/next。

RSocket适配器（可选）

仅在使用RSocket的项目中安装peer包。

bash

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

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

在不使用RSocket的项目中，请勿导入/rsocket。

核心用法

1) 准备Paths类型（通常是OpenAPI paths）

createHttpClient<Paths>()的Paths表示“请求规范（路由）”的类型。文档中习惯上称为paths，但不一定需要是OpenAPI/Swagger，也不一定要命名为paths。

不过，核心内部使用OpenApiPathsLike约束，因此，Paths应类似于OpenAPI paths的形式。

顶级键：URL路径字符串（例如："/users/{id}"）
子键：HTTP方法（get/post/put/delete/patch …）
每个方法中存在parameters.query/path/header/cookie、requestBody、responses等字段（或never）

核心主要参考上述结构中的以下字段来构建ServiceArguments的类型。

url: keyof Paths
method: keyof Paths[url]
queryString: parameters.query
pathVariable: parameters.path
body: requestBody

例如：openapi-typescript 输出通常符合以下规范（部分简化）。

export interface paths {
  "/test/get-hello-world": {
    parameters: {
      query?: never;
      header?: never;
      path?: never;
      cookie?: never;
    };
    get: {
      parameters: {
        query?: never;
        header?: never;
        path?: never;
        cookie?: never;
      };
      requestBody?: never;
      responses: {
        200: {
          content: {
            "application/json": components["schemas"]["TestResponse"];
          };
        };
      };
    };
  };
}

即使不是 OpenAPI 也可以使用

即使不是 OpenAPI，只要以兼容的形式定义类型（或创建转换类型进行映射），也可以直接使用。

// aliasing: OpenAPI `paths` -> `Paths`
import type { paths as Paths } from "./@types/ApiTypes";

2) 创建 HeaderStore

HeaderStore 是一个简单的内存存储，用于管理 CSR 中的基本头部。

import { createHeaderStore } from "@byeolnaerim/typed-rx-http";

export const headerStore = createHeaderStore({
  "Content-Type": "application/json",
});

3) 创建 HTTP 客户端

headerStore 是可选的，但建议在 CSR 中使用基本头部/会话认证时添加。
headersProvider 在需要每个请求计算头部时使用，例如 SSR/多租户。

import { createHttpClient } from "@byeolnaerim/typed-rx-http";
import type { paths as Paths } from "./@types/ApiTypes";

const apiUrl = process.env.NEXT_PUBLIC_API_URL!;

export const client = createHttpClient<Paths>({
  baseUrl: apiUrl,
  headerStore,
});

4) API 调用（类型安全）

请求类型（url/method/pathVariable/queryString/body）由用户注入到 createHttpClient<Paths>() 中的类型（通常是 OpenAPI paths）决定。响应类型由调用者在 callApi<R>() 中选择的泛型 R 决定（核心不会从 responses 中自动推断）。

export const getAdminList = () => {
  return client.callApi<{ items: any[]; total: number }>({
    url: "/api/account/search/admin/get-list",
    method: "get",
    queryString: { pageNumber: 1, pageSize: 20 },
  });
};

响应包装（ResponseWrapper）— 可选

该库不强制响应包装。可以根据 API 选择泛型响应形式。

包装的响应

type ResponseWrapper<T> = { data: T; message?: string; resultType?: string };

export const bidCreate = (body: unknown) => {
  return client.callApi<ResponseWrapper<{ requestId: string }>>({
    url: "/api/bid/create/bid/request",
    method: "post",
    body,
  });
};

未包装的响应

export const bidCreate = (body: unknown) => {
  return client.callApi<{ requestId: string }>({
    url: "/api/bid/create/bid/request",
    method: "post",
    body,
  });
};

流式传输（NDJSON）

当服务器返回 NDJSON（每行一个 JSON）时，使用 callApiStream。如果没有 Accept 头，默认设置为 application/x-ndjson。

export const streamEvents = () => {
  return client.callApiStream<{ eventName: string; content: unknown }>({
    url: "/api/event/stream",
    method: "get",
    // headers: { Accept: "application/x-ndjson" }, // 선택
  });
};

CSR 缓存（客户端缓存）

createCsrCache<CacheName>() 提供的功能：

callApiCsrCache(callApiFn, serviceArgs, cacheOptions)
removeCsrCache(cacheName) — 支持类型缓存名 + 字符串

import { createCsrCache } from "@byeolnaerim/typed-rx-http";
import type { paths as Paths } from "./@types/ApiTypes";

type CacheNames = "adminList" | "profile";

const csrCache = createCsrCache<CacheNames>();

export const callApiClientCache = <R>(
  args: ServiceArguments<Paths, any, any, R>,
  cache: { cacheName: CacheNames; cacheTime: number; cacheSize?: number },
) => csrCache.callApiCsrCache(client.callApi, args as any, cache as any);

// 무효화
csrCache.removeCsrCache("adminList"); // 타입
csrCache.removeCsrCache("any-prefix"); // 문자열

基于会话的认证插件（可选）

createSessionAuth 将会话认证逻辑与核心分离，以选项的方式附加或移除。

操作：

将 Authorization 保持在 headerStore 中
通过 ensureToken$() 同步令牌（/api/auth/token）
发生 401 时尝试刷新 1 次（/api/auth/token/refresh），然后重试原请求
刷新失败时，登出（/api/auth/logout）并传递错误
登录状态的变化通过 onLoginChange 回调在外部处理

import { createSessionAuth } from "@byeolnaerim/typed-rx-http";

const auth = createSessionAuth({
  headerStore,
  onLoginChange: (loggedIn) => {
    // 예: setLogin(loggedIn) 같은 외부 상태 업데이트(선택)
  },
  // tokenUrl / refreshUrl / logoutUrl 커스터마이징 가능
});

export const secureCall = () => {
  return client
    .callApi<{ ok: true }>({ url: "/api/secure", method: "get" })
    .pipe(auth.withSessionAuth());
};

如果只需要令牌同步而不需要刷新/重试：

return client.callApi(...).pipe(auth.withEnsureToken());

错误处理

如果不是 2xx，则抛出 HttpResponseError（包括 status, response, args, data）。

遗留兼容性：如果错误主体是 { resultType: ... } 形式，则直接抛出该对象。

import { isHttpResponseError } from "@byeolnaerim/typed-rx-http";

client.callApi(...).subscribe({
  error: (e) => {
    if (isHttpResponseError(e)) {
      console.log(e.status, e.data);
    }
  },
});

Next.js 适配器（/next）

redirectToUnauthorizedOnServer401

redirectToUnauthorizedOnServer401 是在 Next.js（App Router）SSR 环境中发生 401 时执行重定向的基本实现（便捷函数）。

import { redirectToUnauthorizedOnServer401 } from "@byeolnaerim/typed-rx-http/next";

操作规则（固定）：

重定向目标：/unauthorized
queryString：redirect_uri=<当前页面> + logout=true
当前页面从 x-page-url 头中读取（如果没有则为 /）

也就是说，只有当上述路径/查询规则与项目匹配时，才能直接使用。如果路径不同或查询规则不同，可以像下面这样直接实现 onServer401 并注入。

import { headers } from "next/headers";
import { redirect } from "next/navigation";

export const onServer401 = async () => {
  const h = await headers();
  const pageUrl = h.get("x-page-url") || "/";
  redirect(`/login?next=${encodeURIComponent(pageUrl)}`);
};

callApiSsrCache

Next的基于next/cache(unstable_cache)的SSR缓存助手。

GET + cacheTime > 0 → 强制缓存 + 重新验证
其他情况 → 不存储
通过headersProvider注入请求特定的Cookie / Authorization
发生401时，如果有onServer401则执行（通常是redirect()）

import type { paths as Paths } from "./@types/ApiTypes";
import { firstValueFrom } from "rxjs";
import {
  callApiSsrCache,
  redirectToUnauthorizedOnServer401,
} from "@byeolnaerim/typed-rx-http/next";

const apiUrl = process.env.NEXT_PUBLIC_API_URL!;

const headersProvider = async () => {
  const { cookies } = await import("next/headers");
  const cookieStore = await cookies();

  const cookieHeader = cookieStore
    .getAll()
    .map((c) => `${c.name}=${c.value}`)
    .join("; ");

  const accessToken = cookieStore.get("accessToken")?.value;

  return {
    ...(cookieHeader ? { Cookie: cookieHeader } : {}),
    ...(accessToken ? { Authorization: `Bearer ${accessToken}` } : {}),
  };
};

const obs$ = callApiSsrCache<Paths, { ok: true }, "adminList">({
  baseUrl: apiUrl,
  serviceArguments: { url: "/api/secure", method: "get" },
  cacheForService: { cacheName: "adminList", cacheTime: 10_000 },
  headersProvider,
  onServer401: redirectToUnauthorizedOnServer401,
});

const res = await firstValueFrom(obs$);

Next.js集成示例：项目rxjsHttpService/commonService的完整代码

下面的rxjsHttpService.ts是后续生成器示例中commonServiceFile所指的项目公共HTTP适配器。该库生成的文件不是项目直接管理的，而是核心客户端 + 会话认证 + CSR缓存 + SSR缓存助手集中导出。以下是完整代码。

rxjsHttpService.ts

import {
	createCsrCache,
	createHeaderStore,
	createHttpClient,
	createSessionAuth,
} from "@byeolnaerim/typed-rx-http";

import type {
	CacheForService,
	ServiceArguments,
} from "@byeolnaerim/typed-rx-http";

import { defer, Observable } from "rxjs";

import { setLogin } from "@ui/handler/hooks/useAccounts";

import type { paths } from "./@types/ApiTypes";

import type { CacheNames } from "./@types/CacheNames";

export const redirectToUnauthorizedOnServer401 = async () => {
	const { headers } = await import("next/headers");
	const { redirect } = await import("next/navigation");

	const h = await headers();
	const pageUrl = h.get("x-page-url") || "/";
	const redirectUri = encodeURIComponent(pageUrl);

	redirect(`/unauthorized?redirect_uri=${redirectUri}&logout=true`);
};

export const headerStore = createHeaderStore({
	"Content-Type": "application/json",
});

const headersProvider = async () => {
	if (typeof window !== "undefined") {
		/**
		 * 라이브러리쪽 코드가
		 * if (headersProvider) return headersProvider();
		 * if (headerStore) return headerStore.get();
		 * 이렇게 되어있음 같이 못씀, 방향자체는 맞음. 라이브러리쪽에서 marge해버리면 csr/ssr 구분못함
		 */
		return headerStore.get();
	}

	const { cookies } = await import("next/headers");

	const cookieStore = await cookies();
	const cookieHeader = cookieStore
		.getAll()
		.map((c) => `${c.name}=${c.value}`)
		.join("; ");

	const authorization = cookieStore.get("Authorization")?.value;
	const accessToken = cookieStore.get("accessToken")?.value;

	return {
		...(cookieHeader ? { Cookie: cookieHeader } : {}),
		...(authorization
			? {
					Authorization: authorization.startsWith("Bearer ")
						? authorization
						: `Bearer ${authorization}`,
				}
			: {}),
		...(!authorization && accessToken
			? {
					Authorization: accessToken.startsWith("Bearer ")
						? accessToken
						: `Bearer ${accessToken}`,
				}
			: {}),
	};
};

export const service = createHttpClient<paths>({
	baseUrl: process.env.NEXT_PUBLIC_API_URL || "",
	onServer401: redirectToUnauthorizedOnServer401,
	defaultErrorMessage: () => {
		return "알 수 없는 오류가 발생했습니다.";
	},
	headerStore,
	headersProvider,
	dropAuthWhenCacheControl: true,
});

const sessionAuth = createSessionAuth({
	headerStore,
	tokenUrl: "/api/auth/token",
	refreshUrl: "/api/auth/token/refresh",
	logoutUrl: "/api/auth/logout",
	onLoginChange: (loggedIn) => {
		if (typeof window !== "undefined") {
			setLogin(loggedIn);
		}
	},
});

export const callApi = <
	R,
	TPath extends keyof paths = keyof paths,
	TMethod extends keyof paths[TPath] & string = keyof paths[TPath] & string,
>(
	serviceArguments: ServiceArguments<paths, TPath, TMethod, R>,
): Observable<R> => {
	return defer(() =>
		service.callApi<R, TPath, TMethod>(serviceArguments),
	).pipe(sessionAuth.withSessionAuth());
};

export const callApiStream = <
	R,
	TPath extends keyof paths = keyof paths,
	TMethod extends keyof paths[TPath] & string = keyof paths[TPath] & string,
>(
	serviceArguments: ServiceArguments<paths, TPath, TMethod, R>,
): Observable<R> => {
	return defer(() =>
		service.callApiStream<R, TPath, TMethod>(serviceArguments),
	).pipe(sessionAuth.withSessionAuth());
};

// CSR cache
const csrCache = createCsrCache<CacheNames>();

export const callApiClientCache = <
	R,
	TPath extends keyof paths = keyof paths,
	TMethod extends keyof paths[TPath] & string = keyof paths[TPath] & string,
>(
	serviceArguments: ServiceArguments<paths, TPath, TMethod, R>,
	cacheForService: CacheForService<CacheNames>,
): Observable<R> => {
	return csrCache.callApiCsrCache(callApi, serviceArguments, cacheForService);
};

export const removeCsrCache = csrCache.removeCsrCache;

export const callApiServerCache = <
	R,
	TPath extends keyof paths = keyof paths,
	TMethod extends keyof paths[TPath] & string = keyof paths[TPath] & string,
>(
	serviceArguments: ServiceArguments<paths, TPath, TMethod, R>,
	cacheForService: CacheForService<CacheNames>,
): Observable<R> => {
	if (typeof window !== "undefined") {
		return callApiClientCache(serviceArguments, cacheForService);
	}

	return new Observable<R>((subscriber) => {
		let closed = false;

		import("@byeolnaerim/typed-rx-http/next")
			.then(({ callApiSsrCache }) => {
				if (closed) {
					return;
				}

				const subscription = callApiSsrCache<paths, R, CacheNames>({
					baseUrl: process.env.NEXT_PUBLIC_API_URL || "",
					serviceArguments: serviceArguments as ServiceArguments<
						paths,
						keyof paths,
						keyof paths[keyof paths] & string,
						R
					>,
					cacheForService,
					headersProvider,
					onServer401: redirectToUnauthorizedOnServer401,
				}).subscribe(subscriber);

				subscriber.add(() => {
					closed = true;
					subscription.unsubscribe();
				});
			})
			.catch((error) => {
				if (!closed) {
					subscriber.error(error);
				}
			});

		return () => {
			closed = true;
		};
	});
};

API参考（核心）

`createHttpClient<Paths>(options)`

callApi<R>(args): Observable<R>
callApiStream<RChunk>(args): Observable<RChunk>
uploadFile({ file, url, ifNoneMatch?, headers? }): Observable<Response>
createSSEObservable<R>(args): Observable<R>

选项：

baseUrl: string
headerStore?: HeaderStore
headersProvider?: () => Record<string, string> | Promise<Record<string, string>>
dropAuthWhenCacheControl?: boolean（默认值：true）
onServer401?: () => void | Promise<void>

`createHeaderStore(initial?)`

get(), set(), merge(), remove(), clear()

`createCsrCache<CacheName>()`

callApiCsrCache(callApiFn, serviceArgs, cacheForService)
removeCsrCache(cacheName)（类型 + 字符串）

`createSessionAuth(options)`

withSessionAuth(), withEnsureToken()
ensureToken$(), refreshToken$(), logout$()

运行时要求

使用fetch / Response API（rxjs/fetch）
流式传输（NDJSON）需要ReadableStream + TextDecoder
SSE需要EventSource

大多数现代浏览器和Next.js运行时都提供支持。自定义Node运行时可能需要polyfill。

自动Node脚本：OpenAPI/Swagger代码生成（可选）

该包除了运行时HTTP客户端外，还提供了一个Node脚本，用于从OpenAPI/Swagger JSON生成类型/服务代码。该脚本是可选功能。

一般的@byeolnaerim/typed-rx-http、/next、/rsocket用户无需运行此脚本，也不需要安装openapi-typescript。

import {
  connectToSwaggerEventStream,
  generateSwaggerFromHttp,
  generateSwaggerFromFile,
} from "@byeolnaerim/typed-rx-http/auto-node-script/openapi";

依赖隔离

OpenAPI类型生成需要openapi-typescript CLI。但该包不会将openapi-typescript放入常规依赖中。

@byeolnaerim/typed-rx-http库本身使用devDependencies.typescript的TypeScript 6.0.3。typed-rx-http、/next、/rsocket的入口点和库构建保持此TypeScript 6.0.3标准。

然而，openapi-typescript可能仍然要求特定的TypeScript 5.x版本，因此OpenAPI自动Node脚本仅在单独的npx临时执行环境中与openapi-typescript和[email protected]一起运行。此临时执行环境不会更改库的devDependencies.typescript 6.0.3，也不会使用用户项目中安装的typescript或openapi-typescript版本。

默认情况下，仅在自动脚本执行时生成并执行以下命令。

bash

npx -y -p openapi-typescript@latest -p [email protected] openapi-typescript swagger.json --output ./src/handler/service/@types/ApiTypes.d.ts

因此，使用方法本身并没有改变。可以像以前一样调用自动Node脚本，仅在OpenAPI类型生成阶段使用隔离的TypeScript 5.9.3环境。未使用自动脚本的用户完全不受openapi-typescript或TypeScript 5.9.3的限制。

如有需要，可以直接通过 openApiTypescriptCommand 固定命令。

generateSwaggerFromFile({
  inputFile: "./swagger.json",
  openApiTypescriptCommand:
    "npx -y -p [email protected] -p [email protected] openapi-typescript ./swagger.json --output ./src/handler/service/@types/ApiTypes.d.ts",
});

或者只更改配置基本命令的包版本。

generateSwaggerFromFile({
  inputFile: "./swagger.json",
  openApiTypescriptPackage: "[email protected]",
  openApiTypescriptTypescriptPackage: "[email protected]",
});

生成的文件

默认设置会生成以下文件。

text

swagger.json
src/handler/service/@types/ApiTypes.d.ts
src/handler/service/auto/*Service.ts
src/handler/service/@types/auto/*Types.ts
src/handler/service/apiUnionArrays.ts

apiUnionArrays.ts 不仅生成 OpenAPI schema enum，还生成 query、path、header、cookie 参数的 enum 作为常量数组。也处理数组 query 参数的 items.enum。

EventStream 监控

connectToSwaggerEventStream({
  hostname: "localhost",
  port: 8788,
  path: "/oauth2/for-local/get-swagger",
  serviceDir: "./src/handler/service",
  typesDir: "./src/handler/service/@types",
  commonServiceFile: "./src/handler/service/rxjsHttpService.ts",
});

HTTP 单次请求

await generateSwaggerFromHttp({
  hostname: "localhost",
  port: 8788,
  path: "/oauth2/for-local/get-swagger",
});

从本地文件生成

generateSwaggerFromFile({
  inputFile: "./swagger.json",
});

现有项目集成示例：WebFlux + Swagger 自动生成

从这里开始是后端 Swagger 和自动生成服务一起使用的项目集成示例。上面的核心用法和选择功能可以仅通过 typed-rx-http 使用，下面的流程是在使用 Swagger 基础服务自动生成的项目中额外应用的。

1. 在后端编写 REST 端点

后端代码按常规编写。在此示例中，使用路径变量接收名称，使用查询参数接收消息 Mono并返回响应。

TypedRxHttpExampleRouter.java

java

@Bean
public RouterFunction<ServerResponse> typedRxHttpExample(
  TypedRxHttpExampleHandler handler
) {
  return route()
    .GET("/typed-rx-http/example/{name}", handler::getExample)
    .build();
}

2. 在 Swagger 中生成前端服务

接收后端的 swagger.json生成类型和服务文件。此操作可以在运行开发服务器时调用一次，或通过 watch 脚本连接。

generateSwagger.cjs

const {
  generateSwaggerFromHttp,
} = require("@byeolnaerim/typed-rx-http/auto-node-script/openapi");

generateSwaggerFromHttp({
  hostname: "localhost",
  port: 8123,
  path: "/for-local/get-swagger",
  swaggerFile: "./swagger.json",
  serviceDir: "./src/handler/service",
  typesDir: "./src/handler/service/@types",
  typesFile: "./src/handler/service/@types/ApiTypes.d.ts",
  commonServiceFile:
    "./src/handler/service/rxjsHttpService.ts",
});

commonServiceFile 的含义

该选项不是生成 rxjsHttpService.ts 的设置。它是生成的服务导入 callApi 和 callApiStream 的项目拥有文件的位置。必须先创建公共文件，然后指定此路径。

3. 通过生成的函数发送请求

可以在下面更改 前端代码和 后端代码。结果点击将打开右侧的执行界面。更改值后，点击请求按钮。

TypedRxHttpRequestExample.tsx

tsx

import { byName } from "@handler/service/auto/TypedRxHttpExampleService";
import { firstValueFrom } from "rxjs";

const response = await firstValueFrom(
  byName({
    path: {
      name: name.trim(),
    },
    params: {
      message: message.trim() || undefined,
    },
  }),
);

setResponse(response);

生成结果中需要确认的部分

界面代码中没有 URL 字符串或响应接口。两者都在 Swagger 生成的 byName() 包含在内。