{Audio,Video}Stream: use AsyncIterableIterator

.changeset/real-toes-accept.md

@@ -0,0 +1,5 @@
+---
+"@livekit/rtc-node": minor
+---
+
+make AudioStream use AsyncIterableIterator instead of typed EventEmitter

packages/livekit-rtc/src/audio_stream.ts

@@ -1,39 +1,31 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
-import EventEmitter from 'events';
 import { AudioFrame } from './audio_frame.js';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
 import type { AudioStreamInfo, NewAudioStreamResponse } from './proto/audio_frame_pb.js';
 import { AudioStreamType, NewAudioStreamRequest } from './proto/audio_frame_pb.js';
 import type { Track } from './track.js';
+import { Mutex } from './utils.js';
 
-export type AudioFrameEvent = {
-  frame: AudioFrame;
-};
-
-export type AudioStreamCallbacks = {
-  frameReceived: (frame: AudioFrameEvent) => void;
-};
-
-export enum AudioStreamEvent {
-  FrameReceived = 'frameReceived',
-}
-
-export class AudioStream extends (EventEmitter as new () => TypedEmitter<AudioStreamCallbacks>) {
+export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
   info: AudioStreamInfo;
   /** @internal */
   ffiHandle: FfiHandle;
+  /** @internal */
+  eventQueue: (AudioFrame | null)[] = [];
+  /** @internal */
+  queueResolve: ((value: IteratorResult<AudioFrame>) => void) | null = null;
+  /** @internal */
+  mutex = new Mutex();
 
   track: Track;
   sampleRate: number;
   numChannels: number;
 
   constructor(track: Track, sampleRate: number = 48000, numChannels: number = 1) {
-    super();
     this.track = track;
     this.sampleRate = sampleRate;
     this.numChannels = numChannels;
@@ -70,15 +62,42 @@ export class AudioStream extends (EventEmitter as new () => TypedEmitter<AudioSt
     switch (streamEvent.case) {
       case 'frameReceived':
         const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame);
-        this.emit(AudioStreamEvent.FrameReceived, { frame });
+        if (this.queueResolve) {
+          this.queueResolve({ done: false, value: frame });
+        } else {
+          this.eventQueue.push(frame);
+        }
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
         break;
     }
   };
 
+  async next(): Promise<IteratorResult<AudioFrame>> {
+    const unlock = await this.mutex.lock();
+    if (this.eventQueue.length > 0) {
+      unlock();
+      const value = this.eventQueue.shift();
+      if (value) {
+        return { done: false, value };
+      } else {
+        return { done: true, value: undefined };
+      }
+    }
+    const promise = new Promise<IteratorResult<AudioFrame>>(
+      (resolve) => (this.queueResolve = resolve),
+    );
+    unlock();
+    return promise;
+  }
+
   close() {
+    this.eventQueue.push(null);
     this.ffiHandle.dispose();
   }
+
+  [Symbol.asyncIterator](): AudioStream {
+    return this;
+  }
 }

packages/livekit-rtc/src/index.ts

@@ -17,8 +17,8 @@ export {
 } from './track.js';
 export { VideoFrame } from './video_frame.js';
 export { AudioFrame } from './audio_frame.js';
-export { AudioStream, AudioStreamEvent, AudioFrameEvent } from './audio_stream.js';
-export { VideoStream, VideoStreamEvent, VideoFrameEvent } from './video_stream.js';
+export { AudioStream } from './audio_stream.js';
+export { VideoStream, VideoFrameEvent } from './video_stream.js';
 export { AudioSource } from './audio_source.js';
 export { VideoSource } from './video_source.js';
 export {

packages/livekit-rtc/src/utils.ts

@@ -0,0 +1,36 @@
+// SPDX-FileCopyrightText: 2024 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+export class Mutex {
+  #locking: Promise<void>;
+  #locks: number;
+  #limit: number;
+
+  constructor(limit = 1) {
+    this.#locking = Promise.resolve();
+    this.#locks = 0;
+    this.#limit = limit;
+  }
+
+  isLocked(): boolean {
+    return this.#locks >= this.#limit;
+  }
+
+  async lock(): Promise<() => void> {
+    this.#locks += 1;
+
+    let unlockNext: () => void;
+
+    const willLock = new Promise<void>(
+      (resolve) =>
+        (unlockNext = () => {
+          this.#locks -= 1;
+          resolve();
+        }),
+    );
+
+    const willUnlock = this.#locking.then(() => unlockNext);
+    this.#locking = this.#locking.then(() => willLock);
+    return willUnlock;
+  }
+}

packages/livekit-rtc/src/video_stream.ts

@@ -1,8 +1,6 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import type { TypedEventEmitter as TypedEmitter } from '@livekit/typed-emitter';
-import EventEmitter from 'events';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
 import type {
@@ -12,6 +10,7 @@ import type {
 } from './proto/video_frame_pb.js';
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
+import { Mutex } from './utils.js';
 import { VideoFrame } from './video_frame.js';
 
 export type VideoFrameEvent = {
@@ -20,24 +19,21 @@ export type VideoFrameEvent = {
   rotation: VideoRotation;
 };
 
-export type VideoStreamCallbacks = {
-  frameReceived: (evt: VideoFrameEvent) => void;
-};
-
-export enum VideoStreamEvent {
-  FrameReceived = 'frameReceived',
-}
-
-export class VideoStream extends (EventEmitter as new () => TypedEmitter<VideoStreamCallbacks>) {
+export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
   /** @internal */
   info: VideoStreamInfo;
   /** @internal */
   ffiHandle: FfiHandle;
+  /** @internal */
+  eventQueue: (VideoFrameEvent | null)[] = [];
+  /** @internal */
+  queueResolve: ((value: IteratorResult<VideoFrameEvent>) => void) | null = null;
+  /** @internal */
+  mutex = new Mutex();
 
   track: Track;
 
   constructor(track: Track) {
-    super();
     this.track = track;
 
     const req = new NewVideoStreamRequest({
@@ -72,15 +68,43 @@ export class VideoStream extends (EventEmitter as new () => TypedEmitter<VideoSt
         const rotation = streamEvent.value.rotation;
         const timestampUs = streamEvent.value.timestampUs;
         const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer);
-        this.emit(VideoStreamEvent.FrameReceived, { frame, timestampUs, rotation });
+        const value = { rotation, timestampUs, frame };
+        if (this.queueResolve) {
+          this.queueResolve({ done: false, value });
+        } else {
+          this.eventQueue.push(value);
+        }
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
         break;
     }
   };
 
+  async next(): Promise<IteratorResult<VideoFrameEvent>> {
+    const unlock = await this.mutex.lock();
+    if (this.eventQueue.length > 0) {
+      unlock();
+      const value = this.eventQueue.shift();
+      if (value) {
+        return { done: false, value };
+      } else {
+        return { done: true, value: undefined };
+      }
+    }
+    const promise = new Promise<IteratorResult<VideoFrameEvent>>(
+      (resolve) => (this.queueResolve = resolve),
+    );
+    unlock();
+    return promise;
+  }
+
   close() {
+    this.eventQueue.push(null);
     this.ffiHandle.dispose();
   }
+
+  [Symbol.asyncIterator](): VideoStream {
+    return this;
+  }
 }