Add LocalParticipant.sendBytes

.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.

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<ByteStreamOptions, 'onProgress'> & { 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<StreamSendBytesResponse>({
+      message: { case: 'sendBytes', value: req },
+    });
+
+    const cb = await FfiClient.instance.waitFor<StreamSendBytesCallback>(
+      (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<TextStreamWriterWriteResponse>({
       message: { case: 'textStreamWrite', value: req },

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<Uint8Array>((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