tRPC v11 ReactuseSubscription完全指南端到端类型安全的实时订阅 Hook【免费下载链接】trpc‍♀️ Move Fast and Break Nothing. End-to-end typesafe APIs made easy.项目地址: https://gitcode.com/GitHub_Trending/tr/trpc本文基于本仓库trpc/react-query的useSubscriptionHook讲解如何在 React 组件中订阅服务端 tRPC subscription procedure如 SSE、WebSocket 推送的实时数据涵盖完整签名、基于status的判别联合返回类型、服务端异步生成器写法、真实组件接入示例并结合 packages/react-query/src/shared/hooks/createHooksInternal.tsx 等源码剖析其底层状态机与生命周期。读完你将能在自己的 tRPC React 应用中安全地接入实时订阅并正确处理连接中、出错与重连等状态。一、useSubscription是什么useSubscription是trpc/react-query提供给 React 组件的 Hook用于订阅服务端的 subscription 类型 procedure。与useQuery一次性拉取和useMutation写入不同订阅建立的是一条持续存在的通道服务端可以多次向客户端推送数据客户端通过回调持续接收直到服务端完成或错误中止。在 tRPC 生态里它通常与SSEServer-Sent Events、WebSocket或普通 HTTP 流式传输配合使用——本仓库的 examples/next-sse-chat 与 examples/next-websockets-encoder、examples/lambda-api-gateway-streaming 都展示了这类实时场景。二、函数签名与 Options 详解调用形式useSubscription(input, opts)第一个参数是该 procedure 的入参第二个参数opts是订阅选项对象。Options文档给出的类型定义与 packages/react-query/src/shared/hooks/types.ts 中的UseTRPCSubscriptionOptions完全一致interface UseTRPCSubscriptionOptionsTOutput, TError { /** * Called when the subscription is started. */ onStarted?: () void; /** * Called when new data is received from the subscription. */ onData?: (data: TOutput) void; /** * Called when an **unrecoverable error** occurs and the subscription is stopped. */ onError?: (error: TError) void; /** * Called when the subscription is completed on the server. * The state will transition to idle with data: undefined. */ onComplete?: () void; /** * deprecated Use a skipToken from tanstack/react-query instead. * This will be removed in v12. */ enabled?: boolean; }各选项含义选项触发时机说明onStarted订阅建立成功在源码实现中该回调触发的同时 Hook 内部状态会被置为status: pending并清空erroronData每收到一条服务端推送收到数据后状态更新为status: pending、写入最新data、清空erroronError发生不可恢复错误订阅随即中止状态转为status: error并携带错误对象onComplete服务端正常结束订阅状态转为idledata重置为undefinedenabled已废弃—v12 中将移除请改用tanstack/react-query的skipToken官方使用提示务必遵守需要传选项但无需输入参数时第一个参数传undefined即可例如trpc.onPostAdd.useSubscription(undefined, { onData: ... })传入tanstack/react-query导出的skipToken时订阅会被暂停——这是 v11 推荐的「有条件订阅」方式用于替代废弃的enabled选项想看完整可运行的订阅示例参考本仓库的 examples/next-sse-chat原文档指向上游 SSE 示例工程本仓库内即包含等价实现。实现细节印证在 createHooksInternal.tsx 中Hook 实际计算const enabled opts?.enabled ?? input ! skipToken;即enabled的默认值完全由是否传入skipToken决定enabled只是为兼容旧写法保留的别名。若 Hook 在挂载后才更新opts源码通过optsRefuseEffect见 createHooksInternal.tsx保证回调永远指向最新值避免闭包过期。三、返回值基于status的判别联合useSubscription的返回类型是一个以status字段为判别键的可判别联合类型discriminated unionTypeScript 会根据status精确推断出data/error的联合收窄因此你可以放心地对状态做穷举渲染而不会产生类型漏洞type TRPCSubscriptionResultTOutput, TError | TRPCSubscriptionIdleResultTOutput | TRPCSubscriptionConnectingResultTOutput, TError | TRPCSubscriptionPendingResultTOutput | TRPCSubscriptionErrorResultTOutput, TError; interface TRPCSubscriptionIdleResultTOutput { /** Subscription is disabled or has ended */ status: idle; data: undefined; error: null; reset: () void; } interface TRPCSubscriptionConnectingResultTOutput, TError { /** Trying to establish a connection (may have a previous error from a reconnection attempt) */ status: connecting; data: TOutput | undefined; error: TError | null; reset: () void; } interface TRPCSubscriptionPendingResultTOutput { /** Connected to the server, receiving data */ status: pending; data: TOutput | undefined; error: null; reset: () void; } interface TRPCSubscriptionErrorResultTOutput, TError { /** An unrecoverable error occurred and the subscription is stopped */ status: error; data: TOutput | undefined; error: TError; reset: () void; }四个分支速查表status含义dataerror典型界面idle订阅被暂停skipToken或已结束/完成undefinednull无内容或占位connecting正在尝试建立连接重连失败时可能携带上一次错误TOutput \| undefinedTError \| null“连接中…”提示pending已连接并正在接收数据TOutput \| undefinednull渲染数据流error发生不可恢复错误订阅已停止TOutput \| undefinedTError非空错误信息 重连按钮每种状态下都提供一个reset: () void用于手动重连/重启订阅见下文组件示例。上面这些接口在 types.ts 中均有同名定义并统一继承了一个共享的TRPCSubscriptionBaseResult基类。四、底层状态机与重连原理源码视角useSubscription并不是一个对tanstack/react-query查询的简单封装而是直接在 createHooksInternal.tsx 中独立实现的状态机内部通过client.subscription(...)驱动。梳理源码可还原出如下生命周期挂载进入reset()无输入或传skipToken时保持idle否则先进入connecting再调用client.subscription(path.join(.), input ?? undefined, observer)建立连接源码见 createHooksInternal.tsxonStarted客户端订阅 observer 回调触发状态connecting → pending同时对外调用你的onStartedonData每收到数据更新data并保持在pending同时调用你的onData连接状态回调订阅 observer 还会上报onConnectionStateChange。当底层连接进入idle时 Hook 会把自身状态重置为idle、清空error与data当进入connecting时则记录error这正是「重连失败可携带上一次错误」这一文档说明的来源pending分支被注释为“由 data/onStarted 处理”直接忽略以免重复渲染onError状态进入error订阅停止onComplete服务端完成订阅。源码在此有一处关键注释——WebSocket 场景下底层连接可能并未 idle因此onConnectionStateChange在连接真正关闭前不会触发故此处手动把状态置为idle、data: undefined见 createHooksInternal.tsx卸载清理useEffect的 cleanup 中调用currentSubscriptionRef.current?.unsubscribe()避免内存泄漏见 createHooksInternal.tsx。此外源码使用trackResult基于Proxy做属性级追踪createHooksInternal.tsx组件只读取了status/data中的哪些字段Hook 就只在那些字段变化时触发重渲染——高频推送下未观察的属性更新不会导致无关渲染。从调用链看客户端侧真正干活的是trpc/client内部的TRPCSubscriptionObserver见 TRPCUntypedClient.ts它声明了onStarted、onData、onStopped、onComplete与onConnectionStateChange等回调服务端是否真正以 WebSocket/SSE 推送取决于你使用的订阅传输 link如 SSE 订阅 link、WebSocket link。五、示例一服务端订阅 Procedure要消费订阅首先服务端要暴露一个 subscription procedure。v11 推荐用**异步生成器async generator**声明用for await消费事件把数据yield给客户端并借助opts.signal实现取消。下方是文档中的示例EventEmitterevents.on用于向订阅者实时推送“新增文章”事件// target: esnext import EventEmitter, { on } from events; import { initTRPC } from trpc/server; export const t initTRPC.create(); type Post { id: string; title: string }; const ee new EventEmitter(); export const appRouter t.router({ onPostAdd: t.procedure.subscription(async function* (opts) { for await (const [data] of on(ee, add, { signal: opts.signal, })) { const post data as Post; yield post; } }), }); export type AppRouter typeof appRouter;要点t.procedure.subscription(async function* (opts) {...})声明一个可流式推送多条的 procedureopts.signal来自调用端客户端断开时生成器会收到中止信号并优雅退出从而停止for await循环yield出的每个值都会作为一次推送送到客户端onData。服务端订阅能力的更多细节超时、批量推送、keep-alive 等可查阅 www/docs/server/subscriptions.md。本仓库内多个示例实现了订阅 procedure例如 examples/next-sse-chat/src/server 与 examples/next-websockets-encoder/src/server。六、示例二React 组件中使用在客户端先通过createTRPCReactAppRouter()创建类型安全的 tRPC 客户端对象由于类型是从AppRouter推断出来的trpc.onPostAdd.useSubscription的入参与回调数据类型都自动对齐服务端。下面是一个实时文章流组件把每次推送的post追加进本地posts状态并根据status渲染连接中、错误和正常三种界面import React from react; import { trpc } from ../utils/trpc; type Post { id: string; title: string }; export function PostFeed() { const [posts, setPosts] React.useStatePost[]([]); const subscription trpc.onPostAdd.useSubscription(undefined, { onData: (post) { setPosts((prev) [...prev, post]); }, }); return ( div h1Live Feed/h1 {subscription.status connecting pConnecting.../p} {subscription.status error ( div pError: {subscription.error.message}/p button onClick{() subscription.reset()}Reconnect/button /div )} ul {posts.map((post) ( li key{post.id}{post.title}/li ))} /ul /div ); }需要特别留意的模式第一个参数undefined表示「不需要输入但要传 options」把服务端数据显式地写入本地 statesetPosts是订阅场景的常见写法——useSubscription不负责维护累积列表只负责把单条数据实时递到你的回调里subscription.error只在status error时存在这正是判别联合带来的类型安全红利错误后调用subscription.reset()会走一遍前面源码分析的reset()流程先unsubscribe()旧连接再重新进入connecting→ 重新建立连接实现「重连」按钮。七、条件订阅用skipToken暂停v11 中要“暂停”订阅不要再用废弃的enabled而是把输入参数替换成tanstack/react-query导出的skipTokenimport { skipToken } from tanstack/react-query; const { data, status } trpc.onEvent.useSubscription( isEnabled ? { roomId } : skipToken, { onData: handler }, );从实现看input skipToken会让enabled变为falsereset()直接返回idle状态且不建立连接见 createHooksInternal.tsx当条件满足、输入不再是skipToken时依赖数组中的变化会触发reset()并开始真正订阅。八、测试与更多参考本 Hook 的行为在本仓库有完整测试覆盖例如 packages/react-query/test/useSubscription.test.tsx 中针对数据接收、暂停/恢复、错误处理等场景均有断言底层传输层的订阅行为可参考 packages/tests/server/websockets.test.ts。若想继续深入服务端订阅 API 与服务端推送实现www/docs/server/subscriptions.mdSSE 实时聊天完整示例examples/next-sse-chatWebSocket 编码器推送示例examples/next-websockets-encoder订阅客户端传输层的 observer 协议定义TRPCUntypedClient.ts。一句话总结useSubscription(input, options)返回一个status ∈ { idle, connecting, pending, error }的判别联合通过onData实时收数、onError兜底、reset()重连、skipToken暂停理解其状态机与源码中的reset()/onComplete处理是写出可靠实时界面的关键。【免费下载链接】trpc‍♀️ Move Fast and Break Nothing. End-to-end typesafe APIs made easy.项目地址: https://gitcode.com/GitHub_Trending/tr/trpc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考