add close method to AudioSource and VideoSource (#365)

* add close method to AudioSource and VideoSource

ensures we can clean up allocated resources correctly

* changeset

* async close methods for future-proof

* consistent protocol version

* fix receive-audio example

.changeset/slow-hounds-fail.md

@@ -0,0 +1,5 @@
+---
+'@livekit/rtc-node': patch
+---
+
+add close method to AudioSource and VideoSource

examples/agent-dispatch/index.ts

@@ -21,7 +21,7 @@ async function createExplicitDispatch() {
   });
   console.log('created dispatch', dispatch);
 
-  const dispatches = await agentDispatchClient.listDispatches(roomName);
+  const dispatches = await agentDispatchClient.listDispatch(roomName);
   console.log(`there are ${dispatches.length} dispatches in ${roomName}`);
 }
 

examples/agent-dispatch/package.json

@@ -14,7 +14,7 @@
   "dependencies": {
     "dotenv": "^16.4.5",
     "livekit-server-sdk": "workspace:*",
-    "@livekit/protocol": "^1.28.0"
+    "@livekit/protocol": "^1.30.0"
   },
   "devDependencies": {
     "@types/node": "^20.10.4",

examples/publish-wav/index.ts

@@ -63,6 +63,10 @@ while (written < dataSize) {
   written += frameSize;
 }
 await source.waitForPlayout();
+// release resources allocated for audio publishing
+await source.close();
 
 await room.disconnect();
+
+// disposes all resources, only use if no more sessions are expected
 await dispose();

examples/receive-audio/index.ts

@@ -77,13 +77,13 @@ const room = new Room();
 let trackToProcess: string | null = null;
 let writer: fs.WriteStream | null = null;
 
-room.on(RoomEvent.TrackSubscribed, (track, publication, participant) => {
+room.on(RoomEvent.TrackSubscribed, async (track, publication, participant) => {
   console.log('subscribed to track', track.sid, publication, participant.identity);
   if (track.kind === TrackKind.KIND_AUDIO) {
     const stream = new AudioStream(track);
     trackToProcess = track.sid;
 
-    stream.on('frameReceived', (ev) => {
+    for await (const frame of stream) {
       if (!trackToProcess) {
         return;
       }
@@ -92,14 +92,14 @@ room.on(RoomEvent.TrackSubscribed, (track, publication, participant) => {
         // create file on first frame
         // also guard when track is unsubscribed
         writer = fs.createWriteStream('output.wav');
-        writeWavHeader(writer, ev.frame);
+        writeWavHeader(writer, frame);
       }
 
       if (writer) {
-        const buf = Buffer.from(ev.frame.data.buffer);
+        const buf = Buffer.from(frame.data.buffer);
         writer.write(buf);
       }
-    });
+    }
   }
 });
 

packages/livekit-rtc/src/audio_source.ts

@@ -134,4 +134,8 @@ export class AudioSource {
       throw new Error(cb.error);
     }
   }
+
+  async close() {
+    this.ffiHandle.dispose();
+  }
 }

packages/livekit-rtc/src/video_source.ts

@@ -64,4 +64,8 @@ export class VideoSource {
       message: { case: 'captureVideoFrame', value: req },
     });
   }
+
+  async close() {
+    this.ffiHandle.dispose();
+  }
 }