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: Procedure<{
    type: "subscription";
    ctx: {};
    meta: {};
    input: undefined;
    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>): Procedure<{
    type: "subscription";
    ctx: {};
    meta: {};
    input: undefined;
    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: Procedure<{
        type: "subscription";
        ctx: {};
        meta: {};
        input: undefined;
        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;
    }, CallError<{
        type: "subscription";
        ctx: {};
        meta: {};
        input: undefined;
        output: {
            id: string;
            title: string;
        };
        errors: never;
        services: never;
    }, JsonForm, never>>;
}onPost.function subscribe(input?: undefined, opts?: SubscribeOptions<{}> | undefined): TrpcSubscription<{
    id: string;
    title: string;
}, CallError<{
    type: "subscription";
    ctx: {};
    meta: {};
    input: undefined;
    output: {
        id: string;
        title: string;
    };
    errors: never;
    services: never;
}, JsonForm, never>>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 const onPost: Procedure<{
    type: "subscription";
    ctx: {};
    meta: {};
    input: undefined;
    output: TrackedEnvelope<{
        id: string;
        title: string;
    }>;
    errors: never;
    services: never;
}>onPost = const t: TrpcRoot<{}, {}, RootDef<{}, {}>>t.TrpcRoot<{}, {}, RootDef<{}, {}>>.procedure: ProcedureBuilder<RootDef<{}, {}>>procedure.ProcedureBuilder<RootDef<{}, {}>>.subscription<AsyncGenerator<TrackedEnvelope<{
    id: string;
    title: string;
}>, void, unknown>>(resolver: (opts: SubscriptionResolverOpts<RootDef<{}, {}>>) => AsyncGenerator<TrackedEnvelope<{
    id: string;
    title: string;
}>, void, unknown>): Procedure<{
    type: "subscription";
    ctx: {};
    meta: {};
    input: undefined;
    output: TrackedEnvelope<{
        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: {
    since(lastEventId: string | undefined): AsyncIterable<{
        id: string;
        title: string;
    }>;
}posts.function since(lastEventId: string | undefined): AsyncIterable<{
    id: string;
    title: string;
}>since(opts: SubscriptionResolverOpts<RootDef<{}, {}>>opts.SubscriptionResolverOpts<RootDef<{}, {}>>.lastEventId: string | undefinedlastEventId))
    yield tracked<{
    id: string;
    title: string;
}>(id: string, data: {
    id: string;
    title: string;
}): TrackedEnvelope<{
    id: string;
    title: string;
}>tracked(const post: {
    id: string;
    title: string;
}post.id: stringid, const post: {
    id: string;
    title: string;
}post);
});

const client: TrpcClient<{
    onPost: Procedure<{
        type: "subscription";
        ctx: {};
        meta: {};
        input: undefined;
        output: TrackedEnvelope<{
            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<{
        readonly id: string;
        readonly data: {
            id: string;
            title: string;
        };
    }, CallError<{
        type: "subscription";
        ctx: {};
        meta: {};
        input: undefined;
        output: TrackedEnvelope<{
            id: string;
            title: string;
        }>;
        errors: never;
        services: never;
    }, JsonForm, never>>;
}onPost.function subscribe(input?: undefined, opts?: SubscribeOptions<{}> | undefined): TrpcSubscription<{
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
    };
}, CallError<{
    type: "subscription";
    ctx: {};
    meta: {};
    input: undefined;
    output: TrackedEnvelope<{
        id: string;
        title: string;
    }>;
    errors: never;
    services: never;
}, JsonForm, never>>subscribe(var undefinedundefined, { lastEventId?: string | undefinedlastEventId: '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: for a tracked() event, its { id, data }, with the id also in event.lastEventId. 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 const source: EventSourcesource = new var EventSource: new (url: string | URL, eventSourceInitDict?: EventSourceInit) => EventSourceEventSource('/trpc/onPost');
const source: EventSourcesource.EventSource.onmessage: ((this: EventSource, ev: MessageEvent) => any) | nullonmessage = (event: MessageEvent<any>event) => function render(data: unknown): voidrender(var JSON: JSONJSON.JSON.parse(text: string, reviver?: (this: any, key: string, value: any) => any): anyparse(event: MessageEvent<any>event.MessageEvent<any>.data: anydata));
const source: EventSourcesource.EventSource.addEventListener(type: string, listener: (this: EventSource, event: MessageEvent) => any, options?: boolean | AddEventListenerOptions): void (+2 overloads)addEventListener('done', () => const source: EventSourcesource.EventSource.close(): voidclose());

The examples/realtime app reads a room this way, next to wsLink clients, and both resume after the server restarts.

Over WebSockets

wsLink runs subscriptions (and everything else) over one WebSocket; switchLink can send subscriptions there and the rest over HTTP (Links).