Subscriptions
Introduction
Subscriptions are a type of real-time event stream between the client and server. Use subscriptions when you need to push real-time updates to the client.
With tRPC's subscriptions, the client establishes and maintains a persistent connection to the server plus automatically attempts to reconnect and recover gracefully if disconnected with the help of tracked()
events.
WebSockets or Server-sent Events?
You can either use WebSockets or Server-sent Events (SSE) to setup real-time subscriptions in tRPC.
- For WebSockets, see the WebSockets page
- For SSE, see the httpSubscriptionLink
If you are unsure which one to use, we recommend using SSE for subscriptions as it's easier to setup and don't require setting up a WebSocket server.
Reference projects
Type | Example Type | Link |
---|---|---|
WebSockets | Bare-minimum Node.js WebSockets example | /examples/standalone-server |
SSE | Full-stack SSE implementation | github.com/trpc/examples-next-sse-chat |
WebSockets | Full-stack WebSockets implementation | github.com/trpc/examples-next-prisma-websockets-starter |
Basic example
For a full example, see our full-stack SSE example.
server.tsts
import { initTRPC } from '@trpc/server';const t = initTRPC.create();const ee = new EventEmitter();export const appRouter = router({onPostAdd: publicProcedure.subscription(async function* (opts) {// listen for new eventsfor await (const [data] of on(ee, 'add', {// Passing the AbortSignal from the request automatically cancels the event emitter when the request is abortedsignal: opts.signal,})) {const post = data as Post;yield post;}}),});
server.tsts
import { initTRPC } from '@trpc/server';const t = initTRPC.create();const ee = new EventEmitter();export const appRouter = router({onPostAdd: publicProcedure.subscription(async function* (opts) {// listen for new eventsfor await (const [data] of on(ee, 'add', {// Passing the AbortSignal from the request automatically cancels the event emitter when the request is abortedsignal: opts.signal,})) {const post = data as Post;yield post;}}),});
Automatic tracking of id using tracked()
(recommended)
If you yield
an event using our tracked()
-helper and include an id
, the client will automatically reconnect when it gets disconnected and send the last known ID.
You can send an initial lastEventId
when initializing the subscription and it will be automatically updated as the browser receives data.
- For SSE, this is part of the
EventSource
-spec and will be propagated throughlastEventId
in your.input()
. - For WebSockets, our
wsLink
will automatically send the last known ID and update it as the browser receives data.
If you're fetching data based on the lastEventId
, and capturing all events is critical, make sure you setup the event listener before fetching events from your database as is done in our full-stack SSE example, this can prevent newly emitted events being ignored while yield'ing the original batch based on lastEventId
.
ts
import EventEmitter, { on } from 'events';import type { Post } from '@prisma/client';import { tracked } from '@trpc/server';import { z } from 'zod';import { publicProcedure, router } from '../trpc';const ee = new EventEmitter();export const subRouter = router({onPostAdd: publicProcedure.input(z.object({// lastEventId is the last event id that the client has received// On the first call, it will be whatever was passed in the initial setup// If the client reconnects, it will be the last event id that the client receivedlastEventId: z.string().nullish(),}).optional(),).subscription(async function* (opts) {// We start by subscribing to the ee so that we don't miss any new events while fetchingconst iterable = ee.toIterable('add', {// Passing the AbortSignal from the request automatically cancels the event emitter when the request is abortedsignal: opts.signal,});if (opts.input.lastEventId) {// [...] get the posts since the last event id and yield them// const items = await db.post.findMany({ ... })// for (const item of items) {// yield tracked(item.id, item);// }}// listen for new eventsfor await (const [data] of on(ee, 'add', {signal: opts.signal,})) {const post = data as Post;// tracking the post id ensures the client can reconnect at any time and get the latest events this idyield tracked(post.id, post);}}),});
ts
import EventEmitter, { on } from 'events';import type { Post } from '@prisma/client';import { tracked } from '@trpc/server';import { z } from 'zod';import { publicProcedure, router } from '../trpc';const ee = new EventEmitter();export const subRouter = router({onPostAdd: publicProcedure.input(z.object({// lastEventId is the last event id that the client has received// On the first call, it will be whatever was passed in the initial setup// If the client reconnects, it will be the last event id that the client receivedlastEventId: z.string().nullish(),}).optional(),).subscription(async function* (opts) {// We start by subscribing to the ee so that we don't miss any new events while fetchingconst iterable = ee.toIterable('add', {// Passing the AbortSignal from the request automatically cancels the event emitter when the request is abortedsignal: opts.signal,});if (opts.input.lastEventId) {// [...] get the posts since the last event id and yield them// const items = await db.post.findMany({ ... })// for (const item of items) {// yield tracked(item.id, item);// }}// listen for new eventsfor await (const [data] of on(ee, 'add', {signal: opts.signal,})) {const post = data as Post;// tracking the post id ensures the client can reconnect at any time and get the latest events this idyield tracked(post.id, post);}}),});
Cleanup of side effects
If you need to clean up any side-effects of your subscription you can use the try...finally
pattern, as trpc
invokes the .return()
of the Generator Instance when the subscription stops for any reason.
ts
import EventEmitter, { on } from 'events';import type { Post } from '@prisma/client';import { z } from 'zod';import { publicProcedure, router } from '../trpc';const ee = new EventEmitter();export const subRouter = router({onPostAdd: publicProcedure.subscription(async function* (opts) {let timeout;try {for await (const [data] of on(ee, 'add', {signal: opts.signal,})) {timeout = setTimeout(() => console.log('Pretend like this is useful'));const post = data as Post;yield post;}} finally {if (timeout) clearTimeout(timeout);}}),});
ts
import EventEmitter, { on } from 'events';import type { Post } from '@prisma/client';import { z } from 'zod';import { publicProcedure, router } from '../trpc';const ee = new EventEmitter();export const subRouter = router({onPostAdd: publicProcedure.subscription(async function* (opts) {let timeout;try {for await (const [data] of on(ee, 'add', {signal: opts.signal,})) {timeout = setTimeout(() => console.log('Pretend like this is useful'));const post = data as Post;yield post;}} finally {if (timeout) clearTimeout(timeout);}}),});
Error handling
Throwing an error in a generator function propagates to trpc
's onError()
on the backend, but the error will not be sent to the client - the client will automatically reconnect based on the last event id that is tracked using tracked()
.
If this is surprising behavior to you and you have a finite amount of data to send, you should consider using httpBatchStreamLink instead.
Output validation
Since subscriptions are async iterators, you have to go through the iterator to validate the output.
Example with zod
zAsyncGenerator.tsts
import type { TrackedEnvelope } from '@trpc/server';import { isTrackedEnvelope, tracked } from '@trpc/server';import { z } from 'zod';function isAsyncIterable<TValue, TReturn = unknown>(value: unknown,): value is AsyncIterable<TValue, TReturn> {return !!value && typeof value === 'object' && Symbol.asyncIterator in value;}const trackedEnvelopeSchema =z.custom<TrackedEnvelope<unknown>>(isTrackedEnvelope);/*** A Zod schema helper designed specifically for validating async generators. This schema ensures that:* 1. The value being validated is an async iterable.* 2. Each item yielded by the async iterable conforms to a specified type.* 3. The return value of the async iterable, if any, also conforms to a specified type.*/export function zAsyncGenerator<TYieldIn,TYieldOut,TReturnIn = void,TReturnOut = void,Tracked extends boolean = false,>(opts: {/*** Validate the value yielded by the async generator*/yield: z.ZodType<TYieldIn, any, TYieldOut>;/*** Validate the return value of the async generator* @remark not applicable for subscriptions*/return?: z.ZodType<TReturnIn, any, TReturnOut>;/*** Whether if the yielded values are tracked* @remark only applicable for subscriptions*/tracked?: Tracked;}) {return z.custom<AsyncGenerator<Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,TReturnIn>>((val) => isAsyncIterable(val)).transform(async function* (iter) {const iterator = iter[Symbol.asyncIterator]();let next;while ((next = await iterator.next()) && !next.done) {if (opts.tracked) {const [id, data] = trackedEnvelopeSchema.parse(next.value);yield tracked(id, await opts.yield.parseAsync(data));continue;}yield opts.yield.parseAsync(next.value);}if (opts.return) {return await opts.return.parseAsync(next.value);}return;}) as z.ZodType<AsyncGenerator<Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,TReturnIn,unknown>,any,AsyncGenerator<Tracked extends true ? TrackedEnvelope<TYieldOut> : TYieldOut,TReturnOut,unknown>>;}
zAsyncGenerator.tsts
import type { TrackedEnvelope } from '@trpc/server';import { isTrackedEnvelope, tracked } from '@trpc/server';import { z } from 'zod';function isAsyncIterable<TValue, TReturn = unknown>(value: unknown,): value is AsyncIterable<TValue, TReturn> {return !!value && typeof value === 'object' && Symbol.asyncIterator in value;}const trackedEnvelopeSchema =z.custom<TrackedEnvelope<unknown>>(isTrackedEnvelope);/*** A Zod schema helper designed specifically for validating async generators. This schema ensures that:* 1. The value being validated is an async iterable.* 2. Each item yielded by the async iterable conforms to a specified type.* 3. The return value of the async iterable, if any, also conforms to a specified type.*/export function zAsyncGenerator<TYieldIn,TYieldOut,TReturnIn = void,TReturnOut = void,Tracked extends boolean = false,>(opts: {/*** Validate the value yielded by the async generator*/yield: z.ZodType<TYieldIn, any, TYieldOut>;/*** Validate the return value of the async generator* @remark not applicable for subscriptions*/return?: z.ZodType<TReturnIn, any, TReturnOut>;/*** Whether if the yielded values are tracked* @remark only applicable for subscriptions*/tracked?: Tracked;}) {return z.custom<AsyncGenerator<Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,TReturnIn>>((val) => isAsyncIterable(val)).transform(async function* (iter) {const iterator = iter[Symbol.asyncIterator]();let next;while ((next = await iterator.next()) && !next.done) {if (opts.tracked) {const [id, data] = trackedEnvelopeSchema.parse(next.value);yield tracked(id, await opts.yield.parseAsync(data));continue;}yield opts.yield.parseAsync(next.value);}if (opts.return) {return await opts.return.parseAsync(next.value);}return;}) as z.ZodType<AsyncGenerator<Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,TReturnIn,unknown>,any,AsyncGenerator<Tracked extends true ? TrackedEnvelope<TYieldOut> : TYieldOut,TReturnOut,unknown>>;}
Now you can use this helper to validate the output of your subscription procedures:
_app.tsts
import { publicProcedure, router } from '../trpc';import { zAsyncGenerator } from './zAsyncGenerator';export const appRouter = router({mySubscription: publicProcedure.input(z.object({lastEventId: z.coerce.number().min(0).optional(),}),).output(zAsyncGenerator({yield: z.object({count: z.number(),}),tracked: true,}),).subscription(async function* (opts) {let index = opts.input.lastEventId ?? 0;while (true) {index++;yield tracked(index, {count: index,});await new Promise((resolve) => setTimeout(resolve, 1000));}}),});
_app.tsts
import { publicProcedure, router } from '../trpc';import { zAsyncGenerator } from './zAsyncGenerator';export const appRouter = router({mySubscription: publicProcedure.input(z.object({lastEventId: z.coerce.number().min(0).optional(),}),).output(zAsyncGenerator({yield: z.object({count: z.number(),}),tracked: true,}),).subscription(async function* (opts) {let index = opts.input.lastEventId ?? 0;while (true) {index++;yield tracked(index, {count: index,});await new Promise((resolve) => setTimeout(resolve, 1000));}}),});