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..105d6c09 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, @@ -323,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!; @@ -415,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?: { @@ -454,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!; @@ -540,6 +562,53 @@ 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?: Omit & { streamId?: string }, + ) { + 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 }, + ); + + 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}`, + ); + } + } + 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