From 62b04f4dcb1fa5bc9d09cb4bce41e1716f2e6384 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Mon, 1 Sep 2025 18:42:52 +0200 Subject: [PATCH 1/8] Add Remotion gold sponsor to README --- README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/README.md b/README.md index 8c846a3..5218bec 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,10 @@ Mediabunny is a JavaScript library for reading, writing, and converting media fi ### Gold sponsors
+ + Remotion + +      Gling AI From 4b3fb34a95cc15817d0687fb566cbbb5f023ef30 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 16:21:39 +0200 Subject: [PATCH 2/8] Fix running = false being set too late for ReadOrchestrator workers (fixes #84) --- dev/convert.html | 4 ++-- src/source.ts | 14 +++++++++++--- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/dev/convert.html b/dev/convert.html index ba67311..bdd82c1 100644 --- a/dev/convert.html +++ b/dev/convert.html @@ -24,7 +24,7 @@ chunked: true, chunkSize: 2**20 }); - const outputFormat = new Mediabunny.Mp4OutputFormat({}); + const outputFormat = new Mediabunny.WavOutputFormat({}); const button = document.createElement('button'); button.textContent = 'Cancel'; @@ -74,7 +74,7 @@ }, */ video: () => ({ - codec: 'avc', + //codec: 'avc', //fit: 'contain', //frameRate: 27.123, //width: 320, diff --git a/src/source.ts b/src/source.ts index d0c2e00..7e45f90 100644 --- a/src/source.ts +++ b/src/source.ts @@ -178,6 +178,8 @@ export class BlobSource extends Source { this.onread?.(worker.currentPos, worker.currentPos + value.length); this._orchestrator.supplyWorkerData(worker, value); } + + worker.running = false; } } @@ -409,6 +411,7 @@ export class UrlSource extends Source { ); } + worker.running = false; return; } @@ -418,11 +421,14 @@ export class UrlSource extends Source { if (worker.currentPos >= worker.targetPos || worker.aborted) { abortController.abort(); + worker.running = false; return; } } } + worker.running = false; + // The previous UrlSource had logic for circumventing https://issues.chromium.org/issues/436025873; I haven't // been able to observe this bug with the new UrlSource (maybe because we're using response streaming), so the // logic for that has vanished for now. Leaving a comment here if this becomes relevant again. @@ -681,6 +687,8 @@ export class StreamSource extends Source { throw new TypeError('options.read must return or resolve to a Uint8Array or a ReadableStream.'); } } + + worker.running = false; } } @@ -933,6 +941,7 @@ class ReadOrchestrator { // another one so close to it const gapTolerance = 2 ** 17; + // This check also implies worker.currentPos <= outerHole.start, a critical condition if (closedIntervalsOverlap( outerHole.start - gapTolerance, outerHole.start, worker.currentPos, worker.targetPos, @@ -1024,15 +1033,14 @@ class ReadOrchestrator { void this.options.runWorker(worker) .catch((error) => { + worker.running = false; + if (worker.pendingSlices.length > 0) { worker.pendingSlices.forEach(x => x.reject(error)); // Make sure to propagate any errors worker.pendingSlices.length = 0; } else { throw error; // So it doesn't get swallowed } - }) - .finally(() => { - worker.running = false; }); } From ef95960e135a219f4ea1e129e0f289cfe4b2dc6d Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 16:39:38 +0200 Subject: [PATCH 3/8] Add support for unsized sources, add ReadableStreamSource, fix WAVE packet lookup --- dev/demux.html | 149 +++++++++++++- docs/guide/reading-media-files.md | 80 +++++++- src/adts/adts-demuxer.ts | 7 +- src/index.ts | 2 + src/input.ts | 2 +- src/isobmff/isobmff-demuxer.ts | 6 +- src/matroska/ebml.ts | 14 +- src/matroska/matroska-demuxer.ts | 160 +++++++-------- src/mp3/mp3-demuxer.ts | 7 +- src/mp3/mp3-reader.ts | 6 +- src/ogg/ogg-demuxer.ts | 93 ++++++++- src/reader.ts | 46 ++++- src/source.ts | 318 +++++++++++++++++++++++++++--- src/wave/wave-demuxer.ts | 74 +++++-- 14 files changed, 792 insertions(+), 172 deletions(-) diff --git a/dev/demux.html b/dev/demux.html index 9cb536d..9504e3c 100644 --- a/dev/demux.html +++ b/dev/demux.html @@ -8,8 +8,137 @@ document.body.append(fileInput); fileInput.addEventListener('change', async () => { + /* + const screenStream = await navigator.mediaDevices.getDisplayMedia({ video: true }); + const micStream = await navigator.mediaDevices.getUserMedia({ audio: true }); + + const combinedStream = new MediaStream([ + ...screenStream.getTracks(), + ...micStream.getTracks() + ]); + + const recorder = new MediaRecorder(combinedStream); + + const { writable, readable } = new TransformStream({ + async transform(chunk, controller) { + const arrayBuffer = await chunk.arrayBuffer(); + controller.enqueue(new Uint8Array(arrayBuffer)); + } + }); + const writer = writable.getWriter(); + + const input = new Mediabunny.Input({ + source: new Mediabunny.ReadableStreamSource(readable), + formats: Mediabunny.ALL_FORMATS, + }); + (async () => { + const videoTrack = await input.getPrimaryVideoTrack(); + const sink = new Mediabunny.EncodedPacketSink(videoTrack); + + for await (const packet of sink.packets()) { + console.log(packet); + } + })(); + + recorder.onerror = console.error; + recorder.onstart = () => console.log("yes") + recorder.ondataavailable = async e => { + writer.write(e.data); + }; + recorder.onstop = (e) => { + writer.close(); + }; + + recorder.start(1000); + + + window.addEventListener('click', () => { + recorder.stop() + }, { once: true }) + */ + /* + + const transformStream = new TransformStream(); + const file = fileInput.files[0]; - const source = new Mediabunny.BlobSource(file); + const source = new Mediabunny.ReadableStreamSource(transformStream.readable); + + file.stream().pipeThrough(transformStream); + + const input = new Mediabunny.Input({ + formats: Mediabunny.ALL_FORMATS, + source//: new Mediabunny.BlobSource(file), + }); + + + const audioTrack = await input.getPrimaryAudioTrack(); + console.log(audioTrack); + + const sink = new Mediabunny.EncodedPacketSink(audioTrack); + + + //console.log(await sink.getPacket(100)) + + console.log(await input.computeDuration()) + */ + + /* + const screenStream = await navigator.mediaDevices.getDisplayMedia({ video: true }); + const micStream = await navigator.mediaDevices.getUserMedia({ audio: true }); + + const combinedStream = new MediaStream([ + ...screenStream.getTracks(), + ...micStream.getTracks() + ]); + + const recorder = new MediaRecorder(combinedStream); + + const { writable, readable } = new TransformStream({ + async transform(chunk, controller) { + const arrayBuffer = await chunk.arrayBuffer(); + controller.enqueue(new Uint8Array(arrayBuffer)); + } + }); + const writer = writable.getWriter(); + + const input = new Mediabunny.Input({ + source: new Mediabunny.ReadableStreamSource(readable), + formats: Mediabunny.ALL_FORMATS, + }); + (async () => { + const videoTrack = await input.getPrimaryVideoTrack(); + const sink = new Mediabunny.EncodedPacketSink(videoTrack); + + for await (const packet of sink.packets()) { + console.log(packet); + } + })(); + + recorder.onerror = console.error; + recorder.onstart = () => console.log("yes") + recorder.ondataavailable = async e => { + writer.write(e.data); + }; + recorder.onstop = (e) => { + writer.close(); + }; + + recorder.start(1000); + + window.addEventListener('click', () => { + recorder.stop() + }, { once: true }) + + //setTimeout(() => recorder.stop(), 5000) + */ + + /* + const transformStream = new TransformStream(); + + const file = fileInput.files[0]; + const source = new Mediabunny.ReadableStreamSource(transformStream.readable); + + file.stream().pipeTo(transformStream.writable); const input = new Mediabunny.Input({ formats: Mediabunny.ALL_FORMATS, @@ -17,14 +146,20 @@ }); const audioTrack = await input.getPrimaryAudioTrack(); + console.log(audioTrack); - for (let i = 0; i < 10; i++) { - console.time() - const stats = await audioTrack.computePacketStats(); - console.log(stats) - console.timeEnd() + //return; + + const sink = new Mediabunny.EncodedPacketSink(audioTrack); + + for await (const packet of sink.packets()) { + console.log(packet); } - //console.log(stats); + + console.log("done"); + */ + + //console.log(await sink.getPacket(1)) /* const videoTrack = await input.getPrimaryVideoTrack(); diff --git a/docs/guide/reading-media-files.md b/docs/guide/reading-media-files.md index 30e6753..d9af1c8 100644 --- a/docs/guide/reading-media-files.md +++ b/docs/guide/reading-media-files.md @@ -551,4 +551,82 @@ type MaybePromise = T | Promise; Specifies the prefetch profile that the reader should use with this source. A prefetch propfile specifies the pattern with which bytes outside of the requested range are preloaded to reduce latency for future reads. - `'none'` (default): No prefetching; only the data needed in the moment is requested. - `'fileSystem'`: File system-optimized prefetching: a small amount of data is prefetched bidirectionally, aligned with page boundaries. - - `'network'`: Network-optimized prefetching, or more generally, prefetching optimized for any high-latency environment: tries to minimize the amount of read calls and aggressively prefetches data when sequential access patterns are detected. \ No newline at end of file + - `'network'`: Network-optimized prefetching, or more generally, prefetching optimized for any high-latency environment: tries to minimize the amount of read calls and aggressively prefetches data when sequential access patterns are detected. + +### `ReadableStreamSource` + +This is a source backed by a `ReadableStream` of `Uint8Array`, representing an append-only byte stream of unknown length. This is the source to use for incrementally streaming in input files that are still being constructed and whose size we don't yet know. You could also use it to stream in existing files, but other sources (such as [`BlobSource`](#blobsource) or [`FilePathSource`](#filepathsource)) are recommended instead because they offer random access. + +```ts +import { ReadableStreamSource } from 'mediabunny'; + +const { writable, readable } = new TransformStream(); +const source = new ReadableStreamSource(readable); + +// Append chunks of data +const writer = writable.getWriter(); +writer.write(chunk1); +writer.write(chunk2); +writer.close(); +``` + +This source is *unsized*, meaning calls to `.getSize()` will throw and readers are more limited due to the lack of random file access. You should only use this source with sequential access patterns, such as reading all packets from start to end or doing conversions. This source does not work well with random access patterns unless you increase its max cache size. + +```ts +type ReadableStreamSourceOptions = { + // The maximum number of bytes the cache is allowed to hold + // in memory. Defaults to 16 MiB. + maxCacheSize?: number; +}; +``` + +#### Use with [`MediaRecorder`](https://developer.mozilla.org/en-US/docs/Web/API/MediaRecorder) + +You can combine `MediaRecorder` with `ReadableStreamSource` to stream recorded data into Mediabunny while the recording is taking place. Here's an example where we pipe `MediaRecorder`'s output into Mediabunny's Conversion API to create a WAVE file: +```ts +import { + Input, + Output, + Conversion, + ReadableStreamSource, + ALL_FORMATS, + WavOutputFormat, + BufferTarget, +} from 'mediabunny'; + +// Set up a TransformStream to convert MediaRecorder's Blobs into Uint8Arrays +const { writable, readable } = new TransformStream({ + async transform(chunk, controller) { + const arrayBuffer = await chunk.arrayBuffer(); + controller.enqueue(new Uint8Array(arrayBuffer)); + }, +}); + +const input = new Input({ + source: new ReadableStreamSource(readable), + formats: ALL_FORMATS, +}); +const output = new Output({ + format: new WavOutputFormat(), + target: new BufferTarget(), +}); + +const conversionPromise = Conversion.init({ input, output }) + .then(conversion => conversion.execute()); + +const micStream = await navigator.mediaDevices.getUserMedia({ audio: true }); +const recorder = new MediaRecorder(micStream); +const writer = writable.getWriter(); + +recorder.ondataavailable = e => writer.write(e.data); +recorder.onstop = async () => { + await writer.close(); + await conversionPromise; + + // Get the final .wav file + const wavFile = output.target.buffer!; // => ArrayBuffer +}; + +recorder.start(1000); +setTimeout(() => recorder.stop(), 10_000); // Stop recording after 10s +``` \ No newline at end of file diff --git a/src/adts/adts-demuxer.ts b/src/adts/adts-demuxer.ts index 30f0ef0..794160c 100644 --- a/src/adts/adts-demuxer.ts +++ b/src/adts/adts-demuxer.ts @@ -81,7 +81,7 @@ export class AdtsDemuxer extends Demuxer { return; } - if (header.startPos + header.frameLength > this.reader.fileSize) { + if (this.reader.fileSize !== null && header.startPos + header.frameLength > this.reader.fileSize) { // Frame doesn't fit in the rest of the file this.lastSampleLoaded = true; return; @@ -227,7 +227,10 @@ class AdtsAudioTrackBacking implements InputAudioTrackBacking { } else { let slice = this.demuxer.reader.requestSlice(rawSample.dataStart, rawSample.dataSize); if (slice instanceof Promise) slice = await slice; - assert(slice); + + if (!slice) { + return null; // Data didn't fit into the rest of the file + } data = readBytes(slice, rawSample.dataSize); } diff --git a/src/index.ts b/src/index.ts index 1b9e5ad..e03149e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -102,6 +102,8 @@ export { FilePathSourceOptions, StreamSource, StreamSourceOptions, + ReadableStreamSource, + ReadableStreamSourceOptions, } from './source'; export { InputFormat, diff --git a/src/input.ts b/src/input.ts index d70eea7..1136214 100644 --- a/src/input.ts +++ b/src/input.ts @@ -58,7 +58,7 @@ export class Input { /** @internal */ _getDemuxer() { return this._demuxerPromise ??= (async () => { - this._reader.fileSize = await this._source.getSize(); + this._reader.fileSize = await this._source.getSizeOrNull(); for (const format of this._formats) { const canRead = await format._canReadInput(this); diff --git a/src/isobmff/isobmff-demuxer.ts b/src/isobmff/isobmff-demuxer.ts index 0fa784d..f43c784 100644 --- a/src/isobmff/isobmff-demuxer.ts +++ b/src/isobmff/isobmff-demuxer.ts @@ -272,7 +272,7 @@ export class IsobmffDemuxer extends Demuxer { readMetadata() { return this.metadataPromise ??= (async () => { let currentPos = 0; - while (currentPos < this.reader.fileSize) { + while (true) { let slice = this.reader.requestSliceRange(currentPos, MIN_BOX_HEADER_SIZE, MAX_BOX_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -310,7 +310,7 @@ export class IsobmffDemuxer extends Demuxer { currentPos = startPos + boxInfo.totalSize; } - if (this.isFragmented) { + if (this.isFragmented && this.reader.fileSize !== null) { // The last 4 bytes may contain the size of the mfra box at the end of the file let lastWordSlice = this.reader.requestSlice(this.reader.fileSize - 4, 4); if (lastWordSlice instanceof Promise) lastWordSlice = await lastWordSlice; @@ -2449,7 +2449,7 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { } } - while (currentPos < demuxer.reader.fileSize) { + while (true) { if (prevFragment) { const trackData = prevFragment.trackData.get(this.internalTrack.id); if (trackData && trackData.startTimestamp > latestTimestamp) { diff --git a/src/matroska/ebml.ts b/src/matroska/ebml.ts index 6b14a42..9da08d4 100644 --- a/src/matroska/ebml.ts +++ b/src/matroska/ebml.ts @@ -68,6 +68,7 @@ export enum EBMLId { DocType = 0x4282, DocTypeVersion = 0x4287, DocTypeReadVersion = 0x4285, + Void = 0xec, Segment = 0x18538067, SeekHead = 0x114d9b74, Seek = 0x4dbb, @@ -554,11 +555,16 @@ export const readFloat = (slice: FileSlice, width: number) => { }; /** Returns the byte offset in the file of the next element with a matching ID. */ -export const searchForNextElementId = async (reader: Reader, startPos: number, ids: EBMLId[], until: number) => { +export const searchForNextElementId = async ( + reader: Reader, + startPos: number, + ids: EBMLId[], + until: number | null, +): Promise<{ pos: number; found: boolean }> => { const idsSet = new Set(ids); let currentPos = startPos; - while (currentPos < until) { + while (until === null || currentPos < until) { let slice = reader.requestSliceRange(currentPos, MIN_HEADER_SIZE, MAX_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -569,7 +575,7 @@ export const searchForNextElementId = async (reader: Reader, startPos: number, i } if (idsSet.has(elementHeader.id)) { - return currentPos; + return { pos: currentPos, found: true }; } assertDefinedSize(elementHeader.size); @@ -577,7 +583,7 @@ export const searchForNextElementId = async (reader: Reader, startPos: number, i currentPos = slice.filePos + elementHeader.size; } - return null; + return { pos: (until !== null && until > currentPos) ? until : currentPos, found: false }; }; /** Searches for the next occurrence of an element ID using a naive byte-wise search. */ diff --git a/src/matroska/matroska-demuxer.ts b/src/matroska/matroska-demuxer.ts index 0310131..21ba787 100644 --- a/src/matroska/matroska-demuxer.ts +++ b/src/matroska/matroska-demuxer.ts @@ -85,7 +85,7 @@ type Segment = { cuePoints: CuePoint[]; dataStartPos: number; - elementEndPos: number; + elementEndPos: number | null; clusterSeekStartPos: number; clusters: Cluster[]; @@ -234,9 +234,10 @@ export class MatroskaDemuxer extends Demuxer { readMetadata() { return this.readMetadataPromise ??= (async () => { - // Loop over all top-level elements in the file let currentPos = 0; - while (currentPos < this.reader.fileSize) { + + // Loop over all top-level elements in the file + while (true) { let slice = this.reader.requestSliceRange(currentPos, MIN_HEADER_SIZE, MAX_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -266,7 +267,15 @@ export class MatroskaDemuxer extends Demuxer { // and only segment break; } + + if (this.reader.fileSize === null) { + break; // Stop at the first segment + } } else if (id === EBMLId.Cluster) { + if (this.reader.fileSize === null) { + break; // Shouldn't be reached anyway, since we stop at the first segment + } + // Clusters are not a top-level element in Matroska, but some files contain a Segment whose size // doesn't contain any of the clusters that follow it. In the case, we apply the following logic: if // we find a top-level cluster, attribute it to the previous segment. @@ -280,7 +289,7 @@ export class MatroskaDemuxer extends Demuxer { LEVEL_0_AND_1_EBML_IDS, this.reader.fileSize, ); - size = (nextElementPos ?? this.reader.fileSize) - dataStartPos; + size = nextElementPos.pos - dataStartPos; } const lastSegment = last(this.segments); @@ -312,7 +321,7 @@ export class MatroskaDemuxer extends Demuxer { dataStartPos: segmentDataStart, elementEndPos: dataSize === null - ? await this.input.source.getSize() // Assume it goes until the end of the file + ? null // Assume it goes until the end of the file : segmentDataStart + dataSize, clusterSeekStartPos: segmentDataStart, @@ -321,10 +330,9 @@ export class MatroskaDemuxer extends Demuxer { }; this.segments.push(this.currentSegment); - let currentPos = 0; - let clusterEncountered = false; + let currentPos = segmentDataStart; - while (currentPos < this.currentSegment.elementEndPos) { + while (this.currentSegment.elementEndPos === null || currentPos < this.currentSegment.elementEndPos) { let slice = this.reader.requestSliceRange(currentPos, MIN_HEADER_SIZE, MAX_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -332,14 +340,14 @@ export class MatroskaDemuxer extends Demuxer { const elementStartPos = currentPos; const header = readElementHeader(slice); - if (!header || !LEVEL_1_EBML_IDS.includes(header.id)) { + if (!header || (!LEVEL_1_EBML_IDS.includes(header.id) && header.id !== EBMLId.Void)) { // Potential junk. Let's try to resync const nextPos = await resync( this.reader, elementStartPos, LEVEL_1_EBML_IDS, - Math.min(this.currentSegment.elementEndPos, elementStartPos + MAX_RESYNC_LENGTH), + Math.min(this.currentSegment.elementEndPos ?? Infinity, elementStartPos + MAX_RESYNC_LENGTH), ); if (nextPos) { @@ -367,93 +375,54 @@ export class MatroskaDemuxer extends Demuxer { this.readContiguousElements(slice); } } else if (id === EBMLId.Cluster) { - if (!clusterEncountered) { - clusterEncountered = true; - this.currentSegment.clusterSeekStartPos = elementStartPos; - } - } - - if (size !== null) { - currentPos = dataStartPos + size; - } - - if (this.currentSegment.infoSeen && this.currentSegment.tracksSeen && this.currentSegment.cuesSeen) { - // No need to search anymore, we have everything - break; - } - - if (this.currentSegment.seekHeadSeen) { - let hasInfo = this.currentSegment.infoSeen; - let hasTracks = this.currentSegment.tracksSeen; - let hasCues = this.currentSegment.cuesSeen; - - for (const entry of this.currentSegment.seekEntries) { - if (entry.id === EBMLId.Info) { - hasInfo = true; - } else if (entry.id === EBMLId.Tracks) { - hasTracks = true; - } else if (entry.id === EBMLId.Cues) { - hasCues = true; - } - } - - if (hasInfo && hasTracks && hasCues) { - // No need to search sequentially anymore, we can use the seek head - break; - } + this.currentSegment.clusterSeekStartPos = elementStartPos; + break; // Stop at the first cluster } if (size === null) { break; - } - } - - if (!clusterEncountered) { - const seekEntry = this.currentSegment.seekEntries.find(entry => entry.id === EBMLId.Cluster); - - if (seekEntry) { - // The seek head points us to the first cluster, nice - this.currentSegment.clusterSeekStartPos = segmentDataStart + seekEntry.segmentPosition; } else { - this.currentSegment.clusterSeekStartPos = currentPos; + currentPos = dataStartPos + size; } } - // Sort the seek entries by file position so reading them exhibits a sequential pattern - this.currentSegment.seekEntries.sort((a, b) => a.segmentPosition - b.segmentPosition); + if (this.reader.fileSize !== null) { + // Sort the seek entries by file position so reading them exhibits a sequential pattern + this.currentSegment.seekEntries.sort((a, b) => a.segmentPosition - b.segmentPosition); - // Use the seek head to read missing metadata elements - for (const seekEntry of this.currentSegment.seekEntries) { - const target = METADATA_ELEMENTS.find(x => x.id === seekEntry.id); - if (!target) { - continue; + // Use the seek head to read missing metadata elements + for (const seekEntry of this.currentSegment.seekEntries) { + const target = METADATA_ELEMENTS.find(x => x.id === seekEntry.id); + if (!target) { + continue; + } + + if (this.currentSegment[target.flag]) continue; + + let slice = this.reader.requestSliceRange( + segmentDataStart + seekEntry.segmentPosition, + MIN_HEADER_SIZE, + MAX_HEADER_SIZE, + ); + if (slice instanceof Promise) slice = await slice; + if (!slice) continue; + + const header = readElementHeader(slice); + if (!header) continue; + + const { id, size } = header; + if (id !== target.id) continue; + + assertDefinedSize(size); + + this.currentSegment[target.flag] = true; + + let dataSlice = this.reader.requestSlice(slice.filePos, size); + if (dataSlice instanceof Promise) dataSlice = await dataSlice; + if (!dataSlice) continue; + + this.readContiguousElements(dataSlice); } - - if (this.currentSegment[target.flag]) continue; - - let slice = this.reader.requestSliceRange( - segmentDataStart + seekEntry.segmentPosition, - MIN_HEADER_SIZE, - MAX_HEADER_SIZE, - ); - if (slice instanceof Promise) slice = await slice; - if (!slice) continue; - - const header = readElementHeader(slice); - if (!header) continue; - - const { id, size } = header; - if (id !== target.id) continue; - - assertDefinedSize(size); - - this.currentSegment[target.flag] = true; - - let dataSlice = this.reader.requestSlice(slice.filePos, size); - if (dataSlice instanceof Promise) dataSlice = await dataSlice; - if (!dataSlice) continue; - - this.readContiguousElements(dataSlice); } if (this.currentSegment.timestampScale === -1) { @@ -542,7 +511,7 @@ export class MatroskaDemuxer extends Demuxer { segment.elementEndPos, ); - size = (nextElementPos ?? segment.elementEndPos) - dataStartPos; + size = nextElementPos.pos - dataStartPos; } assert(id === EBMLId.Cluster); @@ -1669,7 +1638,7 @@ abstract class MatroskaTrackBacking implements InputTrackBacking { } } - while (currentPos <= segment.elementEndPos - MIN_HEADER_SIZE) { + while (segment.elementEndPos === null || currentPos <= segment.elementEndPos - MIN_HEADER_SIZE) { if (prevCluster) { const trackData = prevCluster.trackData.get(this.internalTrack.id); if (trackData && trackData.startTimestamp > latestTimestamp) { @@ -1693,14 +1662,17 @@ abstract class MatroskaTrackBacking implements InputTrackBacking { const elementStartPos = currentPos; const elementHeader = readElementHeader(slice); - if (!elementHeader || !LEVEL_1_EBML_IDS.includes(elementHeader.id)) { - // There's an element here that shouldn't be here (or Void). Might be garbage. In this case, let's + if ( + !elementHeader + || (!LEVEL_1_EBML_IDS.includes(elementHeader.id) && elementHeader.id !== EBMLId.Void) + ) { + // There's an element here that shouldn't be here. Might be garbage. In this case, let's // try and resync to the next valid element. const nextPos = await resync( demuxer.reader, elementStartPos, LEVEL_1_EBML_IDS, - Math.min(segment.elementEndPos, elementStartPos + MAX_RESYNC_LENGTH), + Math.min(segment.elementEndPos ?? Infinity, elementStartPos + MAX_RESYNC_LENGTH), ); if (nextPos) { @@ -1764,11 +1736,11 @@ abstract class MatroskaTrackBacking implements InputTrackBacking { segment.elementEndPos, ); - size = (nextElementPos ?? segment.elementEndPos) - dataStartPos; + size = nextElementPos.pos - dataStartPos; } const endPos = dataStartPos + size; - if (endPos > segment.elementEndPos - MIN_HEADER_SIZE) { + if (segment.elementEndPos !== null && endPos > segment.elementEndPos - MIN_HEADER_SIZE) { // No more elements fit in this segment break; } else { diff --git a/src/mp3/mp3-demuxer.ts b/src/mp3/mp3-demuxer.ts index 16537cc..8ab87ef 100644 --- a/src/mp3/mp3-demuxer.ts +++ b/src/mp3/mp3-demuxer.ts @@ -47,7 +47,7 @@ export class Mp3Demuxer extends Demuxer { async readMetadata() { return this.metadataPromise ??= (async () => { // Keep loading until we find the first frame header - while (!this.firstFrameHeader && this.lastLoadedPos < this.reader.fileSize) { + while (!this.firstFrameHeader && !this.lastSampleLoaded) { await this.advanceReader(); } @@ -211,7 +211,10 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking { } else { let slice = this.demuxer.reader.requestSlice(rawSample.dataStart, rawSample.dataSize); if (slice instanceof Promise) slice = await slice; - assert(slice); + + if (!slice) { + return null; // Data didn't fit into the rest of the file + } data = readBytes(slice, rawSample.dataSize); } diff --git a/src/mp3/mp3-reader.ts b/src/mp3/mp3-reader.ts index a28159c..d53708d 100644 --- a/src/mp3/mp3-reader.ts +++ b/src/mp3/mp3-reader.ts @@ -22,20 +22,20 @@ export const readId3 = (slice: FileSlice) => { return { size }; }; -export const readNextFrameHeader = async (reader: Reader, startPos: number, until: number): Promise<{ +export const readNextFrameHeader = async (reader: Reader, startPos: number, until: number | null): Promise<{ header: FrameHeader; startPos: number; } | null> => { let currentPos = startPos; - while (currentPos < until) { + while (until === null || currentPos < until) { let slice = reader.requestSlice(currentPos, FRAME_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; const word = readU32Be(slice); - const result = readFrameHeader(word, reader.fileSize - currentPos); + const result = readFrameHeader(word, reader.fileSize !== null ? reader.fileSize - currentPos : null); if (result.header) { return { header: result.header, startPos: currentPos }; } diff --git a/src/ogg/ogg-demuxer.ts b/src/ogg/ogg-demuxer.ts index 17b70ed..2632e49 100644 --- a/src/ogg/ogg-demuxer.ts +++ b/src/ogg/ogg-demuxer.ts @@ -12,7 +12,16 @@ import { Demuxer } from '../demuxer'; import { Input } from '../input'; import { InputAudioTrack, InputAudioTrackBacking } from '../input-track'; import { PacketRetrievalOptions } from '../media-sink'; -import { assert, findLast, roundToPrecision, toDataView, UNDETERMINED_LANGUAGE } from '../misc'; +import { + assert, + AsyncMutex, + binarySearchLessOrEqual, + findLast, + last, + roundToPrecision, + toDataView, + UNDETERMINED_LANGUAGE, +} from '../misc'; import { EncodedPacket, PLACEHOLDER_DATA } from '../packet'; import { readBytes, Reader } from '../reader'; import { buildOggMimeType, computeOggPageCrc, extractSampleMetadata, OggCodecInfo } from './ogg-misc'; @@ -58,7 +67,8 @@ export class OggDemuxer extends Demuxer { async readMetadata() { return this.metadataPromise ??= (async () => { let currentPos = 0; - while (currentPos <= this.reader.fileSize - MIN_PAGE_HEADER_SIZE) { + + while (true) { let slice = this.reader.requestSliceRange(currentPos, MIN_PAGE_HEADER_SIZE, MAX_PAGE_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -284,10 +294,6 @@ export class OggDemuxer extends Demuxer { // The packet extends to the next page; let's find it let currentPos = currentPage.headerStartPos + currentPage.totalSize; while (true) { - if (currentPos > this.reader.fileSize - MIN_PAGE_HEADER_SIZE) { - return null; - } - let headerSlice = this.reader.requestSliceRange(currentPos, MIN_PAGE_HEADER_SIZE, MAX_PAGE_HEADER_SIZE); if (headerSlice instanceof Promise) headerSlice = await headerSlice; if (!headerSlice) { @@ -343,10 +349,6 @@ export class OggDemuxer extends Demuxer { // Otherwise, search for the next page belonging to the same bitstream let currentPos = lastPacket.endPage.headerStartPos + lastPacket.endPage.totalSize; while (true) { - if (currentPos >= this.reader.fileSize - MIN_PAGE_HEADER_SIZE) { - return null; - } - let slice = this.reader.requestSliceRange(currentPos, MIN_PAGE_HEADER_SIZE, MAX_PAGE_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) { @@ -392,12 +394,15 @@ type EncodedPacketMetadata = { packet: Packet; timestampInSamples: number; durationInSamples: number; + vorbisLastBlockSize: number | null; vorbisBlockSize: number | null; }; class OggAudioTrackBacking implements InputAudioTrackBacking { internalSampleRate: number; encodedPacketToMetadata = new WeakMap(); + sequentialScanCache: EncodedPacketMetadata[] = []; + sequentialScanMutex = new AsyncMutex(); constructor(public bitstream: LogicalBitstream, public demuxer: OggDemuxer) { // Opus always uses a fixed sample rate for its internal calculations, even if the actual rate is different @@ -498,6 +503,7 @@ class OggAudioTrackBacking implements InputAudioTrackBacking { packet, timestampInSamples: additional.timestampInSamples, durationInSamples, + vorbisLastBlockSize: additional.vorbisLastBlocksize, vorbisBlockSize, }); return encodedPacket; @@ -557,6 +563,11 @@ class OggAudioTrackBacking implements InputAudioTrackBacking { } async getPacket(timestamp: number, options: PacketRetrievalOptions) { + if (this.demuxer.reader.fileSize === null) { + // No file size known, can't do binary search, but fall back to sequential algo instead + return this.getPacketSequential(timestamp, options); + } + const timestampInSamples = roundToPrecision(timestamp * this.internalSampleRate, 14); if (timestampInSamples === 0) { // Fast path for timestamp 0 - avoids binary search when playing back from the start @@ -885,6 +896,68 @@ class OggAudioTrackBacking implements InputAudioTrackBacking { return lastEncodedPacket; } + // A slower but simpler and sequential algorithm for finding a packet in a file + async getPacketSequential(timestamp: number, options: PacketRetrievalOptions) { + const release = await this.sequentialScanMutex.acquire(); // Requires exclusivity because we write to a cache + + try { + const timestampInSamples = roundToPrecision(timestamp * this.internalSampleRate, 14); + timestamp = timestampInSamples / this.internalSampleRate; + + const index = binarySearchLessOrEqual( + this.sequentialScanCache, + timestampInSamples, + x => x.timestampInSamples, + ); + + let currentPacket: EncodedPacket | null; + if (index !== -1) { + // We don't need to start from the beginning, we can start at a previous scan point + const cacheEntry = this.sequentialScanCache[index]!; + currentPacket = this.createEncodedPacketFromOggPacket( + cacheEntry.packet, + { + timestampInSamples: cacheEntry.timestampInSamples, + vorbisLastBlocksize: cacheEntry.vorbisLastBlockSize, + }, + options, + ); + } else { + currentPacket = await this.getFirstPacket(options); + } + + let i = 0; + + while (currentPacket && currentPacket.timestamp < timestamp) { + const nextPacket = await this.getNextPacket(currentPacket, options); + if (!nextPacket || nextPacket.timestamp > timestamp) { + break; + } + + currentPacket = nextPacket; + i++; + + if (i === 100) { + // Add "checkpoints" every once in a while to speed up subsequent random accesses + i = 0; + const metadata = this.encodedPacketToMetadata.get(currentPacket); + assert(metadata); + + if (this.sequentialScanCache.length > 0) { + // If we reach this case, we must be at the end of the cache + assert(last(this.sequentialScanCache)!.timestampInSamples <= metadata.timestampInSamples); + } + + this.sequentialScanCache.push(metadata); + } + } + + return currentPacket; + } finally { + release(); + } + } + getKeyPacket(timestamp: number, options: PacketRetrievalOptions) { return this.getPacket(timestamp, options); } diff --git a/src/reader.ts b/src/reader.ts index 6344e91..b0104f0 100644 --- a/src/reader.ts +++ b/src/reader.ts @@ -6,16 +6,16 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ -import { clamp, MaybePromise, toDataView } from './misc'; +import { assert, clamp, MaybePromise, toDataView } from './misc'; import { Source } from './source'; export class Reader { - fileSize!: number; + fileSize!: number | null; constructor(public source: Source) {} requestSlice(start: number, length: number): MaybePromise { - if (start + length > this.fileSize) { + if (this.fileSize !== null && start + length > this.fileSize) { return null; } @@ -40,10 +40,42 @@ export class Reader { } requestSliceRange(start: number, minLength: number, maxLength: number): MaybePromise { - return this.requestSlice( - start, - clamp(this.fileSize - start, minLength, maxLength), - ); + if (this.fileSize !== null) { + return this.requestSlice( + start, + clamp(this.fileSize - start, minLength, maxLength), + ); + } else { + const promisedAttempt = this.requestSlice(start, maxLength); + + const handleAttempt = (attempt: FileSlice | null) => { + if (attempt) { + return attempt; + } + + const handleFileSize = (fileSize: number | null) => { + assert(fileSize !== null); // The slice couldn't fit, meaning we must know the file size now + + return this.requestSlice( + start, + clamp(fileSize - start, minLength, maxLength), + ); + }; + + const promisedFileSize = this.source._retrieveSize(); + if (promisedFileSize instanceof Promise) { + return promisedFileSize.then(handleFileSize); + } else { + return handleFileSize(promisedFileSize); + } + }; + + if (promisedAttempt instanceof Promise) { + return promisedAttempt.then(handleAttempt); + } else { + return handleAttempt(promisedAttempt); + } + } } } diff --git a/src/source.ts b/src/source.ts index 7e45f90..4283eab 100644 --- a/src/source.ts +++ b/src/source.ts @@ -31,19 +31,36 @@ export type ReadResult = { */ export abstract class Source { /** @internal */ - abstract _retrieveSize(): MaybePromise; + abstract _retrieveSize(): MaybePromise; /** @internal */ - abstract _read(start: number, end: number): MaybePromise; + abstract _read(start: number, end: number): MaybePromise; /** @internal */ - private _sizePromise: Promise | null = null; + private _sizePromise: Promise | null = null; /** * Resolves with the total size of the file in bytes. This function is memoized, meaning only the first call * will retrieve the size. + * + * Returns null if the source is unsized. + */ + async getSizeOrNull() { + return this._sizePromise ??= Promise.resolve(this._retrieveSize()); + } + + /** + * Resolves with the total size of the file in bytes. This function is memoized, meaning only the first call + * will retrieve the size. + * + * Throws an error if the source is unsized. */ async getSize() { - return this._sizePromise ??= Promise.resolve(this._retrieveSize()); + const result = await this.getSizeOrNull(); + if (result === null) { + throw new Error('Cannot determine the size of an unsized source.'); + } + + return result; } /** Called each time data is retrieved from the source. Will be called with the retrieved range. */ @@ -692,6 +709,262 @@ export class StreamSource extends Source { } } +type ReadableStreamSourcePendingSlice = { + start: number; + end: number; + bytes: Uint8Array; + resolve: (bytes: ReadResult | null) => void; + reject: (error: unknown) => void; +}; + +/** + * Options for ReadableStreamSource. + * @public + */ +export type ReadableStreamSourceOptions = { + /** The maximum number of bytes the cache is allowed to hold in memory. Defaults to 16 MiB. */ + maxCacheSize?: number; +}; + +/** + * A source backed by a `ReadableStream` of `Uint8Array`, representing an append-only byte stream of unknown + * length. This is the source to use for incrementally streaming in input files that are still being constructed and + * whose size we don't yet know, like for example the output chunks of + * [MediaRecorder](https://developer.mozilla.org/en-US/docs/Web/API/MediaRecorder). + * + * This source is *unsized*, meaning calls to `.getSize()` will throw and readers are more limited due to the + * lack of random file access. You should only use this source with sequential access patterns, such as reading all + * packets from start to end. This source does not work well with random access patterns unless you increase its + * max cache size. + * + * @public + */ +export class ReadableStreamSource extends Source { + /** @internal */ + _stream: ReadableStream; + /** @internal */ + _reader: ReadableStreamDefaultReader | null = null; + /** @internal */ + _cache: CacheEntry[] = []; + /** @internal */ + _maxCacheSize: number; + /** @internal */ + _pendingSlices: ReadableStreamSourcePendingSlice[] = []; + /** @internal */ + _currentIndex = 0; + /** @internal */ + _targetIndex = 0; + /** @internal */ + _maxRequestedIndex = 0; + /** @internal */ + _endIndex: number | null = null; + /** @internal */ + _pulling = false; + + constructor(stream: ReadableStream, options: ReadableStreamSourceOptions = {}) { + if (!(stream instanceof ReadableStream)) { + throw new TypeError('stream must be a ReadableStream.'); + } + if (!options || typeof options !== 'object') { + throw new TypeError('options must be an object.'); + } + if ( + options.maxCacheSize !== undefined + && (!Number.isInteger(options.maxCacheSize) || options.maxCacheSize < 0) + ) { + throw new TypeError('options.maxCacheSize, when provided, must be a non-negative integer.'); + } + + super(); + + this._stream = stream; + this._maxCacheSize = options.maxCacheSize ?? (16 * 2 ** 20 /* 16 MiB */); + } + + /** @internal */ + _retrieveSize() { + return this._endIndex; // Starts out as null, meaning this source is unsized + } + + /** @internal */ + _read(start: number, end: number): MaybePromise { + if (this._endIndex !== null && end > this._endIndex) { + return null; + } + + this._maxRequestedIndex = Math.max(this._maxRequestedIndex, end); + + const cacheStartIndex = binarySearchLessOrEqual(this._cache, start, x => x.start); + const cacheStartEntry = cacheStartIndex !== -1 ? this._cache[cacheStartIndex]! : null; + + if (cacheStartEntry && cacheStartEntry.start <= start && end <= cacheStartEntry.end) { + // The request can be satisfied with a single cache entry + return { + bytes: cacheStartEntry.bytes, + view: cacheStartEntry.view, + offset: cacheStartEntry.start, + }; + } + + let lastEnd = start; + const bytes = new Uint8Array(end - start); + + if (cacheStartIndex !== -1) { + // Walk over the cache to see if we can satisfy the request using multiple cache entries + for (let i = cacheStartIndex; i < this._cache.length; i++) { + const cacheEntry = this._cache[i]!; + if (cacheEntry.start >= end) { + break; + } + + const cappedStart = Math.max(start, cacheEntry.start); + if (cappedStart > lastEnd) { + // We're too far behind + this._throwDueToCacheMiss(); + } + + const cappedEnd = Math.min(end, cacheEntry.end); + + if (cappedStart < cappedEnd) { + bytes.set( + cacheEntry.bytes.subarray(cappedStart - cacheEntry.start, cappedEnd - cacheEntry.start), + cappedStart - start, + ); + + lastEnd = cappedEnd; + } + } + } + + if (lastEnd === end) { + return { + bytes, + view: toDataView(bytes), + offset: start, + }; + } + + // We need to pull more data + + if (this._currentIndex > lastEnd) { + // We're too far behind + this._throwDueToCacheMiss(); + } + + const { promise, resolve, reject } = promiseWithResolvers(); + + this._pendingSlices.push({ + start, + end, + bytes, + resolve, + reject, + }); + + this._targetIndex = Math.max(this._targetIndex, end); + + // Start pulling from the stream if we're not already doing it + if (!this._pulling) { + this._pulling = true; + void this._pull() + .catch((error) => { + this._pulling = false; + + if (this._pendingSlices.length > 0) { + this._pendingSlices.forEach(x => x.reject(error)); // Make sure to propagate any errors + this._pendingSlices.length = 0; + } else { + throw error; // So it doesn't get swallowed + } + }); + } + + return promise; + } + + /** @internal */ + _throwDueToCacheMiss() { + throw new Error( + 'Read is before the cached region. With ReadableStreamSource, you must access the data more' + + ' sequentially or increase the size of its cache.', + ); + } + + /** @internal */ + async _pull() { + this._reader ??= this._stream.getReader(); + + // This is the loop that keeps pulling data from the stream until a target index is reached, filling requests + // in the process + while (this._currentIndex < this._targetIndex) { + const { done, value } = await this._reader.read(); + if (done) { + for (const pendingSlice of this._pendingSlices) { + pendingSlice.resolve(null); + } + this._pendingSlices.length = 0; + this._endIndex = this._currentIndex; // We know how long the file is now! + + break; + } + + const startIndex = this._currentIndex; + const endIndex = this._currentIndex + value.byteLength; + + // Fill the pending slices with the data + for (let i = 0; i < this._pendingSlices.length; i++) { + const pendingSlice = this._pendingSlices[i]!; + + const cappedStart = Math.max(startIndex, pendingSlice.start); + const cappedEnd = Math.min(endIndex, pendingSlice.end); + + if (cappedStart < cappedEnd) { + pendingSlice.bytes.set( + value.subarray(cappedStart - startIndex, cappedEnd - startIndex), + cappedStart - pendingSlice.start, + ); + if (cappedEnd === pendingSlice.end) { + // Pending slice fully filled + pendingSlice.resolve({ + bytes: pendingSlice.bytes, + view: toDataView(pendingSlice.bytes), + offset: pendingSlice.start, + }); + this._pendingSlices.splice(i, 1); + i--; + } + } + } + + this._cache.push({ + start: startIndex, + end: endIndex, + bytes: value, + view: toDataView(value), + age: 0, // Unused + }); + + // Do cache eviction, based on the distance from the last-requested index. It's important that we do it like + // this and not based on where the reader is at, because if the reader is fast, we'll unnecessarily evict + // data that we still might need. + while (this._cache.length > 0) { + const firstEntry = this._cache[0]!; + const distance = this._maxRequestedIndex - firstEntry.end; + + if (distance <= this._maxCacheSize) { + break; + } + + this._cache.shift(); + } + + this._currentIndex += value.byteLength; + } + + this._pulling = false; + } +} + type PrefetchProfile = (start: number, end: number, workers: ReadWorker[]) => { start: number; end: number; @@ -1190,32 +1463,25 @@ class ReadOrchestrator { // LRU eviction of cache entries while (this.currentCacheSize > this.options.maxCacheSize) { - if (this.cache.length > 1) { - let oldestIndex = 0; - let oldestEntry = this.cache[0]!; + let oldestIndex = 0; + let oldestEntry = this.cache[0]!; - for (let i = 1; i < this.cache.length; i++) { - const entry = this.cache[i]!; + for (let i = 1; i < this.cache.length; i++) { + const entry = this.cache[i]!; - if (entry.age < oldestEntry.age) { - oldestIndex = i; - oldestEntry = entry; - } + if (entry.age < oldestEntry.age) { + oldestIndex = i; + oldestEntry = entry; } - - this.cache.splice(oldestIndex, 1); - this.currentCacheSize -= oldestEntry.bytes.length; - } else { - // The single entry that's left is too big for the cache, let's trim it - const entry = this.cache[0]!; - assert(entry.bytes.length > this.options.maxCacheSize); - - entry.bytes = entry.bytes.slice(0, this.options.maxCacheSize); - entry.view = toDataView(entry.bytes); - entry.end = entry.start + entry.bytes.length; - - this.currentCacheSize = entry.bytes.length; } + + if (this.currentCacheSize - oldestEntry.bytes.length <= this.options.maxCacheSize) { + // Don't evict if it would shrink the cache below the max size + break; + } + + this.cache.splice(oldestIndex, 1); + this.currentCacheSize -= oldestEntry.bytes.length; } } } diff --git a/src/wave/wave-demuxer.ts b/src/wave/wave-demuxer.ts index bfbd669..43795f5 100644 --- a/src/wave/wave-demuxer.ts +++ b/src/wave/wave-demuxer.ts @@ -38,6 +38,7 @@ export class WaveDemuxer extends Demuxer { } | null = null; tracks: InputAudioTrack[] = []; + lastKnownPacketIndex = 0; constructor(input: Input) { super(input); @@ -58,7 +59,9 @@ export class WaveDemuxer extends Demuxer { const outerChunkSize = readU32(slice, littleEndian); - let totalFileSize = isRf64 ? this.reader.fileSize : Math.min(outerChunkSize + 8, this.reader.fileSize); + let totalFileSize = isRf64 + ? this.reader.fileSize + : Math.min(outerChunkSize + 8, this.reader.fileSize ?? Infinity); const format = readAscii(slice, 4); if (format !== 'WAVE') { @@ -69,7 +72,7 @@ export class WaveDemuxer extends Demuxer { let dataChunkSize: number | null = null; let currentPos = slice.filePos; - while (currentPos < totalFileSize) { + while (totalFileSize === null || currentPos < totalFileSize) { let slice = this.reader.requestSlice(currentPos, 8); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -88,14 +91,14 @@ export class WaveDemuxer extends Demuxer { dataChunkSize ??= chunkSize; this.dataStart = slice.filePos; - this.dataSize = Math.min(dataChunkSize, totalFileSize - this.dataStart); + this.dataSize = Math.min(dataChunkSize, (totalFileSize ?? Infinity) - this.dataStart); } else if (chunkId === 'ds64') { // File and data chunk sizes are defined in here instead const riffChunkSize = readU64(slice, littleEndian); dataChunkSize = readU64(slice, littleEndian); - totalFileSize = Math.min(riffChunkSize + 8, this.reader.fileSize); + totalFileSize = Math.min(riffChunkSize + 8, this.reader.fileSize ?? Infinity); } currentPos = startPos + chunkSize + (chunkSize & 1); // Handle padding @@ -200,10 +203,11 @@ export class WaveDemuxer extends Demuxer { async computeDuration() { await this.readMetadata(); - assert(this.audioInfo); - const numberOfBlocks = this.dataSize / this.audioInfo.blockSizeInBytes; - return numberOfBlocks / this.audioInfo.sampleRate; + const track = this.tracks[0]; + assert(track); + + return track.computeDuration(); } async getTracks() { @@ -244,8 +248,9 @@ class WaveAudioTrackBacking implements InputAudioTrackBacking { }; } - computeDuration() { - return this.demuxer.computeDuration(); + async computeDuration() { + const lastPacket = await this.getPacket(Infinity, { metadataOnly: true }); + return (lastPacket?.timestamp ?? 0) + (lastPacket?.duration ?? 0); } getNumberOfChannels() { @@ -290,6 +295,19 @@ class WaveAudioTrackBacking implements InputAudioTrackBacking { this.demuxer.dataSize - startOffset, ); + if (this.demuxer.reader.fileSize === null) { + // If the file size is unknown, we weren't able to cap the dataSize in the init logic and we instead have to + // rely on the headers telling us how large the file is. But, these might be wrong, so let's check if the + // requested slice actually exists. + + let slice = this.demuxer.reader.requestSlice(this.demuxer.dataStart + startOffset, sizeInBytes); + if (slice instanceof Promise) slice = await slice; + + if (!slice) { + return null; + } + } + let data: Uint8Array; if (options.metadataOnly) { data = PLACEHOLDER_DATA; @@ -304,6 +322,11 @@ class WaveAudioTrackBacking implements InputAudioTrackBacking { const timestamp = packetIndex * PACKET_SIZE_IN_FRAMES / this.demuxer.audioInfo.sampleRate; const duration = sizeInBytes / this.demuxer.audioInfo.blockSizeInBytes / this.demuxer.audioInfo.sampleRate; + this.demuxer.lastKnownPacketIndex = Math.max( + packetIndex, + timestamp, + ); + return new EncodedPacket( data, 'key', @@ -318,11 +341,38 @@ class WaveAudioTrackBacking implements InputAudioTrackBacking { return this.getPacketAtIndex(0, options); } - getPacket(timestamp: number, options: PacketRetrievalOptions) { + async getPacket(timestamp: number, options: PacketRetrievalOptions) { assert(this.demuxer.audioInfo); - const packetIndex = Math.floor(timestamp * this.demuxer.audioInfo.sampleRate / PACKET_SIZE_IN_FRAMES); - return this.getPacketAtIndex(packetIndex, options); + const packetIndex = Math.floor(Math.min( + timestamp * this.demuxer.audioInfo.sampleRate / PACKET_SIZE_IN_FRAMES, + (this.demuxer.dataSize - 1) / (PACKET_SIZE_IN_FRAMES * this.demuxer.audioInfo.blockSizeInBytes), + )); + + const packet = await this.getPacketAtIndex(packetIndex, options); + if (packet) { + return packet; + } + + if (packetIndex === 0) { + return null; // Empty data chunk + } + + assert(this.demuxer.reader.fileSize === null); + + // The file is shorter than we thought, meaning the packet we were looking for doesn't exist. So, let's find + // the last packet by doing a sequential scan, instead. + let currentPacket = await this.getPacketAtIndex(this.demuxer.lastKnownPacketIndex, options); + while (currentPacket) { + const nextPacket = await this.getNextPacket(currentPacket, options); + if (!nextPacket) { + break; + } + + currentPacket = nextPacket; + } + + return currentPacket; } getNextPacket(packet: EncodedPacket, options: PacketRetrievalOptions) { From a769d099e2b0369560c53abcfd58799a07f2b22f Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 16:50:59 +0200 Subject: [PATCH 4/8] Fix UrlSource for small files --- src/source.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/source.ts b/src/source.ts index 4283eab..a262bf6 100644 --- a/src/source.ts +++ b/src/source.ts @@ -316,7 +316,7 @@ export class UrlSource extends Source { if (response.status === 206) { fileSize = this._getPartialLengthFromRangeResponse(response); - worker = this._orchestrator.createWorker(0, URL_SOURCE_MIN_LOAD_AMOUNT); + worker = this._orchestrator.createWorker(0, Math.min(fileSize, URL_SOURCE_MIN_LOAD_AMOUNT)); } else { // Server probably returned a 200. From 55dd8cdb4c5c6af8dc598602d3e9f9baaeada818 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 17:20:09 +0200 Subject: [PATCH 5/8] Add NullTarget --- dev/convert.html | 6 +++-- docs/guide/writing-media-files.md | 39 +++++++++++++++++++++++++++++++ src/index.ts | 2 +- src/target.ts | 14 ++++++++++- src/writer.ts | 21 +++++++++++++++++ 5 files changed, 78 insertions(+), 4 deletions(-) diff --git a/dev/convert.html b/dev/convert.html index bdd82c1..fd662fd 100644 --- a/dev/convert.html +++ b/dev/convert.html @@ -18,13 +18,15 @@ const file = fileInput.files[0]; const source = new Mediabunny.BlobSource(file); - const target = new Mediabunny.BufferTarget() ?? new Mediabunny.StreamTarget(new WritableStream({ + const target = new Mediabunny.NullTarget() ?? new Mediabunny.BufferTarget() ?? new Mediabunny.StreamTarget(new WritableStream({ write: console.log }), { chunked: true, chunkSize: 2**20 }); - const outputFormat = new Mediabunny.WavOutputFormat({}); + const outputFormat = new Mediabunny.Mp4OutputFormat({ + onMoov: console.log + }); const button = document.createElement('button'); button.textContent = 'Cancel'; diff --git a/docs/guide/writing-media-files.md b/docs/guide/writing-media-files.md index 5a20e6a..384ce9a 100644 --- a/docs/guide/writing-media-files.md +++ b/docs/guide/writing-media-files.md @@ -297,6 +297,45 @@ const output = new Output({ await output.finalize(); // Will automatically close the writable stream ``` +### `NullTarget` + +This target simply discards all data that is passed into it. It is useful for when you need an `Output` but extract data from it differently, for example through output format-specific callbacks or encoder events. + +As an example, here we create a fragmented MP4 file and directly handle the individual fragments: +```ts +import { Output, NullTarget, Mp4OutputFormat } from 'mediabunny'; + +let ftyp: Uint8Array; +let lastMoof: Uint8Array; + +const output = new Output({ + target: new NullTarget(), + format: new Mp4OutputFormat({ + fastStart: 'fragmented', + onFtyp: (data) => { + ftyp = data; + }, + onMoov: (data) => { + const header = new Uint8Array(ftyp.length + data.length); + header.set(ftyp, 0); + header.set(data, ftyp.length); + + // Do something with the header... + }, + onMoof: (data) => { + lastMoof = data; + }, + onMdat: (data) => { + const segment = new Uint8Array(lastMoof.length + data.length); + segment.set(lastMoof, 0); + segment.set(data, lastMoof.length); + + // Do something with the segment... + }, + }), +}); +``` + ## Packet buffering Some [output formats](./output-formats) require *packet buffering* for multi-track outputs. Packet buffering occurs because the `Output` must wait for data from all tracks for a given timestamp to continue writing data. For example, should you first encode all your video frames and then encode the audio afterward, the `Output` will have to hold all of the video frames in memory until the audio packets start coming in. This might lead to memory exhaustion should your video be very long. When there is only one media track, this issue does not arise. diff --git a/src/index.ts b/src/index.ts index e03149e..da1f4ce 100644 --- a/src/index.ts +++ b/src/index.ts @@ -89,7 +89,7 @@ export { getFirstEncodableAudioCodec, getFirstEncodableSubtitleCodec, } from './encode'; -export { Target, BufferTarget, StreamTarget, StreamTargetChunk, StreamTargetOptions } from './target'; +export { Target, BufferTarget, StreamTarget, StreamTargetChunk, StreamTargetOptions, NullTarget } from './target'; export { Rotation, AnyIterable, SetRequired, MaybePromise } from './misc'; export { Source, diff --git a/src/target.ts b/src/target.ts index 78fc777..91d9960 100644 --- a/src/target.ts +++ b/src/target.ts @@ -6,7 +6,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ -import { BufferTargetWriter, StreamTargetWriter, Writer } from './writer'; +import { BufferTargetWriter, NullTargetWriter, StreamTargetWriter, Writer } from './writer'; import { Output } from './output'; /** @@ -104,3 +104,15 @@ export class StreamTarget extends Target { return new StreamTargetWriter(this); } } + +/** + * This target just discards all incoming data. It is useful for when you need an `Output` but extract data from it + * differently, for example through format-specific callbacks (`onMoof`, `onMdat`, ...) or encoder events. + * @public + */ +export class NullTarget extends Target { + /** @internal */ + _createWriter() { + return new NullTargetWriter(); + } +} diff --git a/src/writer.ts b/src/writer.ts index d9853c4..5f2b8f7 100644 --- a/src/writer.ts +++ b/src/writer.ts @@ -459,3 +459,24 @@ export class StreamTargetWriter extends Writer { return this.writer?.close(); } } + +export class NullTargetWriter extends Writer { + private pos = 0; + + write(data: Uint8Array) { + this.maybeTrackWrites(data); + this.pos += data.byteLength; + } + + getPos() { + return this.pos; + } + + seek(newPos: number) { + this.pos = newPos; + } + + async flush() {} + async finalize() {} + async close() {} +} From 9d47b3c5dc016f58d345d628391291d8b83f5cad Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 17:34:09 +0200 Subject: [PATCH 6/8] Add Target.onwrite callback --- docs/guide/writing-media-files.md | 13 ++++++++++++- src/source.ts | 2 +- src/target.ts | 10 +++++++++- src/writer.ts | 12 ++++++++++-- 4 files changed, 32 insertions(+), 5 deletions(-) diff --git a/docs/guide/writing-media-files.md b/docs/guide/writing-media-files.md index 384ce9a..5e58081 100644 --- a/docs/guide/writing-media-files.md +++ b/docs/guide/writing-media-files.md @@ -197,7 +197,18 @@ output.state; // => 'pending' | 'started' | 'canceled' | 'finalizing' | 'finaliz ## Output targets -The _output target_ determines where the data created by the `Output` will be written. This library offers two targets: +The _output target_ determines where the data created by the `Output` will be written. This library offers a couple of targets. + +--- + +All targets have an optional `onwrite` callback you can set to monitor which byte regions are being written to: +```ts +target.onwrite = (start, end) => { + // ... +}; +``` + +You can use this to track the size of the output file as it grows. But be warned, this function is chatty and gets called *extremely* frequently. ### `BufferTarget` diff --git a/src/source.ts b/src/source.ts index a262bf6..1b3d41a 100644 --- a/src/source.ts +++ b/src/source.ts @@ -63,7 +63,7 @@ export abstract class Source { return result; } - /** Called each time data is retrieved from the source. Will be called with the retrieved range. */ + /** Called each time data is retrieved from the source. Will be called with the retrieved range (end exclusive). */ onread: ((start: number, end: number) => unknown) | null = null; } diff --git a/src/target.ts b/src/target.ts index 91d9960..03366bd 100644 --- a/src/target.ts +++ b/src/target.ts @@ -19,6 +19,14 @@ export abstract class Target { /** @internal */ abstract _createWriter(): Writer; + + /** + * Called each time data is written to the target. Will be called with the byte range into which data was written. + * + * Use this callback to track the size of the output file as it grows. But be warned, this function is chatty and + * gets called *extremely* often. + */ + onwrite: ((start: number, end: number) => unknown) | null = null; } /** @@ -113,6 +121,6 @@ export class StreamTarget extends Target { export class NullTarget extends Target { /** @internal */ _createWriter() { - return new NullTargetWriter(); + return new NullTargetWriter(this); } } diff --git a/src/writer.ts b/src/writer.ts index 5f2b8f7..30de003 100644 --- a/src/writer.ts +++ b/src/writer.ts @@ -6,7 +6,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ -import { BufferTarget, StreamTarget, StreamTargetChunk } from './target'; +import { BufferTarget, NullTarget, StreamTarget, StreamTargetChunk } from './target'; import { assert } from './misc'; export abstract class Writer { @@ -156,8 +156,9 @@ export class BufferTargetWriter extends Writer { this.ensureSize(this.pos + data.byteLength); this.bytes.set(data, this.pos); - this.pos += data.byteLength; + this.target.onwrite?.(this.pos, this.pos + data.byteLength); + this.pos += data.byteLength; this.maxPos = Math.max(this.maxPos, this.pos); } @@ -250,6 +251,8 @@ export class StreamTargetWriter extends Writer { data: data.slice(), start: this.pos, }); + this.target.onwrite?.(this.pos, this.pos + data.byteLength); + this.pos += data.byteLength; this.lastWriteEnd = Math.max(this.lastWriteEnd, this.pos); @@ -463,8 +466,13 @@ export class StreamTargetWriter extends Writer { export class NullTargetWriter extends Writer { private pos = 0; + constructor(private target: NullTarget) { + super(); + } + write(data: Uint8Array) { this.maybeTrackWrites(data); + this.target.onwrite?.(this.pos, this.pos + data.byteLength); this.pos += data.byteLength; } From 5ca5fc4633e2ac4176e4a1d5f552db9fc16ac18a Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 17:48:33 +0200 Subject: [PATCH 7/8] Document `onwrite` in conversion docs --- docs/guide/converting-media-files.md | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/guide/converting-media-files.md b/docs/guide/converting-media-files.md index 6e4b8ce..c96e629 100644 --- a/docs/guide/converting-media-files.md +++ b/docs/guide/converting-media-files.md @@ -77,10 +77,19 @@ This callback is called each time the progress of the conversion advances. A progress of `1` doesn't indicate the conversion has finished; the conversion is only finished once the promise returned by `.execute()` resolves. ::: -::: info -Tracking conversion progress may slightly affect performance, as it requires knowledge of the input file's total duration - but this is usually negligible. +::: warning +Tracking conversion progress can slightly affect performance as it requires knowledge of the input file's total duration. This is usually negligible but should be avoided when using append-only input sources such as [`ReadableStreamSource`](./reading-media-files#readablestreamsource). ::: +If you want to monitor the output size of the conversion (in bytes), simply use the `onwrite` callback on your `Target`: +```ts +let currentFileSize = 0; + +output.target.onwrite = (start, end) => { + currentFileSize = Math.max(currentFileSize, end); +}; +``` + ### Canceling a conversion Sometimes, you may want to cancel an ongoing conversion process. For this, use the `cancel` method: From 9d70c4c605efcdacb4ddb678d72f766862b270f0 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Tue, 2 Sep 2025 17:50:05 +0200 Subject: [PATCH 8/8] Bump minor --- package-lock.json | 12 ++++++------ package.json | 2 +- packages/mp3-encoder/package.json | 2 +- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/package-lock.json b/package-lock.json index bd1b3e0..3bec6d7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "mediabunny", - "version": "1.12.1", + "version": "1.13.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "mediabunny", - "version": "1.12.1", + "version": "1.13.0", "license": "MPL-2.0", "workspaces": [ "packages/*" @@ -6147,9 +6147,9 @@ } }, "node_modules/mediabunny": { - "version": "1.12.0", - "resolved": "https://registry.npmjs.org/mediabunny/-/mediabunny-1.12.0.tgz", - "integrity": "sha512-BARnydaYDY9iCar+ZRdnaKER31c2J4otsEVcSRsizXRrFLv97LXSd5ilD5FQF52AaM6FiqLm7iB3Zl1C+a5u9w==", + "version": "1.12.1", + "resolved": "https://registry.npmjs.org/mediabunny/-/mediabunny-1.12.1.tgz", + "integrity": "sha512-Jidi7oARd9/oUVg8tf8bbj5yQHTBAB2vp+iz+txQlqTFwuzqjRrij384pkTezJZRyeE1I1JKE2IY3/GQDMelfA==", "license": "MPL-2.0", "peer": true, "workspaces": [ @@ -9478,7 +9478,7 @@ }, "packages/mp3-encoder": { "name": "@mediabunny/mp3-encoder", - "version": "1.12.1", + "version": "1.13.0", "license": "MPL-2.0", "devDependencies": { "@types/emscripten": "^1.40.1" diff --git a/package.json b/package.json index 8e85294..1d6512f 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "mediabunny", "author": "Vanilagy", - "version": "1.12.1", + "version": "1.13.0", "description": "Pure TypeScript media toolkit for reading, writing, and converting media files, directly in the browser.", "type": "module", "workspaces": [ diff --git a/packages/mp3-encoder/package.json b/packages/mp3-encoder/package.json index f1be561..e512b75 100644 --- a/packages/mp3-encoder/package.json +++ b/packages/mp3-encoder/package.json @@ -1,7 +1,7 @@ { "name": "@mediabunny/mp3-encoder", "author": "Vanilagy", - "version": "1.12.1", + "version": "1.13.0", "description": "MP3 encoder extension for Mediabunny, based on LAME.", "main": "./dist/bundles/mediabunny-mp3-encoder.mjs", "module": "./dist/bundles/mediabunny-mp3-encoder.mjs",