Streaming | tRPC

A query or mutation can return an async generator or an Effect Stream, and the client gets an AsyncIterable. Outputs can also hold promises and async iterables at any depth: the rest of the output arrives first, and each deferred value streams as it settles.

const const countdown: BuildProcedure<RootDef<Context, {}>, "query", {
    output: AsyncGenerator<number, string, unknown>;
    errors: never;
    services: never;
    unmapped: never;
}>countdown = const t: TrpcRoot<Context, {}, RootDef<Context, {}>>t.TrpcRoot<Context, {}, RootDef<Context, {}>>.procedure: ProcedureBuilder<RootDef<Context, {}>>procedure.ProcedureBuilder<RootDef<Context, {}>>.query<AsyncGenerator<number, string, unknown>>(resolver: (opts: ResolverOpts<RootDef<Context, {}>>) => AsyncGenerator<number, string, unknown>): BuildProcedure<RootDef<Context, {}>, "query", {
    output: AsyncGenerator<number, string, unknown>;
    errors: never;
    services: never;
    unmapped: never;
}>query(async function* () {
  for (let let i: numberi = 3; let i: numberi > 0; let i: numberi--) {
    yield let i: numberi;
  }
  return 'liftoff';
});

const const page: BuildProcedure<RootDef<Context, {}>, "query", {
    output: {
        title: string;
        comments: Promise<string[]>;
    };
    errors: never;
    services: never;
    unmapped: never;
}>page = const t: TrpcRoot<Context, {}, RootDef<Context, {}>>t.TrpcRoot<Context, {}, RootDef<Context, {}>>.procedure: ProcedureBuilder<RootDef<Context, {}>>procedure.ProcedureBuilder<RootDef<Context, {}>>.query<{
    title: string;
    comments: Promise<string[]>;
}>(resolver: (opts: ResolverOpts<RootDef<Context, {}>>) => {
    title: string;
    comments: Promise<string[]>;
}): BuildProcedure<RootDef<Context, {}>, "query", {
    output: {
        title: string;
        comments: Promise<string[]>;
    };
    errors: never;
    services: never;
    unmapped: never;
}>query(() => ({
  title: stringtitle: 'Post',
  comments: Promise<string[]>comments: var Promise: PromiseConstructorPromise.PromiseConstructor.resolve<string[]>(value: string[]): Promise<string[]> (+2 overloads)resolve(['First!']),
}));

Over HTTP, httpLink reads such outputs as JSONL (or binary frames, Files); wsLink, experimental_messagePortLink and localLink carry them as they are. Every serializer streams deferred values.

Validating items

To validate each item, wrap the item schema in stream(); a second schema validates the generator's return value, which the client gets as the iterator's final value:

import { function stream<TItem extends AnySchema, TReturn extends AnySchema | undefined = undefined>(item: TItem, returns?: TReturn): StreamSchema<TItem, TReturn>stream } from '@trpc12/trpc/server';

const const chat: BuildProcedure<With<RootDef<Context, {}>, {
    outputIn: StreamOutput<{
        text: string;
    }, {
        tokens: number;
    }>;
    outputOut: StreamOutput<{
        text: string;
    }, {
        tokens: number;
    }>;
    requires: never;
}>, "mutation", {
    output: AsyncGenerator<{
        text: string;
    }, {
        tokens: number;
    }, unknown>;
    errors: never;
    services: never;
    unmapped: never;
}>chat = const t: TrpcRoot<Context, {}, RootDef<Context, {}>>t.TrpcRoot<Context, {}, RootDef<Context, {}>>.procedure: ProcedureBuilder<RootDef<Context, {}>>procedure
  .ProcedureBuilder<RootDef<Context, {}>>.output<StreamSchema<z.ZodObject<{
    text: z.ZodString;
}, z.core.$strip>, z.ZodObject<{
    tokens: z.ZodNumber;
}, z.core.$strip>>>(schema: StreamSchema<z.ZodObject<{
    text: z.ZodString;
}, z.core.$strip>, z.ZodObject<{
    tokens: z.ZodNumber;
}, z.core.$strip>>, opts?: {
    validate?: true;
}): ProcedureBuilder<With<RootDef<Context, {}>, {
    outputIn: StreamOutput<{
        text: string;
    }, {
        tokens: number;
    }>;
    outputOut: StreamOutput<{
        text: string;
    }, {
        tokens: number;
    }>;
    requires: never;
}>> (+1 overload)output(stream<z.ZodObject<{
    text: z.ZodString;
}, z.core.$strip>, z.ZodObject<{
    tokens: z.ZodNumber;
}, z.core.$strip>>(item: z.ZodObject<{
    text: z.ZodString;
}, z.core.$strip>, returns?: z.ZodObject<{
    tokens: z.ZodNumber;
}, z.core.$strip> | undefined): StreamSchema<z.ZodObject<{
    text: z.ZodString;
}, z.core.$strip>, z.ZodObject<{
    tokens: z.ZodNumber;
}, z.core.$strip>>stream(const Token: z.ZodObject<{
    text: z.ZodString;
}, z.core.$strip>Token, const Summary: z.ZodObject<{
    tokens: z.ZodNumber;
}, z.core.$strip>Summary))
  .ProcedureBuilder<With<RootDef<Context, {}>, { outputIn: StreamOutput<{ text: string; }, { tokens: number; }>; outputOut: StreamOutput<{ text: string; }, { tokens: number; }>; requires: never; }>>.mutation<AsyncGenerator<{
    text: string;
}, {
    tokens: number;
}, unknown>>(resolver: (opts: ResolverOpts<With<RootDef<Context, {}>, {
    outputIn: StreamOutput<{
        text: string;
    }, {
        tokens: number;
    }>;
    outputOut: StreamOutput<{
        text: string;
    }, {
        tokens: number;
    }>;
    requires: never;
}>>) => AsyncGenerator<{
    text: string;
}, {
    tokens: number;
}, unknown> & (MaybePromise<AnyTrpcError | AsyncIterable<{
    text: string;
}, {
    tokens: number;
}> | Stream<{
    text: string;
}, any, any>> | Effect<...>)): BuildProcedure<...>mutation(async function* () {
    yield { text: stringtext: 'He' };
    yield { text: stringtext: 'llo' };
    return { tokens: numbertokens: 2 };
  });

const it = await client.chat.mutate(); // AsyncIterable<Token, Summary>

A plain .output(Item) on a streaming resolver is a type error that points to stream(). An invalid item fails the stream with INTERNAL_SERVER_ERROR after the items before it have been delivered. Streams nested inside an object output aren't validated.

Over HTTP the server reads a stream only as fast as the client reads the response: a generator waits at its yield while the client is behind, and its finally runs when the client leaves. WebSocket links don't wait yet; there, a slow client buffers on the server.

Streams and subscriptions

A streamed query or mutation is one call whose result arrives in pieces; the client consumes it once. A subscription is a stream of events the client follows, with tracked() ids to resume after a reconnect. For progress of long jobs in TanStack Query, see Streamed mutations.