tRPC — Subscriptions
Setup
SSE is recommended for most subscription use cases. It is simpler to set up and does not require a WebSocket server.
Server
import EventEmitter, { on } from 'node:events';
import { initTRPC, tracked } from '@trpc/server';
import { createHTTPServer } from '@trpc/server/adapters/standalone';
import { z } from 'zod';
const t = initTRPC.create({
sse: {
ping: {
enabled: true,
intervalMs: 2000,
},
client: {
reconnectAfterInactivityMs: 5000,
},
},
});
type Post = { id: string; title: string };
const ee = new EventEmitter();
const appRouter = t.router({
onPostAdd: t.procedure
.input(z.object({ lastEventId: z.string().nullish() }).optional())
.subscription(async function* (opts) {
for await (const [data] of on(ee, 'add', { signal: opts.signal })) {
const post = data as Post;
yield tracked(post.id, post);
}
}),
});
export type AppRouter = typeof appRouter;
createHTTPServer({
router: appRouter,
createContext() {
return {};
},
}).listen(3000);
Client (SSE)
import {
createTRPCClient,
httpBatchLink,
httpSubscriptionLink,
splitLink,
} from '@trpc/client';
import type { AppRouter } from './server';
const trpc = createTRPCClient<AppRouter>({
links: [
splitLink({
condition: (op) => op.type === 'subscription',
true: httpSubscriptionLink({ url: 'http://localhost:3000' }),
false: httpBatchLink({ url: 'http://localhost:3000' }),
}),
],
});
const subscription = trpc.onPostAdd.subscribe(
{ lastEventId: null },
{
onData(post) {
console.log('New post:', post);
},
onError(err) {
console.error('Subscription error:', err);
},
},
);
Core Patterns
tracked() for reconnection recovery
import EventEmitter, { on } from 'node:events';
import { initTRPC, tracked } from '@trpc/server';
import { z } from 'zod';
const t = initTRPC.create();
const ee = new EventEmitter();
const appRouter = t.router({
onPostAdd: t.procedure
.input(z.object({ lastEventId: z.string().nullish() }).optional())
.subscription(async function* (opts) {
const iterable = on(ee, 'add', { signal: opts.signal });
if (opts.input?.lastEventId) {
}
for await (const [data] of iterable) {
yield tracked(data.id, data);
}
}),
});
When using tracked(id, data), the client automatically sends lastEventId on reconnection. For SSE this is part of the EventSource spec; for WebSocket, wsLink handles it.
Polling loop subscription
import { initTRPC, tracked } from '@trpc/server';
import { z } from 'zod';
const t = initTRPC.create();
const appRouter = t.router({
onNewItems: t.procedure
.input(z.object({ lastEventId: z.coerce.date().nullish() }))
.subscription(async function* (opts) {
let cursor = opts.input?.lastEventId ?? null;
while (!opts.signal?.aborted) {
const items = await db.item.findMany({
where: cursor ? { createdAt: { gt: cursor } } : undefined,
orderBy: { createdAt: 'asc' },
});
for (const item of items) {
yield tracked(item.createdAt.toJSON(), item);
cursor = item.createdAt;
}
await new Promise( (r, ));
}
}),
});
WebSocket setup (when bidirectional communication is required)
import { applyWSSHandler } from '@trpc/server/adapters/ws';
import { WebSocketServer } from 'ws';
import { appRouter } from './router';
const wss = new WebSocketServer({ port: 3001 });
const handler = applyWSSHandler({
wss,
router: appRouter,
createContext() {
return {};
},
keepAlive: {
enabled: true,
pingMs: 30000,
pongWaitMs: 5000,
},
});
process.on('SIGTERM', () => {
handler.broadcastReconnectNotification();
wss.close();
});
import {
createTRPCClient,
createWSClient,
httpBatchLink,
splitLink,
wsLink,
} from '@trpc/client';
import type { AppRouter } from './server';
const wsClient = createWSClient({ url: 'ws://localhost:3001' });
const trpc = createTRPCClient<AppRouter>({
links: [
splitLink({
condition: (op) => op.type === 'subscription',
true: wsLink({ client: wsClient }),
false: httpBatchLink({ url: 'http://localhost:3000' }),
}),
],
});
Cleanup with try...finally
const appRouter = t.router({
events: t.procedure.subscription(async function* (opts) {
const cleanup = registerListener();
try {
for await (const [data] of on(ee, 'event', { signal: opts.signal })) {
yield data;
}
} finally {
cleanup();
}
}),
});
tRPC invokes .return() on the generator when the subscription stops, triggering the finally block.
Common Mistakes
HIGH Using Observable instead of async generator
Wrong:
import { observable } from '@trpc/server/observable';
t.procedure.subscription(({ input }) => {
return observable((emit) => {
emit.next(data);
});
});
Correct:
t.procedure.subscription(async function* ({ input, signal }) {
for await (const [data] of on(ee, 'event', { signal })) {
yield data;
}
});
Observable subscriptions are deprecated and will be removed in v12. Use async generator syntax (async function*).
Source: packages/server/src/unstable-core-do-not-import/procedureBuilder.ts
MEDIUM Empty string as tracked event ID
Wrong:
yield tracked('', data);
Correct:
yield tracked(event.id.toString(), data);
tracked() throws if the ID is an empty string because it conflicts with SSE "no id" semantics.
Source: packages/server/src/unstable-core-do-not-import/stream/tracked.ts
HIGH Fetching history before setting up event listener
Wrong:
t.procedure.subscription(async function* (opts) {
const history = await db.getEvents();
yield* history;
for await (const event of listener) {
yield event;
}
});
Correct:
t.procedure.subscription(async function* (opts) {
const iterable = on(ee, 'event', { signal: opts.signal });
const history = await db.getEvents();
for (const item of history) {
yield tracked(item.id, item);
}
for await (const [event] of iterable) {
yield tracked(event.id, event);
}
});
If you fetch historical data before setting up the event listener, events emitted between the fetch and listener setup are lost.
Source: www/docs/server/subscriptions.md
MEDIUM SSE ping interval >= client reconnect interval
Wrong:
initTRPC.create({
sse: {
ping: { enabled: true, intervalMs: 10000 },
client: { reconnectAfterInactivityMs: 5000 },
},
});
Correct:
initTRPC.create({
sse: {
ping: { enabled: true, intervalMs: 2000 },
client: { reconnectAfterInactivityMs: 5000 },
},
});
If the server ping interval is >= the client reconnect timeout, the client disconnects thinking the connection is dead before receiving a ping.
Source: packages/server/src/unstable-core-do-not-import/stream/sse.ts
HIGH Sending custom headers with SSE without EventSource polyfill
Wrong:
httpSubscriptionLink({
url: 'http://localhost:3000',
});
Correct:
import { EventSourcePolyfill } from 'event-source-polyfill';
httpSubscriptionLink({
url: 'http://localhost:3000',
EventSource: EventSourcePolyfill,
eventSourceOptions: async () => ({
headers: { authorization: 'Bearer token' },
}),
});
The native EventSource API does not support custom headers. Use an EventSource polyfill and pass it via the EventSource option on httpSubscriptionLink.
Source: www/docs/client/links/httpSubscriptionLink.md
MEDIUM Choosing WebSocket when SSE would suffice
SSE (httpSubscriptionLink) is recommended for most subscription use cases. WebSockets add complexity (connection management, reconnection, keepalive, separate server process). Only use wsLink when bidirectional communication or WebSocket-specific features are required.
Source: maintainer interview
MEDIUM WebSocket subscription stale inputs on reconnect
When a WebSocket reconnects, subscriptions re-send the original input parameters. There is no hook to re-evaluate inputs on reconnect, which can cause stale data. Consider using tracked() with lastEventId to mitigate this.
Source: https://github.com/trpc/trpc/issues/4122
See Also
- links --
splitLink, httpSubscriptionLink, wsLink, httpBatchLink
- auth -- authenticating subscription connections (connectionParams, cookies, EventSource polyfill headers)
- server-setup --
initTRPC.create() SSE configuration options
- adapter-fastify -- WebSocket subscriptions via
@fastify/websocket and useWSS
