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: Procedure<{
    type: "query";
    ctx: Context;
    meta: {};
    input: undefined;
    output: AsyncGenerator<number, string, unknown>;
    errors: never;
    services: 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>): Procedure<{
    type: "query";
    ctx: Context;
    meta: {};
    input: undefined;
    output: AsyncGenerator<number, string, unknown>;
    errors: never;
    services: 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: Procedure<{
    type: "query";
    ctx: Context;
    meta: {};
    input: undefined;
    output: {
        title: string;
        comments: Promise<string[]>;
    };
    errors: never;
    services: 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[]>;
}): Procedure<{
    type: "query";
    ctx: Context;
    meta: {};
    input: undefined;
    output: {
        title: string;
        comments: Promise<string[]>;
    };
    errors: never;
    services: 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 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: Procedure<{
    type: "mutation";
    ctx: Context;
    meta: {};
    input: undefined;
    output: AsyncIterable<{
        text: string;
    }, {
        tokens: number;
    }>;
    errors: never;
    services: 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<...>)): Procedure<...>mutation(async function* () {
    yield { text: stringtext: 'He' };
    yield { text: stringtext: 'llo' };
    return { tokens: numbertokens: 2 };
  });

const const it: AsyncIterable<{
    text: string;
}, {
    tokens: number;
}>it = await const client: TrpcClient<{
    chat: typeof chat;
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
}, any>]>>client.chat: {
    mutate(input?: undefined, opts?: CallOptions<{}> | undefined): TrpcPromise<AsyncIterable<{
        text: string;
    }, {
        tokens: number;
    }>, CallError<{
        type: "mutation";
        ctx: Context;
        meta: {};
        input: undefined;
        output: AsyncIterable<{
            text: string;
        }, {
            tokens: number;
        }>;
        errors: never;
        services: never;
    }, JsonForm, never>>;
}chat.function mutate(input?: undefined, opts?: CallOptions<{}> | undefined): TrpcPromise<AsyncIterable<{
    text: string;
}, {
    tokens: number;
}>, CallError<{
    type: "mutation";
    ctx: Context;
    meta: {};
    input: undefined;
    output: AsyncIterable<{
        text: string;
    }, {
        tokens: number;
    }>;
    errors: never;
    services: never;
}, JsonForm, never>>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 Long jobs.

The examples/media-pipeline app streams upload progress from an Effect Stream, a report whose nested promises settle one by one, and an album export from an async generator, each validated with stream().