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] 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) {