refactor: stream API to be similar to non-stream API; created docs for streams

jankarres · jankarres · commit f3497c97f170 · 2025-11-09T16:59:45.000+01:00
diff --git a/README.md b/README.md
@@ -331,6 +331,61 @@ await encoder.encode();
 
 Setting `disptime` changes how often progress is emitted by the LAME CLI, while `silent`, `quiet`, and `verbose` let you align terminal verbosity with your logging needs.
 
+### Encode via Node.js streams
+
+When you already operate on `Readable`/`Writable` streams (for example when piping uploaded audio to disk), use the encoder stream helper to avoid buffering everything in memory.
+
+```js
+import { createReadStream, createWriteStream } from "node:fs";
+import { pipeline } from "node:stream/promises";
+import { LameStream } from "node-lame";
+
+const encoderStream = new LameStream({
+    mode: "encode",
+    bitrate: 192,
+});
+
+encoderStream.getEmitter().on("progress", ([progress, eta]) => {
+    process.stdout.write(
+        `Streaming progress: ${progress}%${eta ? ` – ETA ${eta}` : ""}\r`,
+    );
+});
+
+await pipeline(
+    createReadStream("./audio-files/example.wav"),
+    encoderStream,
+    createWriteStream("./audio-files/example.stream.mp3"),
+);
+```
+
+### Decode via Node.js streams
+
+Decoding an MP3 back to WAV (or raw PCM) works the same way—swap in the decoder helper and wire it into your existing stream graph.
+
+```js
+import { createReadStream, createWriteStream } from "node:fs";
+import { pipeline } from "node:stream/promises";
+import { LameStream } from "node-lame";
+
+const decoderStream = new LameStream({
+    mode: "decode",
+});
+
+decoderStream.getEmitter().on("progress", ([progress, eta]) => {
+    process.stdout.write(
+        `Decoding progress: ${progress}%${eta ? ` – ETA ${eta}` : ""}\r`,
+    );
+});
+
+await pipeline(
+    createReadStream("./audio-files/example.stream.mp3"),
+    decoderStream,
+    createWriteStream("./audio-files/example.stream.wav"),
+);
+```
+
+`LameStream` exposes `getEmitter()` and `getStatus()` so you receive live progress regardless of whether you encode or decode. Choose the direction up front via `mode: "encode"` or `"decode"` when constructing the stream, then treat it like any other duplex pipeline component.
+
 ## All options
 
 | Option                        | Description                                                                                                                                                                                                                                                                                                                                                                                                                   | Values                                                                                                                            | Default                                             |
diff --git a/examples/convert-stream-to-mp3.ts b/examples/convert-stream-to-mp3.ts
@@ -3,7 +3,7 @@ import { resolve } from "node:path";
 import { pipeline } from "node:stream/promises";
 import { fileURLToPath } from "node:url";
 
-import { createLameEncoderStream } from "./helpers/load-node-lame.js";
+import { LameStream } from "./helpers/load-node-lame.js";
 import { removeIfExists } from "./helpers/remove-if-exists.js";
 
 const inputPath = resolve(
@@ -16,7 +16,8 @@ const outputPath = resolve(
 async function main(): Promise<void> {
     await removeIfExists(outputPath);
 
-    const encoderStream = createLameEncoderStream({
+    const encoderStream = new LameStream({
+        mode: "encode",
         bitrate: 192,
     });
 
diff --git a/examples/helpers/load-node-lame.ts b/examples/helpers/load-node-lame.ts
@@ -1,7 +1,7 @@
 type NodeLameModule = typeof import("../../src/index");
 
 const MODULE_NOT_FOUND_CODE = "ERR_MODULE_NOT_FOUND";
-const LOCAL_FALLBACKS = ["../../dist/index.cjs", "../../src/index.ts"] as const;
+const LOCAL_FALLBACKS = ["../../src/index.ts", "../../dist/index.cjs"] as const;
 
 type ErrorWithCode = {
     code?: string;
@@ -53,6 +53,5 @@ const loadNodeLame = async (): Promise<NodeLameModule> => {
 
 const nodeLame = await loadNodeLame();
 
-export const { Lame, createLameDecoderStream, createLameEncoderStream } =
-    nodeLame;
+export const { Lame, LameStream } = nodeLame;
 export default nodeLame;
diff --git a/src/core/lame-process.ts b/src/core/lame-process.ts
@@ -1,5 +1,4 @@
 import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
-import type { ProcessEnv } from "node:process";
 import { delimiter } from "node:path";
 
 import type {
@@ -14,6 +13,7 @@ import {
 import { LameOptions } from "./lame-options";
 
 type ProgressKind = LameStreamMode;
+type ProcessEnv = Record<string, string | undefined>;
 
 const LAME_TAG_MESSAGE = "Writing LAME Tag...done";
 
diff --git a/src/core/lame-stream.ts b/src/core/lame-stream.ts
@@ -8,6 +8,7 @@ import type {
     LameOptionsBag,
     LameProgressEmitter,
     LameStatus,
+    LameStreamMode,
     LameStreamOptions,
 } from "../types";
 import { LameOptions } from "./lame-options";
@@ -20,53 +21,45 @@ import type { ProgressKind } from "./lame-process";
 
 type StreamKind = ProgressKind;
 
-interface LameStreamConfig extends LameStreamOptions {
+type LameStreamConfig = Omit<LameStreamOptions, "output"> & {
     binaryPath?: string;
-}
+};
 
-class LameCodecStream extends Duplex {
+class LameStream extends Duplex {
     private readonly emitter: LameProgressEmitter;
     private readonly status: LameStatus;
-    private readonly kind: StreamKind;
-    private readonly child: ChildProcessWithoutNullStreams;
     private readonly builder: LameOptions;
+    private readonly binaryPath?: string;
+    private readonly kind: StreamKind;
+
+    private child?: ChildProcessWithoutNullStreams;
 
     private isStdoutPaused = false;
     private hasErrored = false;
     private finished = false;
 
-    constructor(kind: StreamKind, options: LameStreamConfig) {
+    constructor(options: LameStreamConfig) {
         super({ allowHalfOpen: false });
 
-        const { binaryPath, ...cliOptions } = options;
+        const { binaryPath, mode, ...cliOptions } = options;
+        if (!isValidStreamMode(mode)) {
+            throw new Error(
+                'lame: LameStream requires a mode of either "encode" or "decode"',
+            );
+        }
+
         const normalizedOptions: LameOptionsBag = {
-            ...(cliOptions as Omit<LameStreamOptions, "output">),
+            ...(cliOptions as Omit<LameStreamOptions, "output" | "mode">),
             output: "stream",
         } as LameOptionsBag;
 
-        this.kind = kind;
+        this.binaryPath = binaryPath;
         this.status = createInitialStatus();
         this.emitter = new EventEmitter() as LameProgressEmitter;
         this.builder = new LameOptions(normalizedOptions);
+        this.kind = mode;
 
-        const spawnArgs = buildLameSpawnArgs(this.builder, this.kind, "-", "-");
-
-        this.child = spawnLameProcess({
-            binaryPath,
-            spawnArgs,
-            kind: this.kind,
-            status: this.status,
-            emitter: this.emitter,
-            progressSources: ["stderr"],
-            completeOnTag: true,
-            onError: (error) => this.emitStreamError(error),
-            onStdoutData: (chunk) => this.forwardStdout(chunk),
-            onStdoutEnd: () => this.push(null),
-            onStdoutError: (error) => this.emitStreamError(error),
-            onStderrError: (error) => this.emitStreamError(error),
-            onStdinError: (error) => this.emitStreamError(error),
-            onSuccess: () => this.handleSuccessfulClose(),
-        });
+        this.initialize(this.kind);
     }
 
     public getEmitter(): LameProgressEmitter {
@@ -78,6 +71,10 @@ class LameCodecStream extends Duplex {
     }
 
     public override _read(): void {
+        if (!this.child) {
+            return;
+        }
+
         if (this.isStdoutPaused && !this.child.stdout.readableEnded) {
             this.isStdoutPaused = false;
             this.child.stdout.resume();
@@ -89,18 +86,24 @@ class LameCodecStream extends Duplex {
         encoding: BufferEncoding,
         callback: (error?: Error | null) => void,
     ): void {
-        if (this.finished || this.child.stdin.destroyed) {
+        const child = this.child;
+        if (!child) {
+            callback(new Error("lame: Stream mode is not initialized yet"));
+            return;
+        }
+
+        if (this.finished || child.stdin.destroyed) {
             callback(new Error("lame: Stream has already finished"));
             return;
         }
 
         try {
-            const flushed = this.child.stdin.write(chunk, encoding);
+            const flushed = child.stdin.write(chunk, encoding);
             if (!flushed) {
                 const cleanup = () => {
-                    this.child.stdin.off("drain", onDrain);
-                    this.child.stdin.off("error", onError);
-                    this.child.stdin.off("close", onClose);
+                    child.stdin.off("drain", onDrain);
+                    child.stdin.off("error", onError);
+                    child.stdin.off("close", onClose);
                 };
 
                 const onDrain = () => {
@@ -116,9 +119,9 @@ class LameCodecStream extends Duplex {
                     callback(new Error("lame: Input stream closed before drain"));
                 };
 
-                this.child.stdin.once("drain", onDrain);
-                this.child.stdin.once("error", onError);
-                this.child.stdin.once("close", onClose);
+                child.stdin.once("drain", onDrain);
+                child.stdin.once("error", onError);
+                child.stdin.once("close", onClose);
                 return;
             }
         } catch (error) {
@@ -130,8 +133,14 @@ class LameCodecStream extends Duplex {
     }
 
     public override _final(callback: (error?: Error | null) => void): void {
+        const child = this.child;
+        if (!child) {
+            callback(new Error("lame: Stream mode is not initialized yet"));
+            return;
+        }
+
         try {
-            this.child.stdin.end();
+            child.stdin.end();
         } catch (error) {
             callback(error as Error);
             return;
@@ -144,9 +153,16 @@ class LameCodecStream extends Duplex {
         error: Error | null,
         callback: (error?: Error | null) => void,
     ): void {
+        const child = this.child;
+
+        if (!child) {
+            callback(error ?? null);
+            return;
+        }
+
         try {
-            if (!this.child.killed) {
-                this.child.kill();
+            if (!child.killed) {
+                child.kill();
             }
         } catch (killError) {
             callback(killError as Error);
@@ -157,13 +173,38 @@ class LameCodecStream extends Duplex {
         callback(error ?? null);
     }
 
+    private initialize(kind: StreamKind): void {
+        if (this.child) {
+            return;
+        }
+
+        const spawnArgs = buildLameSpawnArgs(this.builder, kind, "-", "-");
+
+        this.child = spawnLameProcess({
+            binaryPath: this.binaryPath,
+            spawnArgs,
+            kind,
+            status: this.status,
+            emitter: this.emitter,
+            progressSources: ["stderr"],
+            completeOnTag: true,
+            onError: (error) => this.emitStreamError(error),
+            onStdoutData: (chunk) => this.forwardStdout(chunk),
+            onStdoutEnd: () => this.push(null),
+            onStdoutError: (error) => this.emitStreamError(error),
+            onStderrError: (error) => this.emitStreamError(error),
+            onStdinError: (error) => this.emitStreamError(error),
+            onSuccess: () => this.handleSuccessfulClose(),
+        });
+    }
+
     private forwardStdout(chunk: Buffer) {
         if (this.hasErrored || this.finished) {
             return;
         }
 
         const shouldContinue = this.push(chunk);
-        if (!shouldContinue) {
+        if (!shouldContinue && this.child) {
             this.isStdoutPaused = true;
             this.child.stdout.pause();
         }
@@ -188,8 +229,9 @@ class LameCodecStream extends Duplex {
         this.status.finished = true;
         this.cleanupChildListeners();
         this.emitter.emit("error", error);
+
         try {
-            if (!this.child.killed) {
+            if (this.child && !this.child.killed) {
                 this.child.kill();
             }
         } catch {
@@ -200,24 +242,20 @@ class LameCodecStream extends Duplex {
     }
 
     private cleanupChildListeners() {
+        if (!this.child) {
+            return;
+        }
+
         this.child.stdout.removeAllListeners();
         this.child.stderr.removeAllListeners();
         this.child.stdin.removeAllListeners();
         this.child.removeAllListeners();
     }
 }
 
-const createLameEncoderStream = (options: LameStreamConfig) => {
-    return new LameCodecStream("encode", options);
+const isValidStreamMode = (value: unknown): value is LameStreamMode => {
+    return value === "encode" || value === "decode";
 };
 
-const createLameDecoderStream = (options: LameStreamConfig) => {
-    return new LameCodecStream("decode", options);
-};
-
-export {
-    LameCodecStream,
-    createLameDecoderStream,
-    createLameEncoderStream,
-};
+export { LameStream };
 export type { LameStreamConfig };
diff --git a/src/core/lame.ts b/src/core/lame.ts
@@ -48,7 +48,7 @@ class Lame {
     constructor(options: LameOptionsBag) {
         if (options.output === "stream") {
             throw new Error(
-                "lame: The streaming output mode requires createLameEncoderStream or createLameDecoderStream",
+                'lame: The streaming output mode requires LameStream with mode set to "encode" or "decode"',
             );
         }
 
diff --git a/src/types.ts b/src/types.ts
@@ -86,8 +86,11 @@ interface LameOptionsBag {
     meta?: MetaOptions;
 }
 
+type LameStreamMode = "encode" | "decode";
+
 type LameStreamOptions = Omit<LameOptionsBag, "output"> & {
     output?: "stream";
+    mode: LameStreamMode;
 };
 
 type SampleFrequency = 8 | 11.025 | 12 | 16 | 22.05 | 24 | 32 | 44.1 | 48;
@@ -186,4 +189,5 @@ export type {
     HelpTopic,
     CustomFrameRecord,
     LameStreamOptions,
+    LameStreamMode,
 };
diff --git a/tests/integration/lame.integration.test.ts b/tests/integration/lame.integration.test.ts
diff --git a/tests/unit/lame-stream.test.ts b/tests/unit/lame-stream.test.ts
diff --git a/tests/unit/lame.test.ts b/tests/unit/lame.test.ts

Original file line number	Diff line number	Diff line change
`@@ -48,7 +48,7 @@ class Lame {`
`48`	`48`	`constructor(options: LameOptionsBag) {`
`49`	`49`	`if (options.output === "stream") {`
`50`	`50`	`throw new Error(`
`51`		`- "lame: The streaming output mode requires createLameEncoderStream or createLameDecoderStream",`
	`51`	`+ 'lame: The streaming output mode requires LameStream with mode set to "encode" or "decode"',`
`52`	`52`	`);`
`53`	`53`	`}`
`54`	`54`