Data streams v2

.changeset/slick-geese-happen.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': minor
+---
+
+Add support for data streams v2

package.json

@@ -72,7 +72,7 @@
   },
   "dependencies": {
     "@livekit/mutex": "1.1.1",
-    "@livekit/protocol": "1.46.6",
+    "@livekit/protocol": "1.50.4",
     "events": "^3.3.0",
     "jose": "^6.1.0",
     "loglevel": "^1.9.2",

src/index.ts

@@ -104,6 +104,8 @@ export type {
   TranscriptionSegment,
   ChatMessage,
   SendTextOptions,
+  SendBytesOptions,
+  ByteStreamInfo,
 } from './room/types';
 export * from './version';
 export {

src/room/Room.test.ts

@@ -81,7 +81,10 @@ describe('Room signaling options', () => {
       'wss://test.livekit.io',
       'test-token',
       expect.objectContaining({
-        clientInfoCapabilities: [ClientInfo_Capability.CAP_PACKET_TRAILER],
+        clientInfoCapabilities: [
+          ClientInfo_Capability.CAP_PACKET_TRAILER,
+          ClientInfo_Capability.CAP_COMPRESSION_DEFLATE_RAW,
+        ],
         e2eeEnabled: true,
       }),
       expect.any(AbortSignal),

src/room/Room.ts

@@ -118,6 +118,7 @@ import {
   getEmptyAudioStreamTrack,
   isBrowserSupported,
   isCloud,
+  isCompressionStreamSupported,
   isLocalAudioTrack,
   isLocalParticipant,
   isReactNative,
@@ -268,7 +269,13 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     this.maybeCreateEngine();
 
     this.incomingDataStreamManager = new IncomingDataStreamManager();
-    this.outgoingDataStreamManager = new OutgoingDataStreamManager(this.engine, this.log);
+    this.outgoingDataStreamManager = new OutgoingDataStreamManager(
+      this.engine,
+      this.log,
+      this.getRemoteParticipantClientProtocol,
+      this.getRemoteParticipantCapabilities,
+      this.getAllRemoteParticipantIdentities,
+    );
 
     this.incomingDataTrackManager = new IncomingDataTrackManager({ e2eeManager: this.e2eeManager });
     this.incomingDataTrackManager
@@ -971,11 +978,7 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
         autoSubscribe: connectOptions.autoSubscribe,
         adaptiveStream:
           typeof roomOptions.adaptiveStream === 'object' ? true : roomOptions.adaptiveStream,
-        clientInfoCapabilities:
-          isFrameMetadataSupported(roomOptions.frameMetadata ?? roomOptions.packetTrailer) ||
-          !!this.e2eeManager
-            ? [ClientInfo_Capability.CAP_PACKET_TRAILER]
-            : undefined,
+        clientInfoCapabilities: this.getClientInfoCapabilities(roomOptions),
         maxRetries: connectOptions.maxRetries,
         e2eeEnabled: !!this.e2eeManager,
         websocketTimeout: connectOptions.websocketTimeout,
@@ -2505,10 +2508,39 @@ class Room extends (EventEmitter as new () => TypedEmitter<RoomEventCallbacks>)
     }
   }
 
+  /** The client capabilities this SDK advertises to other participants in its `ClientInfo`. */
+  private getClientInfoCapabilities(
+    roomOptions: InternalRoomOptions,
+  ): Array<ClientInfo_Capability> {
+    const capabilities: Array<ClientInfo_Capability> = [];
+    if (
+      isFrameMetadataSupported(roomOptions.frameMetadata ?? roomOptions.packetTrailer) ||
+      !!this.e2eeManager
+    ) {
+      capabilities.push(ClientInfo_Capability.CAP_PACKET_TRAILER);
+    }
+    // Advertise deflate-raw decompression support so peers know they can send us compressed data
+    // streams (gated separately from clientProtocol — see the data streams v2 spec).
+    if (isCompressionStreamSupported()) {
+      capabilities.push(ClientInfo_Capability.CAP_COMPRESSION_DEFLATE_RAW);
+    }
+    return capabilities;
+  }
+
   private getRemoteParticipantClientProtocol = (identity: Participant['identity']) => {
     return this.remoteParticipants.get(identity)?.clientProtocol ?? CLIENT_PROTOCOL_DEFAULT;
   };
 
+  private getRemoteParticipantCapabilities = (
+    identity: Participant['identity'],
+  ): Array<ClientInfo_Capability> => {
+    return this.remoteParticipants.get(identity)?.capabilities ?? [];
+  };
+
+  private getAllRemoteParticipantIdentities = () => {
+    return Array.from(this.remoteParticipants.keys());
+  };
+
   private registerRpcDataStreamHandler() {
     this.incomingDataStreamManager.registerTextStreamHandler(
       RPC_REQUEST_DATA_STREAM_TOPIC,

src/room/data-stream/compression.ts

@@ -0,0 +1,117 @@
+/**
+ * Compression helpers for data streams. The buffered deflate-raw variant ({@link deflateRawCompress})
+ * is for the inline (single-packet) case where the payload is small and bounded;
+ * {@link deflateRawTransform} / {@link inflateRawTransform} serve the chunked (multi-packet)
+ * `sendText`/`sendBytes`/`sendFile` paths, streaming the bytes through without buffering the whole
+ * payload.
+ *
+ * These operate on bytes (not strings) so a single set of helpers serves both text and byte streams;
+ * the `TextEncoder`/`TextDecoder` boundary lives at the manager/reader edges.
+ *
+ * Both streaming variants are exposed as `ReadableWritablePair`s so they drop straight into a
+ * `pipeThrough` chain. Each needs one localized cast to bridge a DOM lib-type mismatch: the platform
+ * `CompressionStream`/`DecompressionStream` type their `writable` as `WritableStream<BufferSource>`
+ * (a wider element type than `Uint8Array`), and `WritableStream<W>` is covariant in `W`, so neither
+ * is structurally a `ReadableWritablePair<Uint8Array, Uint8Array>` without help.
+ *
+ * @internal
+ */
+import { type NonSharedUint8Array } from '../../type-polyfills/non-shared-typed-arrays';
+import { DataStreamError, DataStreamErrorReason } from '../errors';
+
+/**
+ * A `deflate-raw` compression transform (inverse of {@link inflateRawTransform}): pipe a byte stream
+ * through it to get the compressed bytes without buffering the whole payload. Used for the chunked
+ * `sendText`/`sendBytes`/`sendFile` paths, where the full payload is known up front but is streamed
+ * (e.g. from `file.stream()`) rather than held in memory.
+ */
+export function deflateRawTransform(): ReadableWritablePair<
+  NonSharedUint8Array,
+  NonSharedUint8Array
+> {
+  return new CompressionStream('deflate-raw') as unknown as ReadableWritablePair<
+    NonSharedUint8Array,
+    NonSharedUint8Array
+  >;
+}
+
+/**
+ * A `deflate-raw` decompression transform (inverse of {@link deflateRawTransform}): pipe a
+ * stream of compressed bytes through it to get the decompressed bytes. Inflate emits output greedily,
+ * so as long as the sender flushed at write boundaries each write's content is produced as soon as
+ * its compressed bytes arrive.
+ */
+export function inflateRawTransform(): ReadableWritablePair<
+  NonSharedUint8Array,
+  NonSharedUint8Array
+> {
+  return new DecompressionStream('deflate-raw') as unknown as ReadableWritablePair<
+    NonSharedUint8Array,
+    NonSharedUint8Array
+  >;
+}
+
+/** deflate-raw compresses a byte array in full. Use for inline payloads; prefer the streaming
+ * path for the chunked case. */
+export async function deflateRawCompress(data: NonSharedUint8Array): Promise<NonSharedUint8Array> {
+  const cs = new CompressionStream('deflate-raw');
+  const writer = cs.writable.getWriter();
+  writer.write(data as NonSharedUint8Array);
+  writer.close();
+  return collect(cs.readable);
+}
+
+/**
+ * Decompresses a raw-deflate byte array in full (inverse of {@link deflateRawCompress}).
+ * `maxByteLength` bounds the decompressed output (decompression-bomb guard); exceeding it rejects
+ * with a `PayloadTooLarge` error.
+ */
+export async function deflateRawDecompress(
+  data: NonSharedUint8Array,
+  maxByteLength?: number,
+): Promise<NonSharedUint8Array> {
+  const ds = new DecompressionStream('deflate-raw');
+  const writer = ds.writable.getWriter();
+  // The writer promises are intentionally not awaited (output is consumed via `collect`), but
+  // they reject when the byte cap cancels the readable mid-stream — swallow that so an enforced
+  // cap doesn't surface as an unhandled rejection.
+  writer.write(data as NonSharedUint8Array).catch(() => {});
+  writer.close().catch(() => {});
+  return collect(ds.readable, maxByteLength);
+}
+
+/**
+ * Drains a byte stream, concatenating all of its chunks into a single array. When
+ * `maxByteLength` is given, drops the stream and throws `PayloadTooLarge` as soon as the
+ * accumulated output exceeds it.
+ */
+export async function collect(
+  stream: ReadableStream<NonSharedUint8Array>,
+  maxByteLength?: number,
+): Promise<NonSharedUint8Array> {
+  const reader = stream.getReader();
+  const chunks: Array<NonSharedUint8Array> = [];
+  let total = 0;
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) {
+      break;
+    }
+    chunks.push(value);
+    total += value.byteLength;
+    if (typeof maxByteLength === 'number' && total > maxByteLength) {
+      await reader.cancel();
+      throw new DataStreamError(
+        `Decompressed payload exceeds the maximum payload size of ${maxByteLength} bytes`,
+        DataStreamErrorReason.PayloadTooLarge,
+      );
+    }
+  }
+  const result = new Uint8Array(total);
+  let offset = 0;
+  for (const chunk of chunks) {
+    result.set(chunk, offset);
+    offset += chunk.byteLength;
+  }
+  return result;
+}

src/room/data-stream/constants.ts

@@ -0,0 +1,18 @@
+/**
+ * Maximum size of a single data-stream chunk in bytes, and the budget used to decide whether a
+ * payload can be sent inline as a single header packet. Kept below the ~16k data-channel MTU to
+ * leave headroom for protocol framing and E2EE overhead.
+ *
+ * @internal
+ */
+export const STREAM_CHUNK_SIZE_BYTES = 15_000;
+
+/**
+ * Default cap on the number of decompressed bytes a single incoming compressed data stream may
+ * produce (5 GB). A tiny compressed payload can inflate to an arbitrarily large output
+ * (decompression bomb), so the decompressor's output is bounded rather than trusting the wire
+ * size; streams exceeding the cap error with `PayloadTooLarge`.
+ *
+ * @internal
+ */
+export const DEFAULT_MAX_PAYLOAD_BYTE_LENGTH = 5_000_000_000;