diff --git a/src/conversion.ts b/src/conversion.ts index b07d3ca..d022ccc 100644 --- a/src/conversion.ts +++ b/src/conversion.ts @@ -1319,7 +1319,7 @@ export class Conversion { await tempOutput.start(); - const cursor = new VideoSampleCursor(track, { autoClose: false }); + const cursor = new VideoSampleCursor(track, { closeSamples: false }); using firstSample = await cursor.seekToFirst(); // Let's just use the first sample await cursor.close(); diff --git a/src/cursors.ts b/src/cursors.ts index e057a3f..4e1100d 100644 --- a/src/cursors.ts +++ b/src/cursors.ts @@ -246,7 +246,7 @@ type PendingRequest = { export type SampleTransformer = (sample: Sample) => TransformedSample; export type SampleCursorOptions = { - autoClose?: boolean; + closeSamples?: boolean; transform?: SampleTransformer; skipLiveWait?: boolean; }; @@ -257,8 +257,8 @@ const validateSampleCursorOptions = ( if (!options || typeof options !== 'object') { throw new TypeError('options must an object.'); } - if (options.autoClose !== undefined && typeof options.autoClose !== 'boolean') { - throw new TypeError('options.autoClose, when provided, must be a boolean.'); + if (options.closeSamples !== undefined && typeof options.closeSamples !== 'boolean') { + throw new TypeError('options.closeSamples, when provided, must be a boolean.'); } if (options.transform !== undefined && typeof options.transform !== 'function') { throw new TypeError('options.transform, when provided, must be a function.'); @@ -276,10 +276,11 @@ export abstract class SampleCursor< current: TransformedSample | null = null; private _transform: SampleTransformer; - private _autoClose: boolean; + private _closeSamples: boolean; private _retrievalOptions: PacketRetrievalOptions; - private _mutex = new AsyncMutex(); + /** @internal */ + _mutex = new AsyncMutex(); private _packetReader: PacketReader; private _packetCursor: PacketCursor; @@ -349,7 +350,7 @@ export abstract class SampleCursor< this._retrievalOptions = { skipLiveWait: options.skipLiveWait }; this._packetReader = new PacketReader(track); this._packetCursor = new PacketCursor(track, this._retrievalOptions); - this._autoClose = options.autoClose ?? true; + this._closeSamples = options.closeSamples ?? true; this._transform = options.transform ?? (sample => sample as unknown as TransformedSample); track.input._openSampleCursors.add(this); @@ -405,7 +406,7 @@ export abstract class SampleCursor< validateTimestamp(timestamp); return this._getSample(result => this._seekToPacket( result, - this._packetReader.getKeyAt(timestamp, this._retrievalOptions), + this._packetReader.getKeyAt(timestamp, { ...this._retrievalOptions, verifyKeyPackets: true }), )); } @@ -670,7 +671,7 @@ export abstract class SampleCursor< private _transformSample() { assert(this._currentSample && !this._currentSample.closed); - if (this._autoClose) { + if (this._closeSamples) { // Here, the transformation is memoized: repeated calls will not transform the same sample twice. return this.current ??= this._transform(this._currentSample); } else { @@ -766,11 +767,8 @@ export abstract class SampleCursor< return res.set(this._transformSample()); } - if (targetPacket.timestamp - lastTimestamp < 0.1) { - // TODO this, keep for stuff like prores? Prores is fastest when every frame is just decoded directly - // - // The difference is too small for it to be worth to set up a new pump (especially relevant for - // audio tracks) + if (this.track.type === 'audio' && targetPacket.timestamp - lastTimestamp < 0.1) { + // The difference is too small for it to be worth to set up a new pump, relevant for audio tracks needsNewPump = false; } else { if (this._packetCursor.current) { diff --git a/src/misc.ts b/src/misc.ts index 8228990..f7c2cb4 100644 --- a/src/misc.ts +++ b/src/misc.ts @@ -835,7 +835,7 @@ export class ResultValue { export class AsyncMutex { locked = false; - private resolverQueue: (() => void)[] = []; + resolverQueue: (() => void)[] = []; lock() { if (!this.locked) { @@ -939,6 +939,7 @@ export class ForgivingCallSerializer { if (result instanceof Promise) { this.currentPromise = result + .catch(() => {}) .finally(() => { if (this.queuedCalls === 0) { this.currentPromise = null; diff --git a/src/source.ts b/src/source.ts index 4192d9b..ac7f580 100644 --- a/src/source.ts +++ b/src/source.ts @@ -2446,13 +2446,23 @@ class ReadOrchestrator { } dispose() { + this.disposed = true; + + const error = new InputDisposedError(); + for (const worker of this.workers) { + worker.pendingSlices.forEach(x => x.reject(error)); + } + for (const queued of this.queuedReads) { + queued.pendingSlices.forEach(x => x.reject(error)); + } + for (const worker of this.workers) { worker.aborted = true; } this.workers.length = 0; + this.queuedReads.length = 0; this.cache.length = 0; - this.disposed = true; } } diff --git a/test/browser/flac-encoder-extension.test.ts b/test/browser/flac-encoder-extension.test.ts index eda7459..c5f9c3e 100644 --- a/test/browser/flac-encoder-extension.test.ts +++ b/test/browser/flac-encoder-extension.test.ts @@ -2,7 +2,6 @@ import { expect, test } from 'vitest'; import { Input } from '../../src/input.js'; import { ALL_FORMATS } from '../../src/input-format.js'; import { AudioSampleSource } from '../../src/media-source.js'; -import { AudioSampleCursor } from '../../src/cursors.js'; import { assert } from '../../src/misc.js'; import { Output } from '../../src/output.js'; import { FlacOutputFormat } from '../../src/output-format.js'; @@ -10,6 +9,7 @@ import { AudioSample } from '../../src/sample.js'; import { BufferSource } from '../../src/source.js'; import { BufferTarget } from '../../src/target.js'; import { registerFlacEncoder } from '@mediabunny/flac-encoder'; +import { EncodedPacket, PacketReader } from '../../src/packet.js'; test('FLAC encoder, 24-bit', async () => { registerFlacEncoder(); @@ -19,15 +19,14 @@ test('FLAC encoder, 24-bit', async () => { const durationSeconds = 2; const data = createF32SineWave(sampleRate, channels, durationSeconds); - using sample = await encodeAndDecodeFirstSample(new AudioSample({ + const packet = await encodeSample(new AudioSample({ data, format: 'f32', numberOfChannels: channels, sampleRate, timestamp: 0, })); - - expect(sample.format).toBe('s32'); + expect(getBitDepthFromFlacPacket(packet!)).toBe(0b110); // 0b110 = 24 bit }); test('FLAC encoder, 16-bit', async () => { @@ -38,15 +37,14 @@ test('FLAC encoder, 16-bit', async () => { const durationSeconds = 2; const data = createS16SineWave(sampleRate, channels, durationSeconds); - using sample = await encodeAndDecodeFirstSample(new AudioSample({ + const packet = await encodeSample(new AudioSample({ data, format: 's16', numberOfChannels: channels, sampleRate, timestamp: 0, })); - - expect(sample.format).toBe('s16'); + expect(getBitDepthFromFlacPacket(packet!)).toBe(0b100); // 0b100 = 16 bit }); const createF32SineWave = (sampleRate: number, channels: number, durationSeconds: number) => { @@ -77,7 +75,7 @@ const createS16SineWave = (sampleRate: number, channels: number, durationSeconds return data; }; -const encodeAndDecodeFirstSample = async (audioSample: AudioSample) => { +const encodeSample = async (audioSample: AudioSample) => { const output = new Output({ format: new FlacOutputFormat(), target: new BufferTarget(), @@ -100,10 +98,10 @@ const encodeAndDecodeFirstSample = async (audioSample: AudioSample) => { const track = await input.getPrimaryAudioTrack(); assert(track); - const cursor = new AudioSampleCursor(track, { autoClose: false }); - const sample = await cursor.seekToFirst(); - await cursor.close(); - assert(sample); + const reader = new PacketReader(track); + return reader.getFirst(); +}; - return sample; +const getBitDepthFromFlacPacket = (packet: EncodedPacket) => { + return (packet.data[3]! & 0b1110) >> 1; }; diff --git a/test/browser/sample-cursor.test.ts b/test/browser/sample-cursor.test.ts index 3dee342..2b2cf51 100644 --- a/test/browser/sample-cursor.test.ts +++ b/test/browser/sample-cursor.test.ts @@ -373,7 +373,7 @@ test('Sample cursor sample reuse', async () => { expect(sample2!.closed).toBe(true); const cursor2 = new VideoSampleCursor(videoTrack, { - autoClose: false, + closeSamples: false, }); const sample3 = await cursor2.seekToFirst(); @@ -398,7 +398,7 @@ test('Sample cursor sample reuse', async () => { count = 0; const cursor4 = new VideoSampleCursor(videoTrack, { - autoClose: false, + closeSamples: false, transform: sample => (sample.close(), count++), }); @@ -519,7 +519,7 @@ test('Decoder pump error handling & reset', async () => { }); const videoTrack = (await input.getPrimaryVideoTrack())!; - const cursor = new VideoSampleCursor(videoTrack); + await using cursor = new VideoSampleCursor(videoTrack); cursor._debug.enabled = true; cursor._debug.throwInPump = true; @@ -859,7 +859,7 @@ test('Command queuing', async () => { await promiseAllEnsureOrder(commands7); - const cursor8 = new VideoSampleCursor(videoTrack, { autoClose: false }); + const cursor8 = new VideoSampleCursor(videoTrack, { closeSamples: false }); const firstSample = await cursor8.seekToFirst(); firstSample!.close(); @@ -878,7 +878,7 @@ test('Command queuing', async () => { await cursor8.close(); - const cursor9 = new VideoSampleCursor(videoTrack, { autoClose: false }); + const cursor9 = new VideoSampleCursor(videoTrack, { closeSamples: false }); const commands9 = [ cursor9.seekToFirst(), @@ -929,6 +929,10 @@ test('Automatic cursor disposal', async () => { input.dispose(); expect(cursor.closed).toBe(true); + + // Make sure the close is actually complete + using lock = cursor._mutex.lock(); + if (lock.pending) await lock.ready; }); test('Video with stubborn first sample emit', async () => { diff --git a/test/node/forgiving-call-serializer.test.ts b/test/node/forgiving-call-serializer.test.ts index 40c9590..6be1b56 100644 --- a/test/node/forgiving-call-serializer.test.ts +++ b/test/node/forgiving-call-serializer.test.ts @@ -56,3 +56,16 @@ test('Error handling', async () => { await expect(second).rejects.toThrow(); expect(await third).toBe(3); }); + +test('Dangling rejected call', async () => { + const serializer = new ForgivingCallSerializer(); + + const first = serializer.call(() => executeDelayed(() => { + throw new Error('yo'); + })); + + await expect(first).rejects.toThrow(); + + await new Promise(resolve => setTimeout(resolve, 20)); + // No error +});