feat(client): add iterate() method to subscriptions - #7461
Conversation
Sorry, something went wrong.
…rdown - close the stream when the abort signal fires so a pending reader is not left hanging if the source goes silent after teardown - drop source emissions that arrive after the stream was cancelled (previously threw 'Controller is already closed') - detach the abort listener once the stream has settled - skip subscribing entirely when the signal is already aborted
Expose subscriptions on the vanilla client as an AsyncIterable, e.g. for
use with TanStack Query's streamedQuery:
for await (const data of client.onPostAdd.iterate()) {
// ...
}
The iterable is cold: each new iterator starts a new subscription which
is torn down when iteration stops early. Connection state envelopes
(started/stopped/state) are filtered out and errors reject the iterator.
Aborting the optional signal ends the iteration.
Closes trpc#6868
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Sorry, something went wrong.
WalkthroughThe client now exposes typed subscription results as AsyncIterable values. Iteration supports abort signals, reconnects, early termination, errors, completion, and cleanup. Tests cover HTTP and WebSocket subscriptions, and the vanilla client guide documents both subscription APIs. ChangesAsyncIterable subscription support
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant ClientProxy
participant TRPCUntypedClient
participant SubscriptionObservable
ClientProxy->>TRPCUntypedClient: invoke subscription iterate
TRPCUntypedClient->>SubscriptionObservable: create subscription observable
SubscriptionObservable-->>TRPCUntypedClient: emit subscription data
TRPCUntypedClient-->>ClientProxy: yield async-iterable value
ClientProxy->>TRPCUntypedClient: abort or stop iteration
TRPCUntypedClient->>SubscriptionObservable: unsubscribe and clean up
Possibly related PRs
Suggested reviewers: katt 🚥 Pre-merge checks | ✅ 5 ✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. ❤️ ShareComment @coderabbitai help to get the list of available commands. |
Sorry, something went wrong.
|
@trpc/client
npm i https://pkg.pr.new/@trpc/client@7461
npm i https://pkg.pr.new/@trpc/next@7461
npm i https://pkg.pr.new/@trpc/openapi@7461
npm i https://pkg.pr.new/@trpc/react-query@7461
npm i https://pkg.pr.new/@trpc/server@7461
npm i https://pkg.pr.new/@trpc/tanstack-react-query@7461
npm i https://pkg.pr.new/@trpc/upgrade@7461 commit: f8c4bda |
Sorry, something went wrong.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)packages/server/src/observable/observable.ts (1)165-196: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Do not add an abort listener after synchronous settlement.
If observable.subscribe() completes or errors synchronously, onSettled() runs before the listener registration. Lines 192-196 then add onAbort after settlement, so the listener remains attached to signal.
Only add the listener while the stream is still active.
Proposed fix- } else { + } else if (active) { signal.addEventListener('abort', onAbort, { once: true }); }Add coverage for a synchronous completion followed by ac.abort() to verify that no abort listener remains.
🤖 Prompt for AI AgentsVerify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/server/src/observable/observable.ts` around lines 165 - 196, Update the subscription setup around observable.subscribe and the subsequent signal handling so the abort listener is registered only when the stream remains active after synchronous completion or error settlement. Preserve immediate onAbort handling for an already-aborted signal, and add coverage for synchronous completion followed by ac.abort() verifying no abort listener remains.
packages/client/src/createTRPCClient.ts (1)🤖 Prompt for all review comments with AI agentspackages/tests/server/httpSubscriptionLink.test.ts (1)164-164: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Remove the non-null assertion.
Guard the result of pathCopy.pop() before using it. This lets TypeScript narrow clientCallType without !.
Proposed fix- const clientCallType = pathCopy.pop()!; + const clientCallType = pathCopy.pop(); + if (!clientCallType) { + throw new Error('Missing client call type'); + }Confirm that createRecursiveProxy does not intentionally invoke this handler with an empty path. As per coding guidelines, “Avoid non-null assertions (!). Use proper type guards and optional chaining instead of non-null assertions.”
🤖 Prompt for AI AgentsVerify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/client/src/createTRPCClient.ts` at line 164, Update createRecursiveProxy to guard the result of pathCopy.pop() before assigning or using clientCallType, removing the non-null assertion and allowing TypeScript to narrow the value. Preserve the existing behavior for valid non-empty paths, and explicitly handle an empty path rather than invoking the handler with an undefined clientCallType.Source: Coding guidelines
207-383: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Use a scoped test resource in each test.
These tests need both server and client setup. Create the resource in each test with await using ctx = testServerAndClientResource(...). Use ctx.client directly instead of destructuring client.
This keeps resource cleanup lexical and follows the required test isolation pattern. As per coding guidelines, “ALWAYS use await using ctx = testServerAndClientResource(...) in tests that need both server and client setup”, “Use ctx.client from the test resource for making tRPC calls”, and “Avoid overzealous object destructuring; prefer direct property access.”
🤖 Prompt for AI AgentsVerify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/tests/server/httpSubscriptionLink.test.ts` around lines 207 - 383, Update each async iterable test to create its own scoped resource with await using ctx = testServerAndClientResource(...), ensuring the required setup arguments match the existing test context. Replace destructured client usage with ctx.client, and rely on lexical resource cleanup and isolation within every test.Source: Coding guidelines
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. Outside diff comments: In `@packages/server/src/observable/observable.ts`: - Around line 165-196: Update the subscription setup around observable.subscribe and the subsequent signal handling so the abort listener is registered only when the stream remains active after synchronous completion or error settlement. Preserve immediate onAbort handling for an already-aborted signal, and add coverage for synchronous completion followed by ac.abort() verifying no abort listener remains. --- Nitpick comments: In `@packages/client/src/createTRPCClient.ts`: - Line 164: Update createRecursiveProxy to guard the result of pathCopy.pop() before assigning or using clientCallType, removing the non-null assertion and allowing TypeScript to narrow the value. Preserve the existing behavior for valid non-empty paths, and explicitly handle an empty path rather than invoking the handler with an undefined clientCallType. In `@packages/tests/server/httpSubscriptionLink.test.ts`: - Around line 207-383: Update each async iterable test to create its own scoped resource with await using ctx = testServerAndClientResource(...), ensuring the required setup arguments match the existing test context. Replace destructured client usage with ctx.client, and rely on lexical resource cleanup and isolation within every test.
ℹ️ Review info ⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 15272a9e-6776-4215-a7d8-047ea7fcd4ba
📥 CommitsReviewing files that changed from the base of the PR and between acff823 and c59a1e0.
📒 Files selected for processing (7)- packages/client/src/createTRPCClient.ts
- packages/client/src/internals/TRPCUntypedClient.ts
- packages/server/src/observable/observable.test.ts
- packages/server/src/observable/observable.ts
- packages/tests/server/httpSubscriptionLink.test.ts
- packages/tests/server/websockets.test.ts
- www/docs/client/vanilla/setup.mdx
Sorry, something went wrong.
This comment was marked as resolved.
Adds regression coverage for observableToAsyncIterable: - breaking out of iteration unsubscribes the underlying subscription even when the signal is never aborted (the subscriptionAsIterable fallback path) - a consumer-side cancel() detaches the abort listener without the signal ever being aborted Also clarifies why the ReadableStream cancel() handler delegates to onAbort (which already detaches the listener) rather than onSettled.
|
Good questions - I double-checked both. On early break: teardown doesn't rely on the signal at all. break calls the iterator's return(), which cancels the underlying ReadableStream and unsubscribes the subscription, so the fallback controller being orphaned is harmless. Added a test proving unsubscribe happens with a never-aborted signal. On cancel() vs onSettled: the stream's cancel() handler delegates to onAbort(), whose first line already detaches the abort listener (and unsubscribes), so the listener is never left attached. There's now a test spying on removeEventListener to prove it, plus a comment in cancel() clarifying why onSettled isn't needed there. Best Regards, Tarik |
Sorry, something went wrong.
Reviewers
At least 1 approving review is required to merge this pull request.
Closes #6868
What changed
Subscription procedures on the vanilla client can now be consumed as an AsyncIterable:
iterate(input, opts?) uses the existing observableToAsyncIterable helper. It filters connection-state envelopes, yields subscription data with the same output typing as onData, and propagates link errors unchanged. The iterable is cold, so each iterator owns one subscription and tears it down on break, return, or abort.
While adding this, I also fixed a few teardown cases in observableToReadableStream: late emissions after cancellation are ignored, abort closes a pending reader, listeners are detached after settlement, and an already-aborted signal does not start a request.
The vanilla client docs now show both callback subscriptions and iterate().
Verification
Notes
Checklist
Best Regards, Tarik