From 4f8c8836ec712b914ee1ddef2cf9e4ec35e40344 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Mon, 22 Dec 2025 23:16:05 +0100 Subject: [PATCH] Migrate FLAC demuxer to new packet system, move computeDuration impl to Input --- src/demuxer.ts | 11 -- src/flac/flac-demuxer.ts | 213 +++++++++++++++++++++------------------ src/input.ts | 9 +- test/node/demux.test.ts | 65 +++++++++--- 4 files changed, 172 insertions(+), 126 deletions(-) diff --git a/src/demuxer.ts b/src/demuxer.ts index 0c41306..d5e5ea5 100644 --- a/src/demuxer.ts +++ b/src/demuxer.ts @@ -20,15 +20,4 @@ export abstract class Demuxer { abstract getTracks(): Promise; abstract getMimeType(): Promise; abstract getMetadataTags(): Promise; - - async computeDuration(): Promise { - 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); - } } diff --git a/src/flac/flac-demuxer.ts b/src/flac/flac-demuxer.ts index e362f44..36b0b80 100644 --- a/src/flac/flac-demuxer.ts +++ b/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 { - await this.readMetadata(); - assert(this.track); - return this.track.computeDuration(); - } - override async getMetadataTags(): Promise { 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({ - startPos, - isFirstPacket, - }: { - startPos: number; - isFirstPacket: boolean; - }): Promise { + async readNextFlacFrame( + res: ResultValue, + { + startPos, + isFirstPacket, + }: { + startPos: number; + isFirstPacket: boolean; + }, + ): Promise { 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() { - await this.readMetadata(); + async advanceReader(res: ResultValue): Promise { assert(this.lastLoadedPos !== null); assert(this.audioInfo); const startPos = this.lastLoadedPos; - const frame = await this.readNextFlacFrame({ + + const frameResult = new ResultValue(); + 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 { assert(this.demuxer.audioInfo); @@ -587,101 +582,115 @@ class FlacAudioTrackBacking implements InputAudioTrackBacking { } async getPacket( + res: ResultValue, timestamp: number, options: PacketRetrievalOptions, - ): Promise { + ): Promise { 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 { - 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 advanceResult = new ResultValue(); - const packet = this.demuxer.loadedSamples[packetIndex]!; - const sampleTimestamp + while (true) { + 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 (this.demuxer.lastSampleLoaded) { - return this.getPacketAtIndex( - this.demuxer.loadedSamples.length - 1, - options, - ); - } - - await this.demuxer.advanceReader(); - continue; + if (sampleTimestamp + sampleDuration <= timestamp) { + if (this.demuxer.lastSampleLoaded) { + return this.getPacketAtIndex( + res, + this.demuxer.loadedSamples.length - 1, + options, + ); } - 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, packet: EncodedPacket, options: PacketRetrievalOptions, - ): Promise { - const release = await this.demuxer.readingMutex.acquire(); - try { - const nextIndex = packet.sequenceNumber + 1; - if ( - this.demuxer.lastSampleLoaded - && nextIndex >= this.demuxer.loadedSamples.length - ) { - return null; - } + ): Promise { + using lock = this.demuxer.readingMutex.lock(); + if (lock.pending) await lock.ready; - // Ensure the next sample exists - while ( - nextIndex >= this.demuxer.loadedSamples.length - && !this.demuxer.lastSampleLoaded - ) { - await this.demuxer.advanceReader(); - } - return this.getPacketAtIndex(nextIndex, options); - } finally { - release(); + const nextIndex = packet.sequenceNumber + 1; + if ( + this.demuxer.lastSampleLoaded + && nextIndex >= this.demuxer.loadedSamples.length + ) { + return res.set(null); } + + const advanceResult = new ResultValue(); + + // 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, timestamp: number, options: PacketRetrievalOptions, - ): Promise { - return this.getPacket(timestamp, options); + ): Promise { + return this.getPacket(res, timestamp, options); } getNextKeyPacket( + res: ResultValue, packet: EncodedPacket, options: PacketRetrievalOptions, - ): Promise { - return this.getNextPacket(packet, options); + ): Promise { + return this.getNextPacket(res, packet, options); } async getPacketAtIndex( + res: ResultValue, sampleIndex: number, options: PacketRetrievalOptions, - ): Promise { + ): Promise { 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, options: PacketRetrievalOptions, - ): Promise { + ): Promise { + const advanceResult = new ResultValue(); + // 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); } } diff --git a/src/input.ts b/src/input.ts index 7e9ec3a..d3cff16 100644 --- a/src/input.ts +++ b/src/input.ts @@ -118,8 +118,13 @@ export class Input implements Disposable { * all tracks. */ async computeDuration() { - const demuxer = await this._getDemuxer(); - return demuxer.computeDuration(); + const tracks = await this.getTracks(); + 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. */ diff --git a/test/node/demux.test.ts b/test/node/demux.test.ts index 48c934b..4d69b7d 100644 --- a/test/node/demux.test.ts +++ b/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); +});