= V extends (...args: infer P) => infer R
153 | ? (...args: UnstubifyAll) => Result>
154 | : Result>;
155 |
156 | // Type for the callable part of an `Provider` if `T` is callable.
157 | // This is intersected with methods/properties.
158 | type MaybeCallableProvider = T extends (...args: any[]) => any
159 | ? MethodOrProperty
160 | : unknown;
161 |
162 | // Base type for all other types providing RPC-like interfaces.
163 | // Rewrites all methods/properties to be `MethodOrProperty`s, while preserving callable types.
164 | export type Provider = MaybeCallableProvider &
165 | (T extends Array
166 | ? {
167 | [key: number]: MethodOrProperty;
168 | } & {
169 | map(callback: (elem: U) => V): Result>;
170 | }
171 | : {
172 | [K in Exclude<
173 | keyof T,
174 | symbol | keyof StubBase
175 | >]: MethodOrProperty;
176 | } & {
177 | map(callback: (value: NonNullable) => V): Result>;
178 | });
179 |
--------------------------------------------------------------------------------
/src/batch.ts:
--------------------------------------------------------------------------------
1 | // Copyright (c) 2025 Cloudflare, Inc.
2 | // Licensed under the MIT license found in the LICENSE.txt file or at:
3 | // https://opensource.org/license/mit
4 |
5 | import { RpcStub } from "./core.js";
6 | import { RpcTransport, RpcSession, RpcSessionOptions } from "./rpc.js";
7 | import type { IncomingMessage, ServerResponse, OutgoingHttpHeader, OutgoingHttpHeaders } from "node:http";
8 |
9 | type SendBatchFunc = (batch: string[]) => Promise;
10 |
11 | class BatchClientTransport implements RpcTransport {
12 | constructor(sendBatch: SendBatchFunc) {
13 | this.#promise = this.#scheduleBatch(sendBatch);
14 | }
15 |
16 | #promise: Promise;
17 | #aborted: any;
18 |
19 | #batchToSend: string[] | null = [];
20 | #batchToReceive: string[] | null = null;
21 |
22 | async send(message: string): Promise {
23 | // If the batch was already sent, we just ignore the message, because throwing may cause the
24 | // RPC system to abort prematurely. Once the last receive() is done then we'll throw an error
25 | // that aborts the RPC system at the right time and will propagate to all other requests.
26 | if (this.#batchToSend !== null) {
27 | this.#batchToSend.push(message);
28 | }
29 | }
30 |
31 | async receive(): Promise {
32 | if (!this.#batchToReceive) {
33 | await this.#promise;
34 | }
35 |
36 | let msg = this.#batchToReceive!.shift();
37 | if (msg !== undefined) {
38 | return msg;
39 | } else {
40 | // No more messages. An error thrown here will propagate out of any calls that are still
41 | // open.
42 | throw new Error("Batch RPC request ended.");
43 | }
44 | }
45 |
46 | abort?(reason: any): void {
47 | this.#aborted = reason;
48 | }
49 |
50 | async #scheduleBatch(sendBatch: SendBatchFunc) {
51 | // Wait for microtask queue to clear before sending a batch.
52 | //
53 | // Note that simply waiting for one turn of the microtask queue (await Promise.resolve()) is
54 | // not good enough here as the application needs a chance to call `.then()` on every RPC
55 | // promise in order to explicitly indicate they want the results. Unfortunately, `await`ing
56 | // a thenable does not call `.then()` immediately -- for some reason it waits for a turn of
57 | // the microtask queue first, *then* calls `.then()`.
58 | await new Promise(resolve => setTimeout(resolve, 0));
59 |
60 | if (this.#aborted !== undefined) {
61 | throw this.#aborted;
62 | }
63 |
64 | let batch = this.#batchToSend!;
65 | this.#batchToSend = null;
66 | this.#batchToReceive = await sendBatch(batch);
67 | }
68 | }
69 |
70 | export function newHttpBatchRpcSession(
71 | urlOrRequest: string | Request, options?: RpcSessionOptions): RpcStub {
72 | let sendBatch: SendBatchFunc = async (batch: string[]) => {
73 | let response = await fetch(urlOrRequest, {
74 | method: "POST",
75 | body: batch.join("\n"),
76 | });
77 |
78 | if (!response.ok) {
79 | response.body?.cancel();
80 | throw new Error(`RPC request failed: ${response.status} ${response.statusText}`);
81 | }
82 |
83 | let body = await response.text();
84 | return body == "" ? [] : body.split("\n");
85 | };
86 |
87 | let transport = new BatchClientTransport(sendBatch);
88 | let rpc = new RpcSession(transport, undefined, options);
89 | return rpc.getRemoteMain();
90 | }
91 |
92 | class BatchServerTransport implements RpcTransport {
93 | constructor(batch: string[]) {
94 | this.#batchToReceive = batch;
95 | }
96 |
97 | #batchToSend: string[] = [];
98 | #batchToReceive: string[];
99 | #allReceived: PromiseWithResolvers = Promise.withResolvers();
100 |
101 | async send(message: string): Promise {
102 | this.#batchToSend.push(message);
103 | }
104 |
105 | async receive(): Promise {
106 | let msg = this.#batchToReceive!.shift();
107 | if (msg !== undefined) {
108 | return msg;
109 | } else {
110 | // No more messages.
111 | this.#allReceived.resolve();
112 | return new Promise(r => {});
113 | }
114 | }
115 |
116 | abort?(reason: any): void {
117 | this.#allReceived.reject(reason);
118 | }
119 |
120 | whenAllReceived() {
121 | return this.#allReceived.promise;
122 | }
123 |
124 | getResponseBody(): string {
125 | return this.#batchToSend.join("\n");
126 | }
127 | }
128 |
129 | /**
130 | * Implements the server end of an HTTP batch session, using standard Fetch API types to represent
131 | * HTTP requests and responses.
132 | *
133 | * @param request The request received from the client initiating the session.
134 | * @param localMain The main stub or RpcTarget which the server wishes to expose to the client.
135 | * @param options Optional RPC session options.
136 | * @returns The HTTP response to return to the client. Note that the returned object has mutable
137 | * headers, so you can modify them using e.g. `response.headers.set("Foo", "bar")`.
138 | */
139 | export async function newHttpBatchRpcResponse(
140 | request: Request, localMain: any, options?: RpcSessionOptions): Promise {
141 | if (request.method !== "POST") {
142 | return new Response("This endpoint only accepts POST requests.", { status: 405 });
143 | }
144 |
145 | let body = await request.text();
146 | let batch = body === "" ? [] : body.split("\n");
147 |
148 | let transport = new BatchServerTransport(batch);
149 | let rpc = new RpcSession(transport, localMain, options);
150 |
151 | // TODO: Arguably we should arrange so any attempts to pull promise resolutions from the client
152 | // will reject rather than just hang. But it IS valid to make server->client calls in order to
153 | // then pipeline the result into something returned to the client. We don't want the errors to
154 | // prematurely cancel anything that would eventually complete. So for now we just say, it's the
155 | // app's responsibility to not wait on any server -> client calls since they will never
156 | // complete.
157 |
158 | await transport.whenAllReceived();
159 | await rpc.drain();
160 |
161 | // TODO: Ask RpcSession to dispose everything it is still holding on to?
162 |
163 | return new Response(transport.getResponseBody());
164 | }
165 |
166 | /**
167 | * Implements the server end of an HTTP batch session using traditional Node.js HTTP APIs.
168 | *
169 | * @param request The request received from the client initiating the session.
170 | * @param response The response object, to which the response should be written.
171 | * @param localMain The main stub or RpcTarget which the server wishes to expose to the client.
172 | * @param options Optional RPC session options. You can also pass headers to set on the response.
173 | */
174 | export async function nodeHttpBatchRpcResponse(
175 | request: IncomingMessage, response: ServerResponse,
176 | localMain: any,
177 | options?: RpcSessionOptions & {
178 | headers?: OutgoingHttpHeaders | OutgoingHttpHeader[],
179 | }): Promise {
180 | if (request.method !== "POST") {
181 | response.writeHead(405, "This endpoint only accepts POST requests.");
182 | }
183 |
184 | let body = await new Promise((resolve, reject) => {
185 | let chunks: Buffer[] = [];
186 | request.on("data", chunk => {
187 | chunks.push(chunk);
188 | });
189 | request.on("end", () => {
190 | resolve(Buffer.concat(chunks).toString());
191 | });
192 | request.on("error", reject);
193 | });
194 | let batch = body === "" ? [] : body.split("\n");
195 |
196 | let transport = new BatchServerTransport(batch);
197 | let rpc = new RpcSession(transport, localMain, options);
198 |
199 | await transport.whenAllReceived();
200 | await rpc.drain();
201 |
202 | response.writeHead(200, options?.headers);
203 | response.end(transport.getResponseBody());
204 | }
205 |
--------------------------------------------------------------------------------
/src/index.ts:
--------------------------------------------------------------------------------
1 | // Copyright (c) 2025 Cloudflare, Inc.
2 | // Licensed under the MIT license found in the LICENSE.txt file or at:
3 | // https://opensource.org/license/mit
4 |
5 | import { RpcTarget as RpcTargetImpl, RpcStub as RpcStubImpl, RpcPromise as RpcPromiseImpl } from "./core.js";
6 | import { serialize, deserialize } from "./serialize.js";
7 | import { RpcTransport, RpcSession as RpcSessionImpl, RpcSessionOptions } from "./rpc.js";
8 | import { RpcTargetBranded, RpcCompatible, Stub, Stubify, __RPC_TARGET_BRAND } from "./types.js";
9 | import { newWebSocketRpcSession as newWebSocketRpcSessionImpl,
10 | newWorkersWebSocketRpcResponse } from "./websocket.js";
11 | import { newHttpBatchRpcSession as newHttpBatchRpcSessionImpl,
12 | newHttpBatchRpcResponse, nodeHttpBatchRpcResponse } from "./batch.js";
13 | import { newMessagePortRpcSession as newMessagePortRpcSessionImpl } from "./messageport.js";
14 | import { forceInitMap } from "./map.js";
15 |
16 | forceInitMap();
17 |
18 | // Re-export public API types.
19 | export { serialize, deserialize, newWorkersWebSocketRpcResponse, newHttpBatchRpcResponse,
20 | nodeHttpBatchRpcResponse };
21 | export type { RpcTransport, RpcSessionOptions, RpcCompatible };
22 |
23 | // Hack the type system to make RpcStub's types work nicely!
24 | /**
25 | * Represents a reference to a remote object, on which methods may be remotely invoked via RPC.
26 | *
27 | * `RpcStub` can represent any interface (when using TypeScript, you pass the specific interface
28 | * type as `T`, but this isn't known at runtime). The way this works is, `RpcStub` is actually a
29 | * `Proxy`. It makes itself appear as if every possible method / property name is defined. You can
30 | * invoke any method name, and the invocation will be sent to the server. If it turns out that no
31 | * such method exists on the remote object, an exception is thrown back. But the client does not
32 | * actually know, until that point, what methods exist.
33 | */
34 | export type RpcStub> = Stub;
35 | export const RpcStub: {
36 | new >(value: T): RpcStub;
37 | } = RpcStubImpl;
38 |
39 | /**
40 | * Represents the result of an RPC call.
41 | *
42 | * Also used to represent properties. That is, `stub.foo` evaluates to an `RpcPromise` for the
43 | * value of `foo`.
44 | *
45 | * This isn't actually a JavaScript `Promise`. It does, however, have `then()`, `catch()`, and
46 | * `finally()` methods, like `Promise` does, and because it has a `then()` method, JavaScript will
47 | * allow you to treat it like a promise, e.g. you can `await` it.
48 | *
49 | * An `RpcPromise` is also a proxy, just like `RpcStub`, where calling methods or awaiting
50 | * properties will make a pipelined network request.
51 | *
52 | * Note that and `RpcPromise` is "lazy": the actual final result is not requested from the server
53 | * until you actually `await` the promise (or call `then()`, etc. on it). This is an optimization:
54 | * if you only intend to use the promise for pipelining and you never await it, then there's no
55 | * need to transmit the resolution!
56 | */
57 | export type RpcPromise> = Stub & Promise>;
58 | export const RpcPromise: {
59 | // Note: Cannot construct directly!
60 | } = RpcPromiseImpl;
61 |
62 | /**
63 | * Use to construct an `RpcSession` on top of a custom `RpcTransport`.
64 | *
65 | * Most people won't use this. You only need it if you've implemented your own `RpcTransport`.
66 | */
67 | export interface RpcSession = undefined> {
68 | getRemoteMain(): RpcStub;
69 | getStats(): {imports: number, exports: number};
70 |
71 | // Waits until the peer is not waiting on any more promise resolutions from us. This is useful
72 | // in particular to decide when a batch is complete.
73 | drain(): Promise;
74 | }
75 | export const RpcSession: {
76 | new = undefined>(
77 | transport: RpcTransport, localMain?: any, options?: RpcSessionOptions): RpcSession;
78 | } = RpcSessionImpl;
79 |
80 | // RpcTarget needs some hackage too to brand it properly and account for the implementation
81 | // conditionally being imported from "cloudflare:workers".
82 | /**
83 | * Classes which are intended to be passed by reference and called over RPC must extend
84 | * `RpcTarget`. A class which does not extend `RpcTarget` (and which doesn't have built-in support
85 | * from the RPC system) cannot be passed in an RPC message at all; an exception will be thrown.
86 | *
87 | * Note that on Cloudflare Workers, this `RpcTarget` is an alias for the one exported from the
88 | * "cloudflare:workers" module, so they can be used interchangably.
89 | */
90 | export interface RpcTarget extends RpcTargetBranded {};
91 | export const RpcTarget: {
92 | new(): RpcTarget;
93 | } = RpcTargetImpl;
94 |
95 | /**
96 | * Empty interface used as default type parameter for sessions where the other side doesn't
97 | * necessarily export a main interface.
98 | */
99 | interface Empty {}
100 |
101 | /**
102 | * Start a WebSocket session given either an already-open WebSocket or a URL.
103 | *
104 | * @param webSocket Either the `wss://` URL to connect to, or an already-open WebSocket object to
105 | * use.
106 | * @param localMain The main RPC interface to expose to the peer. Returns a stub for the main
107 | * interface exposed from the peer.
108 | */
109 | export let newWebSocketRpcSession: = Empty>
110 | (webSocket: WebSocket | string, localMain?: any, options?: RpcSessionOptions) => RpcStub =
111 | newWebSocketRpcSessionImpl;
112 |
113 | /**
114 | * Initiate an HTTP batch session from the client side.
115 | *
116 | * The parameters to this method have exactly the same signature as `fetch()`, but the return
117 | * value is an RpcStub. You can customize anything about the request except for the method
118 | * (it will always be set to POST) and the body (which the RPC system will fill in).
119 | */
120 | export let newHttpBatchRpcSession:>
121 | (urlOrRequest: string | Request, options?: RpcSessionOptions) => RpcStub =
122 | newHttpBatchRpcSessionImpl;
123 |
124 | /**
125 | * Initiate an RPC session over a MessagePort, which is particularly useful for communicating
126 | * between an iframe and its parent frame in a browser context. Each side should call this function
127 | * on its own end of the MessageChannel.
128 | */
129 | export let newMessagePortRpcSession: = Empty>
130 | (port: MessagePort, localMain?: any, options?: RpcSessionOptions) => RpcStub =
131 | newMessagePortRpcSessionImpl;
132 |
133 | /**
134 | * Implements unified handling of HTTP-batch and WebSocket responses for the Cloudflare Workers
135 | * Runtime.
136 | *
137 | * SECURITY WARNING: This function accepts cross-origin requests. If you do not want this, you
138 | * should validate the `Origin` header before calling this, or use `newHttpBatchRpcSession()` and
139 | * `newWebSocketRpcSession()` directly with appropriate security measures for each type of request.
140 | * But if your API uses in-band authorization (i.e. it has an RPC method that takes the user's
141 | * credentials as parameters and returns the authorized API), then cross-origin requests should
142 | * be safe.
143 | */
144 | export async function newWorkersRpcResponse(request: Request, localMain: any) {
145 | if (request.method === "POST") {
146 | let response = await newHttpBatchRpcResponse(request, localMain);
147 | // Since we're exposing the same API over WebSocket, too, and WebSocket always allows
148 | // cross-origin requests, the API necessarily must be safe for cross-origin use (e.g. because
149 | // it uses in-band authorization, as recommended in the readme). So, we might as well allow
150 | // batch requests to be made cross-origin as well.
151 | response.headers.set("Access-Control-Allow-Origin", "*");
152 | return response;
153 | } else if (request.headers.get("Upgrade")?.toLowerCase() === "websocket") {
154 | return newWorkersWebSocketRpcResponse(request, localMain);
155 | } else {
156 | return new Response("This endpoint only accepts POST or WebSocket requests.", { status: 400 });
157 | }
158 | }
159 |
--------------------------------------------------------------------------------
/__tests__/workerd.test.ts:
--------------------------------------------------------------------------------
1 | // Copyright (c) 2025 Cloudflare, Inc.
2 | // Licensed under the MIT license found in the LICENSE.txt file or at:
3 | // https://opensource.org/license/mit
4 |
5 | ///
6 | import { expect, it, describe } from "vitest";
7 | import { RpcStub as NativeRpcStub, RpcTarget as NativeRpcTarget, env, DurableObject } from "cloudflare:workers";
8 | import { newHttpBatchRpcSession, newWebSocketRpcSession, RpcStub, RpcTarget } from "../src/index-workers.js";
9 | import { Counter, TestTarget } from "./test-util.js";
10 |
11 | class JsCounter extends RpcTarget {
12 | constructor(private i: number = 0) {
13 | super();
14 | }
15 |
16 | increment(amount: number = 1): number {
17 | this.i += amount;
18 | return this.i;
19 | }
20 |
21 | get value() {
22 | return this.i;
23 | }
24 | }
25 |
26 | class NativeCounter extends RpcTarget {
27 | constructor(private i: number = 0) {
28 | super();
29 | }
30 |
31 | increment(amount: number = 1): number {
32 | this.i += amount;
33 | return this.i;
34 | }
35 |
36 | get value() {
37 | return this.i;
38 | }
39 | }
40 |
41 | class CounterFactory extends RpcTarget {
42 | getNative() {
43 | return new NativeRpcStub(new NativeCounter());
44 | }
45 |
46 | getNativeEmbedded() {
47 | return {stub: new NativeRpcStub(new NativeCounter())};
48 | }
49 |
50 | getJs() {
51 | return new RpcStub(new JsCounter());
52 | }
53 |
54 | getJsEmbedded() {
55 | return {stub: new RpcStub(new JsCounter())};
56 | }
57 | }
58 |
59 | describe("workerd compatibility", () => {
60 | it("allows native RpcStubs to be created using userspace RpcTargets", async () => {
61 | let stub = new NativeRpcStub(new JsCounter());
62 | expect(await stub.increment()).toBe(1);
63 | expect(await stub.increment()).toBe(2);
64 |
65 | expect(await stub.value).toBe(2);
66 | })
67 |
68 | it("allows userspace RpcStubs to be created using native RpcTargets", async () => {
69 | let stub = new RpcStub(new NativeCounter());
70 | expect(await stub.increment()).toBe(1);
71 | expect(await stub.increment()).toBe(2);
72 |
73 | expect(await stub.value).toBe(2);
74 | })
75 |
76 | it("can wrap a native stub in a userspace stub", async () => {
77 | let stub = new RpcStub(new NativeRpcStub(new NativeCounter()));
78 | expect(await stub.increment()).toBe(1);
79 | expect(await stub.increment()).toBe(2);
80 |
81 | expect(await stub.value).toBe(2);
82 | })
83 |
84 | it("can return a native stub from a userspace call", async () => {
85 | // Returning a bare stub.
86 | {
87 | let factory = new RpcStub(new CounterFactory());
88 | let stub = await factory.getNative();
89 | expect(await stub.increment()).toBe(1);
90 | expect(await stub.increment()).toBe(2);
91 |
92 | expect(await stub.value).toBe(2);
93 | }
94 |
95 | // Again with a stub wrapped in an object.
96 | {
97 | let factory = new RpcStub(new CounterFactory());
98 | let obj = await factory.getNativeEmbedded();
99 | expect(await obj.stub.increment()).toBe(1);
100 | expect(await obj.stub.increment()).toBe(2);
101 |
102 | expect(await obj.stub.value).toBe(2);
103 | }
104 | })
105 |
106 | it("can wrap a native promise or property in a userspace stub", async () => {
107 | // Wrap a native RpcPromise in a userspace stub.
108 | {
109 | let factory = new NativeRpcStub(new CounterFactory());
110 | let stub = new RpcStub(factory.getNative());
111 | expect(await stub.increment()).toBe(1);
112 | expect(await stub.increment()).toBe(2);
113 |
114 | expect(await stub.value).toBe(2);
115 | }
116 |
117 | // Wrap a native RpcProperty in a userspace stub.
118 | {
119 | let factory = new NativeRpcStub(new CounterFactory());
120 | let stub = new RpcStub(factory.getNativeEmbedded().stub);
121 | expect(await stub.increment()).toBe(1);
122 | expect(await stub.increment()).toBe(2);
123 |
124 | expect(await stub.value).toBe(2);
125 | }
126 | })
127 |
128 | it("can pipeline on a native stub returned from a userspace call", async () => {
129 | {
130 | let factory = new RpcStub(new CounterFactory());
131 | let obj = factory.getNative();
132 | expect(await obj.increment()).toBe(1);
133 | expect(await obj.increment()).toBe(2);
134 |
135 | expect(await obj.value).toBe(2);
136 | }
137 |
138 | {
139 | let factory = new RpcStub(new CounterFactory());
140 | let obj = factory.getNativeEmbedded();
141 | expect(await obj.stub.increment()).toBe(1);
142 | expect(await obj.stub.increment()).toBe(2);
143 |
144 | expect(await obj.stub.value).toBe(2);
145 | }
146 | })
147 |
148 | it("can wrap a userspace stub in a native stub", async () => {
149 | let stub = new NativeRpcStub(new RpcStub(new JsCounter()));
150 | expect(await stub.increment()).toBe(1);
151 | expect(await stub.increment()).toBe(2);
152 |
153 | expect(await stub.value).toBe(2);
154 | })
155 |
156 | it("can return a userspace stub from a native call", async () => {
157 | // Returning a bare stub.
158 | {
159 | let factory = new NativeRpcStub(new CounterFactory());
160 | let stub = await factory.getJs();
161 | expect(await stub.increment()).toBe(1);
162 | expect(await stub.increment()).toBe(2);
163 |
164 | expect(await stub.value).toBe(2);
165 | }
166 |
167 | // Again with a stub wrapped in an object.
168 | {
169 | let factory = new NativeRpcStub(new CounterFactory());
170 | let obj = await factory.getJsEmbedded();
171 | expect(await obj.stub.increment()).toBe(1);
172 | expect(await obj.stub.increment()).toBe(2);
173 |
174 | expect(await obj.stub.value).toBe(2);
175 | }
176 | })
177 |
178 | it("can wrap a userspace promise or property in a native stub", async () => {
179 | // Wrap a userspace RpcPromise in a native stub.
180 | {
181 | let factory = new RpcStub(new CounterFactory());
182 | let stub = new NativeRpcStub(factory.getJs());
183 | expect(await stub.increment()).toBe(1);
184 | expect(await stub.increment()).toBe(2);
185 |
186 | expect(await stub.value).toBe(2);
187 | }
188 |
189 | // Wrap a userspace property (which is actually also an RpcPromise) in a native stub.
190 | {
191 | let factory = new RpcStub(new CounterFactory());
192 | let stub = new NativeRpcStub(factory.getJsEmbedded().stub);
193 | expect(await stub.increment()).toBe(1);
194 | expect(await stub.increment()).toBe(2);
195 |
196 | expect(await stub.value).toBe(2);
197 | }
198 | })
199 |
200 | it("can pipeline on a userspace stub returned from a native call", async () => {
201 | {
202 | let factory = new NativeRpcStub(new CounterFactory());
203 | let obj = factory.getJs();
204 | expect(await obj.increment()).toBe(1);
205 | expect(await obj.increment()).toBe(2);
206 |
207 | expect(await obj.value).toBe(2);
208 | }
209 |
210 | {
211 | let factory = new NativeRpcStub(new CounterFactory());
212 | let obj = factory.getJsEmbedded();
213 | expect(await obj.stub.increment()).toBe(1);
214 | expect(await obj.stub.increment()).toBe(2);
215 |
216 | expect(await obj.stub.value).toBe(2);
217 | }
218 | })
219 |
220 | it("can wrap a ServiceStub in an RpcStub", async () => {
221 | let result = await new RpcStub((env).testServer).greet("World");
222 | expect(result).toBe("Hello, World!");
223 | });
224 | });
225 |
226 | interface Env {
227 | testServer: Fetcher
228 | }
229 |
230 | interface TestDo extends DurableObject {
231 | setValue(val: any): void;
232 | getValue(): any;
233 | }
234 |
235 | interface WorkerdTestTarget extends TestTarget {
236 | getDurableObject(name: string): DurableObjectStub;
237 | }
238 |
239 | describe("workerd RPC server", () => {
240 | it("can accept WebSocket RPC connections", async () => {
241 | let resp = await (env).testServer.fetch("http://foo", {headers: {Upgrade: "websocket"}});
242 | let ws = resp.webSocket;
243 | expect(ws).toBeTruthy();
244 |
245 | ws!.accept();
246 | let cap = newWebSocketRpcSession(ws!);
247 |
248 | expect(await cap.square(5)).toBe(25);
249 |
250 | {
251 | let counter = cap.makeCounter(2);
252 | expect(await counter.increment(3)).toBe(5);
253 | }
254 |
255 | {
256 | let counter = new Counter(4);
257 | expect(await cap.incrementCounter(counter, 9)).toBe(13);
258 | }
259 |
260 | // Test that we can pass a Durable Object stub over RPC.
261 | {
262 | let foo = cap.getDurableObject("foo");
263 | foo.setValue(123);
264 |
265 | let bar = await cap.getDurableObject("bar");
266 | bar.setValue("abc");
267 |
268 | expect(await foo.getValue()).toBe(123);
269 | expect(await bar.getValue()).toBe("abc");
270 | }
271 | })
272 |
273 | it("can accept HTTP batch RPC connections", async () => {
274 | let cap = newHttpBatchRpcSession(
275 | new Request("http://foo", {fetcher: (env).testServer}));
276 |
277 | let promise1 = cap.square(6);
278 |
279 | let counter = cap.makeCounter(2);
280 | let promise2 = counter.increment(3);
281 | let promise3 = cap.incrementCounter(counter, 4);
282 |
283 | expect(await Promise.all([promise1, promise2, promise3]))
284 | .toStrictEqual([36, 5, 9]);
285 | })
286 | });
287 |
--------------------------------------------------------------------------------
/examples/worker-react/web/src/main/App.tsx:
--------------------------------------------------------------------------------
1 | import { useMemo, useState } from 'react'
2 | import { newHttpBatchRpcSession } from 'capnweb'
3 | import type { Api } from '../../../src/worker'
4 |
5 | type Result = {
6 | posts: number
7 | ms: number
8 | user: any
9 | profile: any
10 | notifications: any
11 | trace: Trace
12 | }
13 |
14 | type CallEvent = { label: string, start: number, end: number }
15 | type NetEvent = { label: string, start: number, end: number }
16 | type Trace = { total: number, calls: CallEvent[], network: NetEvent[] }
17 |
18 | export function App() {
19 | const [pipelined, setPipelined] = useState(null)
20 | const [sequential, setSequential] = useState(null)
21 | const [running, setRunning] = useState(false)
22 |
23 | // Network RTT is now simulated on the server (Worker). See wrangler.toml vars.
24 |
25 | // Count RPC POSTs and capture network timing by wrapping fetch while this component is mounted.
26 | const wrapFetch = useMemo(() => {
27 | let posts = 0
28 | let origin = 0
29 | let events: NetEvent[] = []
30 | const orig = globalThis.fetch
31 | function install() {
32 | ;(globalThis as any).fetch = async (input: RequestInfo, init?: RequestInit) => {
33 | const method = (init?.method) || (input instanceof Request ? input.method : 'GET')
34 | const url = input instanceof Request ? input.url : String(input)
35 | if (url.endsWith('/api') && method === 'POST') {
36 | posts++
37 | const start = performance.now() - origin
38 | const resp = await orig(input as any, init)
39 | const end = performance.now() - origin
40 | events.push({ label: 'POST /api', start, end })
41 | return resp
42 | }
43 | return orig(input as any, init)
44 | }
45 | }
46 | function uninstall() { ;(globalThis as any).fetch = orig }
47 | function get() { return posts }
48 | function reset() { posts = 0; events = [] }
49 | function setOrigin(o: number) { origin = o }
50 | function getEvents(): NetEvent[] { return events.slice() }
51 | return { install, uninstall, get, reset, setOrigin, getEvents }
52 | }, [])
53 |
54 | async function runPipelined() {
55 | wrapFetch.reset()
56 | const t0 = performance.now()
57 | wrapFetch.setOrigin(t0)
58 | const calls: CallEvent[] = []
59 | const api = newHttpBatchRpcSession('/api')
60 | const userStart = 0; calls.push({ label: 'authenticate', start: userStart, end: NaN })
61 | const user = api.authenticate('cookie-123')
62 | user.then(() => { calls.find(c => c.label==='authenticate')!.end = performance.now() - t0 })
63 |
64 | const profStart = performance.now() - t0; calls.push({ label: 'getUserProfile', start: profStart, end: NaN })
65 | const profile = api.getUserProfile(user.id)
66 | profile.then(() => { calls.find(c => c.label==='getUserProfile')!.end = performance.now() - t0 })
67 |
68 | const notiStart = performance.now() - t0; calls.push({ label: 'getNotifications', start: notiStart, end: NaN })
69 | const notifications = api.getNotifications(user.id)
70 | notifications.then(() => { calls.find(c => c.label==='getNotifications')!.end = performance.now() - t0 })
71 |
72 | const [u, p, n] = await Promise.all([user, profile, notifications])
73 | const t1 = performance.now()
74 | const net = wrapFetch.getEvents()
75 | const total = t1 - t0
76 | // Ensure any missing ends are set
77 | calls.forEach(c => { if (!Number.isFinite(c.end)) c.end = total })
78 | return { posts: wrapFetch.get(), ms: total, user: u, profile: p, notifications: n,
79 | trace: { total, calls, network: net } }
80 | }
81 |
82 | async function runSequential() {
83 | wrapFetch.reset()
84 | const t0 = performance.now()
85 | wrapFetch.setOrigin(t0)
86 | const calls: CallEvent[] = []
87 | const api1 = newHttpBatchRpcSession('/api')
88 | const aStart = 0; calls.push({ label: 'authenticate', start: aStart, end: NaN })
89 | const uPromise = api1.authenticate('cookie-123')
90 | uPromise.then(() => { calls.find(c => c.label==='authenticate')!.end = performance.now() - t0 })
91 | const u = await uPromise
92 |
93 | const api2 = newHttpBatchRpcSession('/api')
94 | const pStart = performance.now() - t0; calls.push({ label: 'getUserProfile', start: pStart, end: NaN })
95 | const pPromise = api2.getUserProfile(u.id)
96 | pPromise.then(() => { calls.find(c => c.label==='getUserProfile')!.end = performance.now() - t0 })
97 | const p = await pPromise
98 |
99 | const api3 = newHttpBatchRpcSession('/api')
100 | const nStart = performance.now() - t0; calls.push({ label: 'getNotifications', start: nStart, end: NaN })
101 | const nPromise = api3.getNotifications(u.id)
102 | nPromise.then(() => { calls.find(c => c.label==='getNotifications')!.end = performance.now() - t0 })
103 | const n = await nPromise
104 |
105 | const t1 = performance.now()
106 | const net = wrapFetch.getEvents()
107 | const total = t1 - t0
108 | calls.forEach(c => { if (!Number.isFinite(c.end)) c.end = total })
109 | return { posts: wrapFetch.get(), ms: total, user: u, profile: p, notifications: n,
110 | trace: { total, calls, network: net } }
111 | }
112 |
113 | async function runDemo() {
114 | if (running) return
115 | setRunning(true)
116 | wrapFetch.install()
117 | try {
118 | const piped = await runPipelined()
119 | setPipelined(piped)
120 | const seq = await runSequential()
121 | setSequential(seq)
122 | } finally {
123 | wrapFetch.uninstall()
124 | setRunning(false)
125 | }
126 | }
127 |
128 | return (
129 |
130 |
Cap'n Web: Cloudflare Workers + React
131 |
Network RTT is simulated on the server (configurable via SIMULATED_RTT_MS/SIMULATED_RTT_JITTER_MS in wrangler.toml).
132 |
This demo calls the Worker API in two ways:
133 |
134 | - Pipelined batch: dependent calls in one round trip
135 | - Sequential non-batched: three separate round trips
136 |
137 |
140 |
141 | {pipelined && (
142 |
143 | Pipelined (batched)
144 | HTTP POSTs: {pipelined.posts}
145 | Time: {pipelined.ms.toFixed(1)} ms
146 |
147 | {JSON.stringify({
148 | user: pipelined.user,
149 | profile: pipelined.profile,
150 | notifications: pipelined.notifications,
151 | }, null, 2)}
152 |
153 | )}
154 |
155 | {sequential && (
156 |
157 | Sequential (non-batched)
158 | HTTP POSTs: {sequential.posts}
159 | Time: {sequential.ms.toFixed(1)} ms
160 |
161 | {JSON.stringify({
162 | user: sequential.user,
163 | profile: sequential.profile,
164 | notifications: sequential.notifications,
165 | }, null, 2)}
166 |
167 | )}
168 |
169 | {(pipelined && sequential) && (
170 |
171 | Summary
172 | Pipelined: {pipelined.posts} POST, {pipelined.ms.toFixed(1)} ms
173 | Sequential: {sequential.posts} POSTs, {sequential.ms.toFixed(1)} ms
174 |
175 | )}
176 |