Subscriptions in TanStack Query

A subscription procedure has three helpers. subscriptionOptions() feeds your framework's subscription primitive (useSubscription, createSubscription, injectSubscription or createSubscriptionController), outside the query cache. liveOptions() and streamedOptions() put the subscription in the query cache instead, for useQuery.

Outside the query cache

Each framework's primitive gives status (idle, connecting, active, error), the latest event as data, error and reset():

function function NewPosts(props: {
    channel: string;
}): React.JSX.ElementNewPosts(props: {
    channel: string;
}props: { channel: stringchannel: string }) {
  const const sub: UseSubscriptionResult<{
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | ... 20 more ... | (TrpcError<...> & {
    ...;
})>>sub = useSubscription<{
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | ... 20 more ... | (TrpcError<...> & {
    ...;
})>>(options: TrpcSubscriptionOptions<...>): UseSubscriptionResult<...>useSubscription(
    const trpc: TrpcQueryUtils<TrpcClient<{
    post: {
        onAdd: Procedure<{
            type: "subscription";
            ctx: Context;
            meta: {};
            input: {
                channel: string;
            };
            output: TrackedEnvelope<Post>;
            errors: SchemaValidationError<z.ZodObject<{
                channel: z.ZodString;
            }, z.core.$strip>>;
            services: never;
        }>;
    };
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
}, any>]>>>trpc.post: TrpcQueryUtils<ClientRouter<{
    onAdd: Procedure<{
        type: "subscription";
        ctx: Context;
        meta: {};
        input: {
            channel: string;
        };
        output: TrackedEnvelope<Post>;
        errors: SchemaValidationError<z.ZodObject<{
            channel: z.ZodString;
        }, z.core.$strip>>;
        services: never;
    }>;
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
}, any>]>>>post.onAdd: SubscriptionProcedureUtils<{
    channel: string;
}, {
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | ... 20 more ... | (TrpcError<...> & {
    ...;
})>, Omit<...>>onAdd.SubscriptionProcedureUtils<{ channel: string; }, { readonly id: string; readonly data: { id: string; title: string; author: string | null; createdAt: string; }; }, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & { ...; }) | ... 24 more ... | (TrpcError<...> & { ...; })>, Omit<...>>.subscriptionOptions(opts: {
    input: {
        channel: string;
    } | unique symbol;
} & ContextBag<Omit<NonNullable<{
    signal?: AbortSignal | undefined;
} & BuiltinCallOptions & {} & {
    lastEventId?: string | undefined;
}>, "signal" | "lastEventId">> & {
    enabled?: boolean | undefined;
} & SubscriptionCallbacks<{
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | ... 23 more ... | (TrpcError<...> & {
    ...;
})>>): TrpcSubscriptionOptions<...>subscriptionOptions({
      input: {
    channel: string;
} | unique symbolinput: { channel: stringchannel: props: {
    channel: string;
}props.channel: stringchannel },
      SubscriptionCallbacks<{ readonly id: string; readonly data: { id: string; title: string; author: string | null; createdAt: string; }; }, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & { ...; }) | ... 24 more ... | (TrpcError<...> & { ...; })>>.onData?: ((event: {
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}) => void) | undefinedonData: (event: {
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}event) => function toast(message: string): voidtoast(event: {
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}event.data: {
    id: string;
    title: string;
    author: string | null;
    createdAt: string;
}data.title: stringtitle),
    }),
  );
  if (const sub: UseSubscriptionResult<{
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
}, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | ... 20 more ... | (TrpcError<...> & {
    ...;
})>>sub.status: "idle" | "connecting" | "active" | "error"status === 'error') return <React.JSX.IntrinsicElements.button: React.DetailedHTMLProps<React.ButtonHTMLAttributes<HTMLButtonElement>, HTMLButtonElement>button React.DOMAttributes<HTMLButtonElement>.onClick?: React.MouseEventHandler<HTMLButtonElement> | undefinedonClick={const sub: {
    status: "error";
    data: {
        readonly id: string;
        readonly data: {
            id: string;
            title: string;
            author: string | null;
            createdAt: string;
        };
    } | undefined;
    error: NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
        readonly defined: false;
    }) | (TrpcError<"PARSE_ERROR", undefined> & {
        readonly defined: false;
    }) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
        readonly defined: false;
    }) | (TrpcError<"UNAUTHORIZED", undefined> & {
        readonly defined: false;
    }) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
        readonly defined: false;
    }) | ... 20 more ... | (TrpcError<...> & {
        ...;
    })>;
} & {
    ...;
}sub.reset: () => voidreset}>Retry</React.JSX.IntrinsicElements.button: React.DetailedHTMLProps<React.ButtonHTMLAttributes<HTMLButtonElement>, HTMLButtonElement>button>;
  return <React.JSX.IntrinsicElements.p: React.DetailedHTMLProps<React.HTMLAttributes<HTMLParagraphElement>, HTMLParagraphElement>p>Latest: {const sub: ({
    status: "idle";
    data: {
        readonly id: string;
        readonly data: {
            id: string;
            title: string;
            author: string | null;
            createdAt: string;
        };
    } | undefined;
    error: null;
} & {
    ...;
}) | ({
    ...;
} & {
    ...;
}) | ({
    ...;
} & {
    ...;
})sub.data: {
    readonly id: string;
    readonly data: {
        id: string;
        title: string;
        author: string | null;
        createdAt: string;
    };
} | undefineddata?.data: {
    id: string;
    title: string;
    author: string | null;
    createdAt: string;
}data.title: string | undefinedtitle}</React.JSX.IntrinsicElements.p: React.DetailedHTMLProps<React.HTMLAttributes<HTMLParagraphElement>, HTMLParagraphElement>p>;
}

idle means disabled (enabled: false or input: skipToken), not started yet, or ended by the server; connecting means started with no event yet; error stays until reset() starts it again.
onData, onError and onComplete always run with the latest options, so changing them doesn't restart the subscription. A new input (a new key) or enabled does.
Components with the same client and key share one subscription. A late one starts from the current state, each component's onData runs for every later event, reset() restarts it for all of them, and it closes when the last one unmounts. Pass subscriptionCache: undefined to give an observer its own.
Reconnects resume from the last tracked() id, as with any client (Subscriptions).

SubscriptionObserver from @trpc12/tanstack-query is what the primitives are built on, for a framework without a package.

In the query cache

liveOptions() keeps the latest event as the query's data; streamedOptions() keeps every event, built on TanStack's streamedQuery, capped by maxChunks. The subscription closes when no observer is left, and other components read the same data from the cache:

const const price: UseQueryResult<number, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | (TrpcError<"NOT_FOUND", undefined> & {
    readonly defined: false;
}) | ... 18 more ... | (TrpcError<...> & {
    ...;
})>>price = useQuery<number, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | (TrpcError<"NOT_FOUND", undefined> & {
    readonly defined: false;
}) | ... 18 more ... | (TrpcError<...> & {
    ...;
})>, number, TrpcQueryKey & {
    ...;
}>(options: UndefinedInitialDataOptions<...>, queryClient?: QueryClient): UseQueryResult<...> (+2 overloads)useQuery(
  const trpc: TrpcQueryUtils<TrpcClient<{
    price: {
        onTick: Procedure<{
            type: "subscription";
            ctx: Context;
            meta: {};
            input: {
                symbol: string;
            };
            output: number;
            errors: SchemaValidationError<z.ZodObject<{
                symbol: z.ZodString;
            }, z.core.$strip>>;
            services: never;
        }>;
    };
    chat: {
        onMessage: Procedure<{
            type: "subscription";
            ctx: Context;
            meta: {};
            input: {
                room: string;
            };
            output: {
                text: string;
            };
            errors: SchemaValidationError<z.ZodObject<{
                room: z.ZodString;
            }, z.core.$strip>>;
            services: never;
        }>;
    };
}, {}, undefined, LinksTypes<...>>>trpc.price: TrpcQueryUtils<ClientRouter<{
    onTick: Procedure<{
        type: "subscription";
        ctx: Context;
        meta: {};
        input: {
            symbol: string;
        };
        output: number;
        errors: SchemaValidationError<z.ZodObject<{
            symbol: z.ZodString;
        }, z.core.$strip>>;
        services: never;
    }>;
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
}, any>]>>>price.onTick: SubscriptionProcedureUtils<{
    symbol: string;
}, number, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | ... 19 more ... | (TrpcError<...> & {
    ...;
})>, Omit<...>>onTick.SubscriptionProcedureUtils<{ symbol: string; }, number, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & { readonly defined: false; }) | (TrpcError<"PARSE_ERROR", undefined> & { readonly defined: false; }) | ... 23 more ... | (TrpcError<...> & { ...; })>, Omit<...>>.liveOptions<number>(opts: {
    input: {
        symbol: string;
    } | unique symbol;
} & ContextBag<Omit<NonNullable<{
    signal?: AbortSignal | undefined;
} & BuiltinCallOptions & {} & {
    lastEventId?: string | undefined;
}>, "signal" | "lastEventId">> & QueryOptionsIn<number, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | ... 22 more ... | (TrpcError<...> & {
    ...;
})>, number>): QueryOptionsOut<...>liveOptions({ input: {
    symbol: string;
} | unique symbolinput: { symbol: stringsymbol: 'TRPC' } }),
);
const const chat: UseQueryResult<{
    text: string;
}[], NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | (TrpcError<"NOT_FOUND", undefined> & {
    readonly defined: false;
}) | ... 18 more ... | (TrpcError<...> & {
    ...;
})>>chat = useQuery<{
    text: string;
}[], NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | (TrpcError<"NOT_FOUND", undefined> & {
    readonly defined: false;
}) | ... 18 more ... | (TrpcError<...> & {
    ...;
})>, {
    text: string;
}[], TrpcQueryKey & {
    ...;
}>(options: UndefinedInitialDataOptions<...>, queryClient?: QueryClient): UseQueryResult<...> (+2 overloads)useQuery(
  const trpc: TrpcQueryUtils<TrpcClient<{
    price: {
        onTick: Procedure<{
            type: "subscription";
            ctx: Context;
            meta: {};
            input: {
                symbol: string;
            };
            output: number;
            errors: SchemaValidationError<z.ZodObject<{
                symbol: z.ZodString;
            }, z.core.$strip>>;
            services: never;
        }>;
    };
    chat: {
        onMessage: Procedure<{
            type: "subscription";
            ctx: Context;
            meta: {};
            input: {
                room: string;
            };
            output: {
                text: string;
            };
            errors: SchemaValidationError<z.ZodObject<{
                room: z.ZodString;
            }, z.core.$strip>>;
            services: never;
        }>;
    };
}, {}, undefined, LinksTypes<...>>>trpc.chat: TrpcQueryUtils<ClientRouter<{
    onMessage: Procedure<{
        type: "subscription";
        ctx: Context;
        meta: {};
        input: {
            room: string;
        };
        output: {
            text: string;
        };
        errors: SchemaValidationError<z.ZodObject<{
            room: z.ZodString;
        }, z.core.$strip>>;
        services: never;
    }>;
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
}, any>]>>>chat.onMessage: SubscriptionProcedureUtils<{
    room: string;
}, {
    text: string;
}, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | ... 19 more ... | (TrpcError<...> & {
    ...;
})>, Omit<...>>onMessage.SubscriptionProcedureUtils<{ room: string; }, { text: string; }, NonNullable<(TrpcError<"BAD_REQUEST", undefined> & { readonly defined: false; }) | (TrpcError<"PARSE_ERROR", undefined> & { readonly defined: false; }) | ... 23 more ... | (TrpcError<...> & { ...; })>, Omit<...>>.streamedOptions<{
    text: string;
}[]>(opts: {
    input: {
        room: string;
    } | unique symbol;
} & ContextBag<Omit<NonNullable<{
    signal?: AbortSignal | undefined;
} & BuiltinCallOptions & {} & {
    lastEventId?: string | undefined;
}>, "signal" | "lastEventId">> & StreamedOptions & QueryOptionsIn<{
    text: string;
}[], NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | ... 22 more ... | (TrpcError<...> & {
    ...;
})>, {
    text: string;
}[]>): QueryOptionsOut<...>streamedOptions({
    input: {
    room: string;
} | unique symbolinput: { room: stringroom: 'general' },
    StreamedOptions.maxChunks?: number | undefinedmaxChunks: 100,
  }),
);
const chat: UseQueryResult<{
    text: string;
}[], NonNullable<(TrpcError<"BAD_REQUEST", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PARSE_ERROR", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNSUPPORTED_PROTOCOL", undefined> & {
    readonly defined: false;
}) | (TrpcError<"UNAUTHORIZED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"PAYMENT_REQUIRED", undefined> & {
    readonly defined: false;
}) | (TrpcError<"FORBIDDEN", undefined> & {
    readonly defined: false;
}) | (TrpcError<"NOT_FOUND", undefined> & {
    readonly defined: false;
}) | ... 18 more ... | (TrpcError<...> & {
    ...;
})>>chat.data;     
data: {
    text: string;
}[] | undefined

maxChunks keeps only the newest events. refetchMode says what a refetch does with the events so far: 'reset' (the default) clears them, 'append' keeps them and adds the new ones, and 'replace' swaps them once the new stream ends.