/*! * Copyright (c) 2025-present, Vanilagy and contributors * * This Source Code Form is subject to the terms of the Mozilla Public * License, v. 2.0. If a copy of the MPL was not distributed with this * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ import { PCM_AUDIO_CODECS } from './codec'; import { AudioDecoderWrapper, DecoderWrapper, PcmAudioDecoderWrapper, VideoDecoderWrapper } from './decode'; import { InputDisposedError } from './input'; import { InputAudioTrack, InputTrack, InputVideoTrack } from './input-track'; import { assert, AsyncMutex4, AsyncMutexLock, CallSerializer2, defer, isFirefox, last, MaybePromise, polyfillSymbolDispose, promiseWithResolvers, ResultValue, Rotation, AsyncGate, Yo, isNumber, } from './misc'; import { EncodedPacket } from './packet'; import { AudioSample, clampCropRectangle, CropRectangle, validateCropRectangle, VideoSample } from './sample'; polyfillSymbolDispose(); /** * Additional options for controlling packet retrieval. * @group Media sinks * @public */ export type PacketRetrievalOptions = { /** * When set to `true`, only packet metadata (like timestamp) will be retrieved - the actual packet data will not * be loaded. */ metadataOnly?: boolean; /** * When set to true, key packets will be verified upon retrieval by looking into the packet's bitstream. * If not enabled, the packet types will be determined solely by what's stored in the containing file and may be * incorrect, potentially leading to decoder errors. Since determining a packet's actual type requires looking into * its data, this option cannot be enabled together with `metadataOnly`. */ verifyKeyPackets?: boolean; }; export const validatePacketRetrievalOptions = (options: PacketRetrievalOptions) => { if (!options || typeof options !== 'object') { throw new TypeError('options must be an object.'); } if (options.metadataOnly !== undefined && typeof options.metadataOnly !== 'boolean') { throw new TypeError('options.metadataOnly, when defined, must be a boolean.'); } if (options.verifyKeyPackets !== undefined && typeof options.verifyKeyPackets !== 'boolean') { throw new TypeError('options.verifyKeyPackets, when defined, must be a boolean.'); } if (options.verifyKeyPackets && options.metadataOnly) { throw new TypeError('options.verifyKeyPackets and options.metadataOnly cannot be enabled together.'); } }; export const validateTimestamp = (timestamp: number) => { if (!isNumber(timestamp)) { throw new TypeError('timestamp must be a number.'); // It can be non-finite, that's fine } }; export class PacketReader { track: T; constructor(track: T) { if (!(track instanceof InputTrack)) { throw new TypeError('track must be an InputTrack.'); } this.track = track; } private maybeVerifyPacketType( packet: EncodedPacket | null, options: PacketRetrievalOptions, ): MaybePromise { if (!options.verifyKeyPackets || !packet || packet.type === 'delta') { return packet; } return this.track.determinePacketType(packet).then((determinedType) => { if (determinedType) { // @ts-expect-error Technically readonly packet.type = determinedType; } return packet; }); } readFirst(options: PacketRetrievalOptions = {}): MaybePromise { validatePacketRetrievalOptions(options); if (this.track.input._disposed) { throw new InputDisposedError(); } const result = new ResultValue(); const promise = this.track._backing.getFirstPacket(result, options); if (result.pending) { return promise.then(() => this.maybeVerifyPacketType(result.value, options)); } else { return this.maybeVerifyPacketType(result.value, options); } } readAt(timestamp: number, options: PacketRetrievalOptions = {}): MaybePromise { validateTimestamp(timestamp); validatePacketRetrievalOptions(options); if (this.track.input._disposed) { throw new InputDisposedError(); } const result = new ResultValue(); const promise = this.track._backing.getPacket(result, timestamp, options); if (result.pending) { return promise.then(() => this.maybeVerifyPacketType(result.value, options)); } else { return this.maybeVerifyPacketType(result.value, options); } } readKeyAt(timestamp: number, options: PacketRetrievalOptions = {}): MaybePromise { validateTimestamp(timestamp); validatePacketRetrievalOptions(options); if (this.track.input._disposed) { throw new InputDisposedError(); } if (options.verifyKeyPackets) { return this.readKeyAtVerified(timestamp, options); } const result = new ResultValue(); const promise = this.track._backing.getKeyPacket(result, timestamp, options); if (result.pending) { return promise.then(() => result.value); } else { return result.value; } } private async readKeyAtVerified( timestamp: number, options: PacketRetrievalOptions, ): Promise { const result = new ResultValue(); const promise = this.track._backing.getKeyPacket(result, timestamp, options); if (result.pending) await promise; const packet = result.value; if (!packet) { return null; } const determinedType = await this.track.determinePacketType(packet); if (determinedType === 'delta') { // Try returning the previous key packet (in hopes that it's actually a key packet) return this.readKeyAtVerified(packet.timestamp - 1 / this.track.timeResolution, options); } return packet; } readNext(from: EncodedPacket, options: PacketRetrievalOptions = {}): MaybePromise { if (!(from instanceof EncodedPacket)) { throw new TypeError('from must be an EncodedPacket.'); } validatePacketRetrievalOptions(options); if (this.track.input._disposed) { throw new InputDisposedError(); } const result = new ResultValue(); const promise = this.track._backing.getNextPacket(result, from, options); if (result.pending) { return promise.then(() => this.maybeVerifyPacketType(result.value, options)); } else { return this.maybeVerifyPacketType(result.value, options); } } readNextKey(from: EncodedPacket, options: PacketRetrievalOptions = {}): MaybePromise { if (!(from instanceof EncodedPacket)) { throw new TypeError('from must be an EncodedPacket.'); } validatePacketRetrievalOptions(options); if (this.track.input._disposed) { throw new InputDisposedError(); } if (options.verifyKeyPackets) { return this.readNextKeyVerified(from, options); } const result = new ResultValue(); const promise = this.track._backing.getNextKeyPacket(result, from, options); if (result.pending) { return promise.then(() => result.value); } else { return result.value; } } private async readNextKeyVerified( from: EncodedPacket, options: PacketRetrievalOptions, ): Promise { const result = new ResultValue(); const promise = this.track._backing.getNextKeyPacket(result, from, options); if (result.pending) await promise; const nextPacket = result.value; if (!nextPacket) { return null; } const determinedType = await this.track.determinePacketType(nextPacket); if (determinedType === 'delta') { // Try returning the next key packet (in hopes that it's actually a key packet) return this.readNextKeyVerified(nextPacket, options); } return nextPacket; } } export class PacketCursor { track: T; current: EncodedPacket | null = null; private _reader: PacketReader; private _options: PacketRetrievalOptions; private _nextIsFirst = true; private _callSerializer = new CallSerializer2(); constructor(track: T, options: PacketRetrievalOptions = {}) { if (!(track instanceof InputTrack)) { throw new TypeError('track must be an InputTrack.'); } validatePacketRetrievalOptions(options); this.track = track; this._reader = new PacketReader(track); this._options = options; } private _seekToFirstDirect(): MaybePromise { const result = this._reader.readFirst(this._options); const onPacket = (packet: EncodedPacket | null) => { this._nextIsFirst = false; return this.current = packet; }; if (result instanceof Promise) { return result.then(onPacket); } else { return onPacket(result); } } seekToFirst(): MaybePromise { return this._callSerializer.call(() => this._seekToFirstDirect()); } seekTo(timestamp: number): MaybePromise { validateTimestamp(timestamp); return this._callSerializer.call(() => { const result = this._reader.readAt(timestamp, this._options); const onPacket = (packet: EncodedPacket | null) => { this._nextIsFirst = !packet; return this.current = packet; }; if (result instanceof Promise) { return result.then(onPacket); } else { return onPacket(result); } }); } seekToKey(timestamp: number): MaybePromise { validateTimestamp(timestamp); return this._callSerializer.call(() => { const result = this._reader.readKeyAt(timestamp, this._options); const onPacket = (packet: EncodedPacket | null) => { this._nextIsFirst = !packet; return this.current = packet; }; if (result instanceof Promise) { return result.then(onPacket); } else { return onPacket(result); } }); } next(): MaybePromise { return this._callSerializer.call(() => { if (this._nextIsFirst) { return this._seekToFirstDirect(); } if (!this.current) { return null; } const result = this._reader.readNext(this.current, this._options); const onPacket = (packet: EncodedPacket | null) => { return this.current = packet; }; if (result instanceof Promise) { return result.then(onPacket); } else { return onPacket(result); } }); } nextKey(): MaybePromise { return this._callSerializer.call(() => { if (this._nextIsFirst) { return this._seekToFirstDirect(); } if (!this.current) { return null; } const result = this._reader.readNextKey(this.current, this._options); const onPacket = (packet: EncodedPacket | null) => { return this.current = packet; }; if (result instanceof Promise) { return result.then(onPacket); } else { return onPacket(result); } }); } async iterate( callback: (packet: EncodedPacket) => MaybePromise, ) { if (typeof callback !== 'function') { throw new TypeError('callback must be a function.'); } const donePromise = this._callSerializer.waitUntilIdle(); if (donePromise) await donePromise; while (true) { if (this.current) { let result = callback(this.current); if (result instanceof Promise) result = await result; if (result === false) { break; } } const result = this.next(); if (result instanceof Promise) await result; if (!this.current) { break; } } } // eslint-disable-next-line @stylistic/generator-star-spacing async *[Symbol.asyncIterator]() { const donePromise = this._callSerializer.waitUntilIdle(); if (donePromise) await donePromise; while (true) { if (this.current) { yield this.current; } const result = this.next(); if (result instanceof Promise) await result; if (!this.current) { break; } } } waitUntilIdle() { return this._callSerializer.waitUntilIdle(); } isIdle() { return this._callSerializer.isIdle(); } } type PendingRequest = { timestamp: number; promise: Promise; resolve: (sample: T | null) => void; reject: (error: unknown) => void; successor: PendingRequest | null; }; export type SampleTransformer = (sample: Sample) => TransformedSample; export type SampleCursorOptions = { autoClose?: boolean; transform?: SampleTransformer; }; const validateSampleCursorOptions = ( options: SampleCursorOptions, ) => { 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.transform !== undefined && typeof options.transform !== 'function') { throw new TypeError('options.transform, when provided, must be a function.'); } }; export abstract class SampleCursor< Sample extends VideoSample | AudioSample, TransformedSample = Sample, > implements AsyncDisposable { track: InputTrack; current: TransformedSample | null = null; private _transform: SampleTransformer; private _autoClose: boolean; private _mutex = new AsyncMutex4(); private _packetReader: PacketReader; private _packetCursor: PacketCursor; /** @internal */ _decoder: DecoderWrapper | null = null; private _currentSample: Sample | null = null; /** Updated when _currentSample is updated, but reset when a new pump is started. */ private _currentSampleTimestamp: number | null = null; /** The queue of samples that have been decoded and are now waiting. */ private _sampleQueue: Sample[] = []; private _pendingRequests: PendingRequest[] = []; private _lastPendingRequest: PendingRequest | null = null; /** Whether the next sample is the first sample. */ private _nextIsFirst = true; private _queuedResets = 0; /** @internal */ _pumpRunning = false; /** Used to pause and resume the pump. */ private _pumpGate = new AsyncGate(); private _pumpStopQueued = false; private _pumpStopped = new AsyncGate(); /** The minimum target packet until which the pump should decode. */ private _pumpTarget: EncodedPacket | null = null; private _lastTarget: EncodedPacket | null = null; private _decoderFlushPromise: Promise | null = null; /** * When this value is above 0, the pump is instructed to be lazy: that is, only decode packets until the target and * not further to increase decoder efficiency. */ private _lazyPump = 0; private _closed = false; private _closePromise: Promise | null = null; private _error: unknown = null; private _errorSet = false; /** * A bunch of utilities for simulating certain behaviors for testing. * @internal */ _debug = { enabled: false, pumpsStarted: 0, seekPackets: [] as (EncodedPacket | null)[], decodedPackets: [] as EncodedPacket[], throwInDecoderInit: false, throwInPump: false, throwDecoderError: false, unthrottledPump: false, pumpEnded: new AsyncGate(), }; get closed(): boolean { return this._closed; } get errored(): boolean { return this._errorSet; } protected constructor( track: InputTrack, options: SampleCursorOptions, ) { this.track = track; this._packetReader = new PacketReader(track); this._packetCursor = new PacketCursor(track); this._autoClose = options.autoClose ?? true; this._transform = options.transform ?? (sample => sample as unknown as TransformedSample); track.input._openSampleCursors.add(this); const lock = this._mutex.lock(); assert(!lock.pending); void this._initDecoder() .then(decoder => this._decoder = decoder) .catch(error => this._closeWithError(error, false)) .finally(() => lock.release()); } private _getSample( callback: (result: ResultValue) => Promise, ): MaybePromise { this._ensureWillBeOpen(); try { const result = new ResultValue(); const promise = callback(result); if (result.pending) { return promise .then(() => result.value) .catch(this._closeWithErrorAndThrow.bind(this)); } else { return result.value; } } catch (error) { this._closeWithErrorAndThrow(error); } } seekToFirst(): MaybePromise { return this._getSample(result => this._seekToPacket(result, this._packetReader.readFirst())); } seekTo(timestamp: number): MaybePromise { validateTimestamp(timestamp); return this._getSample(result => this._seekToPacket(result, this._packetReader.readAt(timestamp))); } seekToKey(timestamp: number): MaybePromise { validateTimestamp(timestamp); return this._getSample(result => this._seekToPacket(result, this._packetReader.readKeyAt(timestamp))); } next(): MaybePromise { return this._getSample(result => this._nextInternal(result)); } nextKey(): MaybePromise { return this._getSample(result => this._nextKeyInternal(result)); } hasNext(): MaybePromise { this._ensureWillBeOpen(); try { const result = new ResultValue(); const promise = this._hasNextInternal(result); if (result.pending) { return promise .then(() => result.value) .catch(this._closeWithErrorAndThrow.bind(this)); } else { return result.value; } } catch (error) { this._closeWithErrorAndThrow(error); } } async iterate( callback: (sample: TransformedSample) => MaybePromise, ) { if (typeof callback !== 'function') { throw new TypeError('callback must be a function.'); } this._ensureWillBeOpen(); const waitPromise = this.waitUntilIdle(); if (waitPromise) await waitPromise; this._ensureNotClosed(); while (true) { if (this.current) { let result = callback(this.current); if (result instanceof Promise) result = await result; if (result === false) { break; } } const result = this.next(); if (result instanceof Promise) await result; if (!this.current) { break; } } } // eslint-disable-next-line @stylistic/generator-star-spacing async *[Symbol.asyncIterator]() { this._ensureWillBeOpen(); const waitPromise = this.waitUntilIdle(); if (waitPromise) await waitPromise; this._ensureNotClosed(); while (true) { if (this.current) { yield this.current; } const result = this.next(); if (result instanceof Promise) await result; if (!this.current) { break; } } } close() { return this._closePromise ??= this._closed ? Promise.resolve() : this._closeInternal(); } [Symbol.asyncDispose]() { return this.close(); } async reset() { this._lazyPump++; this._queuedResets++; using _ = defer(() => this._queuedResets--); using lock = this._mutex.lock(); if (lock.pending) await lock.ready; if (!this._closed) { await this._closeInternal(false); } // All of this should automatically be true after a close assert(!this._pumpRunning); assert(!this._pumpStopQueued); assert(!this._currentSample); assert(!this.current); assert(this._sampleQueue.length === 0); assert(this._pendingRequests.length === 0); assert(this._lastPendingRequest === null); assert(this._pumpTarget === null); assert(!this._decoder || this._decoder.closed); this._closed = false; this._closePromise = null; this._error = null; this._errorSet = false; this._nextIsFirst = true; this._lazyPump = 0; this.track.input._openSampleCursors.add(this); try { const newDecoder = await this._initDecoder(); this._decoder = newDecoder; } catch (error) { this._closeWithErrorAndThrow(error, false); } } /** * Returns a Promise that resolves when currently pending operations (at the time of calling this method) * are settled, or `null` if there are none. */ waitUntilIdle(): Promise | null { const lock = this._mutex.lock(); if (!lock.pending && this._pendingRequests.length === 0) { lock.release(); return null; } const getLastPendingPromise = () => { lock.release(); if (this._pendingRequests.length === 0) { return; } let lastRequest = last(this._pendingRequests)!; while (lastRequest.successor) { lastRequest = lastRequest.successor; } return lastRequest.promise .catch(() => {}) .then(() => {}); }; if (lock.pending) { assert(lock.ready); return lock.ready.then(getLastPendingPromise); } else { return getLastPendingPromise() ?? null; } } isIdle(): boolean { return this._pendingRequests.length === 0 && !this._mutex.locked; } /** @internal */ abstract _initDecoder(): Promise>; protected _onDecoderSample(sample: Sample): void { try { if (this._debug.enabled && this._debug.throwDecoderError) { sample.close(); if (!this._closed) { // Let's fake a decoder error this way return this._onDecoderError(new Error('Fake decoder error!')); } else { return; } } if (this._pendingRequests.length === 0) { if (this._pumpStopQueued || !this._pumpRunning) { // Don't care about it anymore sample.close(); } else { // Let's save it for later this._sampleQueue.push(sample); } } else { let given = false; let nextInsertionIndex = 0; // Let's hand the sample to all matching requests for (let i = 0; i < this._pendingRequests.length; i++) { const request = this._pendingRequests[i]!; if (request.timestamp > sample.timestamp) { break; } this._setCurrentRaw(sample); request.resolve(this._transformSample()); this._pendingRequests.splice(i--, 1); given = true; if (request.successor) { // If the request has a successor request, "unlock" that successor and add it to the // start of the queue this._pendingRequests.splice(nextInsertionIndex, 0, request.successor); i++; nextInsertionIndex++; } } if (!given) { sample.close(); } } this._pumpGate.open(); } catch (error) { void this._closeWithError(error); } } protected _onDecoderError(error: unknown): void { void this._closeWithError(error); } protected _onDecoderDequeue() { this._pumpGate.open(); } private _setCurrentRaw(newCurrentRaw: Sample | null) { if (this._currentSample === newCurrentRaw) { return; } this._currentSample?.close(); this._currentSample = newCurrentRaw; this._currentSampleTimestamp = newCurrentRaw?.timestamp ?? null; this.current = null; } private _transformSample() { assert(this._currentSample && !this._currentSample.closed); if (this._autoClose) { // Here, the transformation is memoized: repeated calls will not transform the same sample twice. return this.current ??= this._transform(this._currentSample); } else { // Here, the transformation happens every time since the sample is also cloned every time const clone = this._currentSample.clone(); const transformed = this._transform(clone as Sample); return this.current = transformed; } } private async _seekToPacket( res: ResultValue, targetPacketPromise: MaybePromise, lock?: AsyncMutexLock, ): Promise { this._lazyPump++; if (!lock) { lock = this._mutex.lock(); if (lock.pending) await lock.ready; } using deferred = defer(() => { lock?.release(); this._lazyPump--; }); this._ensureNotClosed(); // First, let's wait for the packet to be retrieved const targetPacket = targetPacketPromise instanceof Promise ? await targetPacketPromise : targetPacketPromise; if (this._debug.enabled) { this._debug.seekPackets.push(targetPacket); } // A null packet means we're before the first packet this._nextIsFirst = !targetPacket; if (!targetPacket) { this._lastTarget = null; this._setCurrentRaw(null); return res.set(null); } if (this._currentSample?.timestamp === targetPacket.timestamp && !this._currentSample.closed) { // We can reuse the current sample, the timestamp is the same return res.set(this._transformSample()); } let needsNewPump: boolean; let lastRequest = last(this._pendingRequests) ?? null; while (lastRequest?.successor) { lastRequest = lastRequest.successor; } if (lastRequest?.timestamp === -Infinity) { // next() requests are queued, so in order to know if we need to start a new pump or not, we'll need to wait await lastRequest.promise .then(() => {}) .catch(() => {}); this._ensureNotClosed(); } const lastTimestamp = Math.max( this._lastTarget?.timestamp ?? -Infinity, this._currentSampleTimestamp ?? -Infinity, // This one's revelant if we had next() requests ); if (lastTimestamp !== -Infinity && lastTimestamp <= targetPacket.timestamp) { // First, let's see if an already-decoded sample can satisfy the request while (this._sampleQueue.length > 0) { const nextSample = this._sampleQueue.shift()!; this._pumpGate.open(); if (targetPacket.timestamp <= nextSample.timestamp) { this._setCurrentRaw(nextSample); return res.set(this._transformSample()); } else { nextSample.close(); } } if (targetPacket.timestamp - lastTimestamp < 0.1) { // The difference is too small for it to be worth to set up a new pump (especially relevant for // audio tracks) needsNewPump = false; } else { if (this._packetCursor.current) { // We need to see if the target packet is ahead of the decoder, GOP-wise let nextKey = this._packetReader.readNextKey( this._packetCursor.current, { verifyKeyPackets: true }, ); if (nextKey instanceof Promise) nextKey = await nextKey; needsNewPump = !!nextKey && targetPacket.sequenceNumber >= nextKey.sequenceNumber; } else { needsNewPump = true; } } } else { // This is the first packet or we went backwards, create a new pump needsNewPump = true; } if (needsNewPump && this._pumpRunning) { await this._stopPump(); } if (!this._pumpTarget || targetPacket.sequenceNumber > this._pumpTarget.sequenceNumber) { this._pumpTarget = targetPacket; } this._lastTarget = targetPacket; if (needsNewPump) { // Set the cursor to the right spot const result = this._packetCursor.seekToKey(targetPacket.timestamp); if (result instanceof Promise) await result; // Start the new pump void this._runPump(); } this._ensureNotClosed(); // Add the request to the queue const request = promiseWithResolvers(); const pendingRequest: PendingRequest = { timestamp: targetPacket.timestamp, promise: request.promise, resolve: request.resolve, reject: request.reject, successor: null, }; this._pendingRequests.push(pendingRequest); this._lastPendingRequest = pendingRequest; this._pumpGate.open(); deferred.execute(); // Waiting for the return would be too long return res.set(await request.promise); } private async _nextInternal(res: ResultValue): Promise { using lock = this._mutex.lock(); if (lock.pending) await lock.ready; this._ensureNotClosed(); if (this._nextIsFirst) { // Easy, just seek to the first sample // await is important so that the lock doesn't release too early return await this._seekToPacket(res, this._packetReader.readFirst(), lock); } // See if the request can be satisfied using already-decoded samples if (this._sampleQueue.length > 0) { const nextSample = this._sampleQueue.shift()!; this._pumpGate.open(); this._setCurrentRaw(nextSample); return res.set(this._transformSample()); } if (!this._pumpRunning) { // No pump is running (but the cursor isn't closed), the pump must've reached the end this._setCurrentRaw(null); return res.set(null); } assert(this._lastPendingRequest); this._ensureNotClosed(); // Instead of figuring out what the next presentation timestamp is (no easy way to do that), we simply add a // new request that can be fulfilled by *any* timestamp. This way, whatever the decoder produces next (and the // decoder is required to output samples in presentation order) is what we'll return. const request = promiseWithResolvers(); const pendingRequest: PendingRequest = { timestamp: -Infinity, promise: request.promise, resolve: request.resolve, reject: request.reject, successor: null, }; if (this._pendingRequests.length === 0) { this._pendingRequests.push(pendingRequest); } else { // The request will only get "unlocked" when the previous request is fulfilled this._lastPendingRequest.successor = pendingRequest; } this._lastPendingRequest = pendingRequest; this._pumpGate.open(); lock.release(); // Waiting for the return would be too long return res.set(await request.promise); } private async _nextKeyInternal(res: ResultValue): Promise { using lock = this._mutex.lock(); if (lock.pending) await lock.ready; this._ensureNotClosed(); if (this._nextIsFirst) { // await is important so that the lock doesn't release too early return await this._seekToPacket(res, this._packetReader.readFirst(), lock); } let timestampToCheck: number; const lastPendingRequest = last(this._pendingRequests); if (lastPendingRequest && !lastPendingRequest.successor) { timestampToCheck = lastPendingRequest.timestamp; } else { if (lastPendingRequest?.successor) { let last = lastPendingRequest.successor; while (last.successor) { last = last.successor; } await last.promise .then(() => {}) .catch(() => {}); this._ensureNotClosed(); } if (this._currentSampleTimestamp !== null) { timestampToCheck = this._currentSampleTimestamp; } else { // We're at the end return res.set(null); } } // The reason we don't just call readNextKey directly is as follows: readNextKey retrieves the next key in // *decode* order, however we want the next key in *presentation* order. We know that at least the key frames // are ascending in timestamp, so we first get the current key (based on a presentation-order search), then // get the next key after that, which will be the answer we're looking for. let key = this._packetReader.readKeyAt(timestampToCheck, { verifyKeyPackets: true }); if (key instanceof Promise) key = await key; assert(key); // Must be let nextKey = this._packetReader.readNextKey(key, { verifyKeyPackets: true }); if (nextKey instanceof Promise) nextKey = await nextKey; if (!nextKey) { this._setCurrentRaw(null); return res.set(null); } return await this._seekToPacket(res, nextKey, lock); } private async _hasNextInternal(res: ResultValue): Promise { using lock = this._mutex.lock(); if (lock.pending) await lock.ready; this._ensureNotClosed(); if (this._nextIsFirst) { let first = this._packetReader.readFirst(); if (first instanceof Promise) first = await first; return res.set(!!first); } if (!this._pumpRunning) { return res.set(false); } if (this._decoderFlushPromise) { await this._decoderFlushPromise; } return res.set(this._sampleQueue.length > 0 || this._pumpRunning); } /** * Starts the "pump process", which handles pushing packets into the decoder. It throttles itself if it is far * enough ahead and must be woken up again by the outside. It also stops itself when the outside tells it to. */ private async _runPump() { assert(this._packetCursor.current); assert(this._pumpTarget); assert(this._decoder); // Close whatever's left from the previous pump run (only relevant if the previous pump ended naturally, i.e. // it wasn't stopped) for (const sample of this._sampleQueue) { sample.close(); } this._sampleQueue.length = 0; try { this._pumpRunning = true; this._currentSampleTimestamp = null; if (this._debug.enabled) { this._debug.pumpsStarted++; } // Main loop while (this._packetCursor.current) { if (this._debug.enabled && this._debug.throwInPump) { throw new Error('Fake pump error!'); } const isAheadOfTarget = this._packetCursor.current.sequenceNumber > this._pumpTarget.sequenceNumber; const nextRequestExists = this._pendingRequests.some(x => x.successor || x.timestamp === -Infinity); if (isAheadOfTarget && !nextRequestExists) { if (this._pumpStopQueued) { break; } if (this._lazyPump > 0) { await this._pumpGate.wait(); continue; } else { // We're eager! That means even if we're past the target, we'll keep decoding samples to // prefill the sample queue to have samples ready. This is the common case when not batching // commands. } } const decodeQueueSize = this._decoder.getDecodeQueueSize(); if ( this._sampleQueue.length + decodeQueueSize >= 4 && !this._pumpStopQueued && !(this._debug.enabled && this._debug.unthrottledPump) ) { await this._pumpGate.wait(); continue; } // Send the packet to the decoder this._decoder.decode(this._packetCursor.current); if (this._debug.enabled) { this._debug.decodedPackets.push(this._packetCursor.current); } // Advance the cursor const maybePromise = this._packetCursor.next(); if (maybePromise instanceof Promise) await maybePromise; } if (!this._closed || this._pendingRequests.length > 0) { const { promise, resolve } = promiseWithResolvers(); this._decoderFlushPromise = promise; try { await this._decoder.flush(); } finally { resolve(); this._decoderFlushPromise = null; } } else { // We're closed with no pending requests, don't bother flushing what's left } // Resolve whatever requests remain with null. The reason this is correct: assume there are still pending // requests. Then the above flush() call ensured that all possible samples that we can get from the decoder // we have received. If some of the pending requests are still unsatisfied after seeing all samples we // decoded, then there is no other way for them to be solved but with null. const resolveWithNull = (request: PendingRequest) => { this._setCurrentRaw(null); // Note that this only runs if there exists at least one pending request request.resolve(null); if (request.successor) { resolveWithNull(request.successor); } }; this._pendingRequests.forEach(resolveWithNull); } catch (error) { if (!this._decoder.closed && this._pendingRequests.length > 0) { // The pump errored but the decoder is still fine, let's first flush the decoder before continuing await this._decoder.flush(); } this._pumpRunning = false; // So that close() doesn't attempt to stop the pump void this._closeWithError(error); } finally { this._pendingRequests.length = 0; this._lastPendingRequest = null; this._pumpRunning = false; this._pumpTarget = null; this._pumpStopQueued = false; this._lastTarget = null; this._pumpStopped.open(); if (this._debug.enabled) { this._debug.pumpEnded.open(); } } } private async _stopPump() { assert(this._pumpRunning); this._pumpStopQueued = true; this._pumpGate.open(); for (const sample of this._sampleQueue) { sample.close(); } this._sampleQueue.length = 0; await this._pumpStopped.wait(); } private async _closeInternal(doLock = true) { this._lazyPump++; this.track.input._openSampleCursors.delete(this); let lock: AsyncMutexLock | null = null; if (doLock) { lock = this._mutex.lock(); if (lock.pending) await lock.ready; } using _ = defer(() => lock?.release()); this._closed = true; if (this._pumpRunning) { await this._stopPump(); } for (const sample of this._sampleQueue) { sample.close(); } this._sampleQueue.length = 0; this._setCurrentRaw(null); this._decoder?.close(); this._decoder = null; } private _closeWithError(error: unknown, doLock?: boolean) { if (this._closed) { return; } this._closed = true; this._error = error; this._errorSet = true; const rejectWithError = (request: PendingRequest) => { request.reject(error); if (request.successor) { rejectWithError(request.successor); } }; this._pendingRequests.forEach(rejectWithError); return this._closeInternal(doLock); } private _closeWithErrorAndThrow(error: unknown, doLock?: boolean): never { void this._closeWithError(error, doLock); throw error; } /** Ensures that the cursor is not currently closed. */ private _ensureNotClosed() { if (this.closed) { if (this._errorSet) { throw this._error; } else { throw new Error('This cursor has been closed and can no longer be used.'); } } } /** Ensures that the cursor is either open or will be open again at some point, even if it currently closed. */ private _ensureWillBeOpen() { if (this._queuedResets > 0) { return; } this._ensureNotClosed(); } } export class VideoSampleCursor extends SampleCursor { override track!: InputVideoTrack; constructor( track: InputVideoTrack, options: SampleCursorOptions = {}, ) { if (!(track instanceof InputVideoTrack)) { throw new TypeError('track must be an InputVideoTrack.'); } validateSampleCursorOptions(options); super(track, options); } /** @internal */ override async _initDecoder(): Promise> { if (!(await this.track.canDecode())) { throw new Error( 'This video track cannot be decoded by this browser. Make sure to check decodability before using' + ' a track.', ); } if (this._debug.enabled && this._debug.throwInDecoderInit) { throw new Error('Fake decoder init error!'); } const decoderConfig = await this.track.getDecoderConfig(); assert(decoderConfig); assert(this.track.codec); const decoder = new VideoDecoderWrapper( sample => this._onDecoderSample(sample), error => this._onDecoderError(error), this.track.codec, decoderConfig, this.track.rotation, this.track.timeResolution, ); decoder.onDequeue = () => this._onDecoderDequeue(); return decoder; } } export class AudioSampleCursor extends SampleCursor { override track!: InputAudioTrack; constructor( track: InputAudioTrack, options: SampleCursorOptions = {}, ) { if (!(track instanceof InputAudioTrack)) { throw new TypeError('track must be an InputAudioTrack.'); } validateSampleCursorOptions(options); super(track, options); } /** @internal */ override async _initDecoder(): Promise> { if (!(await this.track.canDecode())) { throw new Error( 'This audio track cannot be decoded by this browser. Make sure to check decodability before using' + ' a track.', ); } if (this._debug.enabled && this._debug.throwInDecoderInit) { throw new Error('Fake decoder init error!'); } const codec = this.track.codec; const decoderConfig = await this.track.getDecoderConfig(); assert(codec && decoderConfig); let decoder: AudioDecoderWrapper | PcmAudioDecoderWrapper; if ((PCM_AUDIO_CODECS as readonly string[]).includes(decoderConfig.codec)) { decoder = new PcmAudioDecoderWrapper( sample => this._onDecoderSample(sample), error => this._onDecoderError(error), decoderConfig, ); } else { decoder = new AudioDecoderWrapper( sample => this._onDecoderSample(sample), error => this._onDecoderError(error), codec, decoderConfig, ); } decoder.onDequeue = () => this._onDecoderDequeue(); return decoder; } } /** * A canvas with additional timing information (timestamp & duration). * @public */ export class WrappedCanvas { /** A canvas element or offscreen canvas. */ canvas: HTMLCanvasElement | OffscreenCanvas; /** The timestamp of the corresponding video sample, in seconds. */ timestamp: number; /** The duration of the corresponding video sample, in seconds. */ duration: number; constructor(canvas: HTMLCanvasElement | OffscreenCanvas, timestamp: number, duration: number) { this.canvas = canvas; this.timestamp = timestamp; this.duration = duration; } }; /** * Options for constructing a canvas transformer to be used with {@link VideoSampleCursor}. * @public */ export type CanvasTransformerOptions = { /** * Whether the output canvases should have transparency instead of a black background. Defaults to `false`. Set * this to `true` when reading transparent videos. */ alpha?: boolean; /** * The width of the output canvas in pixels, defaulting to the display width of the video track. If height is not * set, it will be deduced automatically based on aspect ratio. */ width?: number; /** * The height of the output canvas in pixels, defaulting to the display height of the video track. If width is not * set, it will be deduced automatically based on aspect ratio. */ height?: number; /** * The fitting algorithm in case both width and height are set. * * - `'fill'` will stretch the image to fill the entire box, potentially altering aspect ratio. * - `'contain'` will contain the entire image within the box while preserving aspect ratio. This may lead to * letterboxing. * - `'cover'` will scale the image until the entire box is filled, while preserving aspect ratio. */ fit?: 'fill' | 'contain' | 'cover'; /** * The clockwise rotation by which to rotate the raw video frame. Defaults to the rotation set in the file metadata. * Rotation is applied before resizing. */ rotation?: Rotation; /** * Specifies the rectangular region of the input video to crop to. The crop region will automatically be clamped to * the dimensions of the input video track. Cropping is performed after rotation but before resizing. */ crop?: CropRectangle; /** * When set, specifies the number of canvases in the pool. These canvases will be reused in a ring buffer / * round-robin type fashion. This keeps the amount of allocated VRAM constant and relieves the browser from * constantly allocating/deallocating canvases. A pool size of 0 or `undefined` disables the pool and means a new * canvas is created each time. */ poolSize?: number; }; export const canvasTransformer = ( options: CanvasTransformerOptions = {}, ): SampleTransformer => { if (options && typeof options !== 'object') { throw new TypeError('options must be an object.'); } if (options.alpha !== undefined && typeof options.alpha !== 'boolean') { throw new TypeError('options.alpha, when provided, must be a boolean.'); } if (options.width !== undefined && (!Number.isInteger(options.width) || options.width <= 0)) { throw new TypeError('options.width, when defined, must be a positive integer.'); } if (options.height !== undefined && (!Number.isInteger(options.height) || options.height <= 0)) { throw new TypeError('options.height, when defined, must be a positive integer.'); } if (options.fit !== undefined && !['fill', 'contain', 'cover'].includes(options.fit)) { throw new TypeError('options.fit, when provided, must be one of "fill", "contain", or "cover".'); } if ( options.width !== undefined && options.height !== undefined && options.fit === undefined ) { throw new TypeError( 'When both options.width and options.height are provided, options.fit must also be provided.', ); } if (options.rotation !== undefined && ![0, 90, 180, 270].includes(options.rotation)) { throw new TypeError('options.rotation, when provided, must be 0, 90, 180 or 270.'); } if (options.crop !== undefined) { validateCropRectangle(options.crop, 'options.'); } if ( options.poolSize !== undefined && (typeof options.poolSize !== 'number' || !Number.isInteger(options.poolSize) || options.poolSize < 0) ) { throw new TypeError('poolSize must be a non-negative integer.'); } let needsSetup = true; let alpha: boolean; let width: number; let height: number; let fit: 'fill' | 'contain' | 'cover'; let rotation: Rotation; let crop: { left: number; top: number; width: number; height: number } | undefined; let canvasPool: (HTMLCanvasElement | OffscreenCanvas | null)[]; let nextCanvasIndex = 0; return (sample) => { if (needsSetup) { rotation = options.rotation ?? sample.rotation; const [rotatedWidth, rotatedHeight] = rotation % 180 === 0 ? [sample.codedWidth, sample.codedHeight] : [sample.codedHeight, sample.codedWidth]; crop = options.crop; if (crop) { clampCropRectangle(crop, rotatedWidth, rotatedHeight); } [width, height] = crop ? [crop.width, crop.height] : [rotatedWidth, rotatedHeight]; const originalAspectRatio = width / height; // If width and height aren't defined together, deduce the missing value using the aspect ratio if (options.width !== undefined && options.height === undefined) { width = options.width; height = Math.round(width / originalAspectRatio); } else if (options.width === undefined && options.height !== undefined) { height = options.height; width = Math.round(height * originalAspectRatio); } else if (options.width !== undefined && options.height !== undefined) { width = options.width; height = options.height; } alpha = options.alpha ?? false; fit = options.fit ?? 'fill'; canvasPool = Array.from({ length: options.poolSize ?? 0 }, () => null); needsSetup = false; } let canvas = canvasPool[nextCanvasIndex]; let canvasIsNew = false; if (!canvas) { if (typeof document !== 'undefined') { // Prefer an HTMLCanvasElement canvas = document.createElement('canvas'); canvas.width = width; canvas.height = height; } else { canvas = new OffscreenCanvas(width, height); } if (canvasPool.length > 0) { canvasPool[nextCanvasIndex] = canvas; } canvasIsNew = true; } if (canvasPool.length > 0) { nextCanvasIndex = (nextCanvasIndex + 1) % canvasPool.length; } const context = canvas.getContext('2d', { alpha: alpha || isFirefox(), // Firefox has VideoFrame glitches with opaque canvases }) as CanvasRenderingContext2D | OffscreenCanvasRenderingContext2D; assert(context); context.resetTransform(); if (!canvasIsNew) { if (!alpha && isFirefox()) { context.fillStyle = 'black'; context.fillRect(0, 0, width, height); } else { context.clearRect(0, 0, width, height); } } sample.drawWithFit(context, { fit, rotation, crop }); sample.close(); return new WrappedCanvas(canvas, sample.timestamp, sample.duration); }; }; /** * An AudioBuffer with additional timing information (timestamp & duration). * @public */ export type WrappedAudioBuffer = { /** An AudioBuffer. */ buffer: AudioBuffer; /** The timestamp of the corresponding audio sample, in seconds. */ timestamp: number; /** The duration of the corresponding audio sample, in seconds. */ duration: number; }; export const audioBufferTransformer = (): SampleTransformer => { return (sample) => { const result: WrappedAudioBuffer = { buffer: sample.toAudioBuffer(), timestamp: sample.timestamp, duration: sample.duration, }; sample.close(); return result; }; };