Migrate FLAC demuxer to new packet system, move computeDuration impl to Input

src/demuxer.ts

@@ -20,15 +20,4 @@ export abstract class Demuxer {
 	abstract getTracks(): Promise<InputTrack[]>;
 	abstract getMimeType(): Promise<string>;
 	abstract getMetadataTags(): Promise<MetadataTags>;
-
-	async computeDuration(): Promise<number> {
-		const tracks = await this.getTracks();
-		if (tracks.length === 0) {
-			return 0;
-		}
-
-		// eslint-disable-next-line @typescript-eslint/await-thenable
-		const trackDurations = await Promise.all(tracks.map(x => x.computeDuration()));
-		return Math.max(...trackDurations);
-	}
 }

src/flac/flac-demuxer.ts

@@ -13,11 +13,13 @@ import { InputAudioTrack, InputAudioTrackBacking } from '../input-track';
 import { PacketRetrievalOptions } from '../media-sink';
 import {
 	assert,
-	AsyncMutex,
+	AsyncMutex4,
 	binarySearchLessOrEqual,
 	Bitstream,
+	ResultValue,
 	textDecoder,
 	UNDETERMINED_LANGUAGE,
+	Yo,
 } from '../misc';
 import { EncodedPacket, PLACEHOLDER_DATA } from '../packet';
 import {
@@ -77,7 +79,7 @@ export class FlacDemuxer extends Demuxer {
 	lastLoadedPos: number | null = null;
 	blockingBit: number | null = null;
 
-	readingMutex = new AsyncMutex();
+	readingMutex = new AsyncMutex4();
 	lastSampleLoaded = false;
 
 	constructor(input: Input) {
@@ -86,12 +88,6 @@ export class FlacDemuxer extends Demuxer {
 		this.reader = input._reader;
 	}
 
-	override async computeDuration(): Promise<number> {
-		await this.readMetadata();
-		assert(this.track);
-		return this.track.computeDuration();
-	}
-
 	override async getMetadataTags(): Promise<MetadataTags> {
 		await this.readMetadata();
 		return this.metadataTags;
@@ -108,9 +104,9 @@ export class FlacDemuxer extends Demuxer {
 	}
 
 	async readMetadata() {
-		let currentPos = 4; // Skip 'fLaC'
-
 		return (this.metadataPromise ??= (async () => {
+			let currentPos = 4; // Skip 'fLaC'
+
 			while (
 				this.reader.fileSize === null
 				|| currentPos < this.reader.fileSize
@@ -261,13 +257,16 @@ export class FlacDemuxer extends Demuxer {
 		})());
 	}
 
-	async readNextFlacFrame({
+	async readNextFlacFrame(
-		startPos,
+		res: ResultValue<NextFlacFrameResult | null>,
-		isFirstPacket,
+		{
-	}: {
+			startPos,
-		startPos: number;
+			isFirstPacket,
-		isFirstPacket: boolean;
+		}: {
-	}): Promise<NextFlacFrameResult | null> {
+			startPos: number;
+			isFirstPacket: boolean;
+		},
+	): Promise<Yo> {
 		assert(this.audioInfo);
 		// we expect that there are at least `minimumFrameSize` bytes left in the file
 
@@ -287,14 +286,15 @@ export class FlacDemuxer extends Demuxer {
 		const maximumSliceLength
 			= this.audioInfo.maximumFrameSize + maximumHeaderSize;
 
-		const slice = await this.reader.requestSliceRange(
+		let slice = this.reader.requestSliceRange(
 			startPos,
 			this.audioInfo.minimumFrameSize,
 			maximumSliceLength,
 		);
+		if (slice instanceof Promise) slice = await slice;
 
 		if (!slice) {
-			return null;
+			return res.set(null);
 		}
 
 		const frameHeader = this.readFlacFrameHeader({
@@ -303,7 +303,7 @@ export class FlacDemuxer extends Demuxer {
 		});
 
 		if (!frameHeader) {
-			return null;
+			return res.set(null);
 		}
 
 		// We don't know exactly how long the packet is, we only know the `minimumFrameSize` and `maximumFrameSize`
@@ -317,13 +317,13 @@ export class FlacDemuxer extends Demuxer {
 		while (true) {
 			// Reached end of the file, packet is over
 			if (slice.filePos > slice.end - minimumHeaderLength) {
-				return {
+				return res.set({
 					num: frameHeader.num,
 					blockSize: frameHeader.blockSize,
 					sampleRate: frameHeader.sampleRate,
 					size: slice.end - startPos,
 					isLastFrame: true,
-				};
+				});
 			}
 
 			const nextByte = readU8(slice);
@@ -369,13 +369,13 @@ export class FlacDemuxer extends Demuxer {
 					}
 				}
 
-				return {
+				return res.set({
 					num: frameHeader.num,
 					blockSize: frameHeader.blockSize,
 					sampleRate: frameHeader.sampleRate,
 					size: lengthIfNextFlacFrameHeaderIsLegit,
 					isLastFrame: false,
-				};
+				});
 			}
 		}
 	}
@@ -483,21 +483,24 @@ export class FlacDemuxer extends Demuxer {
 		return { num, blockSize, sampleRate };
 	}
 
-	async advanceReader() {
+	async advanceReader(res: ResultValue<void>): Promise<Yo> {
-		await this.readMetadata();
 		assert(this.lastLoadedPos !== null);
 		assert(this.audioInfo);
 		const startPos = this.lastLoadedPos;
-		const frame = await this.readNextFlacFrame({
+
+		const frameResult = new ResultValue<NextFlacFrameResult | null>();
+		const promise = this.readNextFlacFrame(frameResult, {
 			startPos,
 			isFirstPacket: this.loadedSamples.length === 0,
 		});
+		if (frameResult.pending) await promise;
 
+		const frame = frameResult.value;
 		if (!frame) {
 			// Unexpected case, failed to read next FLAC frame
 			// handling gracefully
 			this.lastSampleLoaded = true;
-			return;
+			return res.set();
 		}
 
 		const lastSample = this.loadedSamples[this.loadedSamples.length - 1];
@@ -517,8 +520,9 @@ export class FlacDemuxer extends Demuxer {
 
 		if (frame.isLastFrame) {
 			this.lastSampleLoaded = true;
-			return;
 		}
+
+		return res.set();
 	}
 }
 
@@ -542,11 +546,6 @@ class FlacAudioTrackBacking implements InputAudioTrackBacking {
 		return this.demuxer.audioInfo.numberOfChannels;
 	}
 
-	async computeDuration() {
-		const lastPacket = await this.getPacket(Infinity, { metadataOnly: true });
-		return (lastPacket?.timestamp ?? 0) + (lastPacket?.duration ?? 0);
-	}
-
 	getSampleRate() {
 		assert(this.demuxer.audioInfo);
 		return this.demuxer.audioInfo.sampleRate;
@@ -571,10 +570,6 @@ class FlacAudioTrackBacking implements InputAudioTrackBacking {
 		};
 	}
 
-	async getFirstTimestamp() {
-		return 0;
-	}
-
 	async getDecoderConfig(): Promise<AudioDecoderConfig | null> {
 		assert(this.demuxer.audioInfo);
 
@@ -587,101 +582,115 @@ class FlacAudioTrackBacking implements InputAudioTrackBacking {
 	}
 
 	async getPacket(
+		res: ResultValue<EncodedPacket | null>,
 		timestamp: number,
 		options: PacketRetrievalOptions,
-	): Promise<EncodedPacket | null> {
+	): Promise<Yo> {
 		assert(this.demuxer.audioInfo);
 		if (timestamp < 0) {
 			throw new Error('Timestamp cannot be negative');
 		}
 
-		const release = await this.demuxer.readingMutex.acquire();
+		using lock = this.demuxer.readingMutex.lock();
+		if (lock.pending) await lock.ready;
 
-		try {
+		const advanceResult = new ResultValue<void>();
-			while (true) {
-				const packetIndex = binarySearchLessOrEqual(
-					this.demuxer.loadedSamples,
-					timestamp,
-					x => x.blockOffset / this.demuxer.audioInfo!.sampleRate,
-				);
-				if (packetIndex === -1) {
-					await this.demuxer.advanceReader();
-					continue;
-				}
 
-				const packet = this.demuxer.loadedSamples[packetIndex]!;
+		while (true) {
-				const sampleTimestamp
+			const packetIndex = binarySearchLessOrEqual(
+				this.demuxer.loadedSamples,
+				timestamp,
+				x => x.blockOffset / this.demuxer.audioInfo!.sampleRate,
+			);
+			if (packetIndex === -1) {
+				advanceResult.reset();
+				const promise = this.demuxer.advanceReader(advanceResult);
+				if (advanceResult.pending) await promise;
+
+				continue;
+			}
+
+			const packet = this.demuxer.loadedSamples[packetIndex]!;
+			const sampleTimestamp
 					= packet.blockOffset / this.demuxer.audioInfo.sampleRate;
-				const sampleDuration
+			const sampleDuration
 					= packet.blockSize / this.demuxer.audioInfo.sampleRate;
 
-				if (sampleTimestamp + sampleDuration <= timestamp) {
+			if (sampleTimestamp + sampleDuration <= timestamp) {
-					if (this.demuxer.lastSampleLoaded) {
+				if (this.demuxer.lastSampleLoaded) {
-						return this.getPacketAtIndex(
+					return this.getPacketAtIndex(
-							this.demuxer.loadedSamples.length - 1,
+						res,
-							options,
+						this.demuxer.loadedSamples.length - 1,
-						);
+						options,
-					}
+					);
-
-					await this.demuxer.advanceReader();
-					continue;
 				}
 
-				return this.getPacketAtIndex(packetIndex, options);
+				advanceResult.reset();
+				const promise = this.demuxer.advanceReader(advanceResult);
+				if (advanceResult.pending) await promise;
+
+				continue;
 			}
-		} finally {
+
-			release();
+			return this.getPacketAtIndex(res, packetIndex, options);
 		}
 	}
 
 	async getNextPacket(
+		res: ResultValue<EncodedPacket | null>,
 		packet: EncodedPacket,
 		options: PacketRetrievalOptions,
-	): Promise<EncodedPacket | null> {
+	): Promise<Yo> {
-		const release = await this.demuxer.readingMutex.acquire();
+		using lock = this.demuxer.readingMutex.lock();
-		try {
+		if (lock.pending) await lock.ready;
-			const nextIndex = packet.sequenceNumber + 1;
-			if (
-				this.demuxer.lastSampleLoaded
-				&& nextIndex >= this.demuxer.loadedSamples.length
-			) {
-				return null;
-			}
 
-			// Ensure the next sample exists
+		const nextIndex = packet.sequenceNumber + 1;
-			while (
+		if (
-				nextIndex >= this.demuxer.loadedSamples.length
+			this.demuxer.lastSampleLoaded
-				&& !this.demuxer.lastSampleLoaded
+			&& nextIndex >= this.demuxer.loadedSamples.length
-			) {
+		) {
-				await this.demuxer.advanceReader();
+			return res.set(null);
-			}
-			return this.getPacketAtIndex(nextIndex, options);
-		} finally {
-			release();
 		}
+
+		const advanceResult = new ResultValue<void>();
+
+		// Ensure the next sample exists
+		while (
+			nextIndex >= this.demuxer.loadedSamples.length
+			&& !this.demuxer.lastSampleLoaded
+		) {
+			advanceResult.reset();
+			const promise = this.demuxer.advanceReader(advanceResult);
+			if (advanceResult.pending) await promise;
+		}
+
+		return this.getPacketAtIndex(res, nextIndex, options);
 	}
 
 	getKeyPacket(
+		res: ResultValue<EncodedPacket | null>,
 		timestamp: number,
 		options: PacketRetrievalOptions,
-	): Promise<EncodedPacket | null> {
+	): Promise<Yo> {
-		return this.getPacket(timestamp, options);
+		return this.getPacket(res, timestamp, options);
 	}
 
 	getNextKeyPacket(
+		res: ResultValue<EncodedPacket | null>,
 		packet: EncodedPacket,
 		options: PacketRetrievalOptions,
-	): Promise<EncodedPacket | null> {
+	): Promise<Yo> {
-		return this.getNextPacket(packet, options);
+		return this.getNextPacket(res, packet, options);
 	}
 
 	async getPacketAtIndex(
+		res: ResultValue<EncodedPacket | null>,
 		sampleIndex: number,
 		options: PacketRetrievalOptions,
-	): Promise<EncodedPacket | null> {
+	): Promise<Yo> {
 		const rawSample = this.demuxer.loadedSamples[sampleIndex];
 		if (!rawSample) {
-			return null;
+			return res.set(null);
 		}
 
 		let data: Uint8Array;
@@ -695,7 +704,7 @@ class FlacAudioTrackBacking implements InputAudioTrackBacking {
 			if (slice instanceof Promise) slice = await slice;
 
 			if (!slice) {
-				return null; // Data didn't fit into the rest of the file
+				return res.set(null); // Data didn't fit into the rest of the file
 			}
 
 			data = readBytes(slice, rawSample.byteSize);
@@ -704,27 +713,33 @@ class FlacAudioTrackBacking implements InputAudioTrackBacking {
 		assert(this.demuxer.audioInfo);
 		const timestamp = rawSample.blockOffset / this.demuxer.audioInfo.sampleRate;
 		const duration = rawSample.blockSize / this.demuxer.audioInfo.sampleRate;
-		return new EncodedPacket(
+
+		return res.set(new EncodedPacket(
 			data,
 			'key',
 			timestamp,
 			duration,
 			sampleIndex,
 			rawSample.byteSize,
-		);
+		));
 	}
 
 	async getFirstPacket(
+		res: ResultValue<EncodedPacket | null>,
 		options: PacketRetrievalOptions,
-	): Promise<EncodedPacket | null> {
+	): Promise<Yo> {
+		const advanceResult = new ResultValue<void>();
+
 		// Ensure the next sample exists
 		while (
 			this.demuxer.loadedSamples.length === 0
 			&& !this.demuxer.lastSampleLoaded
 		) {
-			await this.demuxer.advanceReader();
+			advanceResult.reset();
+			const promise = this.demuxer.advanceReader(advanceResult);
+			if (advanceResult.pending) await promise;
 		}
 
-		return this.getPacketAtIndex(0, options);
+		return this.getPacketAtIndex(res, 0, options);
 	}
 }

src/input.ts

@@ -118,8 +118,13 @@ export class Input<S extends Source = Source> implements Disposable {
 	 * all tracks.
 	 */
 	async computeDuration() {
-		const demuxer = await this._getDemuxer();
+		const tracks = await this.getTracks();
-		return demuxer.computeDuration();
+		if (tracks.length === 0) {
+			return 0;
+		}
+
+		const trackDurations = await Promise.all(tracks.map(x => x.computeDuration()));
+		return Math.max(...trackDurations);
 	}
 
 	/** Returns the list of all tracks of this input file. */

test/node/demux.test.ts

@@ -3,7 +3,7 @@ import { Input } from '../../src/input.js';
 import { BufferSource, FilePathSource } from '../../src/source.js';
 import path from 'node:path';
 import fs from 'node:fs';
-import { ADTS, ALL_FORMATS, MP3, MP4, OGG, QTFF, WAVE, WEBM } from '../../src/input-format.js';
+import { ADTS, ALL_FORMATS, FLAC, MP3, MP4, OGG, QTFF, WAVE, WEBM } from '../../src/input-format.js';
 import { PacketReader } from '../../src/cursors.js';
 import { InputAudioTrack, InputTrack } from '../../src/input-track.js';
 import { assert } from '../../src/misc.js';
@@ -88,6 +88,19 @@ test('Regular ISOBMFF demuxing', async () => {
 	expect(await input.computeDuration()).toBeCloseTo(5.041666666666667);
 });
 
+test('Regular ISOBMFF sync reading', async () => {
+	using input = new Input({
+		source: new BufferSource(fs.readFileSync(path.join(__dirname, '../public/trim-buck-bunny.mov'))),
+		formats: ALL_FORMATS,
+	});
+
+	const videoTrack = await input.getPrimaryVideoTrack();
+	assert(videoTrack);
+
+	const count = testSyncPacketReading(videoTrack);
+	expect(count).toBe(121);
+});
+
 test('Fragmented ISOBMFF demuxing', async () => {
 	using input = new Input({
 		source: new FilePathSource(path.join(__dirname, '../public/frag-buck-bunny.mp4')),
@@ -108,19 +121,6 @@ test('Fragmented ISOBMFF demuxing', async () => {
 	expect(await input.computeDuration()).toBeCloseTo(5);
 });
 
-test('Regular ISOBMFF sync reading', async () => {
-	using input = new Input({
-		source: new BufferSource(fs.readFileSync(path.join(__dirname, '../public/trim-buck-bunny.mov'))),
-		formats: ALL_FORMATS,
-	});
-
-	const videoTrack = await input.getPrimaryVideoTrack();
-	assert(videoTrack);
-
-	const count = testSyncPacketReading(videoTrack);
-	expect(count).toBe(121);
-});
-
 test('Fragmented ISOBMFF sync reading', async () => {
 	using input = new Input({
 		source: new BufferSource(fs.readFileSync(path.join(__dirname, '../public/frag-buck-bunny.mp4'))),
@@ -316,3 +316,40 @@ test('Ogg sync reading', async () => {
 	const count = testSyncPacketReading(audioTrack);
 	expect(count).toBe(5041);
 });
+
+test('FLAC demuxing', async () => {
+	using input = new Input({
+		source: new FilePathSource(path.join(__dirname, '../public/sample.flac')),
+		formats: ALL_FORMATS,
+	});
+
+	expect(await input.getFormat()).toBe(FLAC);
+	expect(await input.getMimeType()).toBe('audio/flac');
+
+	const tracks = await input.getTracks();
+	expect(tracks).toHaveLength(1);
+	const audioTrack = tracks[0] as InputAudioTrack;
+	expect(audioTrack).toBeInstanceOf(InputAudioTrack);
+
+	expect(audioTrack.codec).toBe('flac');
+	expect(audioTrack.numberOfChannels).toBeGreaterThan(0);
+	expect(audioTrack.sampleRate).toBeGreaterThan(0);
+
+	await testBasicPacketReading(audioTrack);
+
+	const duration = await audioTrack.computeDuration();
+	expect(duration).toBeCloseTo(19.71428571428571);
+});
+
+test('FLAC sync reading', async () => {
+	using input = new Input({
+		source: new BufferSource(fs.readFileSync(path.join(__dirname, '../public/sample.flac'))),
+		formats: ALL_FORMATS,
+	});
+
+	const audioTrack = await input.getPrimaryAudioTrack();
+	assert(audioTrack);
+
+	const count = testSyncPacketReading(audioTrack);
+	expect(count).toBe(213);
+});