Validate types returned by custom coders, refactor custom coder call serialization logic, improve metadata validation

dev/convert.html

@@ -21,10 +21,7 @@
 			chunked: true,
 			chunkSize: 2**20
 		});
-		const outputFormat = new Metamuxer.Mp4OutputFormat({
+		const outputFormat = new Metamuxer.OggOutputFormat();
-			//streamable: true
-			//fastStart: 'fragmented'
-		});
 
 		const button = document.createElement('button');
 		button.textContent = 'Cancel';
@@ -41,7 +38,7 @@
 				target
 			}),
 			audio: {
-				discard: true
+				//discard: true
 				//forceReencode: true,
 			},
 			/*

src/custom-coder.ts

@@ -22,7 +22,7 @@ export abstract class CustomVideoDecoder {
 	}
 
 	/** Called after decoder creation; can be used for custom initialization logic. */
-	abstract init(): void;
+	abstract init(): Promise<void> | void;
 	/** Decodes the provided encoded packet. */
 	abstract decode(packet: EncodedPacket): Promise<void> | void;
 	/** Decodes all remaining packets and then resolves. */
@@ -51,7 +51,7 @@ export abstract class CustomAudioDecoder {
 	}
 
 	/** Called after decoder creation; can be used for custom initialization logic. */
-	abstract init(): void;
+	abstract init(): Promise<void> | void;
 	/** Decodes the provided encoded packet. */
 	abstract decode(packet: EncodedPacket): Promise<void> | void;
 	/** Decodes all remaining packets and then resolves. */
@@ -80,7 +80,7 @@ export abstract class CustomVideoEncoder {
 	}
 
 	/** Called after encoder creation; can be used for custom initialization logic. */
-	abstract init(): void;
+	abstract init(): Promise<void> | void;
 	/** Encodes the provided video sample. */
 	abstract encode(videoSample: VideoSample, options: VideoEncoderEncodeOptions): Promise<void> | void;
 	/** Encodes all remaining video samples and then resolves. */
@@ -109,7 +109,7 @@ export abstract class CustomAudioEncoder {
 	}
 
 	/** Called after encoder creation; can be used for custom initialization logic. */
-	abstract init(): void;
+	abstract init(): Promise<void> | void;
 	/** Encodes the provided audio sample. */
 	abstract encode(audioSample: AudioSample): Promise<void> | void;
 	/** Encodes all remaining audio samples and then resolves. */

src/media-sink.ts

@@ -5,6 +5,7 @@ import {
 	AnyIterable,
 	assert,
 	binarySearchLessOrEqual,
+	CallSerializer,
 	getInt24,
 	getUint24,
 	last,
@@ -640,7 +641,7 @@ class VideoDecoderWrapper extends DecoderWrapper<VideoSample> {
 	decoder: VideoDecoder | null = null;
 
 	customDecoder: CustomVideoDecoder | null = null;
-	lastCustomDecoderPromise = Promise.resolve();
+	customDecoderCallSerializer = new CallSerializer();
 	customDecoderQueueSize = 0;
 
 	sampleQueue: VideoSample[] = [];
@@ -683,9 +684,15 @@ class VideoDecoderWrapper extends DecoderWrapper<VideoSample> {
 			this.customDecoder = new MatchingCustomDecoder() as CustomVideoDecoder;
 			this.customDecoder.codec = codec;
 			this.customDecoder.config = decoderConfig;
-			this.customDecoder.onSample = sampleHandler;
+			this.customDecoder.onSample = (sample) => {
+				if (!(sample instanceof VideoSample)) {
+					throw new TypeError('The argument passed to onSample must be a VideoSample.');
+				}
 
-			this.customDecoder.init();
+				sampleHandler(sample);
+			};
+
+			void this.customDecoderCallSerializer.call(() => this.customDecoder!.init());
 		} else {
 			this.decoder = new VideoDecoder({
 				output: frame => sampleHandler(new VideoSample(frame)),
@@ -715,11 +722,9 @@ class VideoDecoderWrapper extends DecoderWrapper<VideoSample> {
 	decode(packet: EncodedPacket) {
 		if (this.customDecoder) {
 			this.customDecoderQueueSize++;
-			this.lastCustomDecoderPromise = this.lastCustomDecoderPromise.then(() => {
+			void this.customDecoderCallSerializer
-				return this.customDecoder!.decode(packet);
+				.call(() => this.customDecoder!.decode(packet))
-			});
+				.then(() => this.customDecoderQueueSize--);
-
-			void this.lastCustomDecoderPromise.then(() => this.customDecoderQueueSize--);
 		} else {
 			assert(this.decoder);
 			this.decoder.decode(packet.toEncodedVideoChunk());
@@ -728,7 +733,7 @@ class VideoDecoderWrapper extends DecoderWrapper<VideoSample> {
 
 	async flush() {
 		if (this.customDecoder) {
-			await this.lastCustomDecoderPromise.then(() => this.customDecoder!.flush());
+			await this.customDecoderCallSerializer.call(() => this.customDecoder!.flush());
 		} else {
 			assert(this.decoder);
 			await this.decoder.flush();
@@ -742,7 +747,7 @@ class VideoDecoderWrapper extends DecoderWrapper<VideoSample> {
 
 	close() {
 		if (this.customDecoder) {
-			void this.lastCustomDecoderPromise.then(() => this.customDecoder!.close());
+			void this.customDecoderCallSerializer.call(() => this.customDecoder!.close());
 		} else {
 			assert(this.decoder);
 			this.decoder.close();
@@ -1097,7 +1102,7 @@ class AudioDecoderWrapper extends DecoderWrapper<AudioSample> {
 	decoder: AudioDecoder | null = null;
 
 	customDecoder: CustomAudioDecoder | null = null;
-	lastCustomDecoderPromise = Promise.resolve();
+	customDecoderCallSerializer = new CallSerializer();
 	customDecoderQueueSize = 0;
 
 	constructor(
@@ -1123,9 +1128,15 @@ class AudioDecoderWrapper extends DecoderWrapper<AudioSample> {
 			this.customDecoder = new MatchingCustomDecoder() as CustomAudioDecoder;
 			this.customDecoder.codec = codec;
 			this.customDecoder.config = decoderConfig;
-			this.customDecoder.onSample = sampleHandler;
+			this.customDecoder.onSample = (sample) => {
+				if (!(sample instanceof AudioSample)) {
+					throw new TypeError('The argument passed to onSample must be an AudioSample.');
+				}
 
-			this.customDecoder.init();
+				sampleHandler(sample);
+			};
+
+			void this.customDecoderCallSerializer.call(() => this.customDecoder!.init());
 		} else {
 			this.decoder = new AudioDecoder({
 				output: data => sampleHandler(new AudioSample(data)),
@@ -1147,11 +1158,9 @@ class AudioDecoderWrapper extends DecoderWrapper<AudioSample> {
 	decode(packet: EncodedPacket) {
 		if (this.customDecoder) {
 			this.customDecoderQueueSize++;
-			this.lastCustomDecoderPromise = this.lastCustomDecoderPromise.then(() => {
+			void this.customDecoderCallSerializer
-				return this.customDecoder!.decode(packet);
+				.call(() => this.customDecoder!.decode(packet))
-			});
+				.then(() => this.customDecoderQueueSize--);
-
-			void this.lastCustomDecoderPromise.then(() => this.customDecoderQueueSize--);
 		} else {
 			assert(this.decoder);
 			this.decoder.decode(packet.toEncodedAudioChunk());
@@ -1160,7 +1169,7 @@ class AudioDecoderWrapper extends DecoderWrapper<AudioSample> {
 
 	flush() {
 		if (this.customDecoder) {
-			return this.lastCustomDecoderPromise.then(() => this.customDecoder!.flush());
+			return this.customDecoderCallSerializer.call(() => this.customDecoder!.flush());
 		} else {
 			assert(this.decoder);
 			return this.decoder.flush();
@@ -1169,7 +1178,7 @@ class AudioDecoderWrapper extends DecoderWrapper<AudioSample> {
 
 	close() {
 		if (this.customDecoder) {
-			void this.lastCustomDecoderPromise.then(() => this.customDecoder!.close());
+			void this.customDecoderCallSerializer.call(() => this.customDecoder!.close());
 		} else {
 			assert(this.decoder);
 			this.decoder.close();

src/media-source.ts

@@ -16,7 +16,7 @@ import {
 	VideoCodec,
 } from './codec';
 import { OutputAudioTrack, OutputSubtitleTrack, OutputTrack, OutputVideoTrack } from './output';
-import { assert, clamp, promiseWithResolvers, setInt24, setUint24 } from './misc';
+import { assert, CallSerializer, clamp, promiseWithResolvers, setInt24, setUint24 } from './misc';
 import { Muxer } from './muxer';
 import { SubtitleParser } from './subtitles';
 import { toAlaw, toUlaw } from './pcm';
@@ -69,7 +69,7 @@ export abstract class MediaSource {
 	/** @internal */
 	_start() {}
 	/** @internal */
-	async _flush() {}
+	async _flushAndClose() {}
 
 	/**
 	 * Closes this source. This prevents future samples from being added and signals to the output file that no further
@@ -92,7 +92,7 @@ export abstract class MediaSource {
 		}
 
 		return this._closingPromise = (async () => {
-			await this._flush();
+			await this._flushAndClose();
 
 			this._closed = true;
 
@@ -110,7 +110,7 @@ export abstract class MediaSource {
 			// Since closing also flushes, we don't want to do it twice
 			return this._closingPromise;
 		} else {
-			return this._flush();
+			return this._flushAndClose();
 		}
 	}
 }
@@ -161,6 +161,9 @@ export class EncodedVideoPacketSource extends VideoSource {
 		if (packet.isMetadataOnly) {
 			throw new TypeError('Metadata-only packets cannot be added.');
 		}
+		if (meta !== undefined && (!meta || typeof meta !== 'object')) {
+			throw new TypeError('meta, when provided, must be an object.');
+		}
 
 		this._ensureValidAdd();
 		return this._connectedTrack!.output._muxer.addEncodedVideoPacket(this._connectedTrack!, packet, meta);
@@ -250,7 +253,7 @@ class VideoEncoderWrapper {
 	private lastHeight: number | null = null;
 
 	private customEncoder: CustomVideoEncoder | null = null;
-	private lastCustomEncoderPromise = Promise.resolve();
+	private customEncoderCallSerializer = new CallSerializer();
 	private customEncoderQueueSize = 0;
 
 	constructor(private source: VideoSource, private encodingConfig: VideoEncodingConfig) {}
@@ -296,11 +299,9 @@ class VideoEncoderWrapper {
 
 		if (this.customEncoder) {
 			this.customEncoderQueueSize++;
-			this.lastCustomEncoderPromise = this.lastCustomEncoderPromise.then(() => {
+			const promise = this.customEncoderCallSerializer
-				return this.customEncoder!.encode(videoSample, finalEncodeOptions);
+				.call(() => this.customEncoder!.encode(videoSample, finalEncodeOptions))
-			});
+				.then(() => {
-
-			void this.lastCustomEncoderPromise.then(() => {
 					this.customEncoderQueueSize--;
 
 					if (shouldClose) {
@@ -309,7 +310,7 @@ class VideoEncoderWrapper {
 				});
 
 			if (this.customEncoderQueueSize >= 4) {
-				await this.lastCustomEncoderPromise;
+				await promise;
 			}
 		} else {
 			assert(this.encoder);
@@ -371,11 +372,18 @@ class VideoEncoderWrapper {
 			this.customEncoder.codec = this.encodingConfig.codec;
 			this.customEncoder.config = encoderConfig;
 			this.customEncoder.onPacket = (packet, meta) => {
+				if (!(packet instanceof EncodedPacket)) {
+					throw new TypeError('The first argument passed to onPacket must be an EncodedPacket.');
+				}
+				if (meta !== undefined && (!meta || typeof meta !== 'object')) {
+					throw new TypeError('The second argument passed to onPacket must be an object or undefined.');
+				}
+
 				this.encodingConfig.onEncodedPacket?.(packet, meta);
 				void this.muxer!.addEncodedVideoPacket(this.source._connectedTrack!, packet, meta);
 			};
 
-			this.customEncoder.init();
+			await this.customEncoder.init();
 		} else {
 			if (typeof VideoEncoder === 'undefined') {
 				throw new Error('VideoEncoder is not supported by this browser.');
@@ -409,9 +417,10 @@ class VideoEncoderWrapper {
 		resolve();
 	}
 
-	async flush() {
+	async flushAndClose() {
 		if (this.customEncoder) {
-			await this.lastCustomEncoderPromise.then(() => this.customEncoder!.flush());
+			void this.customEncoderCallSerializer.call(() => this.customEncoder!.flush());
+			await this.customEncoderCallSerializer.call(() => this.customEncoder!.close());
 		} else if (this.encoder) {
 			await this.encoder.flush();
 			this.encoder.close();
@@ -459,8 +468,8 @@ export class VideoSampleSource extends VideoSource {
 	}
 
 	/** @internal */
-	override _flush() {
+	override _flushAndClose() {
-		return this._encoder.flush();
+		return this._encoder.flushAndClose();
 	}
 }
 
@@ -511,8 +520,8 @@ export class CanvasSource extends VideoSource {
 	}
 
 	/** @internal */
-	override _flush() {
+	override _flushAndClose() {
-		return this._encoder.flush();
+		return this._encoder.flushAndClose();
 	}
 }
 
@@ -578,13 +587,13 @@ export class MediaStreamVideoTrackSource extends VideoSource {
 	}
 
 	/** @internal */
-	override async _flush() {
+	override async _flushAndClose() {
 		if (this._abortController) {
 			this._abortController.abort();
 			this._abortController = null;
 		}
 
-		await this._encoder.flush();
+		await this._encoder.flushAndClose();
 	}
 }
 
@@ -634,6 +643,9 @@ export class EncodedAudioPacketSource extends AudioSource {
 		if (packet.isMetadataOnly) {
 			throw new TypeError('Metadata-only packets cannot be added.');
 		}
+		if (meta !== undefined && (!meta || typeof meta !== 'object')) {
+			throw new TypeError('meta, when provided, must be an object.');
+		}
 
 		this._ensureValidAdd();
 		return this._connectedTrack!.output._muxer.addEncodedAudioPacket(this._connectedTrack!, packet, meta);
@@ -719,7 +731,7 @@ class AudioEncoderWrapper {
 	private writeOutputValue: ((view: DataView, byteOffset: number, value: number) => void) | null = null;
 
 	private customEncoder: CustomAudioEncoder | null = null;
-	private lastCustomEncoderPromise = Promise.resolve();
+	private customEncoderCallSerializer = new CallSerializer();
 	private customEncoderQueueSize = 0;
 
 	constructor(private source: AudioSource, private encodingConfig: AudioEncodingConfig) {}
@@ -755,11 +767,9 @@ class AudioEncoderWrapper {
 
 		if (this.customEncoder) {
 			this.customEncoderQueueSize++;
-			this.lastCustomEncoderPromise = this.lastCustomEncoderPromise.then(() => {
+			const promise = this.customEncoderCallSerializer
-				return this.customEncoder!.encode(audioSample);
+				.call(() => this.customEncoder!.encode(audioSample))
-			});
+				.then(() => {
-
-			void this.lastCustomEncoderPromise.then(() => {
 					this.customEncoderQueueSize--;
 
 					if (shouldClose) {
@@ -768,7 +778,7 @@ class AudioEncoderWrapper {
 				});
 
 			if (this.customEncoderQueueSize >= 4) {
-				await this.lastCustomEncoderPromise;
+				await promise;
 			}
 
 			await this.muxer!.mutex.currentPromise; // Allow the writer to apply backpressure
@@ -900,11 +910,18 @@ class AudioEncoderWrapper {
 			this.customEncoder.codec = this.encodingConfig.codec;
 			this.customEncoder.config = encoderConfig;
 			this.customEncoder.onPacket = (packet, meta) => {
+				if (!(packet instanceof EncodedPacket)) {
+					throw new TypeError('The first argument passed to onPacket must be an EncodedPacket.');
+				}
+				if (meta !== undefined && (!meta || typeof meta !== 'object')) {
+					throw new TypeError('The second argument passed to onPacket must be an object or undefined.');
+				}
+
 				this.encodingConfig.onEncodedPacket?.(packet, meta);
 				void this.muxer!.addEncodedAudioPacket(this.source._connectedTrack!, packet, meta);
 			};
 
-			this.customEncoder.init();
+			await this.customEncoder.init();
 		} else if ((PCM_AUDIO_CODECS as readonly string[]).includes(this.encodingConfig.codec)) {
 			this.initPcmEncoder();
 		} else {
@@ -1020,9 +1037,10 @@ class AudioEncoderWrapper {
 		}
 	}
 
-	async flush() {
+	async flushAndClose() {
 		if (this.customEncoder) {
-			await this.lastCustomEncoderPromise.then(() => this.customEncoder!.flush());
+			void this.customEncoderCallSerializer.call(() => this.customEncoder!.flush());
+			await this.customEncoderCallSerializer.call(() => this.customEncoder!.close());
 		} else if (this.encoder) {
 			await this.encoder.flush();
 			this.encoder.close();
@@ -1072,8 +1090,8 @@ export class AudioSampleSource extends AudioSource {
 	}
 
 	/** @internal */
-	override _flush() {
+	override _flushAndClose() {
-		return this._encoder.flush();
+		return this._encoder.flushAndClose();
 	}
 }
 
@@ -1153,8 +1171,8 @@ export class AudioBufferSource extends AudioSource {
 	}
 
 	/** @internal */
-	override _flush() {
+	override _flushAndClose() {
-		return this._encoder.flush();
+		return this._encoder.flushAndClose();
 	}
 }
 
@@ -1215,13 +1233,13 @@ export class MediaStreamAudioTrackSource extends AudioSource {
 	}
 
 	/** @internal */
-	override async _flush() {
+	override async _flushAndClose() {
 		if (this._abortController) {
 			this._abortController.abort();
 			this._abortController = null;
 		}
 
-		await this._encoder.flush();
+		await this._encoder.flushAndClose();
 	}
 }
 

src/misc.ts

@@ -494,3 +494,11 @@ export const computeRationalApproximation = (x: number, maxDenominator: number)
 		denominator: currDenominator,
 	};
 };
+
+export class CallSerializer {
+	currentPromise = Promise.resolve();
+
+	call(fn: () => Promise<void> | void) {
+		return this.currentPromise = this.currentPromise.then(fn);
+	}
+}

src/ogg/ogg-muxer.ts

@@ -1,4 +1,4 @@
-import { OPUS_INTERNAL_SAMPLE_RATE, parseOpusIdentificationHeader } from '../codec';
+import { OPUS_INTERNAL_SAMPLE_RATE, parseOpusIdentificationHeader, validateAudioChunkMetadata } from '../codec';
 import { assert, setInt64, toDataView, toUint8Array } from '../misc';
 import { Muxer } from '../muxer';
 import { Output, OutputAudioTrack } from '../output';
@@ -81,9 +81,10 @@ export class OggMuxer extends Muxer {
 
 		assert(track.source._codec === 'vorbis' || track.source._codec === 'opus');
 
+		validateAudioChunkMetadata(meta);
+
 		assert(meta);
 		assert(meta.decoderConfig);
-		assert(meta.decoderConfig.sampleRate);
 
 		const newTrackData: OggTrackData = {
 			track,

src/wave/wave-muxer.ts

@@ -1,11 +1,12 @@
 import { Muxer } from '../muxer';
 import { Output, OutputAudioTrack } from '../output';
-import { parsePcmCodec, PcmAudioCodec } from '../codec';
+import { parsePcmCodec, PcmAudioCodec, validateAudioChunkMetadata } from '../codec';
 import { WaveFormat } from './wave-demuxer';
 import { RiffWriter } from './riff-writer';
 import { Writer } from '../writer';
 import { EncodedPacket } from '../packet';
 import { WavOutputFormat } from '../output-format';
+import { assert } from '../misc';
 
 export class WaveMuxer extends Muxer {
 	private format: WavOutputFormat;
@@ -39,9 +40,10 @@ export class WaveMuxer extends Muxer {
 
 		try {
 			if (!this.headerWritten) {
-				if (!meta?.decoderConfig) {
+				validateAudioChunkMetadata(meta);
-					throw new Error('Decoder config is required for first audio sample.');
+
-				}
+				assert(meta);
+				assert(meta.decoderConfig);
 
 				this.writeHeader(track, meta.decoderConfig);
 				this.headerWritten = true;

todo.txt

@@ -2,4 +2,3 @@
 - cross-track offset for streaming sources
 - textsubtitlesource, chunked piping
 - More efficient MP3 loading when reading sequentially
-- Validate types returned by custom coders