Data tracks - incoming manager

src/index.ts

@@ -12,7 +12,7 @@ import type { ReconnectContext, ReconnectPolicy } from './room/ReconnectPolicy';
 import Room, { ConnectionState, type RoomEventCallbacks } from './room/Room';
 import * as attributes from './room/attribute-typings';
 // FIXME: remove this import in a follow up data track pull request.
-import './room/data-track/depacketizer';
+import './room/data-track/incoming/IncomingDataTrackManager';
 // FIXME: remove this import in a follow up data track pull request.
 import './room/data-track/outgoing/OutgoingDataTrackManager';
 import LocalParticipant from './room/participant/LocalParticipant';

src/room/data-track/track.ts → src/room/data-track/LocalDataTrack.ts

@@ -1,22 +1,25 @@
 import type { DataTrackFrame } from './frame';
-import { type DataTrackHandle } from './handle';
 import type OutgoingDataTrackManager from './outgoing/OutgoingDataTrackManager';
+import {
+  DataTrackSymbol,
+  type IDataTrack,
+  type ILocalTrack,
+  TrackSymbol,
+} from './track-interfaces';
+import type { DataTrackInfo } from './types';
 
-export type DataTrackSid = string;
+export default class LocalDataTrack implements ILocalTrack, IDataTrack {
+  readonly trackSymbol = TrackSymbol;
 
-/** Information about a published data track. */
-export type DataTrackInfo = {
-  sid: DataTrackSid;
-  pubHandle: DataTrackHandle;
-  name: String;
-  usesE2ee: boolean;
-};
+  readonly isLocal = true;
+
+  readonly typeSymbol = DataTrackSymbol;
 
-export class LocalDataTrack {
   info: DataTrackInfo;
 
   protected manager: OutgoingDataTrackManager;
 
+  /** @internal */
   constructor(info: DataTrackInfo, manager: OutgoingDataTrackManager) {
     this.info = info;
     this.manager = manager;

src/room/data-track/RemoteDataTrack.ts

@@ -0,0 +1,82 @@
+import type Participant from '../participant/Participant';
+import type { DataTrackFrame } from './frame';
+import type IncomingDataTrackManager from './incoming/IncomingDataTrackManager';
+import {
+  DataTrackSymbol,
+  type IDataTrack,
+  type IRemoteTrack,
+  TrackSymbol,
+} from './track-interfaces';
+import { type DataTrackInfo } from './types';
+
+type RemoteDataTrackOptions = {
+  publisherIdentity: Participant['identity'];
+};
+
+export type RemoteDataTrackSubscribeOptions = {
+  signal?: AbortSignal;
+
+  /** The number of {@link DataTrackFrame}s to hold in the ReadableStream before disgarding extra
+   * frames. Defaults to 4, but this may not be good enough for especially high frequency data. */
+  highWaterMark?: number;
+};
+
+export default class RemoteDataTrack implements IRemoteTrack, IDataTrack {
+  readonly trackSymbol = TrackSymbol;
+
+  readonly isLocal = false;
+
+  readonly typeSymbol = DataTrackSymbol;
+
+  info: DataTrackInfo;
+
+  publisherIdentity: Participant['identity'];
+
+  protected manager: IncomingDataTrackManager;
+
+  /** @internal */
+  constructor(
+    info: DataTrackInfo,
+    manager: IncomingDataTrackManager,
+    options: RemoteDataTrackOptions,
+  ) {
+    this.info = info;
+    this.manager = manager;
+    this.publisherIdentity = options.publisherIdentity;
+  }
+
+  /** Subscribes to the data track to receive frames.
+   *
+   * # Returns
+   *
+   * A stream that yields {@link DataTrackFrame}s as they arrive.
+   *
+   * # Multiple Subscriptions
+   *
+   * An application may call `subscribe` more than once to process frames in
+   * multiple places. For example, one async task might plot values on a graph
+   * while another writes them to a file.
+   *
+   * Internally, only the first call to `subscribe` communicates with the SFU and
+   * allocates the resources required to receive frames. Additional subscriptions
+   * reuse the same underlying pipeline and do not trigger additional signaling.
+   *
+   * Note that newly created subscriptions only receive frames published after
+   * the initial subscription is established.
+   */
+  async subscribe(
+    options?: RemoteDataTrackSubscribeOptions,
+  ): Promise<ReadableStream<DataTrackFrame>> {
+    try {
+      const stream = await this.manager.subscribeRequest(
+        this.info.sid,
+        options?.signal,
+        options?.highWaterMark,
+      );
+      return stream;
+    } catch (err) {
+      // NOTE: Rethrow errors to break Throws<...> type boundary
+      throw err;
+    }
+  }
+}

src/room/data-track/depacketizer.ts

@@ -21,7 +21,7 @@ type PartialFrame = {
 
 /** An error indicating a frame was dropped. */
 export class DataTrackDepacketizerDropError<
-  Reason extends DataTrackDepacketizerDropReason,
+  Reason extends DataTrackDepacketizerDropReason = DataTrackDepacketizerDropReason,
 > extends LivekitReasonedError<Reason> {
   readonly name = 'DataTrackDepacketizerDropError';
 
@@ -99,13 +99,7 @@ export default class DataTrackDepacketizer {
   push(
     packet: DataTrackPacket,
     options?: PushOptions,
-  ): Throws<
-    DataTrackFrame | null,
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.Interrupted>
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.BufferFull>
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.UnknownFrame>
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.Incomplete>
-  > {
+  ): Throws<DataTrackFrame | null, DataTrackDepacketizerDropError> {
     switch (packet.header.marker) {
       case FrameMarker.Single:
         return this.frameFromSingle(packet, options);
@@ -191,13 +185,7 @@ export default class DataTrackDepacketizer {
   /** Push to the existing partial frame. */
   private pushToPartial(
     packet: DataTrackPacket,
-  ): Throws<
-    DataTrackFrame | null,
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.Interrupted>
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.UnknownFrame>
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.BufferFull>
-    | DataTrackDepacketizerDropError<DataTrackDepacketizerDropReason.Incomplete>
-  > {
+  ): Throws<DataTrackFrame | null, DataTrackDepacketizerDropError> {
     if (packet.header.marker !== FrameMarker.Inter && packet.header.marker !== FrameMarker.Final) {
       // @throws-transformer ignore - this should be treated as a "panic" and not be caught
       throw new Error(

src/room/data-track/e2ee.ts

@@ -10,5 +10,6 @@ export type EncryptionProvider = {
 };
 
 export type DecryptionProvider = {
-  decrypt(payload: Uint8Array, senderIdentity: string): Uint8Array;
+  // FIXME: add in explicit `Throws<..., DecryptionError>`?
+  decrypt(payload: EncryptedPayload, senderIdentity: string): Uint8Array;
 };

src/room/data-track/handle.ts

@@ -43,13 +43,7 @@ export class DataTrackHandleError<
 
 export type DataTrackHandle = number;
 export const DataTrackHandle = {
-  fromNumber(
-    raw: number,
-  ): Throws<
-    DataTrackHandle,
-    | DataTrackHandleError<DataTrackHandleErrorReason.TooLarge>
-    | DataTrackHandleError<DataTrackHandleErrorReason.Reserved>
-  > {
+  fromNumber(raw: number): Throws<DataTrackHandle, DataTrackHandleError> {
     if (raw === 0) {
       throw DataTrackHandleError.reserved(raw);
     }