Set up real-time event streams with async generator subscriptions using .subscription(async function*() { yield }). SSE via httpSubscriptionLink is recommended over WebSocket. Use tracked(id, data) from @trpc/server for reconnection recovery with lastEventId. WebSocket via wsLink and createWSClient from @trpc/client, applyWSSHandler from @trpc/server/adapters/ws. Configure SSE ping with initTRPC.create({ sse: { ping: { enabled, intervalMs } } }). AbortSignal via opts.signal for cleanup. splitLink to route subscriptions.
Set up real-time event streams with async generator subscriptions using .subscription(async function*() { yield }). SSE via httpSubscriptionLink is recommended over WebSocket. Use tracked(id, data) from @trpc/server for reconnection recovery with lastEventId. WebSocket via wsLink and createWSClient from @trpc/client, applyWSSHandler from @trpc/server/adapters/ws. Configure SSE ping with initTRPC.create({ sse: { ping: { enabled, intervalMs } } }). AbortSignal via opts.signal for cleanup. splitLink to route subscriptions.
importEventEmitter, { on } from'node:events';
import { initTRPC, tracked } from'@trpc/server';
import { z } from'zod';
const t = initTRPC.create();
const ee = newEventEmitter();
const appRouter = t.router({
onPostAdd: t.procedure
.input(z.object({ lastEventId: z.string().nullish() }).optional())
.subscription(asyncfunction* (opts) {
const iterable = on(ee, 'add', { signal: opts.signal });
if (opts.input?.lastEventId) {
// Fetch and yield events since lastEventId from your database// const missed = await db.post.findMany({ where: { id: { gt: opts.input.lastEventId } } });// for (const post of missed) { yield tracked(post.id, post); }
}
forawait (const [data] of iterable) {
yieldtracked(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.
HIGH Fetching history before setting up event listener
Wrong:
t.procedure.subscription(asyncfunction* (opts) {
const history = await db.getEvents(); // events may fire here and be lostyield* history;
forawait (const event of listener) {
yield event;
}
});
Correct:
t.procedure.subscription(asyncfunction* (opts) {
const iterable = on(ee, 'event', { signal: opts.signal }); // listen firstconst history = await db.getEvents();
for (const item of history) {
yieldtracked(item.id, item);
}
forawait (const [event] of iterable) {
yieldtracked(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
The native EventSource API does not support custom headers. Use an EventSource polyfill and pass it via the EventSource option on httpSubscriptionLink.
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.
