From fc1bfee67c10fe3c300de531b76a5b743d75d403 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Tue, 20 Oct 2026 13:45:31 +1300 Subject: [PATCH 1/5] feat: add LocalParticipant.sendBytes for one-shot in-memory byte streams Mirrors python-sdks send_bytes: sends a complete in-memory buffer through the FFI StreamSendBytes request so short payloads can be inlined and larger ones compressed, instead of callers going through streamBytes or a temp file with sendFile. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/send-bytes.md | 5 ++ packages/livekit-rtc/src/participant.ts | 58 +++++++++++++++++++ .../src/tests/e2e_data_streams.test.ts | 47 ++++++++++++++- 3 files changed, 107 insertions(+), 3 deletions(-) create mode 100644 .changeset/send-bytes.md diff --git a/.changeset/send-bytes.md b/.changeset/send-bytes.md new file mode 100644 index 00000000..5fcf47af --- /dev/null +++ b/.changeset/send-bytes.md @@ -0,0 +1,5 @@ +--- +'@livekit/rtc-node': patch +--- + +Add `LocalParticipant.sendBytes` to send an in-memory buffer as a complete byte stream in one call, letting short payloads be inlined and larger ones compressed. diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index 07ed90a4..61ccd744 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -47,6 +47,9 @@ import { SetLocalNameRequest, type SetLocalNameResponse, StreamByteOptions, + type StreamSendBytesCallback, + StreamSendBytesRequest, + type StreamSendBytesResponse, type StreamSendFileCallback, StreamSendFileRequest, type StreamSendFileResponse, @@ -540,6 +543,61 @@ export class LocalParticipant extends Participant { } } + /** + * Sends an in-memory buffer as a complete byte stream in one call. + * + * Unlike {@link LocalParticipant.streamBytes}, the whole payload is known up front, so short + * payloads can be sent as a single packet and larger ones compressed, when every recipient + * supports it. `name` defaults to "unknown" and `mimeType` to "application/octet-stream". + */ + async sendBytes( + data: Uint8Array, + options?: { + topic?: string; + name?: string; + mimeType?: string; + attributes?: Record; + destinationIdentities?: Array; + streamId?: string; + /** + * Whether the payload may be compressed on the wire. Defaults to true; + * compression is only applied when it actually reduces the payload size + * and every recipient supports it. + */ + compress?: boolean; + }, + ) { + const req = new StreamSendBytesRequest({ + localParticipantHandle: this.ffi_handle.handle, + options: new StreamByteOptions({ + topic: options?.topic ?? '', + attributes: options?.attributes, + destinationIdentities: options?.destinationIdentities, + id: options?.streamId, + name: options?.name ?? 'unknown', + mimeType: options?.mimeType ?? 'application/octet-stream', + senderIdentity: this.identity, + compress: options?.compress, + }), + bytes: data, + }); + + const res = FfiClient.instance.request({ + message: { case: 'sendBytes', value: req }, + }); + + const cb = await FfiClient.instance.waitFor( + (ev) => ev.message.case == 'sendBytes' && ev.message.value.asyncId == res.asyncId, + { signal: this.disconnectSignal }, + ); + + if (cb.result.case !== 'info') { + throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + } + + return byteStreamInfoFromProto(cb.result.value); + } + private textStreamWriterWrite = async (req: TextStreamWriterWriteRequest) => { const res = FfiClient.instance.request({ message: { case: 'textStreamWrite', value: req }, diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts index 628f429c..61f0fdb5 100644 --- a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -106,9 +106,9 @@ describeE2E('livekit-rtc data streams e2e', () => { testTimeoutMs, ); - // Port of rust `test_send_bytes`. Node has no `sendBytes` (in-memory one-shot) - // wrapper yet, so this uses the incremental `streamBytes` writer — which by - // design is never inlined or compressed on the wire. + // Port of rust `test_send_bytes` using the incremental `streamBytes` writer, + // which by design is never inlined or compressed on the wire. See the + // `sendBytes` test below for the one-shot path. it( 'sends and receives a small byte stream', async () => { @@ -150,6 +150,47 @@ describeE2E('livekit-rtc data streams e2e', () => { testTimeoutMs, ); + // Port of rust `test_send_bytes` using the one-shot `sendBytes` path, which + // (unlike `streamBytes`) may inline or compress the payload on the wire. + it( + 'sends and receives a small in-memory buffer via sendBytes', + async () => { + const { rooms } = await connectTestRooms(2); + const [receivingRoom, sendingRoom] = rooms; + const senderIdentity = sendingRoom!.localParticipant!.identity; + + const topic = 'some-topic'; + const bytesToSend = new TextEncoder().encode('hello world'); + + const receivedBytes = withTimeout( + new Promise((resolve) => { + receivingRoom!.registerByteStreamHandler(topic, async (reader, sender) => { + expect(sender.identity).toBe(senderIdentity); + const chunks = await reader.readAll(); + resolve(concatUint8(chunks)); + }); + }), + testTimeoutMs, + 'Timed out waiting for byte stream', + ); + + const info = await sendingRoom!.localParticipant!.sendBytes(bytesToSend, { + topic, + name: 'hello.bin', + }); + expect(info.streamId).toBeTruthy(); + expect(info.topic).toBe(topic); + expect(info.name).toBe('hello.bin'); + expect(info.totalSize).toBe(bytesToSend.byteLength); + expect(info.mimeType).toBe('application/octet-stream'); + + expect(await receivedBytes).toEqual(bytesToSend); + + await Promise.all(rooms.map((r) => r.disconnect())); + }, + testTimeoutMs, + ); + // Port of rust `test_send_large_compressible_text`: ~50 KB of deterministic // pseudo-random lowercase is too big to inline and compresses well under its // raw size, so the sender emits chunked deflate-raw and the receiver From a5d9c04d0959e1898d411205bc7f441c481f3c9b Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 11:49:50 +1300 Subject: [PATCH 2/5] refactor: take ByteStreamOptions in sendBytes Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/livekit-rtc/src/participant.ts | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index 61ccd744..af21ff3c 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -550,23 +550,7 @@ export class LocalParticipant extends Participant { * payloads can be sent as a single packet and larger ones compressed, when every recipient * supports it. `name` defaults to "unknown" and `mimeType` to "application/octet-stream". */ - async sendBytes( - data: Uint8Array, - options?: { - topic?: string; - name?: string; - mimeType?: string; - attributes?: Record; - destinationIdentities?: Array; - streamId?: string; - /** - * Whether the payload may be compressed on the wire. Defaults to true; - * compression is only applied when it actually reduces the payload size - * and every recipient supports it. - */ - compress?: boolean; - }, - ) { + async sendBytes(data: Uint8Array, options?: ByteStreamOptions & { streamId?: string }) { const req = new StreamSendBytesRequest({ localParticipantHandle: this.ffi_handle.handle, options: new StreamByteOptions({ From 5ea52bfb38da174c7f799eff4374905cc31e0018 Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 11:50:28 +1300 Subject: [PATCH 3/5] feat: report completion through onProgress in sendBytes Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/livekit-rtc/src/participant.ts | 3 +++ packages/livekit-rtc/src/tests/e2e_data_streams.test.ts | 3 +++ 2 files changed, 6 insertions(+) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index af21ff3c..91e1bc23 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -579,6 +579,9 @@ export class LocalParticipant extends Participant { throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); } + // The payload is sent in one call, so the only progress to report is completion. + options?.onProgress?.(1); + return byteStreamInfoFromProto(cb.result.value); } diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts index 61f0fdb5..2d9f8f9d 100644 --- a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -174,10 +174,13 @@ describeE2E('livekit-rtc data streams e2e', () => { 'Timed out waiting for byte stream', ); + const progress: Array = []; const info = await sendingRoom!.localParticipant!.sendBytes(bytesToSend, { topic, name: 'hello.bin', + onProgress: (p) => progress.push(p), }); + expect(progress).toEqual([1]); expect(info.streamId).toBeTruthy(); expect(info.topic).toBe(topic); expect(info.name).toBe('hello.bin'); From a9f3a4cfea50d456bf3b61b4400af74c9a29f62f Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Wed, 21 Oct 2026 11:52:35 +1300 Subject: [PATCH 4/5] refactor: drop onProgress from sendBytes options Matches python-sdks send_bytes, which takes no progress callback. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/livekit-rtc/src/participant.ts | 8 ++++---- packages/livekit-rtc/src/tests/e2e_data_streams.test.ts | 3 --- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index 91e1bc23..a48efc82 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -550,7 +550,10 @@ export class LocalParticipant extends Participant { * payloads can be sent as a single packet and larger ones compressed, when every recipient * supports it. `name` defaults to "unknown" and `mimeType` to "application/octet-stream". */ - async sendBytes(data: Uint8Array, options?: ByteStreamOptions & { streamId?: string }) { + async sendBytes( + data: Uint8Array, + options?: Omit & { streamId?: string }, + ) { const req = new StreamSendBytesRequest({ localParticipantHandle: this.ffi_handle.handle, options: new StreamByteOptions({ @@ -579,9 +582,6 @@ export class LocalParticipant extends Participant { throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); } - // The payload is sent in one call, so the only progress to report is completion. - options?.onProgress?.(1); - return byteStreamInfoFromProto(cb.result.value); } diff --git a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts index 2d9f8f9d..61f0fdb5 100644 --- a/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts +++ b/packages/livekit-rtc/src/tests/e2e_data_streams.test.ts @@ -174,13 +174,10 @@ describeE2E('livekit-rtc data streams e2e', () => { 'Timed out waiting for byte stream', ); - const progress: Array = []; const info = await sendingRoom!.localParticipant!.sendBytes(bytesToSend, { topic, name: 'hello.bin', - onProgress: (p) => progress.push(p), }); - expect(progress).toEqual([1]); expect(info.streamId).toBeTruthy(); expect(info.topic).toBe(topic); expect(info.name).toBe('hello.bin'); From 1b6e3ec2453a2170438400344a7d3f9635325c0d Mon Sep 17 00:00:00 2001 From: Ryan Gaus Date: Thu, 22 Oct 2026 08:31:17 +1300 Subject: [PATCH 5/5] refactor: switch on stream callback result cases in LocalParticipant Handle the expected result case, the error case, and any unexpected case explicitly instead of collapsing everything non-success into 'unknown error'. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/livekit-rtc/src/participant.ts | 48 ++++++++++++++++++------- 1 file changed, 36 insertions(+), 12 deletions(-) diff --git a/packages/livekit-rtc/src/participant.ts b/packages/livekit-rtc/src/participant.ts index a48efc82..105d6c09 100644 --- a/packages/livekit-rtc/src/participant.ts +++ b/packages/livekit-rtc/src/participant.ts @@ -326,8 +326,15 @@ export class LocalParticipant extends Participant { { signal: this.disconnectSignal }, ); - if (cb.result.case !== 'writer') { - throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + switch (cb.result.case) { + case 'writer': + break; + case 'error': + throw new Error(cb.result.value.description); + default: + throw new Error( + `LocalParticipant.streamText - unknown error, cb.result.case expected to be writer or error, found ${cb.result.case}`, + ); } const writerHandle = cb.result.value.handle!.id!; @@ -418,11 +425,16 @@ export class LocalParticipant extends Participant { { signal: this.disconnectSignal }, ); - if (cb.result.case !== 'info') { - throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + switch (cb.result.case) { + case 'info': + return textStreamInfoFromProto(cb.result.value); + case 'error': + throw new Error(cb.result.value.description); + default: + throw new Error( + `LocalParticipant.sendText - unknown error, cb.result.case expected to be info or error, found ${cb.result.case}`, + ); } - - return textStreamInfoFromProto(cb.result.value); } async streamBytes(options?: { @@ -457,8 +469,15 @@ export class LocalParticipant extends Participant { { signal: this.disconnectSignal }, ); - if (cb.result.case !== 'writer') { - throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + switch (cb.result.case) { + case 'writer': + break; + case 'error': + throw new Error(cb.result.value.description); + default: + throw new Error( + `LocalParticipant.streamBytes - unknown error, cb.result.case expected to be writer or error, found ${cb.result.case}`, + ); } const writerHandle = cb.result.value.handle!.id!; @@ -578,11 +597,16 @@ export class LocalParticipant extends Participant { { signal: this.disconnectSignal }, ); - if (cb.result.case !== 'info') { - throw new Error(cb.result.case === 'error' ? cb.result.value.description : 'unknown error'); + switch (cb.result.case) { + case 'info': + return byteStreamInfoFromProto(cb.result.value); + case 'error': + throw new Error(cb.result.value.description); + default: + throw new Error( + `LocalParticipant.sendBytes - unknown error, cb.result.case expected to be info or error, found ${cb.result.case}`, + ); } - - return byteStreamInfoFromProto(cb.result.value); } private textStreamWriterWrite = async (req: TextStreamWriterWriteRequest) => {