From 7db69f6a55fca251958ef8a438139226f31f1163 Mon Sep 17 00:00:00 2001 From: CodeHz Date: Sat, 11 Jul 2026 01:24:10 +0800 Subject: [PATCH 1/3] test: add tests for streaming RpcTarget over RPC --- __tests__/index.test.ts | 113 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts index 486c029..36b2384 100644 --- a/__tests__/index.test.ts +++ b/__tests__/index.test.ts @@ -3005,6 +3005,119 @@ describe("ReadableStream over RPC", () => { expect(cancelCalled).toBe(true); }); + + it("can send an RpcTarget as a stream chunk and call methods on it", async () => { + // The sender enqueues a pass-by-reference RpcTarget into a ReadableStream. The receiver + // reads it back as a stub and can call methods on it, which route back to the sender. + class ChunkTarget extends RpcTarget { + square(i: number) { return i * i; } + get label() { return "chunk-target"; } + } + + let stream = new ReadableStream({ + start(controller) { + controller.enqueue(new ChunkTarget()); + controller.close(); + } + }); + + class StreamReceiver extends RpcTarget { + async receiveStream(stream: ReadableStream) { + let reader = stream.getReader(); + let { value } = await reader.read(); + using chunk = value as any; + let result = { + squared: await chunk.square(5), + label: await chunk.label, + }; + // Drain the closing `done` chunk so the underlying pipe is released. + await reader.read(); + return result; + } + } + + await using harness = new TestHarness(new StreamReceiver()); + let result: any = await harness.stub.receiveStream(stream); + + expect(result.squared).toBe(25); + expect(result.label).toBe("chunk-target"); + }); + + it("can return a ReadableStream containing RpcTargets and call methods on chunks", async () => { + // The server returns a ReadableStream whose chunks are RpcTargets. The client reads them + // as stubs and can call methods that route back to the server. + class ChunkTarget extends RpcTarget { + square(i: number) { return i * i; } + get label() { return "server-chunk"; } + } + + class StreamProvider extends RpcTarget { + getStream(): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue(new ChunkTarget()); + controller.close(); + } + }); + } + } + + await using harness = new TestHarness(new StreamProvider()); + let stream: any = await harness.stub.getStream(); + + let reader = stream.getReader(); + let { value } = await reader.read(); + using chunk = value as any; + + expect(await chunk.square(6)).toBe(36); + expect(await chunk.label).toBe("server-chunk"); + + // Drain the closing `done` chunk so the underlying pipe is released. + let { done } = await reader.read(); + expect(done).toBe(true); + }); + + it("can send multiple RpcTargets as stream chunks", async () => { + class ChunkTarget extends RpcTarget { + constructor(private id: number) { super(); } + getId() { return this.id; } + } + + let stream = new ReadableStream({ + start(controller) { + controller.enqueue(new ChunkTarget(1)); + controller.enqueue(new ChunkTarget(2)); + controller.enqueue(new ChunkTarget(3)); + controller.close(); + } + }); + + class StreamReceiver extends RpcTarget { + async receiveStream(stream: ReadableStream) { + let reader = stream.getReader(); + let ids: number[] = []; + let chunks: any[] = []; + try { + while (true) { + let { done, value } = await reader.read(); + if (done) break; + chunks.push(value); + ids.push(await value.getId()); + } + return ids; + } finally { + for (let chunk of chunks) { + chunk[Symbol.dispose](); + } + } + } + } + + await using harness = new TestHarness(new StreamReceiver()); + let result: any = await harness.stub.receiveStream(stream); + + expect(result).toEqual([1, 2, 3]); + }); }); // ======================================================================================= From 98acb219e32941cdc584e9ac0596534294eefd9e Mon Sep 17 00:00:00 2001 From: CodeHz Date: Sat, 11 Jul 2026 01:30:58 +0800 Subject: [PATCH 2/3] feat: support streaming WritableStream writes with lifecycle-managed RPC payloads --- src/core.ts | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/streams.ts | 7 +++++-- 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/src/core.ts b/src/core.ts index 2ea587e..7361269 100644 --- a/src/core.ts +++ b/src/core.ts @@ -1200,6 +1200,58 @@ export class RpcPayload { } } + // Deliver this payload as the argument list to `WritableStreamDefaultWriter.write(chunk)`. + // + // Unlike deliverCall(), stream chunks are enqueued and read asynchronously. Disposing the + // payload when write() returns would free any stubs nested in the chunk before the reader + // dequeues them. For object chunks we therefore keep the payload alive and attach disposal + // to the chunk (same pattern as deliverResolve). Bare RpcStubs already implement + // Symbol.dispose, so their own disposer owns the capability. Primitive chunks cannot hold + // stubs, so the payload is disposed after write settles. + // + // Expects `this.value` to be a one-element argument array `[chunk]`. Returns a payload + // wrapping the (void) write result. + public async deliverStreamWrite( + writer: { write(chunk: unknown): Promise }): Promise { + try { + let promises: Promise[] = []; + this.deliverTo(this, "value", promises); + + // Same e-order constraint as deliverCall: if there are no nested promises, call write + // synchronously so concurrent writes stay ordered. + if (promises.length > 0) { + await Promise.all(promises); + } + + let chunk = (this.value as unknown[])[0]; + + if (chunk instanceof Object) { + // Object (including RpcStub): do not dispose the payload when write() returns — the + // reader still needs any nested capabilities. Attach a payload disposer when the + // chunk doesn't already have one (RpcStub already disposes its own hook). + if (!(Symbol.dispose in chunk)) { + Object.defineProperty(chunk, Symbol.dispose, { + value: () => this.dispose(), + writable: true, + enumerable: false, + configurable: true, + }); + } + await writer.write(chunk); + return RpcPayload.fromAppReturn(undefined); + } + + // Primitive chunk: nothing to keep alive beyond the write. + await writer.write(chunk); + this.dispose(); + return RpcPayload.fromAppReturn(undefined); + } catch (err) { + // deliverTo / write failed: free any capabilities we still own. + this.dispose(); + throw err; + } + } + // Produce a promise for this payload for return to the application. Any RpcPromises in the // payload are awaited and substituted with their results first. // diff --git a/src/streams.ts b/src/streams.ts index 52a5ae4..2a0ddd3 100644 --- a/src/streams.ts +++ b/src/streams.ts @@ -62,8 +62,11 @@ class WritableStreamStubHook extends StubHook { state.closed = true; } - let func = state.writer[method] as Function; - let promise = args.deliverCall(func, state.writer); + // write(chunk) delivers asynchronously: stubs nested in the chunk must outlive + // writer.write() returning. See RpcPayload.deliverStreamWrite(). + let promise = method === "write" + ? args.deliverStreamWrite(state.writer) + : args.deliverCall(state.writer[method] as Function, state.writer); return new PromiseStubHook(promise.then(payload => new PayloadStubHook(payload))); } catch (err) { return new ErrorStubHook(err); From 4da46608efd580800714e598e81c29a8b3cb3b1f Mon Sep 17 00:00:00 2001 From: CodeHz Date: Sat, 11 Jul 2026 01:33:54 +0800 Subject: [PATCH 3/3] chore: add changeset for RpcTarget streaming support --- .changeset/cool-firm-full.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/cool-firm-full.md diff --git a/.changeset/cool-firm-full.md b/.changeset/cool-firm-full.md new file mode 100644 index 0000000..0009b56 --- /dev/null +++ b/.changeset/cool-firm-full.md @@ -0,0 +1,5 @@ +--- +"capnweb": minor +--- + +Support RpcTargets (and other RPC stubs) as ReadableStream/WritableStream chunks without disposing their capabilities when `write()` returns. Stream chunk payloads now keep lifecycle tied to the chunk (via `Symbol.dispose` when needed) so methods on streamed stubs remain usable after the write resolves.