From 16997da73981b19df86cb4fe61aa7e99263cb3d7 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 21:41:13 -0700
Subject: [PATCH 01/26] init

---
 packages/livekit-rtc/src/utils.test.ts | 30 ++++++++++++++++-
 packages/livekit-rtc/src/utils.ts      | 46 ++++++++++++++++++++++++++
 2 files changed, 75 insertions(+), 1 deletion(-)

diff --git a/packages/livekit-rtc/src/utils.test.ts b/packages/livekit-rtc/src/utils.test.ts
index 35ac904b..af32f49c 100644
--- a/packages/livekit-rtc/src/utils.test.ts
+++ b/packages/livekit-rtc/src/utils.test.ts
@@ -2,7 +2,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import { describe, expect, it } from 'vitest';
-import { splitUtf8 } from './utils.js';
+import { RingQueue, splitUtf8 } from './utils.js';
 
 describe('splitUtf8', () => {
   it('splits a string into chunks of the given size', () => {
@@ -49,3 +49,31 @@ describe('splitUtf8', () => {
     expect(splitUtf8('', 5)).toEqual([]);
   });
 });
+
+describe('RingQueue', () => {
+  it('should push and get items', async () => {
+    const queue = new RingQueue<number>(3);
+    queue.push(1);
+    queue.push(2);
+    expect(await queue.get()).toBe(1);
+    expect(await queue.get()).toBe(2);
+  });
+
+  it('should push and get items with a capacity', async () => {
+    const queue = new RingQueue<number>(3);
+    queue.push(1);
+    queue.push(2);
+    queue.push(3);
+    queue.push(4);
+    expect(await queue.get()).toBe(1);
+    expect(await queue.get()).toBe(2);
+    expect(await queue.get()).toBe(3);
+  });
+
+  it('should push and get items with a capacity of 0', async () => {
+    const queue = new RingQueue<number>(0);
+    queue.push(1);
+    queue.push(2);
+    queue.push(3);
+  });
+});
diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index 7bc5a650..4a68a8fc 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -40,3 +40,49 @@ export function splitUtf8(s: string, n: number): Uint8Array[] {
   }
   return result;
 }
+
+/**
+ * A ring queue that stores a fixed number of items. If capacity is 
+ * set to 0, the buffer will grow indefinitely.
+ */
+export class RingQueue<T> {
+  private buffer: T[] = [];
+  // If the buffer is empty, this will be a promise that resolves when
+  // a new item is pushed, bypassing the internal queue.
+  private resolve: ((value: T) => void) | null = null;
+
+  constructor(private readonly capacity: number = 0) {
+    this.resolve = null;
+    this.capacity = capacity;
+  }
+
+  /**
+   * Push an item into the buffer. If the buffer is full, the oldest item will 
+   * be removed.
+   */
+  push(item: T){
+    if (this.resolve) {
+      this.resolve(item);
+      this.resolve = null;
+    } else {
+      if (this.capacity > 0 && this.buffer.length > this.capacity) {
+        this.buffer.pop();
+      }
+      this.buffer.push(item);
+    }
+
+  }
+
+  /**
+   * Get an item from the buffer. If the buffer is empty, this will return a promise
+   * that resolves when a new item is pushed.
+   */
+  async get(): Promise<T> {
+    if (this.buffer.length === 0) {
+      return new Promise((resolve) => {
+        this.resolve = resolve;
+      });
+    }
+    return this.buffer.shift()!;
+  }
+}

From 695d3ab1afc19218e22858856819602391230ef7 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 21:53:39 -0700
Subject: [PATCH 02/26] update video frame

---
 packages/livekit-rtc/src/utils.ts        |  2 +
 packages/livekit-rtc/src/video_stream.ts | 49 +++++++-----------------
 2 files changed, 15 insertions(+), 36 deletions(-)

diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index 4a68a8fc..49895811 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -44,6 +44,8 @@ export function splitUtf8(s: string, n: number): Uint8Array[] {
 /**
  * A ring queue that stores a fixed number of items. If capacity is 
  * set to 0, the buffer will grow indefinitely.
+ * 
+ * WARNING: THIS IS NOT THREAD SAFE.
  */
 export class RingQueue<T> {
   private buffer: T[] = [];
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 487c99b7..2a8bce98 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -12,6 +12,7 @@ import type {
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
 import { VideoFrame } from './video_frame.js';
+import { RingQueue } from './utils.js';
 
 export type VideoFrameEvent = {
   frame: VideoFrame;
@@ -25,15 +26,13 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  eventQueue: (VideoFrameEvent | null)[] = [];
-  /** @internal */
-  queueResolve: ((value: IteratorResult<VideoFrameEvent>) => void) | null = null;
+  eventQueue: RingQueue<IteratorResult<VideoFrameEvent>>;
   /** @internal */
   mutex = new Mutex();
 
   track: Track;
 
-  constructor(track: Track) {
+  constructor(track: Track, capacity: number = 0) {
     this.track = track;
 
     const req = new NewVideoStreamRequest({
@@ -52,6 +51,8 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
     FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+
+    this.eventQueue = new RingQueue<IteratorResult<VideoFrameEvent>>(capacity);
   }
 
   private onEvent = (ev: FfiEvent) => {
@@ -68,24 +69,11 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
         const rotation = streamEvent.value.rotation;
         const timestampUs = streamEvent.value.timestampUs;
         const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
-        const value = { rotation, timestampUs, frame };
-        if (this.queueResolve) {
-          this.queueResolve({
-            done: false,
-            value: {
-              frame: value.frame,
-              timestampUs: value.timestampUs!,
-              rotation: value.rotation!,
-            },
-          });
-          this.queueResolve = null;
-        } else {
-          this.eventQueue.push({
-            frame: value.frame,
-            timestampUs: value.timestampUs!,
-            rotation: value.rotation!,
-          });
-        }
+        this.eventQueue.push({ done: false, value: { 
+          rotation: rotation!, 
+          timestampUs: timestampUs!, 
+          frame 
+        }});
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
@@ -95,24 +83,13 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
 
   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),
-    );
+    const result = this.eventQueue.get();
     unlock();
-    return promise;
+    return result;
   }
 
   close() {
-    this.eventQueue.push(null);
+    this.eventQueue.push({ done: true, value: undefined });
     this.ffiHandle.dispose();
   }
 

From 70168d228e1625ee1f7cfa345e8737184488afe2 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 22:02:28 -0700
Subject: [PATCH 03/26] format

---
 packages/livekit-rtc/src/utils.ts        |  9 ++++-----
 packages/livekit-rtc/src/video_stream.ts | 15 +++++++++------
 2 files changed, 13 insertions(+), 11 deletions(-)

diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index 49895811..e5af9861 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -42,9 +42,9 @@ export function splitUtf8(s: string, n: number): Uint8Array[] {
 }
 
 /**
- * A ring queue that stores a fixed number of items. If capacity is 
+ * A ring queue that stores a fixed number of items. If capacity is
  * set to 0, the buffer will grow indefinitely.
- * 
+ *
  * WARNING: THIS IS NOT THREAD SAFE.
  */
 export class RingQueue<T> {
@@ -59,10 +59,10 @@ export class RingQueue<T> {
   }
 
   /**
-   * Push an item into the buffer. If the buffer is full, the oldest item will 
+   * Push an item into the buffer. If the buffer is full, the oldest item will
    * be removed.
    */
-  push(item: T){
+  push(item: T) {
     if (this.resolve) {
       this.resolve(item);
       this.resolve = null;
@@ -72,7 +72,6 @@ export class RingQueue<T> {
       }
       this.buffer.push(item);
     }
-
   }
 
   /**
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 2a8bce98..c3b69161 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -11,8 +11,8 @@ import type {
 } from './proto/video_frame_pb.js';
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
-import { VideoFrame } from './video_frame.js';
 import { RingQueue } from './utils.js';
+import { VideoFrame } from './video_frame.js';
 
 export type VideoFrameEvent = {
   frame: VideoFrame;
@@ -69,11 +69,14 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
         const rotation = streamEvent.value.rotation;
         const timestampUs = streamEvent.value.timestampUs;
         const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
-        this.eventQueue.push({ done: false, value: { 
-          rotation: rotation!, 
-          timestampUs: timestampUs!, 
-          frame 
-        }});
+        this.eventQueue.push({
+          done: false,
+          value: {
+            rotation: rotation!,
+            timestampUs: timestampUs!,
+            frame,
+          },
+        });
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);

From 2878e3edb36d804d8be48948d8dfe981e4f29b9e Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 22:05:22 -0700
Subject: [PATCH 04/26] update comment

---
 packages/livekit-rtc/src/utils.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index e5af9861..d92e1efc 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -59,7 +59,7 @@ export class RingQueue<T> {
   }
 
   /**
-   * Push an item into the buffer. If the buffer is full, the oldest item will
+   * Push an item into the buffer. If the buffer is full, the oldest newest will
    * be removed.
    */
   push(item: T) {

From bdb1e14fa6467b57b565236a334c038f2eb7fbc8 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 22:18:16 -0700
Subject: [PATCH 05/26] typo

---
 packages/livekit-rtc/src/utils.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index d92e1efc..079d74d3 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -59,7 +59,7 @@ export class RingQueue<T> {
   }
 
   /**
-   * Push an item into the buffer. If the buffer is full, the oldest newest will
+   * Push an item into the buffer. If the buffer is full, the newest will
    * be removed.
    */
   push(item: T) {

From ec3cd18b3488a9f96ce91f0773c3b70b85506732 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 22:18:49 -0700
Subject: [PATCH 06/26] i can't type for my life

---
 packages/livekit-rtc/src/utils.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index 079d74d3..e4f94431 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -59,7 +59,7 @@ export class RingQueue<T> {
   }
 
   /**
-   * Push an item into the buffer. If the buffer is full, the newest will
+   * Push an item into the buffer. If the buffer is full, the newest item will
    * be removed.
    */
   push(item: T) {

From 83178b3aed80447a0dea120a313cf41184a0c71f Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 3 Apr 2025 22:23:25 -0700
Subject: [PATCH 07/26] update audio stream

---
 packages/livekit-rtc/src/audio_stream.ts | 32 ++++++------------------
 1 file changed, 8 insertions(+), 24 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 9a7c96cc..75931375 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -8,6 +8,7 @@ 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 { RingQueue } from './utils.js';
 
 export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
@@ -15,9 +16,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  eventQueue: (AudioFrame | null)[] = [];
-  /** @internal */
-  queueResolve: ((value: IteratorResult<AudioFrame>) => void) | null = null;
+  eventQueue: RingQueue<IteratorResult<AudioFrame>>;
   /** @internal */
   mutex = new Mutex();
 
@@ -25,7 +24,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   sampleRate: number;
   numChannels: number;
 
-  constructor(track: Track, sampleRate: number = 48000, numChannels: number = 1) {
+  constructor(track: Track, sampleRate: number = 48000, numChannels: number = 1, capacity: number = 0) {
     this.track = track;
     this.sampleRate = sampleRate;
     this.numChannels = numChannels;
@@ -46,6 +45,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
 
     this.info = res.stream!.info!;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
+    this.eventQueue = new RingQueue<IteratorResult<AudioFrame>>(capacity);
 
     FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
   }
@@ -62,12 +62,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
     switch (streamEvent.case) {
       case 'frameReceived':
         const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
-        if (this.queueResolve) {
-          this.queueResolve({ done: false, value: frame });
-          this.queueResolve = null;
-        } else {
-          this.eventQueue.push(frame);
-        }
+        this.eventQueue.push({ done: false, value: frame });
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
@@ -77,24 +72,13 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
 
   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),
-    );
+    const result = this.eventQueue.get();
     unlock();
-    return promise;
+    return result;
   }
 
   close() {
-    this.eventQueue.push(null);
+    this.eventQueue.push({ done: true, value: undefined });
     this.ffiHandle.dispose();
   }
 

From 54f11ea9192dfbdded1e3fa2b8f8437e73544f42 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 20:24:39 -0700
Subject: [PATCH 08/26] update audio stream to use

---
 packages/livekit-rtc/src/audio_stream.ts | 94 +++++++++++++++++-------
 1 file changed, 66 insertions(+), 28 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 75931375..b8d930b9 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -8,7 +8,7 @@ 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 { RingQueue } from './utils.js';
+import { ReadableStream, TransformStream } from 'node:stream/web';
 
 export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
@@ -16,9 +16,11 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  eventQueue: RingQueue<IteratorResult<AudioFrame>>;
-  /** @internal */
   mutex = new Mutex();
+  /** @internal */
+  private reader: ReadableStreamDefaultReader<AudioFrame>;
+  /** @internal */
+  private onEvent: ((ev: FfiEvent) => void) | null = null;
 
   track: Track;
   sampleRate: number;
@@ -45,40 +47,76 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
 
     this.info = res.stream!.info!;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
-    this.eventQueue = new RingQueue<IteratorResult<AudioFrame>>(capacity);
 
-    FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
-  }
+    // Create a readable stream from FfiClient events
+    const source = new ReadableStream<FfiEvent>({
+      start: (controller) => {
+        this.onEvent = (ev: FfiEvent) => {
+          if (
+            ev.message.case === 'audioStreamEvent' &&
+            ev.message.value.streamHandle === this.ffiHandle.handle
+          ) {
+            controller.enqueue(ev);
+          }
+        };
+        FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+      },
+      cancel: () => {
+        if (this.onEvent) {
+          FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+          this.onEvent = null;
+        }
+      }
+    });
 
-  private onEvent = (ev: FfiEvent) => {
-    if (
-      ev.message.case != 'audioStreamEvent' ||
-      ev.message.value.streamHandle != this.ffiHandle.handle
-    ) {
-      return;
-    }
+    const transformStream = new TransformStream<FfiEvent, AudioFrame>(
+      {
+        transform: (event: FfiEvent, controller: TransformStreamDefaultController<AudioFrame>) => {
+          if (
+            event.message.case !== 'audioStreamEvent' ||
+            event.message.value.streamHandle !== this.ffiHandle.handle
+          ) {
+            return;
+          }
 
-    const streamEvent = ev.message.value.message;
-    switch (streamEvent.case) {
-      case 'frameReceived':
-        const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
-        this.eventQueue.push({ done: false, value: frame });
-        break;
-      case 'eos':
-        FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-        break;
-    }
-  };
+          const streamEvent = event.message.value.message;
+          switch (streamEvent.case) {
+            case 'frameReceived':
+              const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
+              controller.enqueue(frame);
+              break;
+            case 'eos':
+              controller.terminate();
+              if (this.onEvent) {
+                FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+                this.onEvent = null;
+              }
+              break;
+          }
+        }
+      }, {
+        highWaterMark: capacity > 0 ? capacity : undefined
+      }
+    );
+    
+    this.reader = source.pipeThrough(transformStream).getReader();
+  }
 
   async next(): Promise<IteratorResult<AudioFrame>> {
     const unlock = await this.mutex.lock();
-    const result = this.eventQueue.get();
-    unlock();
-    return result;
+    try {
+      const result = await this.reader.read();
+      return {
+        done: result.done,
+        value: result.done ? undefined as any : result.value
+      };
+    } finally {
+      unlock();
+    }
   }
 
   close() {
-    this.eventQueue.push({ done: true, value: undefined });
+    this.reader.cancel();
     this.ffiHandle.dispose();
   }
 

From 3409a6a4996f00fd0c36427028a0dc493bbc506e Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 20:38:30 -0700
Subject: [PATCH 09/26] drop frames on backpressure

---
 packages/livekit-rtc/src/audio_stream.ts |   7 +-
 packages/livekit-rtc/src/video_stream.ts | 110 +++++++++++++++--------
 2 files changed, 78 insertions(+), 39 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index b8d930b9..07250d81 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -48,7 +48,6 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
     this.info = res.stream!.info!;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
-    // Create a readable stream from FfiClient events
     const source = new ReadableStream<FfiEvent>({
       start: (controller) => {
         this.onEvent = (ev: FfiEvent) => {
@@ -56,7 +55,11 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
             ev.message.case === 'audioStreamEvent' &&
             ev.message.value.streamHandle === this.ffiHandle.handle
           ) {
-            controller.enqueue(ev);
+            if (controller.desiredSize && controller.desiredSize > 0) {
+              controller.enqueue(ev);
+            } else{
+              console.warn('Dropping audio frame due to low buffer size');
+            }
           }
         };
         FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index c3b69161..1cf55f6d 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -11,7 +11,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 { RingQueue } from './utils.js';
+import { ReadableStream, TransformStream, TransformStreamDefaultController } from 'node:stream/web';
 import { VideoFrame } from './video_frame.js';
 
 export type VideoFrameEvent = {
@@ -26,9 +26,11 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  eventQueue: RingQueue<IteratorResult<VideoFrameEvent>>;
-  /** @internal */
   mutex = new Mutex();
+  /** @internal */
+  private reader: ReadableStreamDefaultReader<VideoFrameEvent>;
+  /** @internal */
+  private onEvent: ((ev: FfiEvent) => void) | null = null;
 
   track: Track;
 
@@ -50,49 +52,83 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
     this.info = res.stream?.info;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
-    FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+    const source = new ReadableStream<FfiEvent>({
+      start: (controller) => {
+        this.onEvent = (ev: FfiEvent) => {
+          if (
+            ev.message.case === 'videoStreamEvent' &&
+            ev.message.value.streamHandle === this.ffiHandle.handle
+          ) {
+            if (controller.desiredSize && controller.desiredSize > 0) {
+              controller.enqueue(ev);
+            } else{
+              console.warn('Dropping video frame due to low buffer size');
+            }
+          }
+        };
+        FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+      },
+      cancel: () => {
+        if (this.onEvent) {
+          FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+          this.onEvent = null;
+        }
+      }
+    });
 
-    this.eventQueue = new RingQueue<IteratorResult<VideoFrameEvent>>(capacity);
-  }
+    const transformStream = new TransformStream<FfiEvent, VideoFrameEvent>({
+      transform: (event: FfiEvent, controller: TransformStreamDefaultController<VideoFrameEvent>) => {
+        if (
+          event.message.case !== 'videoStreamEvent' ||
+          event.message.value.streamHandle !== this.ffiHandle.handle
+        ) {
+          return;
+        }
 
-  private onEvent = (ev: FfiEvent) => {
-    if (
-      ev.message.case != 'videoStreamEvent' ||
-      ev.message.value.streamHandle != this.ffiHandle.handle
-    ) {
-      return;
-    }
+        const streamEvent = event.message.value.message;
+        switch (streamEvent.case) {
+          case 'frameReceived':
+            const rotation = streamEvent.value.rotation;
+            const timestampUs = streamEvent.value.timestampUs;
+            const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
+            controller.enqueue({
+              rotation: rotation!,
+              timestampUs: timestampUs!,
+              frame,
+            });
+            break;
+          case 'eos':
+            controller.terminate();
+            if (this.onEvent) {
+              FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+              this.onEvent = null;
+            }
+            break;
+        }
+      }
+    }, {
+      highWaterMark: capacity > 0 ? capacity : undefined
+    });
 
-    const streamEvent = ev.message.value.message;
-    switch (streamEvent.case) {
-      case 'frameReceived':
-        const rotation = streamEvent.value.rotation;
-        const timestampUs = streamEvent.value.timestampUs;
-        const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
-        this.eventQueue.push({
-          done: false,
-          value: {
-            rotation: rotation!,
-            timestampUs: timestampUs!,
-            frame,
-          },
-        });
-        break;
-      case 'eos':
-        FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-        break;
-    }
-  };
+    // Connect the streams and get the reader directly
+    this.reader = source.pipeThrough(transformStream).getReader();
+  }
 
   async next(): Promise<IteratorResult<VideoFrameEvent>> {
     const unlock = await this.mutex.lock();
-    const result = this.eventQueue.get();
-    unlock();
-    return result;
+    try {
+      const result = await this.reader.read();
+      return {
+        done: result.done,
+        value: result.done ? undefined as any : result.value
+      };
+    } finally {
+      unlock();
+    }
   }
 
   close() {
-    this.eventQueue.push({ done: true, value: undefined });
+    this.reader.cancel();
     this.ffiHandle.dispose();
   }
 

From c98b919a0dc5f05d69e39729409520d8fa0b382b Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 20:41:18 -0700
Subject: [PATCH 10/26] undo

---
 packages/livekit-rtc/src/utils.test.ts | 30 +---------------
 packages/livekit-rtc/src/utils.ts      | 47 --------------------------
 2 files changed, 1 insertion(+), 76 deletions(-)

diff --git a/packages/livekit-rtc/src/utils.test.ts b/packages/livekit-rtc/src/utils.test.ts
index af32f49c..35ac904b 100644
--- a/packages/livekit-rtc/src/utils.test.ts
+++ b/packages/livekit-rtc/src/utils.test.ts
@@ -2,7 +2,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import { describe, expect, it } from 'vitest';
-import { RingQueue, splitUtf8 } from './utils.js';
+import { splitUtf8 } from './utils.js';
 
 describe('splitUtf8', () => {
   it('splits a string into chunks of the given size', () => {
@@ -49,31 +49,3 @@ describe('splitUtf8', () => {
     expect(splitUtf8('', 5)).toEqual([]);
   });
 });
-
-describe('RingQueue', () => {
-  it('should push and get items', async () => {
-    const queue = new RingQueue<number>(3);
-    queue.push(1);
-    queue.push(2);
-    expect(await queue.get()).toBe(1);
-    expect(await queue.get()).toBe(2);
-  });
-
-  it('should push and get items with a capacity', async () => {
-    const queue = new RingQueue<number>(3);
-    queue.push(1);
-    queue.push(2);
-    queue.push(3);
-    queue.push(4);
-    expect(await queue.get()).toBe(1);
-    expect(await queue.get()).toBe(2);
-    expect(await queue.get()).toBe(3);
-  });
-
-  it('should push and get items with a capacity of 0', async () => {
-    const queue = new RingQueue<number>(0);
-    queue.push(1);
-    queue.push(2);
-    queue.push(3);
-  });
-});
diff --git a/packages/livekit-rtc/src/utils.ts b/packages/livekit-rtc/src/utils.ts
index e4f94431..7bc5a650 100644
--- a/packages/livekit-rtc/src/utils.ts
+++ b/packages/livekit-rtc/src/utils.ts
@@ -40,50 +40,3 @@ export function splitUtf8(s: string, n: number): Uint8Array[] {
   }
   return result;
 }
-
-/**
- * A ring queue that stores a fixed number of items. If capacity is
- * set to 0, the buffer will grow indefinitely.
- *
- * WARNING: THIS IS NOT THREAD SAFE.
- */
-export class RingQueue<T> {
-  private buffer: T[] = [];
-  // If the buffer is empty, this will be a promise that resolves when
-  // a new item is pushed, bypassing the internal queue.
-  private resolve: ((value: T) => void) | null = null;
-
-  constructor(private readonly capacity: number = 0) {
-    this.resolve = null;
-    this.capacity = capacity;
-  }
-
-  /**
-   * Push an item into the buffer. If the buffer is full, the newest item will
-   * be removed.
-   */
-  push(item: T) {
-    if (this.resolve) {
-      this.resolve(item);
-      this.resolve = null;
-    } else {
-      if (this.capacity > 0 && this.buffer.length > this.capacity) {
-        this.buffer.pop();
-      }
-      this.buffer.push(item);
-    }
-  }
-
-  /**
-   * Get an item from the buffer. If the buffer is empty, this will return a promise
-   * that resolves when a new item is pushed.
-   */
-  async get(): Promise<T> {
-    if (this.buffer.length === 0) {
-      return new Promise((resolve) => {
-        this.resolve = resolve;
-      });
-    }
-    return this.buffer.shift()!;
-  }
-}

From 8f2009305b520cb349b7e13ba5537c3b1a24e4c3 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 22:29:52 -0700
Subject: [PATCH 11/26] update warning message

---
 packages/livekit-rtc/src/audio_stream.ts | 2 +-
 packages/livekit-rtc/src/video_stream.ts | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 07250d81..59e9a668 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -58,7 +58,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
             if (controller.desiredSize && controller.desiredSize > 0) {
               controller.enqueue(ev);
             } else{
-              console.warn('Dropping audio frame due to low buffer size');
+              console.warn('Audio stream buffer is full, dropping frame');
             }
           }
         };
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 1cf55f6d..8596aefc 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -62,7 +62,7 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
             if (controller.desiredSize && controller.desiredSize > 0) {
               controller.enqueue(ev);
             } else{
-              console.warn('Dropping video frame due to low buffer size');
+              console.warn('Video stream buffer is full, dropping frame');
             }
           }
         };

From 281e7ba2afabad9f495e44f7fa5b19124a69b161 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 22:33:41 -0700
Subject: [PATCH 12/26] format

---
 packages/livekit-rtc/src/audio_stream.ts | 26 +++++---
 packages/livekit-rtc/src/video_stream.ts | 78 +++++++++++++-----------
 2 files changed, 58 insertions(+), 46 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 59e9a668..823bc63b 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -2,13 +2,13 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import { Mutex } from '@livekit/mutex';
+import { ReadableStream, TransformStream } from 'node:stream/web';
 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 { ReadableStream, TransformStream } from 'node:stream/web';
 
 export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
@@ -26,7 +26,12 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   sampleRate: number;
   numChannels: number;
 
-  constructor(track: Track, sampleRate: number = 48000, numChannels: number = 1, capacity: number = 0) {
+  constructor(
+    track: Track,
+    sampleRate: number = 48000,
+    numChannels: number = 1,
+    capacity: number = 0,
+  ) {
     this.track = track;
     this.sampleRate = sampleRate;
     this.numChannels = numChannels;
@@ -57,7 +62,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
           ) {
             if (controller.desiredSize && controller.desiredSize > 0) {
               controller.enqueue(ev);
-            } else{
+            } else {
               console.warn('Audio stream buffer is full, dropping frame');
             }
           }
@@ -69,7 +74,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
           FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
           this.onEvent = null;
         }
-      }
+      },
     });
 
     const transformStream = new TransformStream<FfiEvent, AudioFrame>(
@@ -96,12 +101,13 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
               }
               break;
           }
-        }
-      }, {
-        highWaterMark: capacity > 0 ? capacity : undefined
-      }
+        },
+      },
+      {
+        highWaterMark: capacity > 0 ? capacity : undefined,
+      },
     );
-    
+
     this.reader = source.pipeThrough(transformStream).getReader();
   }
 
@@ -111,7 +117,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
       const result = await this.reader.read();
       return {
         done: result.done,
-        value: result.done ? undefined as any : result.value
+        value: result.done ? (undefined as any) : result.value,
       };
     } finally {
       unlock();
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 8596aefc..3af8f28a 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -2,6 +2,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import { Mutex } from '@livekit/mutex';
+import { ReadableStream, TransformStream, TransformStreamDefaultController } from 'node:stream/web';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
 import type {
@@ -11,7 +12,6 @@ import type {
 } from './proto/video_frame_pb.js';
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
-import { ReadableStream, TransformStream, TransformStreamDefaultController } from 'node:stream/web';
 import { VideoFrame } from './video_frame.js';
 
 export type VideoFrameEvent = {
@@ -61,7 +61,7 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
           ) {
             if (controller.desiredSize && controller.desiredSize > 0) {
               controller.enqueue(ev);
-            } else{
+            } else {
               console.warn('Video stream buffer is full, dropping frame');
             }
           }
@@ -73,42 +73,48 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
           FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
           this.onEvent = null;
         }
-      }
+      },
     });
 
-    const transformStream = new TransformStream<FfiEvent, VideoFrameEvent>({
-      transform: (event: FfiEvent, controller: TransformStreamDefaultController<VideoFrameEvent>) => {
-        if (
-          event.message.case !== 'videoStreamEvent' ||
-          event.message.value.streamHandle !== this.ffiHandle.handle
-        ) {
-          return;
-        }
+    const transformStream = new TransformStream<FfiEvent, VideoFrameEvent>(
+      {
+        transform: (
+          event: FfiEvent,
+          controller: TransformStreamDefaultController<VideoFrameEvent>,
+        ) => {
+          if (
+            event.message.case !== 'videoStreamEvent' ||
+            event.message.value.streamHandle !== this.ffiHandle.handle
+          ) {
+            return;
+          }
 
-        const streamEvent = event.message.value.message;
-        switch (streamEvent.case) {
-          case 'frameReceived':
-            const rotation = streamEvent.value.rotation;
-            const timestampUs = streamEvent.value.timestampUs;
-            const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
-            controller.enqueue({
-              rotation: rotation!,
-              timestampUs: timestampUs!,
-              frame,
-            });
-            break;
-          case 'eos':
-            controller.terminate();
-            if (this.onEvent) {
-              FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-              this.onEvent = null;
-            }
-            break;
-        }
-      }
-    }, {
-      highWaterMark: capacity > 0 ? capacity : undefined
-    });
+          const streamEvent = event.message.value.message;
+          switch (streamEvent.case) {
+            case 'frameReceived':
+              const rotation = streamEvent.value.rotation;
+              const timestampUs = streamEvent.value.timestampUs;
+              const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
+              controller.enqueue({
+                rotation: rotation!,
+                timestampUs: timestampUs!,
+                frame,
+              });
+              break;
+            case 'eos':
+              controller.terminate();
+              if (this.onEvent) {
+                FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+                this.onEvent = null;
+              }
+              break;
+          }
+        },
+      },
+      {
+        highWaterMark: capacity > 0 ? capacity : undefined,
+      },
+    );
 
     // Connect the streams and get the reader directly
     this.reader = source.pipeThrough(transformStream).getReader();
@@ -120,7 +126,7 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
       const result = await this.reader.read();
       return {
         done: result.done,
-        value: result.done ? undefined as any : result.value
+        value: result.done ? (undefined as any) : result.value,
       };
     } finally {
       unlock();

From 527b1d76badc38f9929ea871985964a45cae3cd8 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 22:35:43 -0700
Subject: [PATCH 13/26] less loc

---
 packages/livekit-rtc/src/audio_stream.ts | 4 ++--
 packages/livekit-rtc/src/video_stream.ts | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 823bc63b..0f367c01 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -16,11 +16,11 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  mutex = new Mutex();
-  /** @internal */
   private reader: ReadableStreamDefaultReader<AudioFrame>;
   /** @internal */
   private onEvent: ((ev: FfiEvent) => void) | null = null;
+  /** @internal */
+  mutex = new Mutex();
 
   track: Track;
   sampleRate: number;
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 3af8f28a..cd626485 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -26,11 +26,11 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  mutex = new Mutex();
-  /** @internal */
   private reader: ReadableStreamDefaultReader<VideoFrameEvent>;
   /** @internal */
   private onEvent: ((ev: FfiEvent) => void) | null = null;
+  /** @internal */
+  mutex = new Mutex();
 
   track: Track;
 

From 6c445a30443e39e8961b57d9838d948ba2bc68c6 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 22:40:08 -0700
Subject: [PATCH 14/26] remove coment

---
 packages/livekit-rtc/src/video_stream.ts | 1 -
 1 file changed, 1 deletion(-)

diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index cd626485..4abca325 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -116,7 +116,6 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
       },
     );
 
-    // Connect the streams and get the reader directly
     this.reader = source.pipeThrough(transformStream).getReader();
   }
 

From 6424f532a1d4ef9b29cc1a09550923141aa911b3 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 9 Apr 2025 23:20:47 -0700
Subject: [PATCH 15/26] bugfix

---
 packages/livekit-rtc/src/audio_stream.ts | 4 +++-
 packages/livekit-rtc/src/video_stream.ts | 4 +++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 0f367c01..4d0e0eca 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -53,6 +53,8 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
     this.info = res.stream!.info!;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
+    const infinite_capacity = capacity <= 0;
+
     const source = new ReadableStream<FfiEvent>({
       start: (controller) => {
         this.onEvent = (ev: FfiEvent) => {
@@ -60,7 +62,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
             ev.message.case === 'audioStreamEvent' &&
             ev.message.value.streamHandle === this.ffiHandle.handle
           ) {
-            if (controller.desiredSize && controller.desiredSize > 0) {
+            if (infinite_capacity || (controller.desiredSize && controller.desiredSize > 0)) {
               controller.enqueue(ev);
             } else {
               console.warn('Audio stream buffer is full, dropping frame');
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 4abca325..55861634 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -52,6 +52,8 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
     this.info = res.stream?.info;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
+    const infinite_capacity = capacity <= 0;
+
     const source = new ReadableStream<FfiEvent>({
       start: (controller) => {
         this.onEvent = (ev: FfiEvent) => {
@@ -59,7 +61,7 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
             ev.message.case === 'videoStreamEvent' &&
             ev.message.value.streamHandle === this.ffiHandle.handle
           ) {
-            if (controller.desiredSize && controller.desiredSize > 0) {
+            if (infinite_capacity || (controller.desiredSize && controller.desiredSize > 0)) {
               controller.enqueue(ev);
             } else {
               console.warn('Video stream buffer is full, dropping frame');

From d4234ed43db9ce59c0ff812843d9e906ed1f84db Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Tue, 15 Apr 2025 13:10:22 -0700
Subject: [PATCH 16/26] update video frame

---
 packages/livekit-rtc/src/video_stream.ts | 139 ++++++++---------------
 1 file changed, 49 insertions(+), 90 deletions(-)

diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 55861634..360868f8 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -1,8 +1,6 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import { Mutex } from '@livekit/mutex';
-import { ReadableStream, TransformStream, TransformStreamDefaultController } from 'node:stream/web';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
 import type {
@@ -13,6 +11,7 @@ import type {
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
 import { VideoFrame } from './video_frame.js';
+import { ReadableStream } from 'node:stream/web';
 
 export type VideoFrameEvent = {
   frame: VideoFrame;
@@ -20,21 +19,22 @@ export type VideoFrameEvent = {
   rotation: VideoRotation;
 };
 
-export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
+export class VideoStream extends ReadableStream<VideoFrameEvent> {
   /** @internal */
   info?: VideoStreamInfo;
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  private reader: ReadableStreamDefaultReader<VideoFrameEvent>;
-  /** @internal */
-  private onEvent: ((ev: FfiEvent) => void) | null = null;
-  /** @internal */
-  mutex = new Mutex();
+  controller?: ReadableStreamDefaultController<VideoFrameEvent>;
 
   track: Track;
 
-  constructor(track: Track, capacity: number = 0) {
+  constructor(track: Track) {
+    super({
+      start: (controller) => {
+        this.controller = controller;
+      },
+    });
     this.track = track;
 
     const req = new NewVideoStreamRequest({
@@ -52,94 +52,53 @@ export class VideoStream implements AsyncIterableIterator<VideoFrameEvent> {
     this.info = res.stream?.info;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
-    const infinite_capacity = capacity <= 0;
-
-    const source = new ReadableStream<FfiEvent>({
-      start: (controller) => {
-        this.onEvent = (ev: FfiEvent) => {
-          if (
-            ev.message.case === 'videoStreamEvent' &&
-            ev.message.value.streamHandle === this.ffiHandle.handle
-          ) {
-            if (infinite_capacity || (controller.desiredSize && controller.desiredSize > 0)) {
-              controller.enqueue(ev);
-            } else {
-              console.warn('Video stream buffer is full, dropping frame');
-            }
-          }
-        };
-        FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
-      },
-      cancel: () => {
-        if (this.onEvent) {
-          FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-          this.onEvent = null;
-        }
-      },
-    });
+    FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+  }
 
-    const transformStream = new TransformStream<FfiEvent, VideoFrameEvent>(
-      {
-        transform: (
-          event: FfiEvent,
-          controller: TransformStreamDefaultController<VideoFrameEvent>,
-        ) => {
-          if (
-            event.message.case !== 'videoStreamEvent' ||
-            event.message.value.streamHandle !== this.ffiHandle.handle
-          ) {
-            return;
-          }
+  private onEvent = (ev: FfiEvent) => {
+    if (!this.controller) {
+      throw new Error('Stream controller not initialized');
+    }
 
-          const streamEvent = event.message.value.message;
-          switch (streamEvent.case) {
-            case 'frameReceived':
-              const rotation = streamEvent.value.rotation;
-              const timestampUs = streamEvent.value.timestampUs;
-              const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
-              controller.enqueue({
-                rotation: rotation!,
-                timestampUs: timestampUs!,
-                frame,
-              });
-              break;
-            case 'eos':
-              controller.terminate();
-              if (this.onEvent) {
-                FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-                this.onEvent = null;
-              }
-              break;
-          }
-        },
-      },
-      {
-        highWaterMark: capacity > 0 ? capacity : undefined,
-      },
-    );
+    if (
+      ev.message.case != 'videoStreamEvent' ||
+      ev.message.value.streamHandle != this.ffiHandle.handle
+    ) {
+      return;
+    }
 
-    this.reader = source.pipeThrough(transformStream).getReader();
-  }
+    const streamEvent = ev.message.value.message;
+    switch (streamEvent.case) {
+      case 'frameReceived':
+        const rotation = streamEvent.value.rotation;
+        const timestampUs = streamEvent.value.timestampUs;
+        const frame = VideoFrame.fromOwnedInfo(streamEvent.value.buffer!);
+        const value = { rotation, timestampUs, frame };
+        const videoFrameEvent = {
+          frame: value.frame,
+          timestampUs: value.timestampUs!,
+          rotation: value.rotation!,
+        };
 
-  async next(): Promise<IteratorResult<VideoFrameEvent>> {
-    const unlock = await this.mutex.lock();
-    try {
-      const result = await this.reader.read();
-      return {
-        done: result.done,
-        value: result.done ? (undefined as any) : result.value,
-      };
-    } finally {
-      unlock();
+        this.controller.enqueue(videoFrameEvent);
+        break;
+      case 'eos':
+        FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+        this.controller.close();
+        break;
     }
-  }
+  };
 
   close() {
-    this.reader.cancel();
+    if (!this.controller) {
+      throw new Error('Stream controller not initialized');
+    }
+      try {
+        this.controller.close();
+      } catch (e) {
+        // Controller might already be closed
+        throw e;
+      }
     this.ffiHandle.dispose();
   }
-
-  [Symbol.asyncIterator](): VideoStream {
-    return this;
-  }
 }

From 9e1635392c78628549ce198d27d042dd25579e05 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Tue, 15 Apr 2025 20:22:27 -0700
Subject: [PATCH 17/26] VideoStreamSource

---
 packages/livekit-rtc/src/video_stream.ts | 59 ++++++++++++------------
 1 file changed, 30 insertions(+), 29 deletions(-)

diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 360868f8..09b0eaf8 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -11,7 +11,7 @@ import type {
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
 import { VideoFrame } from './video_frame.js';
-import { ReadableStream } from 'node:stream/web';
+import { ReadableStream, type UnderlyingSource } from 'node:stream/web';
 
 export type VideoFrameEvent = {
   frame: VideoFrame;
@@ -19,24 +19,17 @@ export type VideoFrameEvent = {
   rotation: VideoRotation;
 };
 
-export class VideoStream extends ReadableStream<VideoFrameEvent> {
-  /** @internal */
-  info?: VideoStreamInfo;
-  /** @internal */
-  ffiHandle: FfiHandle;
-  /** @internal */
-  controller?: ReadableStreamDefaultController<VideoFrameEvent>;
 
-  track: Track;
+class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
+  private controller?: ReadableStreamDefaultController<VideoFrameEvent>;
+  private track: Track;
+  private ffiHandle: FfiHandle;
+  private info?: VideoStreamInfo;
+  private closed = false;
+
 
   constructor(track: Track) {
-    super({
-      start: (controller) => {
-        this.controller = controller;
-      },
-    });
     this.track = track;
-
     const req = new NewVideoStreamRequest({
       type: VideoStreamType.VIDEO_STREAM_NATIVE,
       trackHandle: track.ffi_handle.handle,
@@ -49,9 +42,8 @@ export class VideoStream extends ReadableStream<VideoFrameEvent> {
       },
     });
 
-    this.info = res.stream?.info;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
-
+    this.info = res.stream!.info;
     FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
   }
 
@@ -80,25 +72,34 @@ export class VideoStream extends ReadableStream<VideoFrameEvent> {
           rotation: value.rotation!,
         };
 
-        this.controller.enqueue(videoFrameEvent);
+        if (!this.closed) {
+          this.controller.enqueue(videoFrameEvent);
+        }
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-        this.controller.close();
+        if (!this.closed) {
+          this.closed = true;
+          this.controller.close();
+        }
         break;
     }
   };
 
-  close() {
-    if (!this.controller) {
-      throw new Error('Stream controller not initialized');
-    }
-      try {
-        this.controller.close();
-      } catch (e) {
-        // Controller might already be closed
-        throw e;
-      }
+  start(controller: ReadableStreamDefaultController<VideoFrameEvent>) {
+    this.controller = controller;
+  }
+
+  cancel(reason?: any) {
+    this.closed = true;
+    FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
+    console.log('cancelled');
+  }
+}
+
+export class VideoStream extends ReadableStream<VideoFrameEvent> {
+  constructor(track: Track) {
+    super(new VideoStreamSource(track));
   }
 }

From 3c4721400805ee4b91c98d4da2d8c684d7313b88 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Tue, 15 Apr 2025 22:45:02 -0700
Subject: [PATCH 18/26] cleanup

---
 packages/livekit-rtc/src/video_stream.ts | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 09b0eaf8..c767237c 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -22,14 +22,11 @@ export type VideoFrameEvent = {
 
 class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
   private controller?: ReadableStreamDefaultController<VideoFrameEvent>;
-  private track: Track;
   private ffiHandle: FfiHandle;
-  private info?: VideoStreamInfo;
   private closed = false;
 
 
   constructor(track: Track) {
-    this.track = track;
     const req = new NewVideoStreamRequest({
       type: VideoStreamType.VIDEO_STREAM_NATIVE,
       trackHandle: track.ffi_handle.handle,
@@ -43,7 +40,6 @@ class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
     });
 
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
-    this.info = res.stream!.info;
     FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
   }
 

From f077366b7142d261ffae07f493f6cfe52c870892 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Tue, 15 Apr 2025 22:46:12 -0700
Subject: [PATCH 19/26] clean up

---
 packages/livekit-rtc/src/video_stream.ts | 1 -
 1 file changed, 1 deletion(-)

diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index c767237c..800ad67d 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -90,7 +90,6 @@ class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
     this.closed = true;
     FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
-    console.log('cancelled');
   }
 }
 

From cdea12c8154f28dafc3ca8719a58707c139abbd3 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Tue, 15 Apr 2025 23:27:49 -0700
Subject: [PATCH 20/26] undo

---
 packages/livekit-rtc/src/audio_stream.ts | 115 ++++++++---------------
 1 file changed, 41 insertions(+), 74 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 4d0e0eca..9a7c96cc 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -2,7 +2,6 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import { Mutex } from '@livekit/mutex';
-import { ReadableStream, TransformStream } from 'node:stream/web';
 import { AudioFrame } from './audio_frame.js';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
@@ -16,9 +15,9 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   /** @internal */
   ffiHandle: FfiHandle;
   /** @internal */
-  private reader: ReadableStreamDefaultReader<AudioFrame>;
+  eventQueue: (AudioFrame | null)[] = [];
   /** @internal */
-  private onEvent: ((ev: FfiEvent) => void) | null = null;
+  queueResolve: ((value: IteratorResult<AudioFrame>) => void) | null = null;
   /** @internal */
   mutex = new Mutex();
 
@@ -26,12 +25,7 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
   sampleRate: number;
   numChannels: number;
 
-  constructor(
-    track: Track,
-    sampleRate: number = 48000,
-    numChannels: number = 1,
-    capacity: number = 0,
-  ) {
+  constructor(track: Track, sampleRate: number = 48000, numChannels: number = 1) {
     this.track = track;
     this.sampleRate = sampleRate;
     this.numChannels = numChannels;
@@ -53,81 +47,54 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
     this.info = res.stream!.info!;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
-    const infinite_capacity = capacity <= 0;
-
-    const source = new ReadableStream<FfiEvent>({
-      start: (controller) => {
-        this.onEvent = (ev: FfiEvent) => {
-          if (
-            ev.message.case === 'audioStreamEvent' &&
-            ev.message.value.streamHandle === this.ffiHandle.handle
-          ) {
-            if (infinite_capacity || (controller.desiredSize && controller.desiredSize > 0)) {
-              controller.enqueue(ev);
-            } else {
-              console.warn('Audio stream buffer is full, dropping frame');
-            }
-          }
-        };
-        FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
-      },
-      cancel: () => {
-        if (this.onEvent) {
-          FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-          this.onEvent = null;
-        }
-      },
-    });
-
-    const transformStream = new TransformStream<FfiEvent, AudioFrame>(
-      {
-        transform: (event: FfiEvent, controller: TransformStreamDefaultController<AudioFrame>) => {
-          if (
-            event.message.case !== 'audioStreamEvent' ||
-            event.message.value.streamHandle !== this.ffiHandle.handle
-          ) {
-            return;
-          }
+    FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
+  }
 
-          const streamEvent = event.message.value.message;
-          switch (streamEvent.case) {
-            case 'frameReceived':
-              const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
-              controller.enqueue(frame);
-              break;
-            case 'eos':
-              controller.terminate();
-              if (this.onEvent) {
-                FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-                this.onEvent = null;
-              }
-              break;
-          }
-        },
-      },
-      {
-        highWaterMark: capacity > 0 ? capacity : undefined,
-      },
-    );
+  private onEvent = (ev: FfiEvent) => {
+    if (
+      ev.message.case != 'audioStreamEvent' ||
+      ev.message.value.streamHandle != this.ffiHandle.handle
+    ) {
+      return;
+    }
 
-    this.reader = source.pipeThrough(transformStream).getReader();
-  }
+    const streamEvent = ev.message.value.message;
+    switch (streamEvent.case) {
+      case 'frameReceived':
+        const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
+        if (this.queueResolve) {
+          this.queueResolve({ done: false, value: frame });
+          this.queueResolve = null;
+        } 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();
-    try {
-      const result = await this.reader.read();
-      return {
-        done: result.done,
-        value: result.done ? (undefined as any) : result.value,
-      };
-    } finally {
+    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.reader.cancel();
+    this.eventQueue.push(null);
     this.ffiHandle.dispose();
   }
 

From 93b414332497c6c5de7482a9d8a1c94b2d7b8912 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Wed, 16 Apr 2025 22:46:11 -0700
Subject: [PATCH 21/26] cleanup

---
 packages/livekit-rtc/src/video_stream.ts | 14 +++-----------
 1 file changed, 3 insertions(+), 11 deletions(-)

diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 800ad67d..4e266449 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -22,8 +22,7 @@ export type VideoFrameEvent = {
 
 class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
   private controller?: ReadableStreamDefaultController<VideoFrameEvent>;
-  private ffiHandle: FfiHandle;
-  private closed = false;
+  private ffiHandle: FfiHandle; 
 
 
   constructor(track: Track) {
@@ -67,17 +66,11 @@ class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
           timestampUs: value.timestampUs!,
           rotation: value.rotation!,
         };
-
-        if (!this.closed) {
-          this.controller.enqueue(videoFrameEvent);
-        }
+        this.controller.enqueue(videoFrameEvent);
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
-        if (!this.closed) {
-          this.closed = true;
-          this.controller.close();
-        }
+        this.controller.close();
         break;
     }
   };
@@ -87,7 +80,6 @@ class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
   }
 
   cancel(reason?: any) {
-    this.closed = true;
     FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
   }

From 8abd041e1427e98aa888d75cc06a940f56c165aa Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 17 Apr 2025 14:59:13 -0700
Subject: [PATCH 22/26] update final design

---
 packages/livekit-rtc/src/audio_stream.ts | 77 ++++++++----------------
 packages/livekit-rtc/src/video_stream.ts |  3 +-
 2 files changed, 27 insertions(+), 53 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index fe4facca..b2649db5 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -1,13 +1,13 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
-import { Mutex } from '@livekit/mutex';
 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 type { NewAudioStreamResponse } from './proto/audio_frame_pb.js';
 import { AudioStreamType, NewAudioStreamRequest } from './proto/audio_frame_pb.js';
 import type { Track } from './track.js';
+import { ReadableStream, type UnderlyingSource } from 'node:stream/web';
 
 export interface AudioStreamOptions {
   noiseCancellation?: NoiseCancellationOptions;
@@ -20,34 +20,18 @@ export interface NoiseCancellationOptions {
   options: Record<string, any>;
 }
 
-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;
-  ncOptions?: NoiseCancellationOptions;
-
-  constructor(track: Track);
-  constructor(track: Track, sampleRate: number);
-  constructor(track: Track, sampleRate: number, numChannels: number);
-  constructor(track: Track, options: AudioStreamOptions);
+class AudioStreamSource implements UnderlyingSource<AudioFrame> {
+  private controller?: ReadableStreamDefaultController<AudioFrame>;
+  private ffiHandle: FfiHandle;
+  private sampleRate: number;
+  private numChannels: number;
+  private ncOptions?: NoiseCancellationOptions;
 
   constructor(
     track: Track,
     sampleRateOrOptions?: number | AudioStreamOptions,
     numChannels?: number,
   ) {
-    this.track = track;
     if (sampleRateOrOptions !== undefined && typeof sampleRateOrOptions !== 'number') {
       this.sampleRate = sampleRateOrOptions.sampleRate ?? 48000;
       this.numChannels = sampleRateOrOptions.numChannels ?? 1;
@@ -77,13 +61,16 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
       },
     });
 
-    this.info = res.stream!.info!;
     this.ffiHandle = new FfiHandle(res.stream!.handle!.id!);
 
     FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onEvent);
   }
 
   private onEvent = (ev: FfiEvent) => {
+    if (!this.controller) {
+      throw new Error('Stream controller not initialized');
+    }
+
     if (
       ev.message.case != 'audioStreamEvent' ||
       ev.message.value.streamHandle != this.ffiHandle.handle
@@ -95,43 +82,31 @@ export class AudioStream implements AsyncIterableIterator<AudioFrame> {
     switch (streamEvent.case) {
       case 'frameReceived':
         const frame = AudioFrame.fromOwnedInfo(streamEvent.value.frame!);
-        if (this.queueResolve) {
-          this.queueResolve({ done: false, value: frame });
-          this.queueResolve = null;
-        } else {
-          this.eventQueue.push(frame);
-        }
+        this.controller.enqueue(frame);
         break;
       case 'eos':
         FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
+        this.controller.close();
         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;
+  start(controller: ReadableStreamDefaultController<AudioFrame>) {
+    this.controller = controller;
   }
 
-  close() {
-    this.eventQueue.push(null);
+  cancel(_reason?: any) {
+    FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
   }
+}
 
-  [Symbol.asyncIterator](): AudioStream {
-    return this;
+export class AudioStream extends ReadableStream<AudioFrame> {
+  constructor(
+    track: Track,
+    sampleRateOrOptions?: number | AudioStreamOptions,
+    numChannels?: number,
+  ) {
+    super(new AudioStreamSource(track, sampleRateOrOptions, numChannels));
   }
 }
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 4e266449..ff063983 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -6,7 +6,6 @@ import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
 import type {
   NewVideoStreamResponse,
   VideoRotation,
-  VideoStreamInfo,
 } from './proto/video_frame_pb.js';
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
@@ -79,7 +78,7 @@ class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
     this.controller = controller;
   }
 
-  cancel(reason?: any) {
+  cancel(_reason?: any) {
     FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
   }

From 59bca2ad1219727fa1197e70fd2250301af0dbc1 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 17 Apr 2025 15:04:45 -0700
Subject: [PATCH 23/26] fix lint

---
 packages/livekit-rtc/src/audio_stream.ts | 2 +-
 packages/livekit-rtc/src/video_stream.ts | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index b2649db5..6aa2315c 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -95,7 +95,7 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
     this.controller = controller;
   }
 
-  cancel(_reason?: any) {
+  cancel() {
     FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
   }
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index ff063983..1c33c623 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -78,7 +78,7 @@ class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
     this.controller = controller;
   }
 
-  cancel(_reason?: any) {
+  cancel() {
     FfiClient.instance.off(FfiClientEvent.FfiEvent, this.onEvent);
     this.ffiHandle.dispose();
   }

From e687e8f00c4cfc27df65b748dac2f8ad4d6c77b5 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Thu, 17 Apr 2025 15:10:37 -0700
Subject: [PATCH 24/26] format

---
 packages/livekit-rtc/src/audio_stream.ts |  2 +-
 packages/livekit-rtc/src/video_stream.ts | 11 +++--------
 2 files changed, 4 insertions(+), 9 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index 6aa2315c..f0e198bb 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -1,13 +1,13 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import { ReadableStream, type UnderlyingSource } from 'node:stream/web';
 import { AudioFrame } from './audio_frame.js';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
 import type { NewAudioStreamResponse } from './proto/audio_frame_pb.js';
 import { AudioStreamType, NewAudioStreamRequest } from './proto/audio_frame_pb.js';
 import type { Track } from './track.js';
-import { ReadableStream, type UnderlyingSource } from 'node:stream/web';
 
 export interface AudioStreamOptions {
   noiseCancellation?: NoiseCancellationOptions;
diff --git a/packages/livekit-rtc/src/video_stream.ts b/packages/livekit-rtc/src/video_stream.ts
index 1c33c623..92aff82f 100644
--- a/packages/livekit-rtc/src/video_stream.ts
+++ b/packages/livekit-rtc/src/video_stream.ts
@@ -1,16 +1,13 @@
 // SPDX-FileCopyrightText: 2024 LiveKit, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
+import { ReadableStream, type UnderlyingSource } from 'node:stream/web';
 import type { FfiEvent } from './ffi_client.js';
 import { FfiClient, FfiClientEvent, FfiHandle } from './ffi_client.js';
-import type {
-  NewVideoStreamResponse,
-  VideoRotation,
-} from './proto/video_frame_pb.js';
+import type { NewVideoStreamResponse, VideoRotation } from './proto/video_frame_pb.js';
 import { NewVideoStreamRequest, VideoStreamType } from './proto/video_frame_pb.js';
 import type { Track } from './track.js';
 import { VideoFrame } from './video_frame.js';
-import { ReadableStream, type UnderlyingSource } from 'node:stream/web';
 
 export type VideoFrameEvent = {
   frame: VideoFrame;
@@ -18,11 +15,9 @@ export type VideoFrameEvent = {
   rotation: VideoRotation;
 };
 
-
 class VideoStreamSource implements UnderlyingSource<VideoFrameEvent> {
   private controller?: ReadableStreamDefaultController<VideoFrameEvent>;
-  private ffiHandle: FfiHandle; 
-
+  private ffiHandle: FfiHandle;
 
   constructor(track: Track) {
     const req = new NewVideoStreamRequest({

From 6f1d8461f8abd98a5e2dc3761a8d406155933ebd Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Fri, 18 Apr 2025 11:24:18 -0700
Subject: [PATCH 25/26] add constructors

---
 packages/livekit-rtc/src/audio_stream.ts | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/packages/livekit-rtc/src/audio_stream.ts b/packages/livekit-rtc/src/audio_stream.ts
index f0e198bb..94395505 100644
--- a/packages/livekit-rtc/src/audio_stream.ts
+++ b/packages/livekit-rtc/src/audio_stream.ts
@@ -102,6 +102,11 @@ class AudioStreamSource implements UnderlyingSource<AudioFrame> {
 }
 
 export class AudioStream extends ReadableStream<AudioFrame> {
+  
+  constructor(track: Track);
+  constructor(track: Track, sampleRate: number);
+  constructor(track: Track, sampleRate: number, numChannels: number);
+  constructor(track: Track, options: AudioStreamOptions);
   constructor(
     track: Track,
     sampleRateOrOptions?: number | AudioStreamOptions,

From 00d4e588e3569f9b132f7394cbb0d501d9886175 Mon Sep 17 00:00:00 2001
From: Shubhrakanti Ganguly <shubhrakanti.ganguly@livekit.io>
Date: Fri, 18 Apr 2025 11:35:21 -0700
Subject: [PATCH 26/26] linter

---
 packages/livekit-rtc/src/audio_filter.ts | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/packages/livekit-rtc/src/audio_filter.ts b/packages/livekit-rtc/src/audio_filter.ts
index 794105dc..6535b36a 100644
--- a/packages/livekit-rtc/src/audio_filter.ts
+++ b/packages/livekit-rtc/src/audio_filter.ts
@@ -2,9 +2,10 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 import { FfiClient } from './ffi_client.js';
+import type {
+  LoadAudioFilterPluginResponse} from './proto/audio_frame_pb.js';
 import {
-  LoadAudioFilterPluginRequest,
-  LoadAudioFilterPluginResponse,
+  LoadAudioFilterPluginRequest
 } from './proto/audio_frame_pb.js';
 
 export class AudioFilter {