use a shared base for all mediatransforms

src/lib/AudioDecoder.ts

@@ -1,6 +1,5 @@
-import { Transform } from 'node:stream';
 import ffmpeg from '@mmomtchev/ffmpeg';
-import { AudioStreamDefinition, MediaStream } from './MediaStream';
+import { AudioStreamDefinition, MediaStream, MediaTransform } from './MediaStream';
 import { TransformCallback } from 'stream';
 
 const { AudioDecoderContext, Codec } = ffmpeg;
@@ -12,12 +11,12 @@ export const verbose = (process.env.DEBUG_AUDIO_DECODER || process.env.DEBUG_ALL
  * from a Demuxer and write decoded audio samples
  * Its parameters are inherited from the Demuxer.
  */
-export class AudioDecoder extends Transform implements MediaStream {
+export class AudioDecoder extends MediaTransform implements MediaStream {
   protected decoder: any;
   protected busy: boolean;
 
   constructor(options: { _stream: any; }) {
-    super({ objectMode: true });
+    super();
     this.decoder = null;
     if (!options._stream) {
       throw new Error('Input is not a demuxed stream');

src/lib/AudioEncoder.ts

@@ -1,6 +1,5 @@
-import { Transform } from 'node:stream';
 import ffmpeg from '@mmomtchev/ffmpeg';
-import { AudioStreamDefinition, MediaStream } from './MediaStream';
+import { AudioStreamDefinition, MediaStream, MediaTransform } from './MediaStream';
 import { TransformCallback } from 'stream';
 
 const { AudioEncoderContext, AudioSamples } = ffmpeg;
@@ -12,14 +11,14 @@ export const verbose = (process.env.DEBUG_AUDIO_ENCODER || process.env.DEBUG_ALL
  * and write encoded audio data to a Muxer.
  * Its parameters must be explicitly configured.
  */
-export class AudioEncoder extends Transform implements MediaStream {
+export class AudioEncoder extends MediaTransform implements MediaStream {
   protected def: AudioStreamDefinition;
   protected encoder: any;
   protected codec: any;
   protected busy: boolean;
 
   constructor(def: AudioStreamDefinition) {
-    super({ objectMode: true });
+    super();
     this.def = def;
     this.codec = ffmpeg.findEncodingCodec(this.def.codec);
     verbose(`AudioEncoder: using ${this.codec.name()}`);

src/lib/AudioTransform.ts

@@ -1,11 +1,10 @@
-import { Transform, TransformOptions, TransformCallback } from 'node:stream';
+import { TransformCallback } from 'node:stream';
 import ffmpeg from '@mmomtchev/ffmpeg';
-import { AudioStreamDefinition } from './MediaStream';
+import { AudioStreamDefinition, MediaTransform, MediaTransformOptions } from './MediaStream';
 
 export const verbose = (process.env.DEBUG_AUDIO_TRANSFORM || process.env.DEBUG_ALL) ? console.debug.bind(console) : () => undefined;
 
-export interface AudioTransformOptions extends TransformOptions {
-  objectMode?: never;
+export interface AudioTransformOptions extends MediaTransformOptions {
   input: AudioStreamDefinition;
   output: AudioStreamDefinition;
 }
@@ -14,12 +13,12 @@ export interface AudioTransformOptions extends TransformOptions {
  * A stream Transform that uses AudioResampler to rescale/resample the raw Audio.
  * Must receive input from a AudioDecoder and must output to a AudioEncoder
  */
-export class AudioTransform extends Transform {
+export class AudioTransform extends MediaTransform {
   protected resampler: any;
   protected frameSize: number | undefined;
 
   constructor(options: AudioTransformOptions) {
-    super({ ...options, objectMode: true });
+    super(options);
     this.resampler = new ffmpeg.AudioResampler(
       options.output.channelLayout.layout(), options.output.sampleRate, options.output.sampleFormat,
       options.input.channelLayout.layout(), options.input.sampleRate, options.input.sampleFormat

src/lib/Demuxer.ts

@@ -15,7 +15,6 @@ export interface DemuxerOptions extends ReadableOptions {
    * Amount of data to buffer, only when reading from a ReadStream, @default 64Kb
    */
   highWaterMark?: number;
-  objectMode?: never;
 }
 
 /**

src/lib/MediaStream.ts

@@ -1,4 +1,4 @@
-import { EventEmitter, Readable, ReadableOptions, Writable } from 'node:stream';
+import { EventEmitter, Readable, ReadableOptions, Transform, TransformOptions, Writable } from 'node:stream';
 
 export const StreamTypes = {
   'Audio': 'Audio',
@@ -40,6 +40,18 @@ export function isAudioDefinition(def: MediaStreamDefinition): def is AudioStrea
   return def.type === 'Audio';
 }
 
+export interface MediaTransformOptions extends TransformOptions {
+  objectMode?: never;
+}
+
+/**
+ * A generic MediaTransform, uses object mode
+ */
+export class MediaTransform extends Transform {
+  constructor(options?: MediaTransformOptions) {
+    super({ ...(options || {}), objectMode: true });
+  }
+}
 
 /**
  * A generic raw MediaStream, has a definition and it is an EventEmitter.

src/lib/Muxer.ts

@@ -17,7 +17,6 @@ export interface MuxerOptions extends WritableOptions {
   highWaterMark?: number;
   outputFormat?: string;
   streams: MediaEncoder[];
-  objectMode?: never;
 }
 
 /**
@@ -147,13 +146,14 @@ export class Muxer extends EventEmitter {
     if (this.destroyed) return;
     this.destroyed = true;
     verbose(`Muxer: destroy: ${e}`);
-    this.emit('error', e);
-    if (this.output)
+    if (this.output) {
       (this.output as any)._final();
+    }
     for (const s in this.streams) {
       this.streams[s].destroy(e);
     }
     await this.formatContext.closeAsync();
+    this.emit('error', e);
   }
 
   protected async prime(): Promise<void> {

src/lib/VideoDecoder.ts

@@ -1,6 +1,5 @@
-import { Transform } from 'node:stream';
 import ffmpeg from '@mmomtchev/ffmpeg';
-import { VideoStreamDefinition, MediaStream } from './MediaStream';
+import { VideoStreamDefinition, MediaStream, MediaTransform } from './MediaStream';
 import { TransformCallback } from 'stream';
 
 const { VideoDecoderContext, Codec } = ffmpeg;
@@ -12,13 +11,13 @@ export const verbose = (process.env.DEBUG_VIDEO_DECODER || process.env.DEBUG_ALL
  * from a Demuxer and write decoded video frames.
  * Its parameters are inherited from the Demuxer.
  */
-export class VideoDecoder extends Transform implements MediaStream {
+export class VideoDecoder extends MediaTransform implements MediaStream {
   protected decoder: any;
   protected busy: boolean;
   protected stream: any;
 
   constructor(options: { _stream: any; }) {
-    super({ objectMode: true });
+    super();
     this.decoder = null;
     if (!options._stream) {
       throw new Error('Input is not a demuxed stream');

src/lib/VideoEncoder.ts

@@ -1,6 +1,5 @@
-import { Transform } from 'node:stream';
 import ffmpeg from '@mmomtchev/ffmpeg';
-import { VideoStreamDefinition, MediaStream } from './MediaStream';
+import { VideoStreamDefinition, MediaStream, MediaTransform } from './MediaStream';
 import { TransformCallback } from 'stream';
 
 const { VideoEncoderContext, VideoFrame } = ffmpeg;
@@ -12,14 +11,14 @@ export const verbose = (process.env.DEBUG_VIDEO_ENCODER || process.env.DEBUG_ALL
  * and write encoded video data to a Muxer.
  * Its parameters must be explicitly configured.
  */
-export class VideoEncoder extends Transform implements MediaStream {
+export class VideoEncoder extends MediaTransform implements MediaStream {
   protected def: VideoStreamDefinition;
   protected encoder: any;
   protected codec: any;
   protected busy: boolean;
 
   constructor(def: VideoStreamDefinition) {
-    super({ objectMode: true });
+    super();
     this.def = def;
     this.codec = ffmpeg.findEncodingCodec(this.def.codec);
     verbose(`VideoEncoder: using ${this.codec.name()}`);

src/lib/VideoTransform.ts

@@ -1,9 +1,8 @@
-import { Transform, TransformOptions, TransformCallback } from 'node:stream';
+import { TransformCallback } from 'node:stream';
 import ffmpeg from '@mmomtchev/ffmpeg';
-import { VideoStreamDefinition } from './MediaStream';
+import { MediaTransform, MediaTransformOptions, VideoStreamDefinition } from './MediaStream';
 
-export interface VideoTransformOptions extends TransformOptions {
-  objectMode?: never;
+export interface VideoTransformOptions extends MediaTransformOptions {
   input: VideoStreamDefinition;
   output: VideoStreamDefinition;
   interpolation: number;
@@ -13,11 +12,11 @@ export interface VideoTransformOptions extends TransformOptions {
  * A stream Transform that uses VideoRescaler to rescale/resample the raw video.
  * Must receive input from a VideoDecoder and must output to a VideoEncoder
  */
-export class VideoTransform extends Transform {
+export class VideoTransform extends MediaTransform {
   protected rescaler: any;
 
   constructor(options: VideoTransformOptions) {
-    super({ ...options, objectMode: true });
+    super(options);
     this.rescaler = new ffmpeg.VideoRescaler(
       options.output.width, options.output.height, options.output.pixelFormat,
       options.input.width, options.input.height, options.input.pixelFormat,