diff --git a/examples/procedural-generation/procedural-generation.ts b/examples/procedural-generation/procedural-generation.ts index 3c37357..b54f9ef 100644 --- a/examples/procedural-generation/procedural-generation.ts +++ b/examples/procedural-generation/procedural-generation.ts @@ -12,6 +12,7 @@ import { OutputTrackGroup, MpegTsOutputFormat, AdtsOutputFormat, + StreamTarget, } from 'mediabunny'; const durationSlider = document.querySelector('#duration-slider') as HTMLInputElement; @@ -96,19 +97,30 @@ const generateVideo = async () => { // Create a new output file output = new Output({ rootPath: 'master.m3u8', - target: ({ path }) => { + target: async ({ path }) => { + const fileHandle = await dirHandle.getFileHandle(path, { create: true }); + const writable = await fileHandle.createWritable(); + + const target = new StreamTarget(writable); + target.onfinalized = () => console.log('Finalizado', path); + + return target; + + /* const target = new BufferTarget(); target.onfinalized = async () => { const fileHandle = await dirHandle.getFileHandle(path, { create: true }); const writable = await fileHandle.createWritable(); - await writable.write(target.buffer!); + await writable.write(target.buffer); await writable.close(); }; return target; + */ }, format: new HlsOutputFormat({ segmentFormats: [new AdtsOutputFormat(), new MpegTsOutputFormat()], + // singleFilePerPlaylist: true, getPlaylistPath: info => `sussex-${info.n}.m3u8`, }), }); @@ -216,12 +228,14 @@ const generateVideo = async () => { videoInfo.style.display = ''; // Display and play the resulting media file + /* const videoBlob = new Blob([output.target.buffer!], { type: output.format.mimeType }); resultVideo.src = URL.createObjectURL(videoBlob); void resultVideo.play(); const fileSizeMiB = (videoBlob.size / (1024 * 1024)).toPrecision(3); videoInfo.textContent = `File size: ${fileSizeMiB} MiB`; + */ } catch (error) { console.error(error); diff --git a/src/flac/flac-muxer.ts b/src/flac/flac-muxer.ts index 625cf38..23db6de 100644 --- a/src/flac/flac-muxer.ts +++ b/src/flac/flac-muxer.ts @@ -51,7 +51,7 @@ export class FlacMuxer extends Muxer { this.format = format; if (this.format._options.appendOnly) { - this.writer.ensureMonotonicity = true; + this.writer.ensureMonotonicity(); } } diff --git a/src/hls/hls-muxer.ts b/src/hls/hls-muxer.ts index 8471423..2c4a68c 100644 --- a/src/hls/hls-muxer.ts +++ b/src/hls/hls-muxer.ts @@ -12,6 +12,7 @@ import { TrackType, } from '../output'; import { HlsOutputFormat, HlsOutputFormatOptions, HlsOutputSegmentInfo, OutputFormat } from '../output-format'; +import { Writer } from '../writer'; import { EncodedPacket } from '../packet'; import { SubtitleCue, SubtitleMetadata } from '../subtitles'; import { Target } from '../target'; @@ -20,6 +21,8 @@ type HlsTrackData = { track: OutputTrack; packets: EncodedPacket[]; playlist: Playlist; + // We must store it on the TrackData, reading it directly from the track leads to async race conditions! + closed: boolean; info: { type: 'video'; decoderConfig: VideoDecoderConfig; @@ -44,9 +47,17 @@ type Playlist = { path: string; duration: number; byteSize: number; + byteOffset: number | null; }[]; peakBitrate: number | null; averageBitrate: number | null; + + singleFile: { + target: Target; + path: string; + nextOffset: number; + info: HlsOutputSegmentInfo; + } | null; }; type PlaylistDeclaration = { @@ -63,6 +74,7 @@ export class HlsMuxer extends Muxer { targetSegmentDuration: number; trackDatas: HlsTrackData[] = []; + singleFilePerPlaylist: boolean; playlists: Playlist[] = []; playlistDeclarations: PlaylistDeclaration[] = []; @@ -75,12 +87,15 @@ export class HlsMuxer extends Muxer { super(output); this.format = format; - this.targetSegmentDuration = format._options.targetDuration ?? 5; + this.targetSegmentDuration = format._options.targetDuration ?? 2; + this.singleFilePerPlaylist = format._options.singleFilePerPlaylist ?? false; this.getPlaylistPath = format._options.getPlaylistPath ?? (({ n }) => `playlist-${n}.m3u8`); this.getSegmentPath = format._options.getSegmentPath - ?? (info => `segment-${info.playlist.n}-${info.n}${info.format.fileExtension}`); + ?? (info => info.isSingleFile + ? `segments-${info.playlist.n}${info.format.fileExtension}` + : `segment-${info.playlist.n}-${info.n}${info.format.fileExtension}`); } async start(): Promise { @@ -382,6 +397,7 @@ export class HlsMuxer extends Muxer { writtenSegments: [], peakBitrate: null, averageBitrate: null, + singleFile: null, }; this.playlists.push(playlist); @@ -450,6 +466,7 @@ export class HlsMuxer extends Muxer { return; } + trackData.closed = true; await this.advancePlaylist(trackData.playlist); } finally { release(); @@ -474,6 +491,7 @@ export class HlsMuxer extends Muxer { track, packets: [], playlist: playlists[0]!, + closed: false, info: { type: 'video', decoderConfig: meta.decoderConfig, @@ -502,6 +520,7 @@ export class HlsMuxer extends Muxer { track, packets: [], playlist: playlists[0]!, + closed: false, info: { type: 'audio', decoderConfig: meta.decoderConfig, @@ -587,7 +606,7 @@ export class HlsMuxer extends Muxer { throw new Error('Unreachable.'); } - async advancePlaylist(playlist: Playlist, isFinalCall = false) { + async advancePlaylist(playlist: Playlist) { assert(playlist.currentSegmentStartTimestamp !== null); if (!this.allTracksAreKnown(playlist)) { @@ -598,13 +617,6 @@ export class HlsMuxer extends Muxer { const videoTrack = trackDatas.find(x => x.info.type === 'video') as HlsVideoTrackData | undefined; const audioTrack = trackDatas.find(x => x.info.type === 'audio') as HlsAudioTrackData | undefined; - const videoClosed = videoTrack - ? videoTrack.track.source._closed || isFinalCall - : true; - const audioClosed = audioTrack - ? audioTrack.track.source._closed || isFinalCall - : true; - // Loop in case we can finalize multiple segments while (true) { const currentSegmentEndTimestamp = playlist.currentSegmentStartTimestamp + this.targetSegmentDuration; @@ -613,14 +625,14 @@ export class HlsMuxer extends Muxer { let videoEndIndex = 0; let audioEndIndex = 0; - if (videoTrack && (!videoClosed || videoTrack.packets.length > 0)) { + if (videoTrack && (!videoTrack.closed || videoTrack.packets.length > 0)) { const allBelow = videoTrack.packets.every(x => x.timestamp < currentSegmentEndTimestamp); let bestKeyPacket: EncodedPacket | null = null; let bestKeyPacketIndex: number | null = null; if (allBelow) { - if (!videoClosed) { + if (!videoTrack.closed) { // Not enough data yet return; } @@ -649,7 +661,7 @@ export class HlsMuxer extends Muxer { if (index !== -1) { audioEndIndex = index; } else { - if (audioClosed) { + if (audioTrack.closed) { audioEndIndex = audioTrack.packets.length; } else { return; @@ -657,7 +669,7 @@ export class HlsMuxer extends Muxer { } } } else { - if (!videoClosed) { + if (!videoTrack.closed) { return; } @@ -674,7 +686,7 @@ export class HlsMuxer extends Muxer { if (index !== -1) { audioEndIndex = index; } else { - if (audioClosed) { + if (audioTrack.closed) { audioEndIndex = audioTrack.packets.length; } else { return; @@ -686,7 +698,7 @@ export class HlsMuxer extends Muxer { if (index !== -1) { audioEndIndex = index; } else { - if (audioClosed) { + if (audioTrack.closed) { audioEndIndex = audioTrack.packets.length; } else { return; @@ -695,11 +707,11 @@ export class HlsMuxer extends Muxer { } } } - } else if (audioTrack && (!audioClosed || audioTrack.packets.length > 0)) { + } else if (audioTrack && (!audioTrack.closed || audioTrack.packets.length > 0)) { const allBelow = audioTrack.packets.every(x => x.timestamp < currentSegmentEndTimestamp); if (allBelow) { - if (audioClosed) { + if (audioTrack.closed) { audioEndIndex = audioTrack.packets.length; } else { return; @@ -718,29 +730,68 @@ export class HlsMuxer extends Muxer { return; } - // We can finalize a new segment! Let's first get the path - const segmentInfo: HlsOutputSegmentInfo = { - n: playlist.nextSegmentId, - format: playlist.segmentFormat, - playlist: { - n: playlist.id, - tracks: playlist.tracks, - }, - }; + // We can finalize a new segment! - const relativeSegmentPath = await this.getSegmentPath(segmentInfo); - if (typeof relativeSegmentPath !== 'string') { - throw new TypeError('options.getSegmentPath must return or resolve to a string'); - } - if (/[\n\r"]/.test(relativeSegmentPath)) { - throw new TypeError( - 'Segment paths cannot contain line feed or carriage return characters.', - ); - } + let segmentInfo: HlsOutputSegmentInfo | null = null; + let relativeSegmentPath: string; + let fullSegmentPath: string; assert(this.output._rootPath !== null); - const fullSegmentPath = joinPaths(joinPaths(this.output._rootPath, playlist.path), relativeSegmentPath); - playlist.nextSegmentId++; + + if (this.singleFilePerPlaylist) { + if (playlist.singleFile === null) { + const segmentInfo: HlsOutputSegmentInfo = { + n: playlist.nextSegmentId, + format: playlist.segmentFormat, + isSingleFile: true, + playlist: { + n: playlist.id, + tracks: playlist.tracks, + }, + }; + + relativeSegmentPath = await this.getSegmentPath(segmentInfo); + validateSegmentPath(relativeSegmentPath); + + fullSegmentPath = joinPaths( + joinPaths(this.output._rootPath, playlist.path), + relativeSegmentPath, + ); + + const target = await this.output._getTarget({ path: fullSegmentPath, isRoot: false }); + target._start(); + + playlist.singleFile = { + target, + path: relativeSegmentPath, + nextOffset: 0, + info: segmentInfo, + }; + } else { + relativeSegmentPath = playlist.singleFile.path; + fullSegmentPath = joinPaths( + joinPaths(this.output._rootPath, playlist.path), + relativeSegmentPath, + ); + } + } else { + segmentInfo = { + n: playlist.nextSegmentId, + format: playlist.segmentFormat, + isSingleFile: false, + playlist: { + n: playlist.id, + tracks: playlist.tracks, + }, + }; + + relativeSegmentPath = await this.getSegmentPath(segmentInfo); + validateSegmentPath(relativeSegmentPath); + + assert(this.output._rootPath !== null); + fullSegmentPath = joinPaths(joinPaths(this.output._rootPath, playlist.path), relativeSegmentPath); + playlist.nextSegmentId++; + } let segmentSize = 0; let outputTarget: Target | null = null; @@ -749,14 +800,22 @@ export class HlsMuxer extends Muxer { format: playlist.segmentFormat, rootPath: fullSegmentPath, target: async (request) => { - const target = await this.output._getTarget(request); - if (request.path === fullSegmentPath) { - outputTarget = target; - target.onwrite = (_, end) => segmentSize = Math.max(segmentSize, end); + if (playlist.singleFile) { + const slice = playlist.singleFile.target.slice(playlist.singleFile.nextOffset); + slice.onwrite = (_, end) => segmentSize = Math.max(segmentSize, end); + + return slice; + } else { + const target = await this.output._getTarget(request); + outputTarget = target; + target.onwrite = (_, end) => segmentSize = Math.max(segmentSize, end); + + return target; + } } - return target; + return this.output._getTarget(request); }, }); @@ -812,8 +871,10 @@ export class HlsMuxer extends Muxer { throw e; } - assert(outputTarget); - this.format._options.onSegment?.(outputTarget, segmentInfo); + if (segmentInfo) { + assert(outputTarget); + this.format._options.onSegment?.(outputTarget, segmentInfo); + } if (videoEndIndex > 0) { assert(videoTrack); @@ -841,10 +902,17 @@ export class HlsMuxer extends Muxer { path: relativeSegmentPath, duration: nextSegmentStartTimestamp - playlist.currentSegmentStartTimestamp, byteSize: segmentSize, + byteOffset: playlist.singleFile + ? playlist.singleFile.nextOffset + : null, }); playlist.currentSegmentStartTimestamp = nextSegmentStartTimestamp; playlist.currentSegmentStartTimestampIsFixed = true; // After the first segment, the timestamp is now fixed + + if (playlist.singleFile) { + playlist.singleFile.nextOffset += segmentSize; + } } } @@ -852,9 +920,13 @@ export class HlsMuxer extends Muxer { assert(this.output._rootPath !== null); const release = await this.mutex.acquire(); + for (const trackData of this.trackDatas) { + trackData.closed = true; + } + for (const playlist of this.playlists) { if (playlist.currentSegmentStartTimestamp !== null) { - await this.advancePlaylist(playlist, true); + await this.advancePlaylist(playlist); } else { // Never had any data written to it } @@ -876,6 +948,13 @@ export class HlsMuxer extends Muxer { // Write all playlists in parallel const playlistPromises = this.playlists.map(async (playlist) => { + if (playlist.singleFile) { + await playlist.singleFile.target._flush(); + await playlist.singleFile.target._finalize(); + + this.format._options.onSegment?.(playlist.singleFile.target, playlist.singleFile.info); + } + let targetDuration = this.targetSegmentDuration; for (const segment of playlist.writtenSegments) { targetDuration = Math.max(targetDuration, segment.duration); @@ -890,6 +969,9 @@ export class HlsMuxer extends Muxer { + (playlist.writtenSegments .map(segment => ( `#EXTINF:${+segment.duration.toFixed(12)},\n` // Trailing comma mandated by spec + + (segment.byteOffset !== null + ? `#EXT-X-BYTERANGE:${segment.byteSize}@${segment.byteOffset}\n` + : '') + `${segment.path}\n` )) .join('')) @@ -902,8 +984,11 @@ export class HlsMuxer extends Muxer { }); const target = await this.output._getTarget({ path: playlistPath, isRoot: false }); - const writer = target._createWriter(); + const writer = new Writer(target); + writer.start(); writer.write(textEncoder.encode(playlistText)); + + await writer.flush(); await writer.finalize(); }); @@ -1088,3 +1173,14 @@ export class HlsMuxer extends Muxer { release(); } } + +const validateSegmentPath = (path: string) => { + if (typeof path !== 'string') { + throw new TypeError('options.getSegmentPath must return or resolve to a string'); + } + if (/[\n\r"]/.test(path)) { + throw new TypeError( + 'Segment paths cannot contain line feed or carriage return characters.', + ); + } +}; diff --git a/src/hls/hls-segmented-input.ts b/src/hls/hls-segmented-input.ts index e32350c..c54a5a6 100644 --- a/src/hls/hls-segmented-input.ts +++ b/src/hls/hls-segmented-input.ts @@ -107,6 +107,8 @@ export class HlsSegmentedInput extends SegmentedInput { let lastByteRangeEnd: number | null = null; let nextByteRange: { offset: number; length: number } | null = null; let lastProgramDateTimeSeconds: number | null = null; + let targetDuration: number | null = null; + let segmentSeen = false; // Used for repeated parses where our job it is to only add the new segments let prevLastSegment = last(this.segments) ?? null; @@ -220,9 +222,19 @@ export class HlsSegmentedInput extends SegmentedInput { if (line.startsWith('#EXTINF:')) { if (prevLastSegment) { + segmentSeen = true; continue; } + if (!segmentSeen) { + if (lastProgramDateTimeSeconds === null && nextSequenceNumber > 0 && targetDuration !== null) { + // Offset the first segment's start timestamp by the following: + accumulatedTime = nextSequenceNumber * targetDuration; + } + + segmentSeen = true; + } + const extinfContent = line.slice(8); const commaIndex = extinfContent.indexOf(','); const durationStr = commaIndex === -1 ? extinfContent : extinfContent.slice(0, commaIndex); @@ -400,6 +412,7 @@ export class HlsSegmentedInput extends SegmentedInput { } this.refreshInterval = duration; + targetDuration = duration; } else if (line === '#EXT-X-ENDLIST') { this.streamHasEnded = true; break; // No need to keep reading after this @@ -411,6 +424,10 @@ export class HlsSegmentedInput extends SegmentedInput { } } } + + if (!headerRead) { + throw new Error('Invalid M3U8 file; no #EXTM3U header.'); + } } async getFirstSegment() { diff --git a/src/isobmff/isobmff-muxer.ts b/src/isobmff/isobmff-muxer.ts index 6198bed..b6431ae 100644 --- a/src/isobmff/isobmff-muxer.ts +++ b/src/isobmff/isobmff-muxer.ts @@ -9,7 +9,8 @@ import { Box, free, ftyp, IsobmffBoxWriter, mdat, mfra, moof, moov, vtta, vttc, vtte } from './isobmff-boxes'; import { Muxer } from '../muxer'; import { Output, OutputAudioTrack, OutputSubtitleTrack, OutputTrack, OutputVideoTrack } from '../output'; -import { BufferTargetWriter, Writer } from '../writer'; +import { Writer } from '../writer'; +import { BufferTarget } from '../target'; import { assert, computeRationalApproximation, last, promiseWithResolvers, Rational, simplifyRational } from '../misc'; import { IsobmffOutputFormatOptions, IsobmffOutputFormat, MovOutputFormat } from '../output-format'; import { inlineTimestampRegex, SubtitleConfig, SubtitleCue, SubtitleMetadata } from '../subtitles'; @@ -25,7 +26,6 @@ import { } from '../codec'; import { MAX_ADTS_FRAME_HEADER_SIZE, MIN_ADTS_FRAME_HEADER_SIZE, readAdtsFrameHeader } from '../adts/adts-reader'; import { FileSlice } from '../reader'; -import { BufferTarget } from '../target'; import { EncodedPacket, PacketType } from '../packet'; import { concatNalUnitsInLengthPrefixed, @@ -158,7 +158,7 @@ export class IsobmffMuxer extends Muxer { isQuickTime: boolean; private auxTarget = new BufferTarget(); - private auxWriter = this.auxTarget._createWriter(); + private auxWriter = new Writer(this.auxTarget); private auxBoxWriter = new IsobmffBoxWriter(this.auxWriter); private mdat: Box | null = null; @@ -191,12 +191,12 @@ export class IsobmffMuxer extends Muxer { // If the fastStart option isn't defined, enable in-memory fast start if the target is an ArrayBuffer, as the // memory usage remains identical - const fastStartDefault = this.writer instanceof BufferTargetWriter ? 'in-memory' : false; + const fastStartDefault = this.writer.target instanceof BufferTarget ? 'in-memory' : false; this.fastStart = this.format._options.fastStart ?? fastStartDefault; this.isFragmented = this.fastStart === 'fragmented'; if (this.fastStart === 'in-memory' || this.isFragmented) { - this.writer.ensureMonotonicity = true; + this.writer.ensureMonotonicity(); } const holdsAvc = this.output._tracks.some(x => x.isVideoTrack() && x.source._codec === 'avc'); @@ -673,7 +673,7 @@ export class IsobmffMuxer extends Muxer { const box = vtte(); this.auxBoxWriter.writeBox(box); - const body = this.auxWriter.getSlice(0, this.auxWriter.getPos()); + const body = this.auxTarget._getSlice(0, this.auxWriter.getPos()); const sample = this.createSampleForTrack( trackData, body, @@ -728,7 +728,7 @@ export class IsobmffMuxer extends Muxer { } } - const body = this.auxWriter.getSlice(0, this.auxWriter.getPos()); + const body = this.auxTarget._getSlice(0, this.auxWriter.getPos()); const sample = this.createSampleForTrack(trackData, body, sampleStart, sampleEnd - sampleStart, 'key'); await this.registerSample(trackData, sample); diff --git a/src/matroska/matroska-muxer.ts b/src/matroska/matroska-muxer.ts index 242cf19..df9a97f 100644 --- a/src/matroska/matroska-muxer.ts +++ b/src/matroska/matroska-muxer.ts @@ -168,7 +168,7 @@ export class MatroskaMuxer extends Muxer { this.ebmlWriter = new EBMLWriter(this.writer); if (this.format._options.appendOnly) { - this.writer.ensureMonotonicity = true; + this.writer.ensureMonotonicity(); } this.writeEBMLHeader(); diff --git a/src/mpeg-ts/mpeg-ts-muxer.ts b/src/mpeg-ts/mpeg-ts-muxer.ts index 93253b8..a9fa0d2 100644 --- a/src/mpeg-ts/mpeg-ts-muxer.ts +++ b/src/mpeg-ts/mpeg-ts-muxer.ts @@ -96,7 +96,7 @@ export class MpegTsMuxer extends Muxer { const release = await this.mutex.acquire(); this.writer = await this.output._getRootWriter(); - this.writer.ensureMonotonicity = true; + this.writer.ensureMonotonicity(); release(); } diff --git a/src/ogg/ogg-muxer.ts b/src/ogg/ogg-muxer.ts index 4758f0e..17b6c9a 100644 --- a/src/ogg/ogg-muxer.ts +++ b/src/ogg/ogg-muxer.ts @@ -78,7 +78,7 @@ export class OggMuxer extends Muxer { const release = await this.mutex.acquire(); this.writer = await this.output._getRootWriter(); - this.writer.ensureMonotonicity = true; // Ogg is always monotonically written! + this.writer.ensureMonotonicity(); // Ogg is always monotonically written! release(); } diff --git a/src/output-format.ts b/src/output-format.ts index 41c70db..3af1219 100644 --- a/src/output-format.ts +++ b/src/output-format.ts @@ -1102,6 +1102,7 @@ export type HlsOutputPlaylistInfo = { export type HlsOutputSegmentInfo = { n: number; + isSingleFile: boolean; format: OutputFormat; playlist: HlsOutputPlaylistInfo; }; @@ -1109,11 +1110,13 @@ export type HlsOutputSegmentInfo = { export type HlsOutputFormatOptions = { segmentFormats: OutputFormat[]; targetDuration?: number; + singleFilePerPlaylist?: boolean; getPlaylistPath?: (info: HlsOutputPlaylistInfo) => MaybePromise; getSegmentPath?: (info: HlsOutputSegmentInfo) => MaybePromise; onMaster?: (content: string) => unknown; onPlaylist?: (content: string, info: HlsOutputPlaylistInfo) => unknown; + // Document how this is called for the single-file mode onSegment?: (target: Target, info: HlsOutputSegmentInfo) => unknown; }; @@ -1139,6 +1142,9 @@ export class HlsOutputFormat extends OutputFormat { ) { throw new TypeError('options.targetDuration, when provided, must be a positive number.'); } + if (options.singleFilePerPlaylist !== undefined && typeof options.singleFilePerPlaylist !== 'boolean') { + throw new TypeError('options.singleFilePerPlaylist, when provided, must be a boolean.'); + } if (options.getPlaylistPath !== undefined && typeof options.getPlaylistPath !== 'function') { throw new TypeError('options.getPlaylistPath, when provided, must be a function.'); } @@ -1216,5 +1222,3 @@ export class HlsOutputFormat extends OutputFormat { return ` Using different segment formats may grant support for this codec.`; } } - -export const HLS_OUTPUT_FORMATS_DEFAULT = [new AdtsOutputFormat(), new MpegTsOutputFormat()]; diff --git a/src/output.ts b/src/output.ts index 405ba61..6b7bb7a 100644 --- a/src/output.ts +++ b/src/output.ts @@ -381,17 +381,17 @@ export class Output< _getRootWriter() { return this._rootWriterPromise ??= (async () => { - let writer: Writer; + let target: Target; if (typeof this._target === 'function') { assert(this._rootPath !== null); - const rootTarget = await this._getTarget({ path: this._rootPath, isRoot: true }); - writer = rootTarget._createWriter(); + target = await this._getTarget({ path: this._rootPath, isRoot: true }); } else { - writer = this._target._createWriter(); + target = this._target; this.onTarget?.(this._target, null); } + const writer = new Writer(target); writer.start(); return writer; })(); diff --git a/src/target.ts b/src/target.ts index d61df17..c06e399 100644 --- a/src/target.ts +++ b/src/target.ts @@ -7,7 +7,6 @@ */ import type { FileHandle } from 'node:fs/promises'; -import { BufferTargetWriter, NullTargetWriter, StreamTargetWriter, Writer } from './writer'; import { Output } from './output'; import * as nodeAlias from './node'; import { assert } from './misc'; @@ -26,7 +25,18 @@ export abstract class Target { _output: Output | null = null; /** @internal */ - abstract _createWriter(): Writer; + _ensureMonotonicity = false; + + /** @internal */ + abstract _start(): void; + /** @internal */ + abstract _write(data: Uint8Array, pos: number): void; + /** @internal */ + abstract _flush(): Promise; + /** @internal */ + abstract _finalize(): Promise; + /** @internal */ + abstract _close(): Promise; /** * Called each time data is written to the target. Will be called with the byte range into which data was written. @@ -37,8 +47,19 @@ export abstract class Target { onwrite: ((start: number, end: number) => unknown) | null = null; onfinalized: (() => unknown) | null = null; + + slice(offset: number) { + if (!Number.isInteger(offset) && offset < 0) { + throw new TypeError('offset must be a non-negative integer.'); + } + + return new RangedTarget(this, offset); + } } +const ARRAY_BUFFER_INITIAL_SIZE = 2 ** 16; +const ARRAY_BUFFER_MAX_SIZE = 2 ** 32; + /** * A target that writes data directly into an ArrayBuffer in memory. Great for performance, but not suitable for very * large files. The buffer will be available once the output has been finalized. @@ -50,8 +71,92 @@ export class BufferTarget extends Target { buffer: ArrayBuffer | null = null; /** @internal */ - _createWriter() { - return new BufferTargetWriter(this); + _buffer: ArrayBuffer; + /** @internal */ + _bytes: Uint8Array; + /** @internal */ + _maxPos = 0; + /** @internal */ + _supportsResize: boolean; + + constructor() { + super(); + + this._supportsResize = 'resize' in new ArrayBuffer(0); + if (this._supportsResize) { + try { + // @ts-expect-error Don't want to bump "lib" in tsconfig + this._buffer = new ArrayBuffer(ARRAY_BUFFER_INITIAL_SIZE, { maxByteLength: ARRAY_BUFFER_MAX_SIZE }); + } catch { + this._buffer = new ArrayBuffer(ARRAY_BUFFER_INITIAL_SIZE); + this._supportsResize = false; + } + } else { + this._buffer = new ArrayBuffer(ARRAY_BUFFER_INITIAL_SIZE); + } + + this._bytes = new Uint8Array(this._buffer); + } + + /** @internal */ + _ensureSize(size: number) { + let newLength = this._buffer.byteLength; + while (newLength < size) newLength *= 2; + + if (newLength === this._buffer.byteLength) return; + + if (newLength > ARRAY_BUFFER_MAX_SIZE) { + throw new Error( + `ArrayBuffer exceeded maximum size of ${ARRAY_BUFFER_MAX_SIZE} bytes. Please consider using another` + + ` target.`, + ); + } + + if (this._supportsResize) { + // Use resize if it exists + // @ts-expect-error Don't want to bump "lib" in tsconfig + // eslint-disable-next-line @typescript-eslint/no-unsafe-call + this._buffer.resize(newLength); + // The Uint8Array scales automatically + } else { + const newBuffer = new ArrayBuffer(newLength); + const newBytes = new Uint8Array(newBuffer); + newBytes.set(this._bytes, 0); + + this._buffer = newBuffer; + this._bytes = newBytes; + } + } + + /** @internal */ + _start() {} + + /** @internal */ + _write(data: Uint8Array, pos: number) { + this._ensureSize(pos + data.byteLength); + + this._bytes.set(data, pos); + + this._maxPos = Math.max(this._maxPos, pos + data.byteLength); + + this.onwrite?.(pos, pos + data.byteLength); + } + + /** @internal */ + async _flush() {} + + /** @internal */ + async _finalize() { + this.buffer = this._buffer.slice(0, this._maxPos); + this.onfinalized?.(); + } + + /** @internal */ + async _close() {} + + /** @internal */ + _getSlice(start: number, end: number) { + return this._bytes.slice(start, end); } } @@ -85,6 +190,21 @@ export type StreamTargetOptions = { chunkSize?: number; }; +const DEFAULT_CHUNK_SIZE = 2 ** 24; +const MAX_CHUNKS_AT_ONCE = 2; + +type Chunk = { + start: number; + written: ChunkSection[]; + data: Uint8Array; + shouldFlush: boolean; +}; + +type ChunkSection = { + start: number; + end: number; +}; + /** * This target writes data to a [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream), * making it a general-purpose target for writing data anywhere. It is also compatible with @@ -100,6 +220,33 @@ export class StreamTarget extends Target { /** @internal */ _options: StreamTargetOptions; + /** @internal */ + _sections: { + data: Uint8Array; + start: number; + }[] = []; + + /** @internal */ + _lastWriteEnd = 0; + /** @internal */ + _lastFlushEnd = 0; + /** @internal */ + _streamWriter: WritableStreamDefaultWriter | null = null; + /** @internal */ + _writeError: unknown = null; + + // These variables regard chunked mode: + /** @internal */ + _chunked: boolean; + /** @internal */ + _chunkSize: number; + /** + * The data is divided up into fixed-size chunks, whose contents are first filled in RAM and then flushed out. + * A chunk is flushed if all of its contents have been written. + */ + /** @internal */ + _chunks: Chunk[] = []; + /** Creates a new {@link StreamTarget} which writes to the specified `writable`. */ constructor( writable: WritableStream, @@ -122,11 +269,242 @@ export class StreamTarget extends Target { this._writable = writable; this._options = options; + + this._chunked = options.chunked ?? false; + this._chunkSize = options.chunkSize ?? DEFAULT_CHUNK_SIZE; } /** @internal */ - _createWriter() { - return new StreamTargetWriter(this); + _start() { + this._streamWriter = this._writable.getWriter(); + } + + /** @internal */ + _write(data: Uint8Array, pos: number) { + if (pos > this._lastWriteEnd) { + const paddingBytesNeeded = pos - this._lastWriteEnd; + this._write(new Uint8Array(paddingBytesNeeded), this._lastWriteEnd); + } + + this._sections.push({ + data: data.slice(), + start: pos, + }); + + this._lastWriteEnd = Math.max(this._lastWriteEnd, pos + data.byteLength); + + this.onwrite?.(pos, pos + data.byteLength); + } + + /** @internal */ + async _flush() { + if (this._writeError !== null) { + // eslint-disable-next-line @typescript-eslint/only-throw-error + throw this._writeError; + } + + assert(this._streamWriter); + if (this._sections.length === 0) return; + + const chunks: { + start: number; + size: number; + data?: Uint8Array; + }[] = []; + const sorted = [...this._sections].sort((a, b) => a.start - b.start); + + chunks.push({ + start: sorted[0]!.start, + size: sorted[0]!.data.byteLength, + }); + + // Figure out how many contiguous chunks we have + for (let i = 1; i < sorted.length; i++) { + const lastChunk = chunks[chunks.length - 1]!; + const section = sorted[i]!; + + if (section.start <= lastChunk.start + lastChunk.size) { + lastChunk.size = Math.max(lastChunk.size, section.start + section.data.byteLength - lastChunk.start); + } else { + chunks.push({ + start: section.start, + size: section.data.byteLength, + }); + } + } + + for (const chunk of chunks) { + chunk.data = new Uint8Array(chunk.size); + + // Make sure to write the data in the correct order for correct overwriting + for (const section of this._sections) { + // Check if the section is in the chunk + if (chunk.start <= section.start && section.start < chunk.start + chunk.size) { + chunk.data.set(section.data, section.start - chunk.start); + } + } + + if (this._streamWriter.desiredSize !== null && this._streamWriter.desiredSize <= 0) { + await this._streamWriter.ready; // Allow the writer to apply backpressure + } + + if (this._chunked) { + // Let's first gather the data into bigger chunks before writing it + this._writeDataIntoChunks(chunk.data, chunk.start); + this._tryToFlushChunks(); + } else { + if (this._ensureMonotonicity && chunk.start !== this._lastFlushEnd) { + throw new Error('Internal error: Monotonicity violation.'); + } + + void this._streamWriter.write({ + type: 'write', + data: chunk.data, + position: chunk.start, + }).catch((error) => { + this._writeError ??= error; + }); + + this._lastFlushEnd = chunk.start + chunk.data.byteLength; + } + } + + this._sections.length = 0; + } + + /** @internal */ + _writeDataIntoChunks(data: Uint8Array, position: number) { + // First, find the chunk to write the data into, or create one if none exists + let chunkIndex = this._chunks.findIndex(x => x.start <= position && position < x.start + this._chunkSize); + if (chunkIndex === -1) chunkIndex = this._createChunk(position); + const chunk = this._chunks[chunkIndex]!; + + // Figure out how much to write to the chunk, and then write to the chunk + const relativePosition = position - chunk.start; + const toWrite = data.subarray(0, Math.min(this._chunkSize - relativePosition, data.byteLength)); + chunk.data.set(toWrite, relativePosition); + + // Create a section describing the region of data that was just written to + const section: ChunkSection = { + start: relativePosition, + end: relativePosition + toWrite.byteLength, + }; + this._insertSectionIntoChunk(chunk, section); + + // Queue chunk for flushing to target if it has been fully written to + if (chunk.written[0]!.start === 0 && chunk.written[0]!.end === this._chunkSize) { + chunk.shouldFlush = true; + } + + // Make sure we don't hold too many chunks in memory at once to keep memory usage down + if (this._chunks.length > MAX_CHUNKS_AT_ONCE) { + // Flush all but the last chunk + for (let i = 0; i < this._chunks.length - 1; i++) { + this._chunks[i]!.shouldFlush = true; + } + this._tryToFlushChunks(); + } + + // If the data didn't fit in one chunk, recurse with the remaining data + if (toWrite.byteLength < data.byteLength) { + this._writeDataIntoChunks(data.subarray(toWrite.byteLength), position + toWrite.byteLength); + } + } + + /** @internal */ + _insertSectionIntoChunk(chunk: Chunk, section: ChunkSection) { + let low = 0; + let high = chunk.written.length - 1; + let index = -1; + + // Do a binary search to find the last section with a start not larger than `section`'s start + while (low <= high) { + const mid = Math.floor(low + (high - low + 1) / 2); + + if (chunk.written[mid]!.start <= section.start) { + low = mid + 1; + index = mid; + } else { + high = mid - 1; + } + } + + // Insert the new section + chunk.written.splice(index + 1, 0, section); + if (index === -1 || chunk.written[index]!.end < section.start) index++; + + // Merge overlapping sections + while (index < chunk.written.length - 1 && chunk.written[index]!.end >= chunk.written[index + 1]!.start) { + chunk.written[index]!.end = Math.max(chunk.written[index]!.end, chunk.written[index + 1]!.end); + chunk.written.splice(index + 1, 1); + } + } + + /** @internal */ + _createChunk(includesPosition: number) { + const start = Math.floor(includesPosition / this._chunkSize) * this._chunkSize; + const chunk: Chunk = { + start, + data: new Uint8Array(this._chunkSize), + written: [], + shouldFlush: false, + }; + this._chunks.push(chunk); + this._chunks.sort((a, b) => a.start - b.start); + + return this._chunks.indexOf(chunk); + } + + /** @internal */ + _tryToFlushChunks(force = false) { + assert(this._streamWriter); + + for (let i = 0; i < this._chunks.length; i++) { + const chunk = this._chunks[i]!; + if (!chunk.shouldFlush && !force) continue; + + for (const section of chunk.written) { + const position = chunk.start + section.start; + if (this._ensureMonotonicity && position !== this._lastFlushEnd) { + throw new Error('Internal error: Monotonicity violation.'); + } + + void this._streamWriter.write({ + type: 'write', + data: chunk.data.subarray(section.start, section.end), + position, + }).catch((error) => { + this._writeError ??= error; + }); + + this._lastFlushEnd = chunk.start + section.end; + } + + this._chunks.splice(i--, 1); + } + } + + /** @internal */ + async _finalize() { + if (this._chunked) { + this._tryToFlushChunks(true); + } + + if (this._writeError !== null) { + // eslint-disable-next-line @typescript-eslint/only-throw-error + throw this._writeError; + } + + assert(this._streamWriter); + await this._streamWriter.ready; + await this._streamWriter.close(); + + this.onfinalized?.(); + } + + /** @internal */ + async _close() { + return this._streamWriter?.close(); } } @@ -187,8 +565,30 @@ export class FilePathTarget extends Target { } /** @internal */ - _createWriter(): Writer { - return this._streamTarget._createWriter(); + _start() { + this._streamTarget._start(); + } + + /** @internal */ + _write(data: Uint8Array, pos: number) { + this._streamTarget._write(data, pos); + this.onwrite?.(pos, pos + data.byteLength); + } + + /** @internal */ + async _flush() { + return this._streamTarget._flush(); + } + + /** @internal */ + async _finalize() { + await this._streamTarget._finalize(); + this.onfinalized?.(); + } + + /** @internal */ + async _close() { + return this._streamTarget._close(); } } @@ -200,7 +600,59 @@ export class FilePathTarget extends Target { */ export class NullTarget extends Target { /** @internal */ - _createWriter() { - return new NullTargetWriter(this); + _start() {} + + /** @internal */ + + _write(data: Uint8Array, pos: number) { + this.onwrite?.(pos, pos + data.byteLength); } + + /** @internal */ + async _flush() {} + + /** @internal */ + async _finalize() { + this.onfinalized?.(); + } + + /** @internal */ + async _close() {} +} + +export class RangedTarget extends Target { + /** @internal */ + _baseTarget: Target; + /** @internal */ + _offset: number; + + /** @internal */ + constructor(baseTarget: Target, offset: number) { + super(); + + this._baseTarget = baseTarget; + this._offset = offset; + } + + /** @internal */ + _start() {} + + /** @internal */ + _write(data: Uint8Array, pos: number): void { + this._baseTarget._write(data, this._offset + pos); + this.onwrite?.(pos, pos + data.byteLength); + } + + /** @internal */ + _flush() { + return this._baseTarget._flush(); + } + + /** @internal */ + async _finalize() { + this.onfinalized?.(); + } + + /** @internal */ + async _close() {} } diff --git a/src/writer.ts b/src/writer.ts index fbbf026..88b3e30 100644 --- a/src/writer.ts +++ b/src/writer.ts @@ -6,33 +6,63 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ -import { BufferTarget, NullTarget, StreamTarget, StreamTargetChunk } from './target'; -import { assert } from './misc'; +import { Target } from './target'; -export abstract class Writer { - /** Setting this to true will cause the writer to ensure data is written in a strictly monotonic, streamable way. */ - ensureMonotonicity = false; +export class Writer { + target: Target; - start() {} + private pos = 0; + + constructor(target: Target) { + this.target = target; + } + + start() { + this.target._start(); + } + + ensureMonotonicity() { + this.target._ensureMonotonicity = true; + // Note that this currently is without effect for RangedTarget. But, should be fine since its use is rare + } /** Writes the given data to the target, at the current position. */ - abstract write(data: Uint8Array): void; + write(data: Uint8Array) { + this.maybeTrackWrites(data); + this.target._write(data, this.pos); + this.pos += data.byteLength; + } + /** Sets the current position for future writes to a new one. */ - abstract seek(newPos: number): void; + seek(newPos: number) { + this.pos = newPos; + } + /** Returns the current position. */ - abstract getPos(): number; + getPos() { + return this.pos; + } + /** Signals to the writer that it may be time to flush. */ - abstract flush(): Promise; + async flush() { + return this.target._flush(); + } + /** Called after muxing has finished. */ - abstract finalize(): Promise; + async finalize() { + await this.target._finalize(); + } + /** Closes the writer. */ - abstract close(): Promise; + async close() { + return this.target._close(); + } private trackedWrites: Uint8Array | null = null; private trackedStart = -1; private trackedEnd = -1; - protected maybeTrackWrites(data: Uint8Array) { + private maybeTrackWrites(data: Uint8Array) { if (!this.trackedWrites) { return; } @@ -88,425 +118,3 @@ export abstract class Writer { return result; } } - -const ARRAY_BUFFER_INITIAL_SIZE = 2 ** 16; -const ARRAY_BUFFER_MAX_SIZE = 2 ** 32; - -export class BufferTargetWriter extends Writer { - private pos = 0; - private target: BufferTarget; - private buffer: ArrayBuffer; - private bytes: Uint8Array; - private maxPos = 0; - private supportsResize: boolean; - - constructor(target: BufferTarget) { - super(); - - this.target = target; - - this.supportsResize = 'resize' in new ArrayBuffer(0); - if (this.supportsResize) { - try { - // @ts-expect-error Don't want to bump "lib" in tsconfig - this.buffer = new ArrayBuffer(ARRAY_BUFFER_INITIAL_SIZE, { maxByteLength: ARRAY_BUFFER_MAX_SIZE }); - } catch { - this.buffer = new ArrayBuffer(ARRAY_BUFFER_INITIAL_SIZE); - this.supportsResize = false; - } - } else { - this.buffer = new ArrayBuffer(ARRAY_BUFFER_INITIAL_SIZE); - } - - this.bytes = new Uint8Array(this.buffer); - } - - private ensureSize(size: number) { - let newLength = this.buffer.byteLength; - while (newLength < size) newLength *= 2; - - if (newLength === this.buffer.byteLength) return; - - if (newLength > ARRAY_BUFFER_MAX_SIZE) { - throw new Error( - `ArrayBuffer exceeded maximum size of ${ARRAY_BUFFER_MAX_SIZE} bytes. Please consider using another` - + ` target.`, - ); - } - - if (this.supportsResize) { - // Use resize if it exists - // @ts-expect-error Don't want to bump "lib" in tsconfig - // eslint-disable-next-line @typescript-eslint/no-unsafe-call - this.buffer.resize(newLength); - // The Uint8Array scales automatically - } else { - const newBuffer = new ArrayBuffer(newLength); - const newBytes = new Uint8Array(newBuffer); - newBytes.set(this.bytes, 0); - - this.buffer = newBuffer; - this.bytes = newBytes; - } - } - - write(data: Uint8Array) { - this.maybeTrackWrites(data); - - this.ensureSize(this.pos + data.byteLength); - - this.bytes.set(data, this.pos); - this.target.onwrite?.(this.pos, this.pos + data.byteLength); - - this.pos += data.byteLength; - this.maxPos = Math.max(this.maxPos, this.pos); - } - - seek(newPos: number) { - this.pos = newPos; - } - - getPos() { - return this.pos; - } - - async flush() {} - - async finalize() { - this.ensureSize(this.pos); - this.target.buffer = this.buffer.slice(0, Math.max(this.maxPos, this.pos)); - this.target.onfinalized?.(); - } - - async close() {} - - getSlice(start: number, end: number) { - return this.bytes.slice(start, end); - } -} - -const DEFAULT_CHUNK_SIZE = 2 ** 24; -const MAX_CHUNKS_AT_ONCE = 2; - -interface Chunk { - start: number; - written: ChunkSection[]; - data: Uint8Array; - shouldFlush: boolean; -} - -interface ChunkSection { - start: number; - end: number; -} - -/** - * Writes to a StreamTarget every time it is flushed, sending out all of the new data written since the - * last flush. This is useful for streaming applications, like piping the output to disk. When using the chunked mode, - * data will first be accumulated in larger chunks, and then the entire chunk will be flushed out at once when ready. - */ -export class StreamTargetWriter extends Writer { - private pos = 0; - private target: StreamTarget; - private sections: { - data: Uint8Array; - start: number; - }[] = []; - - private lastWriteEnd = 0; - private lastFlushEnd = 0; - private writer: WritableStreamDefaultWriter | null = null; - private writeError: unknown = null; - - // These variables regard chunked mode: - private chunked: boolean; - private chunkSize: number; - /** - * The data is divided up into fixed-size chunks, whose contents are first filled in RAM and then flushed out. - * A chunk is flushed if all of its contents have been written. - */ - private chunks: Chunk[] = []; - - constructor(target: StreamTarget) { - super(); - - this.target = target; - - this.chunked = target._options.chunked ?? false; - this.chunkSize = target._options.chunkSize ?? DEFAULT_CHUNK_SIZE; - } - - override start() { - this.writer = this.target._writable.getWriter(); - } - - write(data: Uint8Array) { - if (this.pos > this.lastWriteEnd) { - const paddingBytesNeeded = this.pos - this.lastWriteEnd; - this.pos = this.lastWriteEnd; - this.write(new Uint8Array(paddingBytesNeeded)); - } - - this.maybeTrackWrites(data); - - this.sections.push({ - 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); - } - - seek(newPos: number) { - this.pos = newPos; - } - - getPos() { - return this.pos; - } - - async flush() { - if (this.writeError !== null) { - // eslint-disable-next-line @typescript-eslint/only-throw-error - throw this.writeError; - } - - if (this.pos > this.lastWriteEnd) { - // There's a "void" between the last written byte and the next byte we're about to write. Let's pad that - // void with zeroes explicitly. - const paddingBytesNeeded = this.pos - this.lastWriteEnd; - this.pos = this.lastWriteEnd; - this.write(new Uint8Array(paddingBytesNeeded)); - } - - assert(this.writer); - if (this.sections.length === 0) return; - - const chunks: { - start: number; - size: number; - data?: Uint8Array; - }[] = []; - const sorted = [...this.sections].sort((a, b) => a.start - b.start); - - chunks.push({ - start: sorted[0]!.start, - size: sorted[0]!.data.byteLength, - }); - - // Figure out how many contiguous chunks we have - for (let i = 1; i < sorted.length; i++) { - const lastChunk = chunks[chunks.length - 1]!; - const section = sorted[i]!; - - if (section.start <= lastChunk.start + lastChunk.size) { - lastChunk.size = Math.max(lastChunk.size, section.start + section.data.byteLength - lastChunk.start); - } else { - chunks.push({ - start: section.start, - size: section.data.byteLength, - }); - } - } - - for (const chunk of chunks) { - chunk.data = new Uint8Array(chunk.size); - - // Make sure to write the data in the correct order for correct overwriting - for (const section of this.sections) { - // Check if the section is in the chunk - if (chunk.start <= section.start && section.start < chunk.start + chunk.size) { - chunk.data.set(section.data, section.start - chunk.start); - } - } - - if (this.writer.desiredSize !== null && this.writer.desiredSize <= 0) { - await this.writer.ready; // Allow the writer to apply backpressure - } - - if (this.chunked) { - // Let's first gather the data into bigger chunks before writing it - this.writeDataIntoChunks(chunk.data, chunk.start); - this.tryToFlushChunks(); - } else { - if (this.ensureMonotonicity && chunk.start !== this.lastFlushEnd) { - throw new Error('Internal error: Monotonicity violation.'); - } - - void this.writer.write({ - type: 'write', - data: chunk.data, - position: chunk.start, - }).catch((error) => { - this.writeError ??= error; - }); - - this.lastFlushEnd = chunk.start + chunk.data.byteLength; - } - } - - this.sections.length = 0; - } - - private writeDataIntoChunks(data: Uint8Array, position: number) { - // First, find the chunk to write the data into, or create one if none exists - let chunkIndex = this.chunks.findIndex(x => x.start <= position && position < x.start + this.chunkSize); - if (chunkIndex === -1) chunkIndex = this.createChunk(position); - const chunk = this.chunks[chunkIndex]!; - - // Figure out how much to write to the chunk, and then write to the chunk - const relativePosition = position - chunk.start; - const toWrite = data.subarray(0, Math.min(this.chunkSize - relativePosition, data.byteLength)); - chunk.data.set(toWrite, relativePosition); - - // Create a section describing the region of data that was just written to - const section: ChunkSection = { - start: relativePosition, - end: relativePosition + toWrite.byteLength, - }; - this.insertSectionIntoChunk(chunk, section); - - // Queue chunk for flushing to target if it has been fully written to - if (chunk.written[0]!.start === 0 && chunk.written[0]!.end === this.chunkSize) { - chunk.shouldFlush = true; - } - - // Make sure we don't hold too many chunks in memory at once to keep memory usage down - if (this.chunks.length > MAX_CHUNKS_AT_ONCE) { - // Flush all but the last chunk - for (let i = 0; i < this.chunks.length - 1; i++) { - this.chunks[i]!.shouldFlush = true; - } - this.tryToFlushChunks(); - } - - // If the data didn't fit in one chunk, recurse with the remaining data - if (toWrite.byteLength < data.byteLength) { - this.writeDataIntoChunks(data.subarray(toWrite.byteLength), position + toWrite.byteLength); - } - } - - private insertSectionIntoChunk(chunk: Chunk, section: ChunkSection) { - let low = 0; - let high = chunk.written.length - 1; - let index = -1; - - // Do a binary search to find the last section with a start not larger than `section`'s start - while (low <= high) { - const mid = Math.floor(low + (high - low + 1) / 2); - - if (chunk.written[mid]!.start <= section.start) { - low = mid + 1; - index = mid; - } else { - high = mid - 1; - } - } - - // Insert the new section - chunk.written.splice(index + 1, 0, section); - if (index === -1 || chunk.written[index]!.end < section.start) index++; - - // Merge overlapping sections - while (index < chunk.written.length - 1 && chunk.written[index]!.end >= chunk.written[index + 1]!.start) { - chunk.written[index]!.end = Math.max(chunk.written[index]!.end, chunk.written[index + 1]!.end); - chunk.written.splice(index + 1, 1); - } - } - - private createChunk(includesPosition: number) { - const start = Math.floor(includesPosition / this.chunkSize) * this.chunkSize; - const chunk: Chunk = { - start, - data: new Uint8Array(this.chunkSize), - written: [], - shouldFlush: false, - }; - this.chunks.push(chunk); - this.chunks.sort((a, b) => a.start - b.start); - - return this.chunks.indexOf(chunk); - } - - private tryToFlushChunks(force = false) { - assert(this.writer); - - for (let i = 0; i < this.chunks.length; i++) { - const chunk = this.chunks[i]!; - if (!chunk.shouldFlush && !force) continue; - - for (const section of chunk.written) { - const position = chunk.start + section.start; - if (this.ensureMonotonicity && position !== this.lastFlushEnd) { - throw new Error('Internal error: Monotonicity violation.'); - } - - void this.writer.write({ - type: 'write', - data: chunk.data.subarray(section.start, section.end), - position, - }).catch((error) => { - this.writeError ??= error; - }); - - this.lastFlushEnd = chunk.start + section.end; - } - - this.chunks.splice(i--, 1); - } - } - - async finalize() { - if (this.chunked) { - this.tryToFlushChunks(true); - } - - if (this.writeError !== null) { - // eslint-disable-next-line @typescript-eslint/only-throw-error - throw this.writeError; - } - - assert(this.writer); - await this.writer.ready; - await this.writer.close(); - - this.target.onfinalized?.(); - } - - async close() { - return this.writer?.close(); - } -} - -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; - } - - getPos() { - return this.pos; - } - - seek(newPos: number) { - this.pos = newPos; - } - - async flush() {} - - async finalize() { - this.target.onfinalized?.(); - } - - async close() {} -} diff --git a/test/node/hls-input.test.ts b/test/node/hls-input.test.ts index 58680b8..745ec53 100644 --- a/test/node/hls-input.test.ts +++ b/test/node/hls-input.test.ts @@ -476,8 +476,8 @@ test.concurrent('fMP4 Bitmovin', { timeout: 15_000 }, async () => { const videoTrack = await input.pluckVideoTrack(); assert(videoTrack); - expect(await videoTrack.getFirstTimestamp()).toBe(0); - expect(await videoTrack.computeDuration()).toBe(210.28); + expect(await videoTrack.getFirstTimestamp()).toBe(4); + expect(await videoTrack.computeDuration()).toBe(214.28); expect(sourceCount).toBe(5); }); diff --git a/test/node/hls-output.test.ts b/test/node/hls-output.test.ts index cdd7bb1..8ed6cda 100644 --- a/test/node/hls-output.test.ts +++ b/test/node/hls-output.test.ts @@ -1,7 +1,7 @@ import { expect, test, vi } from 'vitest'; import { Output, OutputTrackGroup } from '../../src/output.js'; -import { HLS_OUTPUT_FORMATS_DEFAULT, HlsOutputFormat, MpegTsOutputFormat } from '../../src/output-format.js'; -import { BufferTarget, NullTarget } from '../../src/target.js'; +import { HlsOutputFormat, MpegTsOutputFormat } from '../../src/output-format.js'; +import { BufferTarget, NullTarget, StreamTarget, StreamTargetChunk } from '../../src/target.js'; import { EncodedAudioPacketSource, EncodedVideoPacketSource } from '../../src/media-source.js'; import { HlsMuxer } from '../../src/hls/hls-muxer.js'; import { AudioCodec, VideoCodec } from '../../src/codec.js'; @@ -19,7 +19,7 @@ const audioSource = (codec: AudioCodec = 'aac') => new EncodedAudioPacketSource( test('Playlist assignment, single video', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -41,7 +41,7 @@ test('Playlist assignment, single video', async () => { test('Playlist assignment, single audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -63,7 +63,7 @@ test('Playlist assignment, single audio', async () => { test('Playlist assignment, multiple video', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -91,7 +91,7 @@ test('Playlist assignment, multiple video', async () => { test('Playlist assignment, multiple audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -119,7 +119,7 @@ test('Playlist assignment, multiple audio', async () => { test('Playlist assignment, multiple video with different metadata #1', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -155,7 +155,7 @@ test('Playlist assignment, multiple video with different metadata #1', async () test('Playlist assignment, multiple video with different metadata #2', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -191,7 +191,7 @@ test('Playlist assignment, multiple video with different metadata #2', async () test('Playlist assignment, multiple audio with different metadata', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -227,7 +227,7 @@ test('Playlist assignment, multiple audio with different metadata', async () => test('Playlist assignment, video and audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -250,7 +250,7 @@ test('Playlist assignment, video and audio', async () => { test('Playlist assignment, one video and multiple audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -292,7 +292,7 @@ test('Playlist assignment, one video and multiple audio', async () => { test('Playlist assignment, multiple video and one audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -334,7 +334,7 @@ test('Playlist assignment, multiple video and one audio', async () => { test('Playlist assignment, multiple video and audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -388,7 +388,7 @@ test('Playlist assignment, multiple video and audio', async () => { test('Playlist assignment, video and audio in different groups', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -421,7 +421,7 @@ test('Playlist assignment, video and audio in different groups', async () => { test('Playlist assignment, multiple video and audio in pairs', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -467,7 +467,7 @@ test('Playlist assignment, multiple video and audio in pairs', async () => { test('Playlist assignment, multiple video and audio with some unpaired', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -525,7 +525,7 @@ test('Playlist assignment, multiple video and audio with some unpaired', async ( test('Playlist assignment, multiple video and audio with multiple groups', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -594,7 +594,7 @@ test('Playlist assignment, multiple video and audio with multiple groups', async test('Playlist assignment, video with multiple audio codecs', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -639,7 +639,7 @@ test('Playlist assignment, video with multiple audio codecs', async () => { test('Playlist assignment, audio with multiple video codecs', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -677,7 +677,7 @@ test('Playlist assignment, audio with multiple video codecs', async () => { test('Playlist assignment, multiple video with conflicting audio interests', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -730,7 +730,7 @@ test('Playlist assignment, video paired with video', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -767,7 +767,7 @@ test('Playlist assignment, audio paired with audio', async () => { const output = new Output({ format: new HlsOutputFormat({ - segmentFormats: HLS_OUTPUT_FORMATS_DEFAULT, + segmentFormats: [new MpegTsOutputFormat()], }), target: () => new NullTarget(), rootPath: '', @@ -831,7 +831,6 @@ const setUpSegmentationEnvironment = async (options: { const output = new Output({ format: new HlsOutputFormat({ segmentFormats: [new MpegTsOutputFormat()], // No ADTS for simplicity - targetDuration: 2, }), target: (request) => { const target = new BufferTarget(); @@ -1826,7 +1825,6 @@ test('onSegment, onPlaylist, onMaster events', async () => { const output = new Output({ format: new HlsOutputFormat({ segmentFormats: [new MpegTsOutputFormat()], - targetDuration: 2, onSegment, onPlaylist, onMaster, @@ -1879,3 +1877,94 @@ test('onSegment, onPlaylist, onMaster events', async () => { expect(typeof onMaster.mock.calls[0]![0]).toBe('string'); expect(onMaster.mock.calls[0]![0]).toContain('#EXTM3U'); }); + +test('Single-file mode', async () => { + let playlistText: string | null = null; + const segmentPaths = new Set(); + + const output = new Output({ + format: new HlsOutputFormat({ + segmentFormats: [new MpegTsOutputFormat()], + singleFilePerPlaylist: true, + }), + target: (request) => { + const target = new BufferTarget(); + + if (request.path.includes('playlist')) { + target.onfinalized = () => { + playlistText = new TextDecoder().decode(target.buffer!); + }; + } else if (request.path.includes('segment')) { + segmentPaths.add(request.path); + } + + return target; + }, + rootPath: '', + }); + + const source = videoSource(); + output.addVideoTrack(source); + + await output.start(); + + await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 2.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3.5, 0), avcMetadata); + + await output.finalize(); + + // Only one segment file should have been created + expect(segmentPaths.size).toBe(1); + + expect(playlistText).not.toBeNull(); + expect(playlistText!.match(/#EXT-X-BYTERANGE/g)).toHaveLength(2); +}); + +test('StreamTarget, write is called for each target', async () => { + const writeCounts = new Map(); + + const output = new Output({ + format: new HlsOutputFormat({ + segmentFormats: [new MpegTsOutputFormat()], + }), + target: (request) => { + writeCounts.set(request.path, 0); + + const writable = new WritableStream({ + write() { + writeCounts.set(request.path, writeCounts.get(request.path)! + 1); + }, + }); + + return new StreamTarget(writable); + }, + rootPath: '', + }); + + const source = videoSource(); + output.addVideoTrack(source); + + await output.start(); + + await source.add(new EncodedPacket(avcPacketData, 'key', 0, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 0.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 1.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 2.5, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3, 0), avcMetadata); + await source.add(new EncodedPacket(avcPacketData, 'delta', 3.5, 0), avcMetadata); + + await output.finalize(); + + // Every StreamTarget should have been written to at least once + for (const [path, count] of writeCounts) { + expect(count, `Expected writes for ${path}`).toBeGreaterThanOrEqual(1); + } +}); diff --git a/todo.txt b/todo.txt index cc24b55..b8cbe0c 100644 --- a/todo.txt +++ b/todo.txt @@ -6,6 +6,5 @@ Also, why not just have the packet metadata on the packet? I think that would ma - Add an HLS "live mode". Challenge, where does bitrate come from? -- hls muxer starting offset -- hls demuxer offset by segment number * targetduration -- fmp4 muxing (CMAF??) \ No newline at end of file +- fmp4 muxing (CMAF??) +- ext-x-i-frames-only writing \ No newline at end of file