Add track processor API (livekit#711)

* model insertable streams api

* kitten processor, virtual background api/processor

* init support for stream processors

* processor stability updates

* simplify processor handling

* add notes

* fix types

* Update src/room/track/LocalTrack.ts

Co-authored-by: David Zhao <dz@livekit.io>

* remove console log

* Create lovely-trains-visit.md

* change changeset to minor

---------

Co-authored-by: David Zhao <dz@livekit.io>

Author: lukasIO

.DS_Store

@@ -0,0 +1,5 @@
+---
+'livekit-client': minor
+---
+
+Add track processor API

.changeset/lovely-trains-visit.md

@@ -40,6 +40,7 @@ export * from './room/track/options';
 export * from './room/track/types';
 export type { DataPublishOptions, SimulationScenario } from './room/types';
 export * from './version';
+export * from './room/track/processor/types';
 export {
   setLogLevel,
   setLogExtension,

src/index.ts

@@ -6,6 +6,7 @@ import { TrackEvent } from '../events';
 import { Mutex, compareVersions, isMobile, sleep } from '../utils';
 import { Track, attachToElement, detachTrack } from './Track';
 import type { VideoCodec } from './options';
+import type { TrackProcessor } from './processor/types';
 
 const defaultDimensionsTimeout = 1000;
 
@@ -26,6 +27,12 @@ export default abstract class LocalTrack extends Track {
 
   protected pauseUpstreamLock: Mutex;
 
+  protected processorElement?: HTMLMediaElement;
+
+  protected processor?: TrackProcessor<typeof this.kind>;
+
+  protected isSettingUpProcessor: boolean = false;
+
   /**
    *
    * @param mediaTrack
@@ -77,6 +84,10 @@ export default abstract class LocalTrack extends Track {
     return this.providedByUser;
   }
 
+  get mediaStreamTrack() {
+    return this.processor?.processedTrack ?? this._mediaStreamTrack;
+  }
+
   async waitForDimensions(timeout = defaultDimensionsTimeout): Promise<Track.Dimensions> {
     if (this.kind === Track.Kind.Audio) {
       throw new Error('cannot get dimensions for audio tracks');
@@ -153,6 +164,9 @@ export default abstract class LocalTrack extends Track {
 
     this.mediaStream = new MediaStream([track]);
     this.providedByUser = userProvidedTrack;
+    if (this.processor) {
+      await this.stopProcessor();
+    }
     return this;
   }
 
@@ -175,7 +189,7 @@ export default abstract class LocalTrack extends Track {
 
     // detach
     this.attachedElements.forEach((el) => {
-      detachTrack(this._mediaStreamTrack, el);
+      detachTrack(this.mediaStreamTrack, el);
     });
     this._mediaStreamTrack.removeEventListener('ended', this.handleEnded);
     // on Safari, the old audio track must be stopped before attempting to acquire
@@ -198,12 +212,16 @@ export default abstract class LocalTrack extends Track {
 
     await this.resumeUpstream();
 
-    this.attachedElements.forEach((el) => {
-      attachToElement(newTrack, el);
-    });
-
     this.mediaStream = mediaStream;
     this.constraints = constraints;
+    if (this.processor) {
+      const processor = this.processor;
+      await this.setProcessor(processor);
+    } else {
+      this.attachedElements.forEach((el) => {
+        attachToElement(this._mediaStreamTrack, el);
+      });
+    }
     this.emit(TrackEvent.Restarted, this);
     return this;
   }
@@ -248,6 +266,12 @@ export default abstract class LocalTrack extends Track {
     this.emit(TrackEvent.Ended, this);
   };
 
+  stop() {
+    super.stop();
+    this.processor?.destroy();
+    this.processor = undefined;
+  }
+
   /**
    * pauses publishing to the server without disabling the local MediaStreamTrack
    * this is used to display a user's own video locally while pausing publishing to
@@ -297,5 +321,81 @@ export default abstract class LocalTrack extends Track {
     }
   }
 
+  /**
+   * Sets a processor on this track.
+   * See https://github.com/livekit/track-processors-js for example usage
+   *
+   * @experimental
+   *
+   * @param processor
+   * @param showProcessedStreamLocally
+   * @returns
+   */
+  async setProcessor(
+    processor: TrackProcessor<typeof this.kind>,
+    showProcessedStreamLocally = true,
+  ) {
+    if (this.isSettingUpProcessor) {
+      log.warn('already trying to set up a processor');
+      return;
+    }
+    log.debug('setting up processor');
+    this.isSettingUpProcessor = true;
+    if (this.processor) {
+      await this.stopProcessor();
+    }
+    if (this.kind === 'unknown') {
+      throw TypeError('cannot set processor on track of unknown kind');
+    }
+    this.processorElement = this.processorElement ?? document.createElement(this.kind);
+    this.processorElement.muted = true;
+
+    attachToElement(this._mediaStreamTrack, this.processorElement);
+    this.processorElement.play().catch((e) => log.error(e));
+
+    const processorOptions = {
+      kind: this.kind,
+      track: this._mediaStreamTrack,
+      element: this.processorElement,
+    };
+
+    await processor.init(processorOptions);
+    this.processor = processor;
+    if (this.processor.processedTrack) {
+      for (const el of this.attachedElements) {
+        if (el !== this.processorElement && showProcessedStreamLocally) {
+          detachTrack(this._mediaStreamTrack, el);
+          attachToElement(this.processor.processedTrack, el);
+        }
+      }
+      await this.sender?.replaceTrack(this.processor.processedTrack);
+    }
+    this.isSettingUpProcessor = false;
+  }
+
+  getProcessor() {
+    return this.processor;
+  }
+
+  /**
+   * Stops the track processor
+   * See https://github.com/livekit/track-processors-js for example usage
+   *
+   * @experimental
+   * @returns
+   */
+  async stopProcessor() {
+    if (!this.processor) return;
+
+    log.debug('stopping processor');
+    this.processor.processedTrack?.stop();
+    await this.processor.destroy();
+    this.processor = undefined;
+    this.processorElement?.remove();
+    this.processorElement = undefined;
+
+    await this.restart();
+  }
+
   protected abstract monitorSender(): void;
 }

src/room/track/LocalTrack.ts

@@ -118,7 +118,7 @@ export abstract class Track extends (EventEmitter as new () => TypedEventEmitter
     // even if we believe it's already attached to the element, it's possible
     // the element's srcObject was set to something else out of band.
     // we'll want to re-attach it in that case
-    attachToElement(this._mediaStreamTrack, element);
+    attachToElement(this.mediaStreamTrack, element);
 
     // handle auto playback failures
     const allMediaStreamTracks = (element.srcObject as MediaStream).getTracks();
@@ -167,7 +167,7 @@ export abstract class Track extends (EventEmitter as new () => TypedEventEmitter
     try {
       // detach from a single element
       if (element) {
-        detachTrack(this._mediaStreamTrack, element);
+        detachTrack(this.mediaStreamTrack, element);
         const idx = this.attachedElements.indexOf(element);
         if (idx >= 0) {
           this.attachedElements.splice(idx, 1);
@@ -179,7 +179,7 @@ export abstract class Track extends (EventEmitter as new () => TypedEventEmitter
 
       const detached: HTMLMediaElement[] = [];
       this.attachedElements.forEach((elm) => {
-        detachTrack(this._mediaStreamTrack, elm);
+        detachTrack(this.mediaStreamTrack, elm);
         detached.push(elm);
         this.recycleElement(elm);
         this.emit(TrackEvent.ElementDetached, elm);

src/room/track/Track.ts

@@ -0,0 +1,20 @@
+import type { Track } from '../Track';
+
+/**
+ * @experimental
+ */
+export type ProcessorOptions<T extends Track.Kind> = {
+  kind: T;
+  track: MediaStreamTrack;
+  element?: HTMLMediaElement;
+};
+
+/**
+ * @experimental
+ */
+export interface TrackProcessor<T extends Track.Kind> {
+  name: string;
+  init: (opts: ProcessorOptions<T>) => void;
+  destroy: () => Promise<void>;
+  processedTrack?: MediaStreamTrack;
+}