Add AbortSignal to prepareStream and playStream (#153)

package.json

@@ -16,7 +16,6 @@
         "@libav.js/variant-webcodecs": "^6.4.7",
         "debug-level": "^3.2.1",
         "fluent-ffmpeg": "^2.1.3",
-        "p-cancelable": "^4.0.1",
         "p-debounce": "^4.0.0",
         "sodium-plus": "^0.9.0",
         "uid": "^2.0.2",

src/media/newApi.ts

@@ -83,8 +83,10 @@ export type EncoderOptions = {
 
 export function prepareStream(
     input: string | Readable,
-    options: Partial<EncoderOptions> = {}
+    options: Partial<EncoderOptions> = {},
+    cancelSignal?: AbortSignal
 ) {
+    cancelSignal?.throwIfAborted();
     const defaultOptions = {
         noTranscoding: false,
         // negative values = resize by aspect ratio, see https://trac.ffmpeg.org/wiki/Scaling
@@ -281,8 +283,26 @@ export function prepareStream(
             .audioCodec("libopus")
             .audioBitrate(`${bitrateAudio}k`);
 
+    // exit handling
+    const promise = new Promise<void>((resolve, reject) => {
+        command.on("error", (err) => {
+            if (cancelSignal?.aborted)
+                /**
+                 * fluent-ffmpeg might throw an error when SIGTERM is sent to
+                 * the process, so we check if the abort signal is triggered
+                 * and throw that instead
+                 */
+                reject(cancelSignal.reason);
+            else
+                reject(err);
+        });
+        command.on("end", () => resolve());
+    })
+    promise.catch(() => {});
+    cancelSignal?.addEventListener("abort", () => command.kill("SIGTERM"), { once: true });
     command.run();
-    return { command, output }
+
+    return { command, output, promise }
 }
 
 export type PlayStreamOptions = {
@@ -321,12 +341,18 @@ export type PlayStreamOptions = {
 }
 
 export async function playStream(
-    input: Readable, streamer: Streamer, options: Partial<PlayStreamOptions> = {})
+    input: Readable, streamer: Streamer,
+    options: Partial<PlayStreamOptions> = {},
+    cancelSignal?: AbortSignal
+)
 {
+    cancelSignal?.throwIfAborted();
     if (!streamer.voiceConnection)
         throw new Error("Bot is not connected to a voice channel");
 
     const { video, audio } = await demux(input);
+    cancelSignal?.throwIfAborted();
+
     if (!video)
         throw new Error("No video stream in media");
 
@@ -421,12 +447,21 @@ export async function playStream(
             vStream.on("pts", stopBurst);
         }
     }
-    return new Promise<void>((resolve) => {
-        vStream.once("finish", () => {
+    return new Promise<void>((resolve, reject) => {
+        const cleanup = () => {
             stopStream();
             udp.mediaConnection.setSpeaking(false);
             udp.mediaConnection.setVideoAttributes(false);
+        }
+        cancelSignal?.addEventListener("abort", () => {
+            cleanup();
+            reject(cancelSignal.reason);
+        }, { once: true })
+        vStream.once("finish", () => {
+            if (cancelSignal?.aborted)
+                return;
+            cleanup();
             resolve();
         });
-    });
+    }).catch(() => {});
 }