Subscriptions

A subscription returns an async generator, an AsyncIterable or an Effect Stream. Clients iterate it with for await. Breaking out of the loop, or aborting signal, closes the connection.

const const onPost: BuildProcedure<RootDef<{}, {}>, "subscription", {
    output: {
        id: string;
        title: string;
    };
    errors: never;
    services: never;
}>onPost = const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure.ProcedureBuilder<RootDef<{}, {}>>.subscription<AsyncGenerator<{
    id: string;
    title: string;
}, void, unknown>>(resolver: (opts: SubscriptionResolverOpts<RootDef<{}, {}>>) => AsyncGenerator<{
    id: string;
    title: string;
}, void, unknown>): BuildProcedure<RootDef<{}, {}>, "subscription", {
    output: {
        id: string;
        title: string;
    };
    errors: never;
    services: never;
}>subscription(async function* (opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts) {
  for await (const const post: {
    id: string;
    title: string;
}post of const posts: {
    watch(opts: {
        signal: AbortSignal;
    }): AsyncIterable<{
        id: string;
        title: string;
    }>;
}posts.function watch(opts: {
    signal: AbortSignal;
}): AsyncIterable<{
    id: string;
    title: string;
}>watch({ signal: AbortSignalsignal: opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts.ResolverOpts<RootDef<{}, {}>>.signal: AbortSignalsignal })) {
    yield const post: {
    id: string;
    title: string;
}post;
  }
});

for await (const const post: {
    id: string;
    title: string;
}post of const client: TrpcClient<{
    onPost: BuildProcedure<RootDef<{}, {}>, "subscription", {
        output: {
            id: string;
            title: string;
        };
        errors: never;
        services: never;
    }>;
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
}, any>]>>client.onPost: {
    subscribe(input?: undefined, opts?: SubscribeOptions<{}> | undefined): TrpcSubscription<{
        id: string;
        title: string;
    }, UnexpectedError>;
}onPost.function subscribe(input?: undefined, opts?: SubscribeOptions<{}> | undefined): TrpcSubscription<{
    id: string;
    title: string;
}, UnexpectedError>subscribe()) {
  var console: Consoleconsole.Console.log(...data: any[]): voidlog(const post: {
    id: string;
    title: string;
}post.title);                   
title: string
}

.output(Event) validates each event (the data of tracked() events); .output(stream(Event)) does the same, matching streamed queries.

Resuming

Yield tracked(id, data) to give events ids. SSE sends them as id:, and on reconnect the resolver gets lastEventId in its options (not in its input). An id can't be empty or hold a line break or NUL, which would end the SSE field:

const onPost = t.procedure.subscription(async function* ({ lastEventId }) {
  for await (const post of posts.since(lastEventId))
    yield tracked(post.id, post);
});

client.onPost.subscribe(undefined, { lastEventId: '42' }); // resume from a known id

httpLink reconnects a dropped subscription with backoff (reconnect: { attempts, delayMs }). wsLink resends lastEventId after reconnecting. A subscription whose resolver has side effects that must not rerun can opt out with subscribe(input, { reconnect: false }); a dropped connection then ends it with NETWORK_ERROR.

While a subscription is quiet, the handler sends a : ping comment every 15 seconds so proxies don't close it as idle. Set createFetchHandler({ sse: { pingMs } }), or false to turn it off.

Reading with EventSource

The stream is standard SSE, so a browser EventSource (or curl) can read a subscription without the client. Input goes in the query string as httpLink sends it (/trpc/onPost?input={"channel":"news"}, URL-encoded). Each message event's data is a serialized value. An EventSource reconnects with Last-Event-ID by itself, so close it on the done event, or it reconnects when the stream ends. Failures arrive as error events with data:

const source = new EventSource('/trpc/onPost');
source.onmessage = (event) => render(JSON.parse(event.data));
source.addEventListener('done', () => source.close());

Over WebSockets

wsLink runs subscriptions (and everything else) over one WebSocket; switchLink can send subscriptions there and the rest over HTTP (Links). On Cloudflare, Durable Objects can hibernate between events.

Publishing events

To publish from mutations to subscribers, across processes and with resumable ids, use a Publisher.