diff --git a/dev/convert.html b/dev/convert.html
index 110cbc3..a16477b 100644
--- a/dev/convert.html
+++ b/dev/convert.html
@@ -23,7 +23,9 @@
chunked: true,
chunkSize: 2**20
});
- const outputFormat = new Mediabunny.Mp4OutputFormat();
+ const outputFormat = new Mediabunny.HlsOutputFormat({
+ segmentFormat: new Mediabunny.MpegTsOutputFormat(),
+ });
const p = document.createElement('p');
p.textContent = 'Capturing...';
@@ -50,7 +52,7 @@
const output = new Mediabunny.Output({
format: outputFormat,
- target
+ target: new Mediabunny.PathedTarget('master.m3u8', ({ path }) => new Mediabunny.BufferTarget()),
});
let input;
@@ -119,76 +121,13 @@
bitrate: 320000
},
*/
- video: (track) => ({
- discard: true,
- //discard: true,
- //discard: !tracks.includes(track),
- //forceTranscode: true,
- //forceTranscode: true,
- //width: 1280,
- //forceTranscode: true,
- //allowRotationMetadata: false,
- //width: 720,
- //frameRate: 30,
- //bitrate: Mediabunny.QUALITY_VERY_LOW,
- //discard: true,
- /*
- process: (sample) => {
- if (!ctx) {
- // Create a canvas for image compositing
- const canvas = new OffscreenCanvas(
- sample.displayWidth,
- sample.displayHeight,
- );
- ctx = canvas.getContext('2d');
- }
-
- console.log(ctx.canvas.width, ctx.canvas.height);
-
- ctx.clearRect(0, 0, ctx.canvas.width, ctx.canvas.height);
- sample.drawWithFit(ctx, { fit: 'fill' });
- //ctx.drawImage(watermark, 32, 32);
-
- return ctx.canvas;
- },
- */
- //width: 300,
- //alpha: 'keep',
- //width: 320,
- //discard: true,
- //discard: true,
- //crop: {
- // left: 0,
- // top: 0,
- // width: 500,
- // height: 500,
- //},
- //rotate: 90,
- //width: 200,
- //height: 500,
- //fit: 'contain',
- //forceTranscode: true,
- //codec: 'avc',
- //fit: 'contain',
- //frameRate: 27.123,
- //width: 320,
- //forceTranscode: true,
- //codec: 'av1',
- //discard: true,
- //width: 1280,
- //discard: true,
- //width: 640
- //forceTranscode: true,
- //rotate: 90
- //width: 720 ?? 2160,
- //height: 1280 ?? 3840,
- //fit: 'contain',
- //rotate: 90,
- //width: 512,
- //height: 512,
- //width: 200,
- //height: 100,
- }),
+ video: [
+ { height: 1080 },
+ { height: 720 },
+ { height: 480 },
+ { height: 360 },
+ { height: 240 },
+ ],
tags: {} ?? {
title: 'Bigggy',
artist: 'Buck Bunny',
@@ -207,7 +146,7 @@
}
},
trim: {
- end: 5,
+ //end: 5,
//start,
//end: start + 5,
////start: 0,
diff --git a/src/hls/hls-muxer.ts b/src/hls/hls-muxer.ts
index a79c383..f5d88e9 100644
--- a/src/hls/hls-muxer.ts
+++ b/src/hls/hls-muxer.ts
@@ -898,6 +898,11 @@ export class HlsMuxer extends Muxer {
target: new PathedTarget(
fullSegmentPath,
async (request: TargetRequest) => {
+ const proxiedRequest: TargetRequest = {
+ ...request,
+ isRoot: false,
+ };
+
if (request.isRoot) {
if (playlist.singleFile) {
const slice = playlist.singleFile.target.slice(playlist.singleFile.nextOffset);
@@ -905,7 +910,7 @@ export class HlsMuxer extends Muxer {
return slice;
} else {
- const target = await this.output._getTarget(request);
+ const target = await this.output._getTarget(proxiedRequest);
outputTarget = target;
target.on('write', ({ end }) => segmentSize = Math.max(segmentSize, end));
@@ -913,7 +918,7 @@ export class HlsMuxer extends Muxer {
}
}
- return this.output._getTarget(request);
+ return this.output._getTarget(proxiedRequest);
},
),
initTarget: async () => {
@@ -955,8 +960,12 @@ export class HlsMuxer extends Muxer {
info: null,
};
+ const fullInitPath = joinPaths(
+ joinPaths(pathedTarget.rootPath, playlist.path),
+ initPath,
+ );
const target = await this.output._getTarget({
- path: initPath,
+ path: fullInitPath,
isRoot: false,
mimeType: playlist.segmentFormat.mimeType,
});
diff --git a/src/hls/hls-segmented-input.ts b/src/hls/hls-segmented-input.ts
index 48669f2..adec23c 100644
--- a/src/hls/hls-segmented-input.ts
+++ b/src/hls/hls-segmented-input.ts
@@ -11,7 +11,7 @@ import { ENCRYPTION_KEY_CACHE_GROUP, Input } from '../input';
import { Segment, SegmentedInput, SegmentedInputTrackDeclaration, SegmentRetrievalOptions } from '../segmented-input';
import { toDataView, joinPaths, last, assert, binarySearchLessOrEqual, arrayArgmin, wait } from '../misc';
import { readAllLines, readBytes, Reader } from '../reader';
-import { CustomPathedSource, ReadableStreamSource, SourceRef } from '../source';
+import { CustomPathedSource, ReadableStreamSource, SourceRef, SourceRequest } from '../source';
import { HlsDemuxer } from './hls-demuxer';
import {
AttributeList,
@@ -568,6 +568,11 @@ export class HlsSegmentedInput extends SegmentedInput {
async (request) => {
assert(request.isRoot); // Shouldn't fail since we don't allow recursive HLS
+ const proxiedRequest: SourceRequest = {
+ ...request,
+ isRoot: false,
+ };
+
let ref: SourceRef;
const needsSlice = hlsSegment.location.offset > 0 || hlsSegment.location.length !== null;
@@ -576,7 +581,7 @@ export class HlsSegmentedInput extends SegmentedInput {
|| hlsSegment.encryption.method === 'SAMPLE-AES'
|| hlsSegment.encryption.method === 'SAMPLE-AES-CTR'
) {
- ref = await this.input._getSourceCached(request);
+ ref = await this.input._getSourceCached(proxiedRequest);
if (needsSlice) {
const slice = ref.source.slice(
@@ -591,7 +596,7 @@ export class HlsSegmentedInput extends SegmentedInput {
const encryption = hlsSegment.encryption;
assert(encryption.iv);
- let ciphertextRef = await this.input._getSourceCached(request);
+ let ciphertextRef = await this.input._getSourceCached(proxiedRequest);
if (needsSlice) {
// Slice before decrypting
const slice = ciphertextRef.source.slice(
diff --git a/src/index.ts b/src/index.ts
index 99b7473..064bf36 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -129,16 +129,17 @@ export {
Target,
TargetEvents,
TargetRequest,
+ AppendOnlyStreamTarget,
BufferTarget,
BufferTargetOptions,
FilePathTarget,
FilePathTargetOptions,
NullTarget,
+ PathedTarget,
RangedTarget,
StreamTarget,
StreamTargetOptions,
StreamTargetChunk,
- PathedTarget,
} from './target';
export {
AnyIterable,
diff --git a/src/output.ts b/src/output.ts
index e1f7b67..5f4624a 100644
--- a/src/output.ts
+++ b/src/output.ts
@@ -378,7 +378,7 @@ export class Output<
/** @internal */
_muxer: Muxer;
/** @internal */
- _targets = new Set();
+ _unfinalizedTargets = new Set();
/** @internal */
_rootWriterPromise: Promise | null = null;
/** @internal */
@@ -439,12 +439,7 @@ export class Output<
throw new TypeError('options.target must be a Target or a PathedTarget.');
}
if (options.target instanceof Target) {
- if (options.target._output) {
- throw new Error('Target is already used for another output.');
- }
-
- options.target._output = this;
- this._targets.add(options.target);
+ this._rememberTarget(options.target);
}
if (
options.initTarget !== undefined
@@ -466,8 +461,7 @@ export class Output<
this._initTarget = options.initTarget ?? null;
if (this._initTarget instanceof Target) {
- this._initTarget._output = this;
- this._targets.add(this._initTarget);
+ this._rememberTarget(this._initTarget);
}
this._muxer = options.format._createMuxer(this);
@@ -498,18 +492,23 @@ export class Output<
assert(this._target instanceof PathedTarget);
const target = await this._getTargetValidated(request);
- target._output = this;
this._emit('target', { target, request, isRoot: request.isRoot });
if (this.state === 'canceled') {
await target._close();
} else {
- this._targets.add(target);
+ this._rememberTarget(target);
}
return target;
}
+ /** @internal */
+ _rememberTarget(target: Target) {
+ this._unfinalizedTargets.add(target);
+ target.on('finalized', () => this._unfinalizedTargets.delete(target), { once: true });
+ }
+
/** @internal */
async _getInitTarget(): Promise {
assert(this._initTarget !== null);
@@ -519,12 +518,11 @@ export class Output<
}
const target = await this._initTarget();
- target._output = this;
if (this.state === 'canceled') {
await target._close();
} else {
- this._targets.add(target);
+ this._rememberTarget(target);
}
return target;
@@ -558,13 +556,11 @@ export class Output<
const result = this._getTargetValidated(request);
const handleResult = (target: T) => {
- target._output = this;
-
if (this.state === 'canceled') {
// Promise thrown away here, but no way to surface it to the user really
void target._close();
} else {
- this._targets.add(target);
+ this._rememberTarget(target);
}
this._emit('target', { target, request, isRoot: true });
@@ -849,8 +845,8 @@ export class Output<
const promises = this._tracks.map(x => x.source._flushOrWaitForOngoingClose(true)); // Force close
await Promise.all(promises);
- await Promise.all([...this._targets].map(target => target._close()));
- this._targets.clear();
+ await Promise.all([...this._unfinalizedTargets].map(target => target._close()));
+ this._unfinalizedTargets.clear();
} finally {
release();
}
diff --git a/src/source.ts b/src/source.ts
index 733723f..7e2d298 100644
--- a/src/source.ts
+++ b/src/source.ts
@@ -558,7 +558,7 @@ export class BlobSource extends Source {
}
}
- worker.running = false;
+ this._orchestrator.signalWorkerStoppedRunning(worker);
if (worker.aborted) {
// MDN: "Calling this method signals a loss of interest in the stream by a consumer."
@@ -853,7 +853,7 @@ export class UrlSource extends PathedSource {
while (true) {
if (worker.currentPos >= worker.targetPos || worker.aborted) {
abortController.abort();
- worker.running = false;
+ this._orchestrator.signalWorkerStoppedRunning(worker);
return;
}
@@ -1201,7 +1201,7 @@ export class StreamSource extends Source {
}
}
- worker.running = false;
+ this._orchestrator.signalWorkerStoppedRunning(worker);
}
/** @internal */
@@ -1608,6 +1608,8 @@ class ReadOrchestrator {
minReadPosition: number,
maxReadPosition: number,
): MaybePromise {
+ assert(!this.disposed);
+
const prefetchRange = this.options.prefetchProfile(innerStart, innerEnd, this.workers);
const outerStart = Math.max(prefetchRange.start, minReadPosition);
const outerEnd = Math.min(prefetchRange.end, this.fileSize ?? Infinity, maxReadPosition);
@@ -1868,7 +1870,11 @@ class ReadOrchestrator {
for (let i = 0; i < this.workers.length; i++) {
const worker = this.workers[i]!;
- if (!worker.running && (!oldestWorker || worker.age < oldestWorker.age)) {
+ if (
+ !worker.running
+ && worker.pendingSlices.length === 0
+ && (!oldestWorker || worker.age < oldestWorker.age)
+ ) {
oldestIndex = i;
oldestWorker = worker;
}
@@ -1952,13 +1958,18 @@ class ReadOrchestrator {
// Here we merge everything into one "megaworker" that spans the entire file. We assume the passed-in worker
// is already configured to be a megaworker.
+ const uniqueSlices = new Set(worker.pendingSlices);
+
for (let i = 0; i < this.workers.length; i++) {
const otherWorker = this.workers[i]!;
if (otherWorker === worker) {
continue;
}
- worker.pendingSlices.push(...otherWorker.pendingSlices);
+ for (const slice of otherWorker.pendingSlices) {
+ uniqueSlices.add(slice);
+ }
+
otherWorker.aborted = true;
otherWorker.pendingSlices.length = 0;
this.workers.splice(i, 1);
@@ -1967,9 +1978,13 @@ class ReadOrchestrator {
for (let i = 0; i < this.queuedReads.length; i++) {
const queuedRead = this.queuedReads[i]!;
- worker.pendingSlices.push(...queuedRead.pendingSlices);
+
+ for (const slice of queuedRead.pendingSlices) {
+ uniqueSlices.add(slice);
+ }
}
+ worker.pendingSlices = [...uniqueSlices];
this.queuedReads.length = 0;
}
@@ -2101,6 +2116,16 @@ class ReadOrchestrator {
}
}
+ signalWorkerStoppedRunning(worker: ReadWorker) {
+ worker.running = false;
+
+ // When a worker stops running, that means it has hit its targetPos. It might still have pendingSlices assigned,
+ // but this is because those pending slices cover data that other workers are assigned to fill. Since targetPos
+ // has been reached, we can confidently say that this worker has completed its share of work on the pending
+ // slices and must no longer care about them.
+ worker.pendingSlices.length = 0;
+ }
+
/** Called when a worker reaches the end of the underlying data and must be cleaned up. */
onWorkerFinished(worker: ReadWorker) {
const index = this.workers.indexOf(worker);
diff --git a/src/target.ts b/src/target.ts
index ee008cb..20a6aeb 100644
--- a/src/target.ts
+++ b/src/target.ts
@@ -7,7 +7,6 @@
*/
import type { FileHandle } from 'node:fs/promises';
-import { Output } from './output';
import * as nodeAlias from './node';
import { assert, EventEmitter, FilePath, MaybePromise } from './misc';
@@ -39,7 +38,7 @@ export type TargetEvents = {
*/
export abstract class Target extends EventEmitter {
/** @internal */
- _output: Output | null = null;
+ _writerAcquired = false;
/** @internal */
_monotonicity: boolean | null = null; // null = unknown
@@ -582,6 +581,17 @@ export class StreamTarget extends Target {
}
}
+/**
+ * This target writes to a `WritableStream`, meaning all writes are necessarily append-only and involve no
+ * seeking. Great for streaming data to a source that can only accept sequential data, like an HTTP server processing
+ * an incoming upload.
+ *
+ * Note that using this target *requires* that the underlying format write data sequentially. Not all formats do this,
+ * and this target will throw for the formats that don't. Check the guide for more.
+ *
+ * @group Output targets
+ * @public
+ */
export class AppendOnlyStreamTarget extends Target {
/** @internal */
_writable: WritableStream;
@@ -787,7 +797,6 @@ export class RangedTarget extends Target {
this._baseTarget = baseTarget;
this._offset = offset;
- this._output = baseTarget._output;
}
/** @internal */
diff --git a/src/writer.ts b/src/writer.ts
index 2f610fd..d6e3731 100644
--- a/src/writer.ts
+++ b/src/writer.ts
@@ -17,8 +17,13 @@ export class Writer {
private pos = 0;
constructor(target: Target, isMonotonic: boolean) {
+ if (target._writerAcquired) {
+ throw new Error('Can\'t have multiple Writers for the same Target.');
+ }
+
this.target = target;
target._setMonotonicity(isMonotonic);
+ target._writerAcquired = true;
}
start() {
@@ -55,10 +60,8 @@ export class Writer {
/** Called after muxing has finished. */
async finalize() {
assert(this.started && !this.finalized);
- assert(this.target._output);
await this.target._finalize();
- this.target._output._targets.delete(this.target);
this.finalized = true;
}
diff --git a/test/node/hls-input.test.ts b/test/node/hls-input.test.ts
index e974a76..dcbe907 100644
--- a/test/node/hls-input.test.ts
+++ b/test/node/hls-input.test.ts
@@ -948,3 +948,24 @@ test.concurrent('Widevine encryption (SAMPLE-AES-CTR) succeeds with buffer keys'
assert(lastPacket);
expect(lastPacket.timestamp + lastPacket.duration).toBe(60);
});
+
+test.concurrent('SourceRequest.isRoot', async () => {
+ using input = new Input({
+ source: new CustomPathedSource(
+ 'https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8',
+ ({ path, isRoot }) => {
+ if (isRoot) {
+ expect(path).toBe('https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8');
+ }
+
+ return new UrlSource(path);
+ },
+ ),
+ formats: ALL_FORMATS,
+ });
+
+ const videoTrack = await input.getPrimaryVideoTrack();
+ assert(videoTrack);
+
+ await videoTrack.computeDuration();
+});
diff --git a/test/node/hls-output.test.ts b/test/node/hls-output.test.ts
index 6777b80..e275fe0 100644
--- a/test/node/hls-output.test.ts
+++ b/test/node/hls-output.test.ts
@@ -2903,3 +2903,37 @@ test('Append-only stream with monotonicity violation', async () => {
await expect(source.add(new EncodedPacket(avcPacketData, 'key', 2, 0), avcMetadata))
.rejects.toThrow('AppendOnlyStreamTarget');
});
+
+test('Relative paths & isRoot', async () => {
+ const output = new Output({
+ format: new HlsOutputFormat({
+ segmentFormat: new CmafOutputFormat(),
+ getPlaylistPath: () => `a/folder/playlist.m3u8`,
+ }),
+ target: new PathedTarget('path/to/master.m3u8', (request) => {
+ if (request.isRoot) {
+ expect(request.path).toBe('path/to/master.m3u8');
+ } else {
+ expect(request.path.startsWith('path/to/a/folder/')).toBe(true);
+ }
+
+ return new BufferTarget();
+ }),
+ });
+
+ 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();
+});