From 9e9937747a78915d224bdef4a0c43f4d18165871 Mon Sep 17 00:00:00 2001 From: Steven Chong <25894545+teamchong@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:37:18 -0500 Subject: [PATCH 1/2] feat: support zero-copy byte views and transferable ArrayBuffers --- .changeset/send-exact-byte-views.md | 4 +- __tests__/index.test.ts | 185 +++++++++++++++++++++++++++- src/index.ts | 8 +- src/messageport.ts | 40 +++++- src/serialize.ts | 173 +++++++++++++++++--------- 5 files changed, 341 insertions(+), 69 deletions(-) diff --git a/.changeset/send-exact-byte-views.md b/.changeset/send-exact-byte-views.md index 331cb1f4..1f9fc158 100644 --- a/.changeset/send-exact-byte-views.md +++ b/.changeset/send-exact-byte-views.md @@ -1,5 +1,5 @@ --- -"capnweb": patch +"capnweb": minor --- -Fix `MessagePort` sessions leaking a byte view's entire backing buffer instead of only the view's bytes. +Fix `MessagePort` sessions leaking a byte view's entire backing buffer, avoid copying aligned byte views at `jsonCompatibleWithBytes`, and add `transferArrayBuffer()` for zero-copy `MessagePort` transfers. diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts index 48008efd..a9d3cc0d 100644 --- a/__tests__/index.test.ts +++ b/__tests__/index.test.ts @@ -5,7 +5,7 @@ import { expect, it, describe, inject } from "vitest" import { deserialize, serialize, RpcSession, type RpcSessionOptions, RpcTransport, type RpcTransportWithCustomEncoding, RpcTarget, RpcStub, RpcPromise, newWebSocketRpcSession, - newMessagePortRpcSession, + newMessagePortRpcSession, MessagePortTransport, transferArrayBuffer, newHttpBatchRpcSession} from "../src/index.js" import { swapByteOrder } from "../src/serialize.js" import { MAX_CLOSE_REASON_BYTES } from "../src/websocket.js" @@ -2968,6 +2968,80 @@ describe("MessagePorts", () => { expect(await stub.getBytes()).toStrictEqual([new Uint8Array(buffer), int16]); expect(crossedBufferSizes).toStrictEqual([4, 4]); }); + + it("transfers marked ArrayBuffers across MessagePort without copying", async () => { + // workerd's MessagePort does not yet support postMessage transfer lists. + if (navigator.userAgent === "Cloudflare-Workers") return; + + let sentUint8 = transferArrayBuffer(new Uint8Array([10, 20, 30, 40])); + let sentFloat32 = transferArrayBuffer(new Float32Array([1.5, -2.5, 3.25])); + let sentBuffer = transferArrayBuffer(new Uint8Array([7, 8, 9]).buffer); + let untransferred = new Uint8Array([1, 2, 3]); + + let receivedOnServer: { + uint8: Uint8Array; + float32: Float32Array; + rawBuffer: ArrayBuffer; + untransferred: Uint8Array; + } | undefined; + let serverReplyBuffer: ArrayBuffer | undefined; + + class TransferServer extends RpcTarget { + exchange( + uint8: Uint8Array, + float32: Float32Array, + rawBuffer: ArrayBuffer, + copyBytes: Uint8Array + ) { + receivedOnServer = { uint8, float32, rawBuffer, untransferred: copyBytes }; + let reply = transferArrayBuffer(new Float32Array([42.5, 99.5])); + serverReplyBuffer = reply.buffer; + return reply; + } + } + + let channel = new MessageChannel(); + let crossedWireBuffers: ArrayBuffer[] = []; + function collectBuffers(value: unknown) { + if (ArrayBuffer.isView(value)) crossedWireBuffers.push(value.buffer); + else if (value instanceof ArrayBuffer) crossedWireBuffers.push(value); + else if (Array.isArray(value)) value.forEach(collectBuffers); + } + channel.port1.addEventListener("message", event => collectBuffers(event.data)); + + newMessagePortRpcSession(channel.port1, new TransferServer()); + using stub = newMessagePortRpcSession(channel.port2); + + let resultPromise = stub.exchange(sentUint8, sentFloat32, sentBuffer, untransferred); + + // Marked buffers are detached on the sender immediately after postMessage. + expect(sentUint8.byteLength).toBe(0); + expect(sentUint8.buffer.byteLength).toBe(0); + expect(sentFloat32.byteLength).toBe(0); + expect(sentFloat32.buffer.byteLength).toBe(0); + expect(sentBuffer.byteLength).toBe(0); + + // Unmarked buffer is NOT detached on the sender. + expect(untransferred.byteLength).toBe(3); + expect(untransferred.buffer.byteLength).toBe(3); + + let reply = await resultPromise; + expect(receivedOnServer).toBeDefined(); + expect(receivedOnServer!.uint8).toStrictEqual(new Uint8Array([10, 20, 30, 40])); + expect(receivedOnServer!.float32).toStrictEqual(new Float32Array([1.5, -2.5, 3.25])); + expect(new Uint8Array(receivedOnServer!.rawBuffer)).toStrictEqual(new Uint8Array([7, 8, 9])); + expect(receivedOnServer!.untransferred).toStrictEqual(new Uint8Array([1, 2, 3])); + + // On the receiver, the evaluated views point directly at the transferred ArrayBuffers + // that arrived in the MessageEvent without an extra slice() copy. + expect(receivedOnServer!.uint8.buffer).toBe(crossedWireBuffers[0]); + expect(receivedOnServer!.float32.buffer).toBe(crossedWireBuffers[1]); + expect(receivedOnServer!.rawBuffer).toBe(crossedWireBuffers[2]); + + // And the server's transferred reply buffer was detached on the server and received intact. + expect(serverReplyBuffer!.byteLength).toBe(0); + expect(reply).toStrictEqual(new Float32Array([42.5, 99.5])); + }); }); // ======================================================================================= @@ -3567,6 +3641,115 @@ describe("transport encoding levels", () => { }); } + it("preserves exact Uint8Array and Float32Array subviews without extra buffer copies at jsonCompatibleWithBytes", async () => { + // Pass messages directly without structuredClone in the transport so we can verify + // that Devaluator and Evaluator themselves do not copy or slice backing ArrayBuffers. + class ZeroCopyBytesTransport implements RpcTransportWithCustomEncoding { + readonly encodingLevel = "jsonCompatibleWithBytes" as const; + partner?: ZeroCopyBytesTransport; + #queue: unknown[] = []; + #waiter?: (msg: unknown) => void; + + send(message: unknown): void { + if (this.partner!.#waiter) { + let resolve = this.partner!.#waiter; + this.partner!.#waiter = undefined; + resolve(message); + } else { + this.partner!.#queue.push(message); + } + } + + async receive(): Promise { + if (this.#queue.length > 0) { + return this.#queue.shift()!; + } + let { promise, resolve } = Promise.withResolvers(); + this.#waiter = resolve; + return promise; + } + } + + let clientTransport = new ZeroCopyBytesTransport(); + let serverTransport = new ZeroCopyBytesTransport(); + clientTransport.partner = serverTransport; + serverTransport.partner = clientTransport; + + let receivedViews: { + uint8Sub: Uint8Array; + float32Sub: Float32Array; + } | undefined; + + class SubviewServer extends RpcTarget { + sendSubviews(uint8Sub: Uint8Array, float32Sub: Float32Array) { + receivedViews = { uint8Sub, float32Sub }; + return [uint8Sub, float32Sub]; + } + } + + let client = new RpcSession(clientTransport); + new RpcSession(serverTransport, new SubviewServer()); + using stub = client.getRemoteMain(); + + let sharedArena = new ArrayBuffer(128); + let arenaBytes = new Uint8Array(sharedArena); + for (let i = 0; i < arenaBytes.length; i++) arenaBytes[i] = i; + + let uint8Sub = new Uint8Array(sharedArena, 12, 10); + let float32Sub = new Float32Array(sharedArena, 32, 4); + float32Sub.set([1.25, -2.5, 3.75, 4.0]); + + let [echoedUint8, echoedFloat32] = await stub.sendSubviews(uint8Sub, float32Sub); + + expect(receivedViews).toBeDefined(); + + // Neither Devaluator nor Evaluator copied the aligned subviews: + // they still reference `sharedArena` with the exact byteOffset and byteLength. + expect(receivedViews!.uint8Sub.buffer).toBe(sharedArena); + expect(receivedViews!.uint8Sub.byteOffset).toBe(12); + expect(receivedViews!.uint8Sub.byteLength).toBe(10); + expect(receivedViews!.uint8Sub).toStrictEqual(uint8Sub); + + expect(receivedViews!.float32Sub.buffer).toBe(sharedArena); + expect(receivedViews!.float32Sub.byteOffset).toBe(32); + expect(receivedViews!.float32Sub.byteLength).toBe(16); + expect(receivedViews!.float32Sub).toStrictEqual(float32Sub); + + expect(echoedUint8.buffer).toBe(sharedArena); + expect(echoedUint8.byteOffset).toBe(12); + expect(echoedUint8.byteLength).toBe(10); + + expect(echoedFloat32.buffer).toBe(sharedArena); + expect(echoedFloat32.byteOffset).toBe(32); + expect(echoedFloat32.byteLength).toBe(16); + + // Also test an unaligned Uint8Array backing an Int16Array wire payload to verify + // Evaluator safely copies when byteOffset is not a multiple of elementSize. + let unalignedBacking = new Uint8Array(16); + new DataView(unalignedBacking.buffer, 1, 4).setInt16(0, 0x1234, true); + new DataView(unalignedBacking.buffer, 1, 4).setInt16(2, 0x5678, true); + let unalignedWireView = new Uint8Array(unalignedBacking.buffer, 1, 4); + + let unalignedClient = new ZeroCopyBytesTransport(); + let unalignedServer = new ZeroCopyBytesTransport(); + unalignedClient.partner = unalignedServer; + unalignedServer.partner = unalignedClient; + + let { promise: decodedPromise, resolve: resolveDecoded } = Promise.withResolvers(); + class UnalignedReceiver extends RpcTarget { + take(arr: Int16Array) { + resolveDecoded(arr); + } + } + new RpcSession(unalignedServer, new UnalignedReceiver()); + // Inject a raw ["bytes", unalignedUint8Array, "Int16Array"] frame at jsonCompatibleWithBytes. + unalignedServer.partner.send(["push", ["pipeline", 0, ["take"], [["bytes", unalignedWireView, "Int16Array"]]]]); + let decodedUnaligned = await decodedPromise; + expect(decodedUnaligned).toStrictEqual(new Int16Array([0x1234, 0x5678])); + expect(decodedUnaligned.buffer).not.toBe(unalignedBacking.buffer); + expect(decodedUnaligned.byteOffset).toBe(0); + }); + it("aborts the receive loop when the transport is aborted", async () => { let transport = new ObjectTestTransport(); let pending = transport.receive(); diff --git a/src/index.ts b/src/index.ts index 0a45747c..f26e98af 100644 --- a/src/index.ts +++ b/src/index.ts @@ -3,7 +3,7 @@ // https://opensource.org/license/mit import { RpcTarget as RpcTargetImpl, RpcStub as RpcStubImpl, RpcPromise as RpcPromiseImpl } from "./core.js"; -import { serialize, deserialize, EncodingLevel } from "./serialize.js"; +import { serialize, deserialize, EncodingLevel, transferArrayBuffer } from "./serialize.js"; import { RpcTransport, RpcTransportWithCustomEncoding, AnyRpcTransport, RpcSession as RpcSessionImpl, RpcSessionOptions } from "./rpc.js"; import { RpcLimits, DEFAULT_LIMITS, DEFAULT_MAX_DEPTH } from "./serialize.js"; import { RpcTargetBranded, RpcCompatible, Stub, ElideStub, PayloadOrStub, @@ -12,7 +12,8 @@ import { newWebSocketRpcSession as newWebSocketRpcSessionImpl, newWorkersWebSocketRpcResponse, WebSocketTransport } from "./websocket.js"; import { newHttpBatchRpcSession as newHttpBatchRpcSessionImpl, newHttpBatchRpcResponse, nodeHttpBatchRpcResponse } from "./batch.js"; -import { newMessagePortRpcSession as newMessagePortRpcSessionImpl } from "./messageport.js"; +import { newMessagePortRpcSession as newMessagePortRpcSessionImpl, + MessagePortTransport } from "./messageport.js"; import { forceInitMap } from "./map.js"; import { forceInitStreams } from "./streams.js"; @@ -21,7 +22,8 @@ forceInitStreams(); // Re-export public API types. export { serialize, deserialize, newWorkersWebSocketRpcResponse, newHttpBatchRpcResponse, - nodeHttpBatchRpcResponse, WebSocketTransport, DEFAULT_LIMITS, DEFAULT_MAX_DEPTH }; + nodeHttpBatchRpcResponse, WebSocketTransport, MessagePortTransport, + transferArrayBuffer, DEFAULT_LIMITS, DEFAULT_MAX_DEPTH }; export type { RpcTransport, RpcTransportWithCustomEncoding, AnyRpcTransport, RpcSessionOptions, RpcCompatible, EncodingLevel, RpcLimits }; diff --git a/src/messageport.ts b/src/messageport.ts index 11d2ce3e..d565651b 100644 --- a/src/messageport.ts +++ b/src/messageport.ts @@ -4,6 +4,7 @@ import { RpcStub } from "./core.js"; import { RpcTransportWithCustomEncoding, RpcSession, RpcSessionOptions } from "./rpc.js"; +import { consumeTransferableArrayBuffer, hasMarkedTransferableBuffers } from "./serialize.js"; // Start a MessagePort session given a MessagePort or a pair of MessagePorts. // @@ -16,7 +17,7 @@ export function newMessagePortRpcSession( return rpc.getRemoteMain(); } -class MessagePortTransport implements RpcTransportWithCustomEncoding { +export class MessagePortTransport implements RpcTransportWithCustomEncoding { readonly encodingLevel = "structuredClonable" as const; constructor (port: MessagePort) { @@ -58,7 +59,18 @@ class MessagePortTransport implements RpcTransportWithCustomEncoding { if (this.#error) { throw this.#error; } - this.#port.postMessage(message); + let transfer: ArrayBuffer[] | undefined; + if (hasMarkedTransferableBuffers()) { + collectTransferableBuffers(message, (buffer) => { + transfer ??= []; + transfer.push(buffer); + }); + } + if (transfer !== undefined) { + this.#port.postMessage(message, transfer); + } else { + this.#port.postMessage(message); + } } async receive(): Promise { @@ -101,3 +113,27 @@ class MessagePortTransport implements RpcTransportWithCustomEncoding { } } } + +function collectTransferableBuffers( + value: unknown, onTransferable: (buffer: ArrayBuffer) => void): void { + if (Array.isArray(value)) { + for (let i = 0; i < value.length; i++) { + collectTransferableBuffers(value[i], onTransferable); + } + } else if (ArrayBuffer.isView(value)) { + if (consumeTransferableArrayBuffer(value.buffer)) { + onTransferable(value.buffer); + } + } else if (consumeTransferableArrayBuffer(value)) { + onTransferable(value); + } else if (value !== null && typeof value === "object") { + let proto = Object.getPrototypeOf(value); + if (proto === Object.prototype || proto === null) { + for (let key in value) { + if (Object.prototype.hasOwnProperty.call(value, key)) { + collectTransferableBuffers((value as Record)[key], onTransferable); + } + } + } + } +} diff --git a/src/serialize.ts b/src/serialize.ts index c8f7118d..10ddc419 100644 --- a/src/serialize.ts +++ b/src/serialize.ts @@ -136,6 +136,33 @@ const BYTE_CONTAINER_TYPE_BY_PROTOTYPE = new Map(); +let hasAnyTransferableBuffers = false; + +/** + * Mark an `ArrayBuffer` (or the backing `ArrayBuffer` of an `ArrayBufferView`) to be transferred + * rather than copied when sent over a `MessagePort` transport. + */ +export function transferArrayBuffer(value: T): T { + let buffer = ArrayBuffer.isView(value) ? value.buffer : value; + if (buffer instanceof ArrayBuffer) { + TRANSFERABLE_BUFFERS.add(buffer); + hasAnyTransferableBuffers = true; + } + return value; +} + +/** Returns true if any `ArrayBuffer` has been marked via `transferArrayBuffer()`. */ +export function hasMarkedTransferableBuffers(): boolean { + return hasAnyTransferableBuffers; +} + +/** + * If `buffer` was marked via `transferArrayBuffer()`, removes the mark and returns true. + */ +export function consumeTransferableArrayBuffer(buffer: unknown): buffer is ArrayBuffer { + return buffer instanceof ArrayBuffer && TRANSFERABLE_BUFFERS.delete(buffer); +} // Reverse each element's bytes in place. Production callers only need this on // big-endian hosts; it is exported solely so the behavior can be tested on @@ -388,14 +415,18 @@ export class Devaluator { } // At structuredClonable or jsonCompatibleWithBytes level, keep the bytes raw. - if (this.encodingLevel === "structuredClonable" || - this.encodingLevel === "jsonCompatibleWithBytes") { - // Copy partial views so the transport can't send the whole backing buffer. - if (bytes.byteLength !== bytes.buffer.byteLength) { + if (this.encodingLevel === "structuredClonable") { + // Copy partial views of un-transferred buffers so the transport doesn't + // structured-clone the whole backing buffer. + if ((bytes.byteOffset !== 0 || bytes.byteLength !== bytes.buffer.byteLength) && + !(bytes.buffer instanceof ArrayBuffer && TRANSFERABLE_BUFFERS.has(bytes.buffer))) { bytes = new Uint8Array(bytes); } return alternateTypeName === undefined - ? ["bytes", bytes] : ["bytes", bytes, alternateTypeName]; + ? ["bytes", bytes] : ["bytes", bytes, alternateTypeName]; + } else if (this.encodingLevel === "jsonCompatibleWithBytes") { + return alternateTypeName === undefined + ? ["bytes", bytes] : ["bytes", bytes, alternateTypeName]; } let b64: string; @@ -906,70 +937,90 @@ export class Evaluator { return set; } break; - case "bytes": { - let bytes: Uint8Array; - // At jsonCompatibleWithBytes/structuredClonable level, bytes may already be raw. - if (value[1] instanceof Uint8Array) { - bytes = value[1]; - } else if (typeof value[1] == "string") { - if (typeof Buffer !== "undefined") { - bytes = Buffer.from(value[1], "base64"); - } else if (Uint8Array.fromBase64) { - bytes = Uint8Array.fromBase64(value[1]); - } else { - let bs = atob(value[1]); - let len = bs.length; - bytes = new Uint8Array(len); - for (let i = 0; i < len; i++) { - bytes[i] = bs.charCodeAt(i); - } - } - } else { - break; - } - if (value.length === 2) { - return bytes; - } - if (typeof value[2] !== "string") { - throw new TypeError(`Unknown bytes type marker type: ${typeof value[2]}`); + case "bytes": { + let bytes: Uint8Array; + let isRawBytes = false; + // At jsonCompatibleWithBytes/structuredClonable level, bytes may already be raw. + if (value[1] instanceof Uint8Array) { + bytes = value[1]; + isRawBytes = true; + } else if (typeof value[1] === "string") { + if (typeof Buffer !== "undefined") { + bytes = Buffer.from(value[1], "base64"); + } else if (Uint8Array.fromBase64) { + bytes = Uint8Array.fromBase64(value[1]); + } else { + let bs = atob(value[1]); + let len = bs.length; + bytes = new Uint8Array(len); + for (let i = 0; i < len; i++) { + bytes[i] = bs.charCodeAt(i); + } } - - if (!isValidByteContainerName(value[2])) { - let marker = value[2].slice(0, 64); - throw new TypeError(`Unknown bytes type marker: ${marker}`); + } else { + break; + } + if (value.length === 2) { + // If decoded via Buffer.from(..., "base64"), copy out of Node's shared slab pool. + if (!isRawBytes && (bytes.byteOffset !== 0 || bytes.byteLength !== bytes.buffer.byteLength)) { + return new Uint8Array(bytes); } + return bytes; + } - let marker = value[2]; - let elementSize = TYPED_ARRAY_ELEMENT_SIZE[marker]; - if (elementSize !== undefined && bytes.byteLength % elementSize !== 0) { - throw new TypeError( - `Invalid byte length ${bytes.byteLength} for ${marker}; ` + - `expected a multiple of ${elementSize}`); - } + if (typeof value[2] !== "string") { + throw new TypeError(`Unknown bytes type marker type: ${typeof value[2]}`); + } + if (!isValidByteContainerName(value[2])) { + let marker = value[2].slice(0, 64); + throw new TypeError(`Unknown bytes type marker: ${marker}`); + } + let marker = value[2]; + let elementSize = TYPED_ARRAY_ELEMENT_SIZE[marker]; + if (elementSize !== undefined && bytes.byteLength % elementSize !== 0) { + throw new TypeError( + `Invalid byte length ${bytes.byteLength} for ${marker}; ` + + `expected a multiple of ${elementSize}`); + } - // Copy exactly the decoded range rather than exposing or aliasing a pooled Buffer. - let buffer = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength); + let buffer: ArrayBufferLike; + let byteOffset: number; + let byteLength = bytes.byteLength; + if (isRawBytes && + (NATIVE_LITTLE_ENDIAN || elementSize === undefined) && + (elementSize === undefined || bytes.byteOffset % elementSize === 0)) { + buffer = bytes.buffer; + byteOffset = bytes.byteOffset; + } else { + // Copy when base64-decoded, unaligned, or endianness-swapped. + let copy = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength); if (!NATIVE_LITTLE_ENDIAN && elementSize !== undefined) { - swapByteOrder(new Uint8Array(buffer), elementSize); - } - switch (marker) { - case "ArrayBuffer": return buffer; - case "DataView": return new DataView(buffer); - case "Int8Array": return new Int8Array(buffer); - case "Uint8Array": return new Uint8Array(buffer); - case "Uint8ClampedArray": return new Uint8ClampedArray(buffer); - case "Int16Array": return new Int16Array(buffer); - case "Uint16Array": return new Uint16Array(buffer); - case "Int32Array": return new Int32Array(buffer); - case "Uint32Array": return new Uint32Array(buffer); - case "BigInt64Array": return new BigInt64Array(buffer); - case "BigUint64Array": return new BigUint64Array(buffer); - case "Float32Array": return new Float32Array(buffer); - case "Float64Array": return new Float64Array(buffer); - default: marker satisfies never; + swapByteOrder(new Uint8Array(copy), elementSize); } + buffer = copy; + byteOffset = 0; } + switch (marker) { + case "ArrayBuffer": + return byteOffset === 0 && byteLength === buffer.byteLength && buffer instanceof ArrayBuffer + ? buffer + : buffer.slice(byteOffset, byteOffset + byteLength); + case "DataView": return new DataView(buffer, byteOffset, byteLength); + case "Int8Array": return new Int8Array(buffer, byteOffset, byteLength); + case "Uint8Array": return new Uint8Array(buffer, byteOffset, byteLength); + case "Uint8ClampedArray": return new Uint8ClampedArray(buffer, byteOffset, byteLength); + case "Int16Array": return new Int16Array(buffer, byteOffset, byteLength / 2); + case "Uint16Array": return new Uint16Array(buffer, byteOffset, byteLength / 2); + case "Int32Array": return new Int32Array(buffer, byteOffset, byteLength / 4); + case "Uint32Array": return new Uint32Array(buffer, byteOffset, byteLength / 4); + case "BigInt64Array": return new BigInt64Array(buffer, byteOffset, byteLength / 8); + case "BigUint64Array": return new BigUint64Array(buffer, byteOffset, byteLength / 8); + case "Float32Array": return new Float32Array(buffer, byteOffset, byteLength / 4); + case "Float64Array": return new Float64Array(buffer, byteOffset, byteLength / 8); + default: marker satisfies never; + } + } case "error": if (value.length >= 3 && typeof value[1] === "string" && typeof value[2] === "string") { let cls = ERROR_TYPES[value[1]] || Error; From 46e7785ebd78b65089e502e378fb65e6c0160643 Mon Sep 17 00:00:00 2001 From: Steven Chong <25894545+teamchong@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:52:09 -0500 Subject: [PATCH 2/2] chore: keep MessagePortTransport internal --- __tests__/index.test.ts | 2 +- src/index.ts | 7 +++---- src/messageport.ts | 2 +- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts index a9d3cc0d..02178c22 100644 --- a/__tests__/index.test.ts +++ b/__tests__/index.test.ts @@ -5,7 +5,7 @@ import { expect, it, describe, inject } from "vitest" import { deserialize, serialize, RpcSession, type RpcSessionOptions, RpcTransport, type RpcTransportWithCustomEncoding, RpcTarget, RpcStub, RpcPromise, newWebSocketRpcSession, - newMessagePortRpcSession, MessagePortTransport, transferArrayBuffer, + newMessagePortRpcSession, transferArrayBuffer, newHttpBatchRpcSession} from "../src/index.js" import { swapByteOrder } from "../src/serialize.js" import { MAX_CLOSE_REASON_BYTES } from "../src/websocket.js" diff --git a/src/index.ts b/src/index.ts index f26e98af..2009073a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -12,8 +12,7 @@ import { newWebSocketRpcSession as newWebSocketRpcSessionImpl, newWorkersWebSocketRpcResponse, WebSocketTransport } from "./websocket.js"; import { newHttpBatchRpcSession as newHttpBatchRpcSessionImpl, newHttpBatchRpcResponse, nodeHttpBatchRpcResponse } from "./batch.js"; -import { newMessagePortRpcSession as newMessagePortRpcSessionImpl, - MessagePortTransport } from "./messageport.js"; +import { newMessagePortRpcSession as newMessagePortRpcSessionImpl } from "./messageport.js"; import { forceInitMap } from "./map.js"; import { forceInitStreams } from "./streams.js"; @@ -22,8 +21,8 @@ forceInitStreams(); // Re-export public API types. export { serialize, deserialize, newWorkersWebSocketRpcResponse, newHttpBatchRpcResponse, - nodeHttpBatchRpcResponse, WebSocketTransport, MessagePortTransport, - transferArrayBuffer, DEFAULT_LIMITS, DEFAULT_MAX_DEPTH }; + nodeHttpBatchRpcResponse, WebSocketTransport, transferArrayBuffer, + DEFAULT_LIMITS, DEFAULT_MAX_DEPTH }; export type { RpcTransport, RpcTransportWithCustomEncoding, AnyRpcTransport, RpcSessionOptions, RpcCompatible, EncodingLevel, RpcLimits }; diff --git a/src/messageport.ts b/src/messageport.ts index d565651b..f78d085c 100644 --- a/src/messageport.ts +++ b/src/messageport.ts @@ -17,7 +17,7 @@ export function newMessagePortRpcSession( return rpc.getRemoteMain(); } -export class MessagePortTransport implements RpcTransportWithCustomEncoding { +class MessagePortTransport implements RpcTransportWithCustomEncoding { readonly encodingLevel = "structuredClonable" as const; constructor (port: MessagePort) {