mirror of
https://github.com/zoriya/v10.git
synced 2026-08-16 02:45:09 +00:00
249 lines
9.9 KiB
JavaScript
249 lines
9.9 KiB
JavaScript
import { peek } from "../../../core/signals/primitives.js";
|
||
import { SerialRunner, Task } from "../../../core/tasks/task.js";
|
||
import { createMachineActor } from "../../../core/actors/create-machine-actor.js";
|
||
import "../../../media/types/index.js";
|
||
import { DEFAULT_FORWARD_BUFFER_CONFIG, calculateForwardFlushPoint, getSegmentsToLoad, isTimeRangeCovered, mergeTimeRanges } from "../../../media/buffer/forward-buffer.js";
|
||
import { DEFAULT_BACK_BUFFER_CONFIG, calculateBackBufferFlushPoint } from "../../../media/buffer/back-buffer.js";
|
||
import { DEFAULT_MESSAGE_PIPELINES } from "../../primitives/segment-load-pipeline.js";
|
||
//#region src/playback/actors/dom/segment-loader.ts
|
||
/**
|
||
* Wraps a LoadTask descriptor into a Task that runs the op's message pipeline
|
||
* (fetch/discover/stamp/dispatch, per the composition's `messagePipelines`).
|
||
* Updates in-flight context around the async region so the loading handler can
|
||
* make accurate continue/preempt decisions at any point, and checks the abort
|
||
* signal before each step.
|
||
*/
|
||
function makeLoadTask(op, { getContext, setContext, pipelines, deps }) {
|
||
return new Task(async (taskSignal) => {
|
||
if (taskSignal.aborted) return;
|
||
const frame = op.type === "append-segment" ? {
|
||
op,
|
||
meta: op.meta
|
||
} : { op };
|
||
try {
|
||
if (op.type === "append-init") setContext({
|
||
...getContext(),
|
||
inFlightInitTrackId: op.meta.trackId
|
||
});
|
||
else if (op.type === "append-segment") setContext({
|
||
...getContext(),
|
||
inFlightSegment: {
|
||
id: op.meta.id,
|
||
trackId: op.meta.trackId
|
||
}
|
||
});
|
||
for (const step of pipelines[op.type]) {
|
||
if (taskSignal.aborted) return;
|
||
await step(frame, taskSignal, deps);
|
||
}
|
||
} finally {
|
||
if (op.type === "append-init") setContext({
|
||
...getContext(),
|
||
inFlightInitTrackId: null
|
||
});
|
||
else if (op.type === "append-segment") setContext({
|
||
...getContext(),
|
||
inFlightSegment: null
|
||
});
|
||
}
|
||
});
|
||
}
|
||
/**
|
||
* Creates a SegmentLoaderActor for one track type (video or audio).
|
||
*
|
||
* Receives load assignments via `send()` and owns all execution: planning,
|
||
* removes, fetches, and appends. Coordinates with the SourceBufferActor for
|
||
* all physical SourceBuffer operations.
|
||
*
|
||
* Planning (Cases 1–3) happens in the `load` handler on every incoming
|
||
* message, producing an ordered LoadTask list. The runner drains that list
|
||
* sequentially via SerialRunner. When a new message arrives mid-run, the
|
||
* handler replans and either continues the in-flight operation (abortPending
|
||
* + schedule new remainder) or preempts it (abortAll + cancel SourceBuffer
|
||
* if needed + schedule new plan).
|
||
*
|
||
* @param sourceBufferActor - Shared SourceBufferActor reference (not owned)
|
||
* @param fetchBytes - Tracked fetch closure (owns throughput sampling for segments).
|
||
* Accepts an optional `minChunkSize` in options; init segments pass `Infinity`
|
||
* so the entire body accumulates as one chunk before appending.
|
||
* @param compositionDeps - The composition's `state`/`context`/`config`, threaded
|
||
* opaquely into each step's {@link StepDeps} (the loader never reads them). Lets
|
||
* injected steps (relocation) read composition signals at call time. Defaults to
|
||
* empty for standalone / base-pipeline use.
|
||
*/
|
||
function createSegmentLoaderActor(sourceBufferActor, fetchBytes, config = {}, compositionDeps = {
|
||
state: {},
|
||
context: {},
|
||
config: {}
|
||
}) {
|
||
const forwardBufferConfig = {
|
||
...DEFAULT_FORWARD_BUFFER_CONFIG,
|
||
...config.forwardBuffer
|
||
};
|
||
const backBufferConfig = {
|
||
...DEFAULT_BACK_BUFFER_CONFIG,
|
||
...config.backBuffer
|
||
};
|
||
const deps = {
|
||
state: compositionDeps.state,
|
||
context: compositionDeps.context,
|
||
config: {
|
||
...compositionDeps.config,
|
||
sourceBufferActor,
|
||
fetch: fetchBytes
|
||
}
|
||
};
|
||
const pipelines = (config.messagePipelines ?? DEFAULT_MESSAGE_PIPELINES)();
|
||
const getBufferedSegments = (allSegments) => {
|
||
const merged = mergeTimeRanges(peek(sourceBufferActor.snapshot).context.segments.filter((s) => !s.partial).map((s) => ({
|
||
start: s.startTime,
|
||
end: s.startTime + s.duration
|
||
})));
|
||
return allSegments.filter((s) => isTimeRangeCovered(s.startTime, s.startTime + s.duration, merged));
|
||
};
|
||
/**
|
||
* Translate a load message into an ordered LoadTask list based on committed
|
||
* actor state. In-flight awareness is handled separately in the load handler.
|
||
*
|
||
* @todo Rename alongside LoadTask (e.g. planOps).
|
||
*
|
||
* Case 1 — Removes: forward and back buffer flush points, segment-aligned.
|
||
* ABR-style track switches (same content, different bitrate) do not flush:
|
||
* appending new content overwrites existing buffer ranges, and the actor's
|
||
* time-aligned deduplication keeps the segment model accurate as new
|
||
* segments arrive.
|
||
*
|
||
* Cross-rendition track switches (audio language change, text language
|
||
* change) do flush: the buffered content is semantically incompatible with
|
||
* the newly-selected track, so overwrite-on-append would leave stale
|
||
* content playing until each replacement segment lands. Today's predicate:
|
||
* `actorCtx.initTrackLanguage !== track.language` — fires for language
|
||
* changes, no-ops for video / same-language audio bitrate switches.
|
||
* Future stage: pluggable predicate / strategy at actor construction time
|
||
* for codec-change (5.1 surround) and other cross-rendition shapes.
|
||
*
|
||
* Case 2 — Init: schedule if not yet committed for this track.
|
||
*
|
||
* Case 3 — Segments: all segments in the load window not yet committed.
|
||
*/
|
||
const planTasks = (message) => {
|
||
const { track, range } = message;
|
||
const actorCtx = peek(sourceBufferActor.snapshot).context;
|
||
const bufferedSegments = getBufferedSegments(track.segments);
|
||
const currentTime = range?.start ?? 0;
|
||
const tasks = [];
|
||
const isCrossRenditionSwitch = actorCtx.initTrackId !== void 0 && actorCtx.initTrackId !== track.id && actorCtx.initTrackLanguage !== track.language;
|
||
const removes = [];
|
||
const staleRanges = [];
|
||
if (range) {
|
||
if (isCrossRenditionSwitch) {
|
||
const staleStart = actorCtx.segments.find((s) => s.startTime <= currentTime && s.startTime + s.duration > currentTime)?.startTime ?? actorCtx.segments.find((s) => s.startTime > currentTime)?.startTime;
|
||
if (staleStart !== void 0) staleRanges.push({
|
||
start: staleStart,
|
||
end: Infinity
|
||
});
|
||
}
|
||
const forwardFlushStart = calculateForwardFlushPoint(bufferedSegments, currentTime, forwardBufferConfig);
|
||
if (forwardFlushStart < Infinity) removes.push({
|
||
start: forwardFlushStart,
|
||
end: Infinity
|
||
});
|
||
const backFlushEnd = calculateBackBufferFlushPoint(bufferedSegments, currentTime, backBufferConfig);
|
||
if (backFlushEnd > 0) removes.push({
|
||
start: 0,
|
||
end: backFlushEnd
|
||
});
|
||
for (const r of removes) tasks.push({
|
||
type: "remove",
|
||
start: r.start,
|
||
end: r.end
|
||
});
|
||
}
|
||
const overlapsStale = (seg) => {
|
||
const segEnd = seg.startTime + seg.duration;
|
||
return removes.some((r) => seg.startTime < r.end && segEnd > r.start) || staleRanges.some((r) => seg.startTime < r.end && segEnd > r.start);
|
||
};
|
||
const effectiveBuffered = removes.length + staleRanges.length > 0 ? bufferedSegments.filter((s) => !overlapsStale(s)) : bufferedSegments;
|
||
if (actorCtx.initTrackId !== track.id) tasks.push({
|
||
type: "append-init",
|
||
meta: {
|
||
trackId: track.id,
|
||
language: track.language
|
||
},
|
||
url: track.initialization.url,
|
||
...track.initialization.byteRange !== void 0 && { byteRange: track.initialization.byteRange }
|
||
});
|
||
if (range) {
|
||
const segmentsToLoad = getSegmentsToLoad(track.segments, effectiveBuffered, currentTime, forwardBufferConfig).filter((seg) => {
|
||
const existing = actorCtx.segments.find((s) => !overlapsStale(s) && Math.abs(s.startTime - seg.startTime) < 1e-4);
|
||
if (existing?.partial) return true;
|
||
if (!existing?.trackBandwidth || !track.bandwidth) return true;
|
||
return track.bandwidth > existing.trackBandwidth;
|
||
});
|
||
for (const segment of segmentsToLoad) tasks.push({
|
||
type: "append-segment",
|
||
meta: {
|
||
id: segment.id,
|
||
startTime: segment.startTime,
|
||
duration: segment.duration,
|
||
trackId: track.id,
|
||
trackBandwidth: track.bandwidth
|
||
},
|
||
url: segment.url,
|
||
...segment.byteRange !== void 0 && { byteRange: segment.byteRange }
|
||
});
|
||
}
|
||
return tasks;
|
||
};
|
||
const scheduleAll = (tasks, { getContext, setContext, runner }) => {
|
||
tasks.forEach((op) => {
|
||
runner.schedule(makeLoadTask(op, {
|
||
getContext,
|
||
setContext,
|
||
pipelines,
|
||
deps
|
||
})).then(void 0, (e) => {
|
||
if (e instanceof Error && e.name === "AbortError") return;
|
||
console.error("Unexpected error in segment loader:", e);
|
||
runner.abortPending();
|
||
});
|
||
});
|
||
};
|
||
return createMachineActor({
|
||
runner: () => new SerialRunner(),
|
||
initial: "idle",
|
||
context: {
|
||
inFlightInitTrackId: null,
|
||
inFlightSegment: null
|
||
},
|
||
states: {
|
||
idle: { on: { load: (msg, ctx) => {
|
||
const allTasks = planTasks(msg);
|
||
if (allTasks.length === 0) return;
|
||
ctx.transition("loading");
|
||
scheduleAll(allTasks, ctx);
|
||
} } },
|
||
loading: {
|
||
onSettled: "idle",
|
||
on: { load: (msg, ctx) => {
|
||
const { context, runner } = ctx;
|
||
const allTasks = planTasks(msg);
|
||
const inFlight = context.inFlightSegment;
|
||
const segmentInFlightStillNeeded = (t) => t.type === "append-segment" && inFlight !== null && t.meta.id === inFlight.id && t.meta.trackId === inFlight.trackId;
|
||
if (inFlight !== null && allTasks.some(segmentInFlightStillNeeded) || context.inFlightInitTrackId !== null && allTasks.some((t) => t.type === "append-init" && t.meta.trackId === context.inFlightInitTrackId)) {
|
||
runner.abortPending();
|
||
scheduleAll(allTasks.filter((t) => !segmentInFlightStillNeeded(t) && !(t.type === "append-init" && t.meta.trackId === context.inFlightInitTrackId)), ctx);
|
||
} else {
|
||
runner.abortAll();
|
||
if (context.inFlightSegment !== null || context.inFlightInitTrackId !== null && allTasks.some((t) => t.type === "append-init" && t.meta.trackId !== context.inFlightInitTrackId)) sourceBufferActor.send({ type: "cancel" });
|
||
scheduleAll(allTasks, ctx);
|
||
}
|
||
} }
|
||
}
|
||
}
|
||
});
|
||
}
|
||
//#endregion
|
||
export { createSegmentLoaderActor };
|
||
|
||
//# sourceMappingURL=segment-loader.js.map
|