From ae5a5838d63acc26a9ef7dfd8acf38e85806001b Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Wed, 22 Jul 2026 15:30:37 +0200 Subject: [PATCH] Add ability to pause conversions or step them deliberately --- docs/guide/converting-media-files.md | 73 +++++- docs/guide/quick-start.md | 42 ++++ src/conversion.ts | 321 ++++++++++++++++++++------- src/index.ts | 1 + test/browser/conversion.test.ts | 99 +++++++++ 5 files changed, 451 insertions(+), 85 deletions(-) diff --git a/docs/guide/converting-media-files.md b/docs/guide/converting-media-files.md index 8136295..169131e 100644 --- a/docs/guide/converting-media-files.md +++ b/docs/guide/converting-media-files.md @@ -114,6 +114,43 @@ await conversion.cancel(); // Resolves once the conversion is canceled This automatically frees up all resources used by the conversion process and will cause any ongoing call to `execute` to throw a `ConversionCanceledError`. +If the conversion is [composable](#composable-conversions), the corresponding `Output` is not canceled and remains usable after conversion cancellation. + +### Pausing a conversion + +You can pause a conversion mid-execution and resume it later. For this, pass a pause signal to the `execute` method: +```ts +const controller = new AbortController(); +button.onclick = () => controller.abort(); + +await conversion.execute({ + pauseSignal: controller.signal, +}); + +if (conversion.state === 'idle') { + // Paused before completion +} else if (conversion.state === 'done') { + // Ran to completion +} +``` + +An unfinished conversion can simply be resumed with another call to `execute`: +```ts +await conversion.execute(); +``` + +### Partial execution + +Instead of running a conversion in full, you can execute it only until a certain timestamp is reached: +```ts +await conversion.execute({ + // Pauses execution once an output timestamp of 10 seconds is reached + until: 10, +}); +``` + +The conversion can then be resumed and continued by calling `execute` again. This feature is especially useful for [composable conversions](#composable-conversions). + ## Video options You can set the `video` property in the conversion options to configure the converter's behavior for video tracks. The options are: @@ -555,9 +592,41 @@ await Promise.all([ await output.finalize(); ``` -### Cancellation +### Running in lockstep -[Canceling](#canceling-a-conversion) a composable conversion does *not* cancel the output, it only stops media data from being added and closes its tracks. For a full abort, you must cancel the output manually. +To prevent high memory usage due to buffering needs, it's important to add media data at roughly the same speed across all tracks. To achieve this, you can step the conversion deliberately by calling `execute` multiple times: +```ts +await output.start(); + +for (using sample of generateAudioSamples()) { + await audioSource.add(sample); + await conversion.execute({ until: sample.timestamp }); +} + +// Convert whatever's left +await conversion.execute(); + +await output.finalize(); +``` + +When running multiple composable conversions that target the same output, you can use a pattern like this: + +```ts +await output.start(); + +for (let until = 1; true; until += 1) { + await Promise.all([ + conversion1.execute({ until }), + conversion2.execute({ until }), + ]); + + if (conversion1.state === 'done' && conversion2.state === 'done') { + break; + } +} + +await output.finalize(); +``` ## Converting live streams diff --git a/docs/guide/quick-start.md b/docs/guide/quick-start.md index cfd85ee..c0a62e5 100644 --- a/docs/guide/quick-start.md +++ b/docs/guide/quick-start.md @@ -627,6 +627,48 @@ await conversion.execute(); // Conversion is complete ``` +## Combine multiple files into one + +```ts +import { + Input, + Output, + Conversion, +} from 'mediabunny'; + +// Let's take the video track from one file... +const videoInput = new Input(...); +// ...and the audio track from another +const audioInput = new Input(...); + +const output = new Output(...); + +const videoConversion = await Conversion.init({ + input: videoInput, + output, + composable: true, // Ensure the conversion doesn't own the output + audio: { discard: true }, +}); +const audioConversion = await Conversion.init({ + input: audioInput, + output, + composable: true, + video: { discard: true }, +}); + +await output.start(); +await Promise.all([ + videoConversion.execute(), + audioConversion.execute(), +]); +await output.finalize(); +// Conversion is complete +``` + +::: info +See [Composable conversions](./converting-media-files#composable-conversions) for the full documentation. +::: + ## Reading HLS playlists ```ts diff --git a/src/conversion.ts b/src/conversion.ts index d7cd8ca..d9ba4b0 100644 --- a/src/conversion.ts +++ b/src/conversion.ts @@ -539,6 +539,33 @@ export type DiscardedTrack = { trackOptions: ConversionVideoOptions | ConversionAudioOptions; }; +/** + * Options for controlling a single call to {@link Conversion.execute}. + * @group Conversion + * @public + */ +export type ConversionExecuteOptions = { + /** + * The timestamp in seconds, in the output's timescale, until which the conversion should advance. Defaults to + * `Infinity`, meaning the conversion runs until the end. + * + * This field is especially useful for composable conversions, as it allows you to advance the conversion in + * lockstep with other media data sources. + */ + until?: number; + /** + * A signal that, when triggered, pauses the conversion as soon as possible. + */ + pauseSignal?: AbortSignal; +}; + +type TrackPump = { + done: boolean; + resolvers: ReturnType>; + wake: (() => void) | null; + start: () => void; +}; + /** * Represents a media file conversion process, used to convert one media file into another. In addition to conversion, * this class can be used to resize and rotate video, resample audio, drop tracks, or trim to a specific time range. @@ -551,6 +578,16 @@ export class Conversion { /** The output file. */ readonly output: Output; + /** + * The current state of the conversion. + * + * - `'idle'`: The conversion is not currently executing and isn't done; `execute` can be called. + * - `'executing'`: A call to `execute` is currently running. + * - `'canceled'`: The conversion has been canceled and can no longer be executed. + * - `'done'`: The conversion has run to completion. Subsequent calls to `execute` do nothing. + */ + state: 'idle' | 'executing' | 'canceled' | 'done' = 'idle'; + /** @internal */ _options: ConversionOptions; /** @internal */ @@ -566,28 +603,25 @@ export class Conversion { _outputOwnTrackGroups: (OutputTrackGroup | null)[] = []; /** @internal */ - _trackPromises: Promise[] = []; + _trackPumps: TrackPump[] = []; /** @internal */ _composable = false; - /** @internal */ - _started: Promise; - /** @internal */ - _start: () => void; /** @internal */ _executed = false; + /** @internal */ + _executionUntil = Infinity; + /** @internal */ + _pauseRequested = false; /** @internal */ - _synchronizer = new TrackSynchronizer(); + _synchronizer = new TrackSynchronizer(this); /** @internal */ _totalDuration: number | null = null; /** @internal */ _maxTimestamps = new Map(); // Track ID -> timestamp - /** @internal */ - _canceled = false; - /** * A callback that is fired whenever the conversion progresses. Gets passed as first argument a number between * 0 and 1, indicating the completion of the conversion. Note that a progress of 1 doesn't necessarily mean the @@ -731,10 +765,6 @@ export class Conversion { this._composable = composable; this.input = options.input; this.output = options.output; - - const { promise: started, resolve: start } = promiseWithResolvers(); - this._started = started; - this._start = start; } /** @internal */ @@ -1076,11 +1106,24 @@ export class Conversion { } /** - * Executes the conversion process. Resolves once conversion is complete. + * Executes the conversion process and resolves when the conversion is complete. When + * {@link ConversionExecuteOptions.until} is provided, the conversion will be suspended once that output timestamp + * is reached and can be resumed with another call to `execute`. An ongoing execution may also be suspended via + * {@link ConversionExecuteOptions.pauseSignal}. * - * Will throw if `isValid` is `false`. + * Execution will throw if `isValid` is `false`. */ - async execute() { + async execute(options: ConversionExecuteOptions = {}) { + if (!options || typeof options !== 'object') { + throw new TypeError('options must be an object.'); + } + if (options.until !== undefined && (typeof options.until !== 'number' || Number.isNaN(options.until))) { + throw new TypeError('options.until, when provided, must be a number.'); + } + if (options.pauseSignal !== undefined && !(options.pauseSignal instanceof AbortSignal)) { + throw new TypeError('options.pauseSignal, when provided, must be an AbortSignal.'); + } + if (!this.isValid) { throw new Error( 'Cannot execute this conversion because its output configuration is invalid. Make sure to always check' @@ -1089,6 +1132,19 @@ export class Conversion { ); } + if (this.state === 'executing') { + throw new Error('Cannot call execute() while a previous call to execute() is still running.'); + } + + if (this.state === 'canceled') { + throw new ConversionCanceledError(); + } + + if (this.state === 'done') { + // The conversion already ran to completion, nothing left to do + return; + } + if (this._composable && this.output.state === 'pending') { throw new Error( 'A composable conversion requires the output to be started. Call start() on the output before executing' @@ -1096,68 +1152,103 @@ export class Conversion { ); } - if (this._executed) { - throw new Error('Conversion cannot be executed twice.'); - } - this._executed = true; + this.state = 'executing'; + this._executionUntil = options.until ?? Infinity; + this._pauseRequested = options.pauseSignal?.aborted ?? false; - for (const id of this._outputTrackIds) { - this._synchronizer.declareTrack(id); - } - - if (this.onProgress) { - // Compute duration using only the utilized tracks - const uniqueUtilizedTracks = new Set(this.utilizedTracks); - const durationPromises = [...uniqueUtilizedTracks].map(async (track) => { - if (await track.isLive()) { - return Infinity; // Upper bound (assuming no universe heat death) - } - - return (await track.getDurationFromMetadata()) ?? (await track.computeDuration()); - }); - const duration = Math.max(0, ...await Promise.all(durationPromises)); - - this._computeProgress = true; - this._totalDuration = Math.min( - duration - this._startTimestamp, - this._endTimestamp - this._startTimestamp, - ); - - for (const id of this._outputTrackIds) { - this._maxTimestamps.set(id, 0); + const onPause = () => { + if (this.state !== 'executing') { + return; } - this.onProgress?.(0, 0); + this._pauseRequested = true; + + // Release any pumps stuck in the synchronizer so they can reach their next checkpoint and suspend + this._synchronizer.resolveAll(); + }; + options.pauseSignal?.addEventListener('abort', onPause); + + for (const pump of this._trackPumps) { + if (!pump.done) { + pump.resolvers = promiseWithResolvers(); + } } - if (!this._composable) { - await this.output.start(); - } + if (!this._executed) { + this._executed = true; - this._start(); + for (const id of this._outputTrackIds) { + this._synchronizer.declareTrack(id); + } + + if (this.onProgress) { + // Compute duration using only the utilized tracks + const uniqueUtilizedTracks = new Set(this.utilizedTracks); + const durationPromises = [...uniqueUtilizedTracks].map(async (track) => { + if (await track.isLive()) { + return Infinity; // Upper bound (assuming no universe heat death) + } + + return (await track.getDurationFromMetadata()) ?? (await track.computeDuration()); + }); + const duration = Math.max(0, ...await Promise.all(durationPromises)); + + this._computeProgress = true; + this._totalDuration = Math.min( + duration - this._startTimestamp, + this._endTimestamp - this._startTimestamp, + ); + + for (const id of this._outputTrackIds) { + this._maxTimestamps.set(id, 0); + } + + this.onProgress?.(0, 0); + } + + if (!this._composable) { + await this.output.start(); + } + + for (const pump of this._trackPumps) { + pump.start(); + } + } else { + // Wake all suspended track pumps + for (const pump of this._trackPumps) { + pump.wake?.(); + } + } try { - await Promise.all(this._trackPromises); + await Promise.all(this._trackPumps.map(x => x.resolvers.promise)); } catch (error) { - if (!this._canceled) { + if ((this.state as Conversion['state']) !== 'canceled') { // Make sure to cancel to stop other encoding processes and clean up resources void this.cancel(); } throw error; + } finally { + options.pauseSignal?.removeEventListener('abort', onPause); } - if (this._canceled) { + if ((this.state as Conversion['state']) === 'canceled') { throw new ConversionCanceledError(); } - if (!this._composable) { - await this.output.finalize(); - } + const isDone = this._trackPumps.every(x => x.done); + this.state = isDone ? 'done' : 'idle'; - if (this._computeProgress) { - const minTimestamp = Math.min(...this._maxTimestamps.values()); - this.onProgress?.(1, minTimestamp); + if (isDone) { + if (!this._composable) { + await this.output.finalize(); + } + + if (this._computeProgress) { + const minTimestamp = Math.min(...this._maxTimestamps.values()); + this.onProgress?.(1, minTimestamp); + } } } @@ -1166,16 +1257,23 @@ export class Conversion { * Does nothing if the conversion is already complete. */ async cancel() { - if (this.output.state === 'finalizing' || this.output.state === 'finalized') { + if (this.state === 'done') { return; } - if (this._canceled) { + if (this.state === 'canceled') { Logging._warn('Conversion already canceled.'); return; } - this._canceled = true; + this.state = 'canceled'; + + // Wake all suspended track pumps so they can wind down + for (const pump of this._trackPumps) { + pump.wake?.(); + } + + this._synchronizer.resolveAll(); if (!this._composable) { await this.output.cancel(); @@ -1260,15 +1358,13 @@ export class Conversion { const source = new EncodedVideoPacketSource(sourceCodec); videoSource = source; - this._trackPromises.push((async () => { - await this._started; - + this._registerTrackPump(async (pump) => { const sink = new EncodedPacketSink(track); const decoderConfig = await track.getDecoderConfig(); const meta: EncodedVideoChunkMetadata = { decoderConfig: decoderConfig ?? undefined }; for await (const packet of sink.packets(undefined, undefined, { verifyKeyPackets: true })) { - if (this._canceled) { + if (this.state === 'canceled') { break; } @@ -1290,11 +1386,13 @@ export class Conversion { if (this._synchronizer.shouldWait(outputTrackId, modifiedPacket.timestamp)) { await this._synchronizer.wait(modifiedPacket.timestamp); } + + await this._checkpoint(pump, modifiedPacket.timestamp); } source.close(); this._synchronizer.closeTrack(outputTrackId); - })()); + }); } else { // We need to decode & reencode the video @@ -1420,13 +1518,11 @@ export class Conversion { const source = new VideoSampleSource(encodingConfig); videoSource = source; - this._trackPromises.push((async () => { - await this._started; - + this._registerTrackPump(async (pump) => { const sink = new VideoSampleSink(track); for await (using sample of sink.samples(this._startTimestamp, this._endTimestamp)) { - if (this._canceled) { + if (this.state === 'canceled') { break; } @@ -1441,12 +1537,14 @@ export class Conversion { if (this._synchronizer.shouldWait(outputTrackId, lastSampleTimestamp)) { await this._synchronizer.wait(lastSampleTimestamp); } + + await this._checkpoint(pump, lastSampleTimestamp); } } source.close(); this._synchronizer.closeTrack(outputTrackId); - })()); + }); } let ownGroup: OutputTrackGroup | null = null; @@ -1515,15 +1613,13 @@ export class Conversion { const source = new EncodedAudioPacketSource(sourceCodec); audioSource = source; - this._trackPromises.push((async () => { - await this._started; - + this._registerTrackPump(async (pump) => { const sink = new EncodedPacketSink(track); const decoderConfig = await track.getDecoderConfig(); const meta: EncodedAudioChunkMetadata = { decoderConfig: decoderConfig ?? undefined }; for await (const packet of sink.packets()) { - if (this._canceled) { + if (this.state === 'canceled') { break; } @@ -1542,11 +1638,13 @@ export class Conversion { if (this._synchronizer.shouldWait(outputTrackId, modifiedPacket.timestamp)) { await this._synchronizer.wait(modifiedPacket.timestamp); } + + await this._checkpoint(pump, modifiedPacket.timestamp); } source.close(); this._synchronizer.closeTrack(outputTrackId); - })()); + }); } else { // We need to decode & reencode the audio @@ -1639,12 +1737,10 @@ export class Conversion { const source = new AudioSampleSource(encodingConfig); audioSource = source; - this._trackPromises.push((async () => { - await this._started; - + this._registerTrackPump(async (pump) => { const sink = new AudioSampleSink(track); for await (using sample of sink.samples(this._startTimestamp, this._endTimestamp)) { - if (this._canceled) { + if (this.state === 'canceled') { break; } @@ -1668,7 +1764,9 @@ export class Conversion { sampleRate: originalSampleRate, timestamp: 0, }); - await this._registerAudioSample(silentSample, source, outputTrackId, () => lastSampleTimestamp); + await this._registerAudioSample( + pump, silentSample, source, outputTrackId, () => lastSampleTimestamp, + ); needsPadding = false; } @@ -1704,12 +1802,14 @@ export class Conversion { // Offset the timestamp as needed finalSample.setTimestamp(finalSample.timestamp - this._startTimestamp); - await this._registerAudioSample(finalSample, source, outputTrackId, () => lastSampleTimestamp); + await this._registerAudioSample( + pump, finalSample, source, outputTrackId, () => lastSampleTimestamp, + ); } source.close(); this._synchronizer.closeTrack(outputTrackId); - })()); + }); } let ownGroup: OutputTrackGroup | null = null; @@ -1735,6 +1835,7 @@ export class Conversion { /** @internal */ async _registerAudioSample( + pump: TrackPump, sample: AudioSample, source: AudioSampleSource, outputTrackId: number, @@ -1750,6 +1851,39 @@ export class Conversion { if (this._synchronizer.shouldWait(outputTrackId, lastSampleTimestamp)) { await this._synchronizer.wait(lastSampleTimestamp); } + + await this._checkpoint(pump, lastSampleTimestamp); + } + } + + /** @internal */ + _registerTrackPump(fn: (pump: TrackPump) => Promise) { + const pump: TrackPump = { + done: false, + resolvers: promiseWithResolvers(), + wake: null, + start: () => { + void fn(pump).then(() => { + pump.done = true; + pump.resolvers.resolve(); + }, (error) => { + pump.resolvers.reject(error); + }); + }, + }; + + this._trackPumps.push(pump); + } + + /** @internal */ + async _checkpoint(pump: TrackPump, timestamp: number) { + while (this.state !== 'canceled' && (timestamp >= this._executionUntil || this._pauseRequested)) { + // We've reached the target; signal it and suspend until the next execution wakes us up + pump.resolvers.resolve(); + + const { promise, resolve } = promiseWithResolvers(); + pump.wake = resolve; + await promise; } } @@ -1797,12 +1931,17 @@ const MAX_TIMESTAMP_GAP = 1; // in seconds * slowest consumer. */ class TrackSynchronizer { + conversion: Conversion; maxTimestamps = new Map(); // Track ID -> timestamp resolvers: { timestamp: number; resolve: () => void; }[] = []; + constructor(conversion: Conversion) { + this.conversion = conversion; + } + declareTrack(trackId: number) { this.maxTimestamps.set(trackId, 0); } @@ -1814,6 +1953,15 @@ class TrackSynchronizer { this.maxTimestamps.set(trackId, Math.max(timestamp, currentValue)); const newMin = this.computeMinAndMaybeResolve(); + if ( + this.conversion.state === 'canceled' + || this.conversion._pauseRequested + || timestamp >= this.conversion._executionUntil + ) { + // No point in throttling consumers that are about to suspend or wind down anyway + return false; + } + return timestamp - newMin > MAX_TIMESTAMP_GAP; // Should wait if it is too far ahead of the slowest consumer } @@ -1833,6 +1981,13 @@ class TrackSynchronizer { this.computeMinAndMaybeResolve(); } + resolveAll() { + for (const entry of this.resolvers) { + entry.resolve(); + } + this.resolvers.length = 0; + } + computeMinAndMaybeResolve() { let newMin = Infinity; for (const [, timestamp] of this.maxTimestamps) { diff --git a/src/index.ts b/src/index.ts index 1e41cd4..8933d0e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -287,6 +287,7 @@ export { type ConversionOptions, type ConversionVideoOptions, type ConversionAudioOptions, + type ConversionExecuteOptions, ConversionCanceledError, type DiscardedTrack, } from './conversion'; diff --git a/test/browser/conversion.test.ts b/test/browser/conversion.test.ts index cb998e3..955ab0a 100644 --- a/test/browser/conversion.test.ts +++ b/test/browser/conversion.test.ts @@ -618,6 +618,7 @@ test('Canceling a composable conversion leaves the output usable', async () => { const executePromise = conversion.execute(); void conversion.cancel(); + expect(conversion.state).toBe('canceled'); await expect(executePromise).rejects.toBeInstanceOf(ConversionCanceledError); @@ -660,3 +661,101 @@ test('Track capacity works correctly with composable conversions', async () => { // WAVE allows only one track in total, so the total-count check fires before the per-type one expect(conversion.discardedTracks[0]!.reason).toBe('max_track_count_reached'); }); + +test('Blank execute', async () => { + using input = new Input({ + source: new UrlSource('/video.mp4'), + formats: ALL_FORMATS, + }); + + const output = new Output({ format: new Mp4OutputFormat(), target: new BufferTarget() }); + const conversion = await Conversion.init({ input, output }); + expect(conversion.state).toBe('idle'); + + const promise = conversion.execute(); + expect(conversion.state).toBe('executing'); + await promise; + expect(conversion.state).toBe('done'); + expect(output.state).toBe('finalized'); + + await conversion.execute(); + expect(conversion.state).toBe('done'); +}); + +test('Stepwise until', async () => { + using input = new Input({ + source: new UrlSource('/video.mp4'), + formats: ALL_FORMATS, + }); + + const output = new Output({ format: new Mp4OutputFormat(), target: new BufferTarget() }); + const conversion = await Conversion.init({ input, output }); + + await conversion.execute({ until: 2 }); + expect(conversion.state).toBe('idle'); + expect(output.state).toBe('started'); + await conversion.execute({ until: 4 }); + expect(conversion.state).toBe('idle'); + await conversion.execute({ until: 6 }); + expect(conversion.state).toBe('done'); + expect(output.state).toBe('finalized'); + + await conversion.execute(); + expect(conversion.state).toBe('done'); + + using result = new Input({ source: new BufferSource(output.target.buffer!), formats: ALL_FORMATS }); + const videoTrack = await result.getPrimaryVideoTrack(); + expect(await videoTrack!.computeDuration()).toBeGreaterThan(4); +}); + +test('Pause signal', async () => { + using input = new Input({ + source: new UrlSource('/video.mp4'), + formats: ALL_FORMATS, + }); + + const output = new Output({ format: new Mp4OutputFormat(), target: new BufferTarget() }); + const conversion = await Conversion.init({ input, output }); + + const controller = new AbortController(); + conversion.onProgress = (progress) => { + if (progress >= 0.5 && !controller.signal.aborted) { + controller.abort(); + } + }; + + await conversion.execute({ pauseSignal: controller.signal }); + expect(conversion.state).toBe('idle'); + expect(output.state).toBe('started'); + + await conversion.execute(); + expect(conversion.state).toBe('done'); + expect(output.state).toBe('finalized'); + + await conversion.execute(); + expect(conversion.state).toBe('done'); +}); + +test('Pre-signaled pause signal', async () => { + using input = new Input({ + source: new UrlSource('/video.mp4'), + formats: ALL_FORMATS, + }); + + const output = new Output({ format: new Mp4OutputFormat(), target: new BufferTarget() }); + const conversion = await Conversion.init({ input, output }); + + const controller = new AbortController(); + controller.abort(); + + await conversion.execute({ pauseSignal: controller.signal }); + expect(conversion.state).toBe('idle'); + expect(output.state).toBe('started'); + + await conversion.execute(); + expect(conversion.state).toBe('done'); + expect(output.state).toBe('finalized'); + + await conversion.execute(); + expect(conversion.state).toBe('done'); +});