/*! @license
* Shaka Player
* Copyright 2016 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
goog.provide('shaka.msf.MSFParser');
goog.require('goog.asserts');
goog.require('shaka.log');
goog.require('shaka.drm.DrmUtils');
goog.require('shaka.drm.PlayReady');
goog.require('shaka.media.ManifestParser');
goog.require('shaka.media.PresentationTimeline');
goog.require('shaka.media.SegmentIndex');
goog.require('shaka.media.SegmentReference');
goog.require('shaka.msf.CatalogStore');
goog.require('shaka.msf.Compression');
goog.require('shaka.msf.EventTimeline');
goog.require('shaka.msf.MediaTimeline');
goog.require('shaka.msf.MSFTransport');
goog.require('shaka.msf.MSFPresentationTimeline');
goog.require('shaka.msf.PackagingRegistry');
goog.require('shaka.msf.Utils');
goog.require('shaka.net.NetworkingEngine');
goog.require('shaka.util.Error');
goog.require('shaka.util.EventManager');
goog.require('shaka.util.Functional');
goog.require('shaka.util.LanguageUtils');
goog.require('shaka.util.ManifestParserUtils');
goog.require('shaka.util.MimeUtils');
goog.require('shaka.util.PlayerConfiguration');
goog.require('shaka.util.Scte35');
goog.require('shaka.util.StringUtils');
goog.require('shaka.util.Uint8ArrayUtils');
goog.requireType('shaka.media.InitSegmentReference');
/**
* MOQT Streaming Format.
*
* @see https://datatracker.ietf.org/doc/draft-ietf-moq-transport/
* @see https://datatracker.ietf.org/doc/draft-ietf-moq-msf/
* @see https://datatracker.ietf.org/doc/draft-ietf-moq-cmsf/
*
* @implements {shaka.extern.ManifestParser}
* @export
*/
shaka.msf.MSFParser = class {
constructor() {
/** @private {?shaka.extern.ManifestParser.PlayerInterface} */
this.playerInterface_ = null;
/** @private {?shaka.extern.ManifestConfiguration} */
this.config_ = null;
/** @private {function():boolean} */
this.isPreloadFn_ = () => false;
/** @private {?shaka.extern.Manifest} */
this.manifest_ = null;
/** @private {?shaka.msf.MSFTransport} */
this.msfTransport_ = null;
/** @private {?shaka.extern.MsfSession} */
this.connection_ = null;
/** @private {!Array<Array<string>>} */
this.publishNamespaces_ = [];
/** @private {?function()} */
this.unregisterPublishNamespaceCallback_ = null;
/** @private {?function()} */
this.unregisterCatalogCallback_ = null;
/** @private {!Promise.PromiseWithResolvers} */
this.catalogPromise_ = Promise.withResolvers();
/**
* Whether the catalog the presentation is built from has been received.
* Only the first complete catalog is used; what the catalog track says
* after it is not applied yet.
* @private {boolean}
*/
this.catalogReceived_ = false;
/** @private {number} */
this.globalId_ = 1;
/**
* The in-progress ABR bandwidth sample for each track, keyed by track key.
* Tracks are accumulated separately because each has its own group
* sequence.
* @private {!Map<string, {bytes: number, readMs: number, group: bigint}>}
*/
this.bandwidthSamples_ = new Map();
/** @private {?shaka.media.PresentationTimeline} */
this.presentationTimeline_ = null;
/** @private {!Array<!shaka.extern.Variant>} */
this.variants_ = [];
/** @private {!Array<!shaka.extern.Stream>} */
this.audioStreams_ = [];
/** @private {!Array<!shaka.extern.Stream>} */
this.videoStreams_ = [];
/** @private {!Array<!shaka.extern.Stream>} */
this.textStreams_ = [];
/** @private {!Map<string, function()>} */
this.unregisterTracksCallback_ = new Map();
/**
* Subscriptions whose SUBSCRIBE_OK has not arrived yet, by track key.
* A subscription can only be withdrawn once it has an alias, so a track
* closed before then has to leave a note for the in-flight subscribe to
* find; without it the subscription is orphaned and the publisher keeps
* sending objects nobody listens to for the rest of the session.
*
* @private {!Map<string, {cancelled: boolean}>}
*/
this.pendingSubscriptions_ = new Map();
/** @private {boolean} */
this.isFirstVideoSegment_ = true;
/**
* Tracks whether the first segment has been received for each content type.
* Used to delay locking the presentation timeline until all expected
* stream types have started producing data.
* @private {!Set<shaka.util.ManifestParserUtils.ContentType>}
*/
this.receivedFirstSegment_ = new Set();
/**
* The media timeline of each track that has one, by catalog track name.
*
* A track gets one from a media timeline track that names it in `depends`,
* from its own `template` field, or from both, in which case the one
* instance holds both and prefers the explicit records.
*
* @private {!Map<string, !shaka.msf.MediaTimeline>}
*/
this.mediaTimelines_ = new Map();
/**
* The tracks each media timeline track describes, by the timeline track's
* name. An event timeline names the timeline track in its `depends`, while
* the timelines themselves are kept by the media track they describe, so
* this is what gets from one to the other.
*
* @private {!Map<string, !Array<string>>}
*/
this.mediaTimelineDepends_ = new Map();
/**
* The event timelines of the event timeline tracks subscribed to.
*
* @private {!Array<!shaka.msf.EventTimeline>}
*/
this.eventTimelines_ = [];
/**
* The streams that are currently subscribed, by track key. A seek has to
* find them to re-point their subscriptions, which is why the state that
* used to live in createSegmentIndex's closure is out here.
*
* @private {!Map<string, shaka.msf.MSFParser.StreamState>}
*/
this.activeStreams_ = new Map();
/** @private {HTMLMediaElement} */
this.mediaElement_ = null;
/** @private {shaka.util.EventManager} */
this.eventManager_ = null;
}
/**
* @param {shaka.extern.ManifestConfiguration} config
* @param {(function():boolean)=} isPreloadFn
* @override
* @exportInterface
*/
configure(config, isPreloadFn) {
this.config_ = config;
if (isPreloadFn) {
this.isPreloadFn_ = isPreloadFn;
}
if (this.msfTransport_) {
this.msfTransport_.configure(this.config_.msf);
}
}
/**
* @override
* @exportInterface
*/
async start(uri, playerInterface) {
goog.asserts.assert(this.config_, 'Must call configure() before start()!');
this.playerInterface_ = playerInterface;
this.msfTransport_ = new shaka.msf.MSFTransport(this.config_.msf);
/** @type {?Uint8Array} */
let fingerprint = null;
if (this.config_.msf.fingerprintUri) {
const requestType = shaka.net.NetworkingEngine.RequestType.FINGERPRINT;
const request = shaka.net.NetworkingEngine.makeRequest(
[this.config_.msf.fingerprintUri], this.config_.retryParameters);
const response = await this.playerInterface_.networkingEngine.request(
requestType, request).promise;
// Make sure that the parser has not been destroyed.
if (!this.playerInterface_) {
throw new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.PLAYER,
shaka.util.Error.Code.OPERATION_ABORTED);
}
const hexString = shaka.util.StringUtils.fromUTF8(response.data).trim();
const hexBytes = new Uint8Array(hexString.length / 2);
for (let i = 0; i < hexBytes.length; i += 1) {
hexBytes[i] = parseInt(hexString.slice(2 * i, 2 * i + 2), 16);
}
fingerprint = hexBytes;
}
try {
this.connection_ = await this.msfTransport_.connect(
uri, fingerprint, this.config_.msf.authorizationToken);
} catch (error) {
if (error instanceof shaka.util.Error) {
throw error;
}
throw new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.MANIFEST,
shaka.util.Error.Code.WEBTRANSPORT_INITIALIZATION_FAILED,
error);
}
if (this.config_.msf.namespaces.length) {
// Get catalog in this namespace
this.subscribeToCatalog_(this.config_.msf.namespaces);
} else {
// Listen for announcements
// Catalog subscription will happen after announcement is received
this.listenForAnnouncements_();
}
this.presentationTimeline_ = new shaka.msf.MSFPresentationTimeline();
this.presentationTimeline_.setStatic(true);
this.manifest_ = {
presentationTimeline: this.presentationTimeline_,
variants: [],
textStreams: [],
imageStreams: [],
chapterStreams: [],
offlineSessionIds: [],
sequenceMode: false,
ignoreManifestTimestampsInSegmentsMode: false,
type: shaka.media.ManifestParser.MSF,
serviceDescription: null,
nextUrl: null,
periodCount: 1,
gapCount: 0,
isLowLatency: false,
startTime: null,
};
// A session the server hangs up on never answers, so wait for the end of
// the session as well as for the catalog: the reason the session gives is
// worth reporting, and waiting out the timeout for it is not.
const sessionEndPromise = this.msfTransport_.waitForSessionEnd().then(
(reason) => {
throw new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.MANIFEST,
shaka.util.Error.Code.MSF_CONNECTION_CLOSED,
reason);
});
let catalog;
try {
catalog = await shaka.util.Functional.promiseWithTimeout(
/* seconds= */ 10,
Promise.race([this.catalogPromise_.promise, sessionEndPromise]));
} catch (error) {
if (error instanceof shaka.util.Error) {
throw error;
}
if (error) {
// The timeout rejects with nothing at all, so anything here is the
// failure that the subscription or the transport itself reported.
throw new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.MANIFEST,
shaka.util.Error.Code.MSF_CONNECTION_CLOSED,
String(error));
}
throw new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.MANIFEST,
shaka.util.Error.Code.MSF_CATALOG_TIMEOUT);
}
await this.processCatalog_(catalog);
if (!this.presentationTimeline_.isLive()) {
throw new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.MANIFEST,
shaka.util.Error.Code.MSF_VOD_CONTENT_NOT_SUPPORTED);
}
this.createVariants_();
this.manifest_.isLowLatency = this.presentationTimeline_.isDynamic();
this.manifest_.variants = this.variants_;
this.manifest_.textStreams = this.textStreams_;
this.playerInterface_.makeTextStreamsForClosedCaptions(this.manifest_);
return this.manifest_;
}
/**
* @override
* @exportInterface
*/
async stop() {
this.unregisterCatalogCallback_?.();
this.unregisterCatalogCallback_ = null;
this.unregisterPublishNamespaceCallback_?.();
this.unregisterPublishNamespaceCallback_ = null;
this.unregisterTracksCallback_.forEach((callback, key) => {
callback();
});
this.unregisterTracksCallback_.clear();
for (const token of this.pendingSubscriptions_.values()) {
token.cancelled = true;
}
this.pendingSubscriptions_.clear();
this.eventManager_?.release();
this.eventManager_ = null;
this.mediaElement_ = null;
this.activeStreams_.clear();
this.mediaTimelines_.clear();
this.mediaTimelineDepends_.clear();
this.eventTimelines_ = [];
if (this.connection_) {
// Try to close the connection if it's not already closed
try {
// WebTransport requires some time to close connections, so we set
// 1 second here, but this is based on experimental testing only.
await shaka.util.Functional.delay(/* seconds= */ 1);
await this.connection_.close();
this.connection_ = null;
} catch (error) {
// Ignore error
}
}
this.msfTransport_?.release();
this.msfTransport_ = null;
this.playerInterface_ = null;
this.config_ = null;
this.manifest_ = null;
}
/**
* @override
* @exportInterface
*/
update() {}
/**
* @override
* @exportInterface
*/
onExpirationUpdated(sessionId, expiration) {
// No-op
}
/**
* @override
* @exportInterface
*/
onInitialVariantChosen(variant) {
// No-op
}
/**
* @override
* @exportInterface
*/
banLocation(uri) {
// No-op
}
/**
* @override
* @exportInterface
*/
setMediaElement(mediaElement) {
this.eventManager_?.release();
this.eventManager_ = null;
this.mediaElement_ = mediaElement;
if (!mediaElement) {
return;
}
// A seek is the one thing that makes this parser issue a request of its
// own: there is no manifest to consult and no URL to fetch, so the target
// has to be turned into a Location and subscribed from.
this.eventManager_ = new shaka.util.EventManager();
this.eventManager_.listen(mediaElement, 'seeking', () => this.onSeeking_());
}
/**
* Listen for announcements from the server
* @private
*/
listenForAnnouncements_() {
try {
shaka.log.debug('Listening for announcements...');
// Subscribe to announcements
const unregister =
this.msfTransport_.registerPublishNamespaceCallback((namespace) => {
const namespaceStr = namespace.join('/');
shaka.log.debug(`Received publish namespace callback with namespace:
${namespaceStr}`);
const isAlreadyProcessed = this.publishNamespaces_.some((ns) =>
ns.join('/') === namespaceStr);
if (isAlreadyProcessed) {
shaka.log.debug(`Already processed namespace: ${namespaceStr}`);
return;
}
// Store the namespace
this.publishNamespaces_.push(namespace);
// Get catalog in this namespace
this.subscribeToCatalog_(namespace);
});
// Save the unregister function
this.unregisterPublishNamespaceCallback_ = unregister;
// Log that we've registered the callback
shaka.log.debug('Announcement listener registered successfully');
} catch (error) {
shaka.log.error(`Error listening for announcements:
${error instanceof Error ? error.message : String(error)}`);
}
}
/**
* Subscribe to the catalog in the given namespace.
*
* MSF requires a SUBSCRIBE that also retrieves the current Group from its
* first Object: that Object is the latest complete catalog, and the
* subscription alone would only deliver what is published after it, which
* may be nothing at all.
*
* @param {Array<string>} namespace
* @return {!Promise}
* @private
*/
async subscribeToCatalog_(namespace) {
try {
const namespaceStr = namespace.join('/');
shaka.log.debug(`Subscribing to catalog in namespace: ${namespaceStr}`);
// Subscribe to the "catalog" track in the given namespace
const trackAlias = await this.msfTransport_.subscribeTrack(
namespace, 'catalog',
this.createCatalogCallback_(new shaka.msf.CatalogStore(namespaceStr)),
/* startLocation= */ null, /* joinCurrentGroup= */ true);
// Create an unregister function that uses the track alias
const unregisterFunc = () => {
// Don't try to unsubscribe if we're already disconnecting
if (!this.playerInterface_) {
shaka.log.debug('Skipping catalog unsubscribe during disconnect');
return;
}
shaka.log.debug(`Unsubscribing from catalog track with alias
${trackAlias}`);
this.msfTransport_.unsubscribeTrack(trackAlias).catch((err) => {
shaka.log.error(`Failed to unsubscribe from catalog: ${err}`);
});
};
if (this.unregisterCatalogCallback_) {
shaka.log.debug('Unregistering previous catalog callback');
this.unregisterCatalogCallback_();
}
this.unregisterCatalogCallback_ = unregisterFunc;
shaka.log.debug(`Successfully subscribed to catalog in namespace:
${namespaceStr} with track alias: ${trackAlias}`);
} catch (error) {
shaka.log.error('Error subscribing to catalog:', error);
this.catalogPromise_.reject(error);
}
}
/**
* Builds the callback for the catalog track's Objects, which rebuilds the
* catalog from them and resolves the catalog promise with the first
* complete one.
*
* @param {!shaka.msf.CatalogStore} store
* @return {shaka.msf.Utils.ObjectCallback}
* @private
*/
createCatalogCallback_(store) {
const onError = (error) => {
if (!this.catalogReceived_) {
// Nothing to play without a catalog, and MSF forbids reading this
// one: its compression or its version is not one we understand.
this.catalogPromise_.reject(error);
} else {
shaka.log.warning('Ignoring a catalog update:', error);
}
};
return this.createDecodingCallback_((obj, data) => {
/** @type {!msfCatalog.Catalog} */
let document;
try {
const text = shaka.util.StringUtils.fromUTF8(data);
document = /** @type {!msfCatalog.Catalog} */(JSON.parse(text));
} catch (e) {
shaka.log.error(`Failed to decode catalog data:
${e instanceof Error ? e.message : String(e)}`);
return;
}
if (!store.push(obj.location, document)) {
return;
}
const catalog = store.getCatalog();
if (!catalog) {
return;
}
if (this.catalogReceived_) {
shaka.log.alwaysWarn('MSF catalog updates are not supported yet; ' +
`ignoring the one in group ${obj.location.group}, object ` +
`${obj.location.object}`);
return;
}
this.catalogReceived_ = true;
// The preprocessor may change what it is given, and the store has to
// keep applying updates to the catalog as it was published.
this.catalogPromise_.resolve(structuredClone(catalog));
}, (error) => {
if (error instanceof shaka.util.Error) {
onError(error);
} else {
shaka.log.error('Failed to decompress catalog data:', error);
}
});
}
/**
* Wraps a handler for the Objects of a JSON track -- the catalog or a media
* timeline -- so that each payload reaches it with its MSF_COMPRESSION
* undone, and in the order the Objects arrived. Order matters: the first
* Object of a Group is a complete document and the ones after it are
* updates to it, and a compressed complete document takes longer to decode
* than the uncompressed update that follows it.
*
* An uncompressed payload with nothing waiting ahead of it is handled in the
* same turn it arrives.
*
* Objects with no payload carry an Object status (e.g. an end-of-group
* marker), not a document, and are skipped.
*
* @param {function(!shaka.msf.Utils.MOQObject, !Uint8Array)} handler
* @param {function(*)} onError Receives the reason a payload could not be
* decoded, which is a shaka.util.Error when its compression is not
* supported. The Object is dropped.
* @return {shaka.msf.Utils.ObjectCallback}
* @private
*/
createDecodingCallback_(handler, onError) {
/**
* Settles when the last payload queued so far has been handled, or null
* when none is waiting.
* @type {?Promise}
*/
let queue = null;
const handle = (obj, data) => {
try {
handler(obj, data);
} catch (error) {
onError(error);
}
};
return (obj) => {
if (!obj.data.byteLength) {
return;
}
/** @type {!Uint8Array|!Promise<!Uint8Array>} */
let decoded;
try {
decoded = this.decodePayload_(obj);
} catch (error) {
onError(error);
return;
}
if (!queue && ArrayBuffer.isView(decoded)) {
handle(obj, /** @type {!Uint8Array} */ (decoded));
return;
}
// Observe the outcome right away, so a failure waiting for its turn is
// never reported as an unhandled rejection.
const settled = Promise.resolve(decoded).then(
(data) => () => handle(obj, data),
(error) => () => onError(error));
const current = (queue || Promise.resolve())
.then(() => settled)
.then((deliver) => deliver());
queue = current;
current.then(() => {
if (queue == current) {
queue = null;
}
});
};
}
/**
* An Object's payload with its MSF_COMPRESSION undone.
*
* @param {!shaka.msf.Utils.MOQObject} obj
* @return {!Uint8Array|!Promise<!Uint8Array>}
* @private
*/
decodePayload_(obj) {
const codec = this.msfTransport_.getCodec();
if (!codec) {
return obj.data;
}
return shaka.msf.Compression.decode(obj, codec);
}
/**
* Feed the ABR bandwidth estimator.
*
* MoQ delivers media as many small per-object (often per-frame) chunks, which
* are individually too small for the bandwidth estimator to draw a useful
* throughput sample from. A group is the natural aggregation unit, so
* accumulate a group's objects and report one sample once the group is
* complete, which we detect by the object's group changing.
*
* The time reported is the sum of the per-object <em>active read</em>
* durations, not wall clock. A push stream delivers at (roughly) the encoded
* rate, so wall-clock timing would only ever measure the current variant's
* bitrate and ABR could never learn there is spare capacity. When the link
* has headroom the objects sit ready in the transport buffer and read almost
* instantly, so the active read time reflects the true link speed and lets
* ABR climb to a variant the connection can actually sustain.
*
* @param {string} trackKey
* @param {shaka.msf.Utils.Location} location
* @param {number} byteLength
* @param {number} readMs Active time spent reading this object from the
* stream.
* @private
*/
recordBandwidthSample_(trackKey, location, byteLength, readMs) {
const sample = this.bandwidthSamples_.getOrInsert(trackKey, {
bytes: 0,
readMs: 0,
group: location.group,
});
if (location.group !== sample.group) {
// A new group started, so the previous one is complete: report it.
if (sample.bytes > 0) {
this.playerInterface_.onSegmentReceived(
// Guard against a zero total; Date.now() has millisecond
// resolution, so a small group can read in "no" time.
Math.max(sample.readMs, 1), sample.bytes);
}
sample.bytes = 0;
sample.readMs = 0;
sample.group = location.group;
}
sample.bytes += byteLength;
sample.readMs += readMs;
}
/**
* @param {msfCatalog.Track} track
* @param {string} trackKey
* @param {shaka.msf.Utils.ObjectCallback} callback
* @param {?shaka.msf.Utils.Location=} startLocation Where delivery should
* begin, or null to start at the live edge. A start at the live edge
* joins the current Group from its first Object, which is where media
* can start decoding and where a timeline track restates itself in full,
* rather than starting mid-group or waiting for the next Group.
* @return {!Promise<?Error>} Why the publisher refused the subscription,
* or null when it did not: either the track is subscribed, or it was
* closed before the answer came and nothing is wanted any more. A
* publisher can refuse a request, and one that starts at a Location is
* the likeliest to be refused: the Group asked for may be older than
* anything the publisher still holds, whatever its media timeline said.
*/
async subscribeToTrack(track, trackKey, callback, startLocation) {
try {
// Fall back to the session namespace when the catalog track object
// does not include an explicit namespace field (the namespace is
// typically established at the transport level via PUBLISH_NAMESPACE).
let sessionNamespace = [];
if (this.config_.msf.namespaces.length) {
sessionNamespace = this.config_.msf.namespaces;
} else if (this.publishNamespaces_.length) {
sessionNamespace = this.publishNamespaces_[0];
}
const namespace = track.namespace ?
shaka.msf.MSFParser.namespaceTuple(
track.namespace, sessionNamespace) :
sessionNamespace;
const trackName = track.name;
shaka.log.debug(`Subscribing to track: ${trackKey}`);
const token = {cancelled: false};
this.pendingSubscriptions_.set(trackKey, token);
const trackAlias = await this.msfTransport_.subscribeTrack(
namespace, trackName, (obj) => {
if (token.cancelled) {
return;
}
shaka.log.v1(
`Received object for track ${trackKey} with`, obj);
callback(obj);
}, startLocation, /* joinCurrentGroup= */ !startLocation);
if (this.pendingSubscriptions_.get(trackKey) == token) {
this.pendingSubscriptions_.delete(trackKey);
}
// Create an unregister function that uses the track alias
const unregisterFunc = () => {
// Don't try to unsubscribe if we're already disconnecting
if (!this.playerInterface_) {
shaka.log.debug('Skipping catalog unsubscribe during disconnect');
return;
}
shaka.log.debug(`Unsubscribing from catalog track with alias
${trackAlias}`);
this.msfTransport_.unsubscribeTrack(trackAlias).catch((err) => {
shaka.log.error(`Failed to unsubscribe from catalog: ${err}`);
});
};
if (token.cancelled) {
// The track was closed while we were waiting for the SUBSCRIBE_OK.
// Now that there is an alias, withdraw it rather than registering a
// subscription nobody will ever close.
shaka.log.debug(
`Track ${trackKey} was closed before it finished subscribing`);
unregisterFunc();
return null;
}
// Store the subscription
this.unregisterTracksCallback_.set(trackKey, unregisterFunc);
shaka.log.debug(
`Subscribed to track ${trackKey} with alias ${trackAlias}`);
return null;
} catch (error) {
shaka.log.debug(`Error subscribing to track:
${error instanceof Error ? error.message : String(error)}`);
return error instanceof Error ? error : new Error(String(error));
}
}
/**
* @param {msfCatalog.Catalog} catalog
* @return {!Promise}
* @private
*/
processCatalog_(catalog) {
// A preprocessor mutates the catalog, and a console renders a logged
// object when it is expanded, so this logs a copy or both lines would show
// the processed one. goog.DEBUG keeps the copy out of release builds,
// which drop the log call but not its arguments.
if (goog.DEBUG) {
shaka.log.info('MSF Catalog:', structuredClone(catalog));
}
const catalogPreprocessor = this.config_.msf.catalogPreprocessor;
const defaultCatalogPreprocessor =
shaka.util.PlayerConfiguration.defaultCatalogPreprocessor;
if (catalogPreprocessor != defaultCatalogPreprocessor) {
catalogPreprocessor(catalog);
shaka.log.info('MSF Catalog after preprocessor:', catalog);
}
const promises = [];
let isLive = true;
let duration = Infinity;
let targetLatency = 0;
const contentProtections = this.getContentProtections_(catalog);
const initDataList = new Map();
for (const init of catalog.initDataList || []) {
initDataList.set(init.id, init.data);
}
for (const track of catalog.tracks) {
if (track.packaging == shaka.msf.MSFParser.MEDIA_TIMELINE_PACKAGING_) {
// A media timeline track carries no media, so none of what follows
// applies to it: it says nothing about how long the presentation is
// or how far behind the live edge to play, and a publisher is not
// required to mark it live at all, so letting it set isLive here
// would let a metadata track decide the presentation is VOD.
this.processMediaTimelineTrack_(track);
continue;
}
if (track.packaging == shaka.msf.MSFParser.EVENT_TIMELINE_PACKAGING_) {
// Nor does an event timeline track, for the same reasons.
this.processEventTimelineTrack_(track);
continue;
}
if ('isLive' in track) {
isLive = track.isLive;
}
if (track.targetLatency) {
targetLatency = Math.max(targetLatency, track.targetLatency);
}
if (track.trackDuration) {
duration = Math.min(duration, track.trackDuration);
}
if (track.template) {
this.getOrCreateTimeline_(track.name).setTemplate(track.template);
}
promises.push(this.processTrack_(
track, contentProtections, initDataList));
}
if (targetLatency) {
/** @type {shaka.extern.ServiceDescription} */
const serviceDescription = {
maxLatency: null,
maxPlaybackRate: null,
minLatency: null,
minPlaybackRate: null,
targetLatency: targetLatency / 1000,
clientDataReporting: null,
};
this.manifest_.serviceDescription = serviceDescription;
}
if (isLive) {
this.presentationTimeline_.setStatic(false);
}
this.presentationTimeline_.setDuration(duration);
return Promise.all(promises);
}
/**
* Subscribes to a media timeline track and feeds what it publishes to the
* timelines of the tracks it describes.
*
* @param {msfCatalog.Track} track
* @private
*/
processMediaTimelineTrack_(track) {
const depends = track.depends || [];
if (!depends.length) {
// MSF requires it, and without it there is nothing to say which tracks
// these records address, so the map they carry cannot be used.
shaka.log.warning(
'Ignoring a media timeline track that declares no "depends"', track);
return;
}
this.mediaTimelineDepends_.set(track.name, depends);
const timelines = depends.map((name) => this.getOrCreateTimeline_(name));
const trackKey = `${track.namespace || ''}/${track.name}`;
this.subscribeToOptionalJsonTrack_(track, trackKey, 'media timeline',
(obj, data) => {
// The first Object of a Group is a complete timeline and replaces
// what we hold; the ones after it in that Group only add to it
// (MSF 7.3).
const independent = !obj.location.object;
let changed = false;
for (const timeline of timelines) {
changed = timeline.addObject(data, independent) || changed;
}
if (changed) {
this.updateAvailabilityWindow_();
// A record an event timeline could not place yet may be placeable
// now.
for (const eventTimeline of this.eventTimelines_) {
eventTimeline.resolve();
}
}
});
}
/**
* Subscribes to a track of JSON documents the presentation can do without,
* such as a media or event timeline, and hands each payload to the handler
* with its MSF_COMPRESSION undone and in the order the Objects arrived.
*
* An Object that fails to decompress is dropped. A compression we cannot
* undo at all is a different matter: MSF forbids reading the track and says
* to unsubscribe, and since the track is optional, playback carries on
* without it.
*
* @param {msfCatalog.Track} track
* @param {string} trackKey
* @param {string} description What the track is, for the log.
* @param {function(!shaka.msf.Utils.MOQObject, !Uint8Array)} handler
* @private
*/
subscribeToOptionalJsonTrack_(track, trackKey, description, handler) {
// Set once the track turns out to be compressed in a way we cannot
// undo, so the Objects already on their way are not read either.
let unsupported = false;
this.subscribeToTrack(track, trackKey, this.createDecodingCallback_(
(obj, data) => {
if (!unsupported) {
handler(obj, data);
}
},
(error) => {
if (!(error instanceof shaka.util.Error)) {
shaka.log.warning(
`Discarding a ${description} object that could not be ` +
'decompressed', error);
return;
}
if (unsupported) {
return;
}
unsupported = true;
shaka.log.warning(
`Unsubscribing from ${description} track ${trackKey}, whose ` +
'compression is not supported', error);
const unregister = this.unregisterTracksCallback_.get(trackKey);
if (unregister) {
unregister();
this.unregisterTracksCallback_.delete(trackKey);
}
}));
}
/**
* Subscribes to an event timeline track whose `eventType` this player
* consumes, and hands each of its records to the consumer once it can be
* placed on the presentation timeline.
*
* @param {msfCatalog.Track} track
* @private
*/
processEventTimelineTrack_(track) {
const eventType = track.eventType || '';
const onRecord = this.getEventRecordHandler_(eventType);
if (!onRecord) {
shaka.log.info(
`Skipping event timeline track with unsupported eventType ` +
`"${eventType}"`, track);
return;
}
const depends = track.depends || [];
const eventTimeline = new shaka.msf.EventTimeline(
(record) => this.getEventRecordTime_(record, depends), onRecord);
this.eventTimelines_.push(eventTimeline);
const trackKey = `${track.namespace || ''}/${track.name}`;
this.subscribeToOptionalJsonTrack_(track, trackKey, 'event timeline',
(obj, data) => {
eventTimeline.addObject(data, !obj.location.object);
});
}
/**
* What consumes the records of an event timeline of the given type, or null
* when nothing here does. What a record's `data` means is defined by the
* type, so this is the one place that knows about any of them.
*
* @param {string} eventType
* @return {?function(shaka.msf.EventTimeline.Record, number)}
* @private
*/
getEventRecordHandler_(eventType) {
if (shaka.util.Scte35.isMsfEventType(eventType)) {
return (record, time) => this.onScte35Record_(eventType, record, time);
}
return null;
}
/**
* Reports a record of an SCTE-35 event timeline to the player
* (draft-wilaw-moq-scte35-event-timeline).
*
* @param {string} eventType
* @param {shaka.msf.EventTimeline.Record} record
* @param {number} time
* @private
*/
onScte35Record_(eventType, record, time) {
if (!this.playerInterface_) {
return;
}
const data = record.data;
const payload = data && typeof data == 'object' ?
data['scte35_payload'] : null;
const event = typeof payload == 'string' ?
shaka.util.Scte35.fromMsf(eventType, payload, time, record.id) : null;
if (event) {
this.playerInterface_.onScte35Event(event);
} else {
shaka.log.warning(
'Discarding an SCTE-35 record whose payload cannot be decoded',
record);
}
}
/**
* Where an event timeline record falls on the presentation timeline, in
* seconds, or null when it cannot be placed yet.
*
* A media time is a presentation time already (MSF 7.1.1). A Location or a
* wallclock time has to go through the media timeline of a track the
* record's track depends on.
*
* @param {shaka.msf.EventTimeline.Record} record
* @param {!Array<string>} depends The event timeline track's `depends`.
* @return {?number}
* @private
*/
getEventRecordTime_(record, depends) {
if (record.index == 'm') {
return record.time / 1000;
}
for (const timeline of this.getMediaTimelinesFor_(depends)) {
const time = record.location ?
timeline.timeAtOrBefore(record.location) :
timeline.timeForWallClock(record.time);
if (time != null) {
return time;
}
}
return null;
}
/**
* The media timelines a track that depends on the given tracks can use.
*
* `depends` names either media timeline tracks, as in the MSF catalog
* examples, whose timelines are kept by the media tracks they in turn
* depend on, or media tracks that carry a template. MSF only requires
* `depends` of an event timeline whose records refer to a media timeline,
* so an empty one falls back to every timeline the presentation has, in
* catalog order.
*
* @param {!Array<string>} depends
* @return {!Array<!shaka.msf.MediaTimeline>}
* @private
*/
getMediaTimelinesFor_(depends) {
if (!depends.length) {
return Array.from(this.mediaTimelines_.values());
}
const timelines = [];
for (const name of depends) {
for (const trackName of this.mediaTimelineDepends_.get(name) || [name]) {
const timeline = this.mediaTimelines_.get(trackName);
if (timeline && !timelines.includes(timeline)) {
timelines.push(timeline);
}
}
}
return timelines;
}
/**
* @param {string} trackName
* @return {!shaka.msf.MediaTimeline}
* @private
*/
getOrCreateTimeline_(trackName) {
let timeline = this.mediaTimelines_.get(trackName);
if (!timeline) {
timeline = new shaka.msf.MediaTimeline();
this.mediaTimelines_.set(trackName, timeline);
}
return timeline;
}
/**
* The earliest presentation time the player can seek to and actually be
* served, or null when there is no such point and the window should stay at
* the live edge.
*
* Every subscribed track has to be able to reach it, so this is the latest
* of their starts and is null as soon as one of them has no timeline: a seek
* range is a promise that all of the media is there, and half of it is a
* stall rather than a seek.
*
* @return {?number}
* @private
*/
getTimelineWindowStart_() {
let start = null;
for (const state of this.activeStreams_.values()) {
const timeline = this.mediaTimelines_.get(state.track.name);
const trackStart = timeline ? timeline.getStartTime() : null;
if (trackStart == null) {
return null;
}
start = start == null ? trackStart : Math.max(start, trackStart);
}
return start;
}
/**
* Sizes the segment availability window, which is what the player offers as
* the seek range.
*
* Without a media timeline the window is a floor around the live edge and
* the presentation has no DVR, because nothing says where to subscribe from
* to get anything older. With one, the window reaches back to the oldest
* point the timeline still describes.
*
* @private
*/
updateAvailabilityWindow_() {
if (!this.presentationTimeline_ ||
!this.presentationTimeline_.isDynamic()) {
return;
}
const override = this.config_.availabilityWindowOverride;
if (!isNaN(override)) {
this.presentationTimeline_.setSegmentAvailabilityDuration(override);
return;
}
let duration = 0;
const maxSegmentDuration =
this.presentationTimeline_.getMaxSegmentDuration();
if (maxSegmentDuration > 0) {
// No DVR -- the window sits at the live edge. But it still has to be
// wide enough to CONTAIN the playhead, which trails the live edge,
// and to hold the trailing track when the two are not published in
// lockstep.
//
// "availability window = one segment" is right when a segment is
// seconds long. Where one object is one frame it is tens of
// milliseconds, so the playhead falls outside its own seek range
// immediately and shaka re-seeks it back into range over and over,
// while the trailing track is never inside the window at all.
duration = Math.max(
maxSegmentDuration, shaka.msf.MSFParser.MIN_AVAILABILITY_WINDOW_SEC_);
}
const windowStart = this.getTimelineWindowStart_();
if (windowStart != null) {
const availabilityEnd =
this.presentationTimeline_.getSegmentAvailabilityEnd();
duration = Math.max(duration, availabilityEnd - windowStart);
}
if (duration > 0) {
this.presentationTimeline_.setSegmentAvailabilityDuration(duration);
}
}
/**
* Subscribes a stream's track and turns the Objects it delivers into
* segments, from the live edge or from a given Location.
*
* @param {shaka.msf.MSFParser.StreamState} state
* @param {?shaka.msf.Utils.Location} startLocation Where to start, or null
* for the live edge.
* @return {!Promise} Resolves once the subscription has produced its first
* segment, and rejects with MSF_SUBSCRIBE_FAILED if the publisher refuses
* to serve the track from the live edge.
* @private
*/
startSubscription_(state, startLocation) {
const stream = state.stream;
// Objects of the subscription being replaced can still be in flight, and
// they describe a part of the presentation we have just left. The
// generation is what tells them apart from the ones asked for here.
const generation = ++state.generation;
state.startLocation = startLocation;
const segmenter =
state.packaging.createSegmenter(this.msfTransport_.getCodec());
let promiseWithResolvers = Promise.withResolvers();
stream.segmentIndex?.release();
stream.segmentIndex = new shaka.media.SegmentIndex([]);
this.subscribeToTrack(state.track, state.trackKey, (obj) => {
if (state.generation != generation || !stream.segmentIndex) {
return;
}
// An object with no payload carries an object status rather than
// media, so it contributes no bytes to measure. It is still handed to
// the segmenter, which may need it to know a group has ended.
if (obj.data.byteLength) {
this.recordBandwidthSample_(state.trackKey, obj.location,
obj.data.byteLength,
obj.receiveTimestampMs - obj.payloadReadStartMs);
}
for (const segment of segmenter.push(obj)) {
// recordBandwidthSample_() above reports to ABR, which can pick a
// new variant and switch to it synchronously; StreamingEngine then
// calls closeSegmentIndex() on this stream before we get here.
// Re-check on every segment instead of only on entry, or we append
// to a segment index that is already gone.
if (state.generation != generation || !stream.segmentIndex) {
return;
}
this.addSegment_(stream, state.type, segment,
state.description.initSegmentReference);
promiseWithResolvers?.resolve();
promiseWithResolvers = null;
}
}, startLocation).then((error) => {
if (!error || state.generation != generation ||
!stream.segmentIndex ||
!this.activeStreams_.has(state.trackKey)) {
// Nothing failed that is still wanted: the subscription is up, or
// the stream has been closed or re-pointed since.
return;
}
if (!startLocation) {
// Nothing else to fall back to. Waiting for the first segment would
// wait forever, and the player with it. No segment can have arrived
// yet: Objects only flow once a subscription has been accepted.
promiseWithResolvers?.reject(new shaka.util.Error(
shaka.util.Error.Severity.CRITICAL,
shaka.util.Error.Category.MANIFEST,
shaka.util.Error.Code.MSF_SUBSCRIBE_FAILED,
state.track.name,
error.message));
promiseWithResolvers = null;
return;
}
// The publisher refused to start where the media timeline said it
// could. Following the live edge is not what the viewer asked for, but
// it plays, whereas leaving the subscription withdrawn leaves the
// stream waiting for content that is never coming.
shaka.log.warning(
`Track ${state.trackKey} could not be subscribed from ` +
`group ${startLocation.group}; returning to the live edge`);
this.resubscribe_(state, /* startLocation= */ null);
});
return /** @type {!Promise} */(promiseWithResolvers?.promise);
}
/**
* Starts a subscription that nothing waits on, reporting its failure to the
* player instead of leaving it to reject unheard.
*
* @param {shaka.msf.MSFParser.StreamState} state
* @param {?shaka.msf.Utils.Location} startLocation
* @private
*/
resubscribe_(state, startLocation) {
this.startSubscription_(state, startLocation).catch((error) => {
this.playerInterface_?.onError(
/** @type {!shaka.util.Error} */ (error));
});
}
/**
* Withdraws a stream's subscription, leaving its segment index alone.
*
* @param {shaka.msf.MSFParser.StreamState} state
* @private
*/
stopSubscription_(state) {
const pending = this.pendingSubscriptions_.get(state.trackKey);
if (pending) {
pending.cancelled = true;
this.pendingSubscriptions_.delete(state.trackKey);
}
const unregister = this.unregisterTracksCallback_.get(state.trackKey);
if (unregister) {
unregister();
this.unregisterTracksCallback_.delete(state.trackKey);
}
// The next subscription starts a new group sequence, so an accumulated
// sample from this one would be reported as one enormous group.
this.bandwidthSamples_.delete(state.trackKey);
}
/**
* Re-points the subscriptions after a seek to content that has been
* published but not received.
*
* A MoQT subscription is a position in a track, not a URL to fetch, so this
* is what a seek costs here: the subscription is withdrawn and asked for
* again from the Location the media timeline gives for the target, and the
* segment index starts over from there. Seeking within what has already
* arrived costs nothing and is left alone.
*
* @private
*/
onSeeking_() {
if (!this.mediaElement_ || !this.msfTransport_ ||
!this.presentationTimeline_) {
return;
}
const time = this.mediaElement_.currentTime;
const liveEdge = this.presentationTimeline_.getSegmentAvailabilityEnd();
for (const state of this.activeStreams_.values()) {
const stream = state.stream;
if (!stream.segmentIndex ||
shaka.msf.MSFParser.isIndexed_(stream.segmentIndex, time)) {
// Nothing to ask for: either the stream is not subscribed, or the
// target is already indexed and the media is on its way.
continue;
}
/** @type {?shaka.msf.Utils.Location} */
let startLocation = null;
if (time < liveEdge - shaka.msf.MSFParser.LIVE_EDGE_TOLERANCE_SEC_) {
const timeline = this.mediaTimelines_.get(state.track.name);
startLocation = timeline ? timeline.locationForTime(time) : null;
if (!startLocation) {
continue;
}
if (state.startLocation &&
state.startLocation.group == startLocation.group &&
state.startLocation.object == startLocation.object) {
// Already waiting on exactly this; asking again would restart the
// wait and drop the Objects that are on their way.
continue;
}
} else if (!state.startLocation) {
// Back at the live edge on a subscription that already follows it.
continue;
}
shaka.log.debug(`Re-subscribing ${state.trackKey} at ${time}s`,
startLocation);
this.stopSubscription_(state);
this.resubscribe_(state, startLocation);
}
}
/**
* @param {msfCatalog.Catalog} catalog
* @return {!Map<string, !shaka.extern.DrmInfo>}
* @private
*/
getContentProtections_(catalog) {
const uuidMap = shaka.drm.DrmUtils.getUuidMap();
/** @type {!Map<string, !shaka.extern.DrmInfo>} */
const mapContentProtections = new Map();
const contentProtections = catalog.contentProtections || [];
for (const contentProtection of contentProtections) {
const refId = contentProtection.refID;
const drmSystem = contentProtection.drmSystem;
if (!drmSystem) {
continue;
}
const keySystem = uuidMap[drmSystem.systemID.toLowerCase()];
if (!keySystem) {
continue;
}
const encryptionScheme = contentProtection.scheme || 'cenc';
let initData = null;
if (drmSystem.pssh) {
initData = [{
initDataType: 'cenc',
initData: shaka.util.Uint8ArrayUtils.fromBase64(drmSystem.pssh),
}];
}
const drmInfo = shaka.util.ManifestParserUtils.createDrmInfo(
keySystem, encryptionScheme, initData);
if (drmSystem.laURL?.url) {
drmInfo.licenseServerUri = drmSystem.laURL.url;
} else if (initData &&
shaka.drm.DrmUtils.isPlayReadyKeySystem(keySystem)) {
drmInfo.licenseServerUri =
shaka.drm.PlayReady.getLicenseUrlFromPssh(initData[0].initData);
}
if (drmSystem.certURL?.url) {
drmInfo.serverCertificateUri = drmSystem.certURL.url;
}
if (drmSystem.robustness) {
drmInfo.videoRobustness = drmSystem.robustness;
drmInfo.audioRobustness = drmSystem.robustness;
}
if (contentProtection.defaultKID) {
for (const kid of contentProtection.defaultKID) {
const normalizedKid = kid.replace(/-/g, '').toLowerCase();
drmInfo.keyIds.add(normalizedKid);
}
}
mapContentProtections.set(refId, drmInfo);
}
return mapContentProtections;
}
/**
* @param {msfCatalog.Track} track
* @param {!Map<string, !shaka.extern.DrmInfo>} contentProtections
* @param {!Map<string, string>} initDataList
* @private
*/
processTrack_(track, contentProtections, initDataList) {
const ContentType = shaka.util.ManifestParserUtils.ContentType;
const ManifestParserUtils = shaka.util.ManifestParserUtils;
let initData = new Uint8Array([]);
if (track.initData) {
initData = shaka.util.Uint8ArrayUtils.fromBase64(track.initData);
} else if (track.initRef) {
if (initDataList.has(track.initRef)) {
initData = shaka.util.Uint8ArrayUtils.fromBase64(
initDataList.get(track.initRef));
}
}
const packaging = shaka.msf.PackagingRegistry.create(track.packaging);
if (!packaging) {
shaka.log.info(
`Skipping track with unsupported packaging "${track.packaging}"`,
track);
return;
}
const description = packaging.describeTrack(track, initData);
if (!description) {
shaka.log.info('Skipping incompatible track', track);
return;
}
const basicInfo = description.basicInfo;
const mimeType = basicInfo.mimeType || track.mimeType || '';
const codecs = basicInfo.codecs;
let language = track.lang;
if (!track.lang || track.lang == 'und') {
language = basicInfo.language || track.lang;
}
const frameRate = Number(basicInfo.frameRate) || track.framerate;
const width = Number(basicInfo.width) || track.width;
const height = Number(basicInfo.height) || track.height;
let channelsCount = basicInfo.channelCount;
if (!channelsCount && track.channelConfig) {
channelsCount = parseInt(track.channelConfig, 10);
}
const audioSamplingRate = basicInfo.sampleRate || track.samplerate || null;
const hdr = basicInfo.videoRange || undefined;
const colorGamut = basicInfo.colorGamut || undefined;
let type = ContentType.TEXT;
for (const format of ManifestParserUtils.VIDEO_CODEC_REGEXPS) {
if (format.test(codecs.trim())) {
type = ContentType.VIDEO;
}
}
if (type == ContentType.TEXT) {
for (const format of ManifestParserUtils.AUDIO_CODEC_REGEXPS) {
if (format.test(codecs.trim())) {
type = ContentType.AUDIO;
}
}
}
let kind = undefined;
let accessibilityPurpose = null;
const roles = [];
if (track.role) {
roles.push(track.role);
switch (track.role) {
case 'audiodescription':
if (type == ContentType.AUDIO) {
accessibilityPurpose =
shaka.media.ManifestParser.AccessibilityPurpose.VISUALLY_IMPAIRED;
}
break;
case 'caption':
if (type == ContentType.TEXT) {
kind = ManifestParserUtils.TextStreamKind.CLOSED_CAPTION;
}
break;
case 'subtitle':
if (type == ContentType.TEXT) {
kind = ManifestParserUtils.TextStreamKind.SUBTITLE;
}
break;
}
}
let drmInfos = [];
const contentProtectionRefIDs = track.contentProtectionRefIDs;
if (contentProtectionRefIDs) {
for (const refId of contentProtectionRefIDs) {
const drmInfo = contentProtections.get(refId);
if (drmInfo) {
drmInfos.push(drmInfo);
} else {
shaka.log.alwaysWarn('Unrecognized contentProtectionRefID', refId);
return;
}
}
} else {
drmInfos = basicInfo.drmInfos;
}
const closedCaptions = new Map();
if (!this.config_.disableText) {
for (const accessibility of (track.accessibility || [])) {
// draft-ietf-moq-msf-01 5.2.44 names this field 'scheme'. The
// corresponding DASH attribute is 'schemeIdUri', which is where the
// wrong name came from; reading it left closedCaptions empty for
// every conforming catalog, so embedded captions were never exposed.
const scheme = accessibility.scheme;
const value = accessibility.value;
if (scheme == 'urn:scte:dash:cc:cea-608:2015') {
ManifestParserUtils.parseCEA608Captions(value, closedCaptions);
} else if (scheme == 'urn:scte:dash:cc:cea-708:2015') {
ManifestParserUtils.parseCEA708Captions(value, closedCaptions);
}
}
}
/** @type {shaka.extern.Stream} */
const stream = {
id: this.globalId_++,
originalId: track.name,
groupId: null,
createSegmentIndex: () => Promise.resolve(),
segmentIndex: null,
mimeType,
codecs,
supplementalCodecs: '',
kind,
encrypted: false,
drmInfos,
keyIds: new Set(),
language: shaka.util.LanguageUtils.normalize(language || 'und'),
originalLanguage: language || null,
label: track.label || null,
type,
primary: false,
trickModeVideo: null,
dependencyStream: null,
emsgSchemeIdUris: null,
frameRate,
pixelAspectRatio: undefined,
width,
height,
bandwidth: track.bitrate,
roles,
forced: false,
channelsCount,
audioSamplingRate,
spatialAudio: false,
closedCaptions,
hdr,
colorGamut,
videoLayout: undefined,
tilesLayout: undefined,
accessibilityPurpose,
external: false,
fastSwitching: false,
fullMimeTypes: new Set([shaka.util.MimeUtils.getFullType(
mimeType, codecs)]),
isAudioMuxedInVideo: false,
baseOriginalId: null,
isIframe: false,
preselection: null,
};
const trackKey = `${track.namespace || ''}/${track.name}`;
/** @type {shaka.msf.MSFParser.StreamState} */
const state = {
stream,
type,
track,
trackKey,
packaging,
description,
startLocation: null,
generation: 0,
};
stream.createSegmentIndex = () => {
this.activeStreams_.set(trackKey, state);
return this.startSubscription_(state, /* startLocation= */ null);
};
stream.closeSegmentIndex = () => {
this.activeStreams_.delete(trackKey);
// Whatever the subscription still has in flight belongs to a stream
// that is going away.
state.generation++;
this.stopSubscription_(state);
// If we have a segment index, release it.
stream.segmentIndex?.release();
stream.segmentIndex = null;
};
switch (type) {
case ContentType.AUDIO:
if (!this.config_.disableAudio) {
this.audioStreams_.push(stream);
}
break;
case ContentType.VIDEO:
if (!this.config_.disableVideo) {
this.videoStreams_.push(stream);
}
break;
case ContentType.TEXT:
if (!this.config_.disableText) {
this.textStreams_.push(stream);
}
break;
}
}
/**
* Adds a segment produced by a packaging's segmenter to a stream's segment
* index and advances the presentation timeline to cover it.
*
* @param {shaka.extern.Stream} stream
* @param {shaka.util.ManifestParserUtils.ContentType} type
* @param {!shaka.extern.MsfSegment} segment
* @param {shaka.media.InitSegmentReference} initSegmentReference
* @private
*/
addSegment_(stream, type, segment, initSegmentReference) {
const ContentType = shaka.util.ManifestParserUtils.ContentType;
const reference = new shaka.media.SegmentReference(
/* startTime= */ segment.startTime,
/* endTime= */ segment.startTime + segment.duration,
/* getUris= */ () => [],
/* startByte= */ 0,
/* endByte= */ null,
/* initSegmentReference= */ initSegmentReference,
/* timestampOffset= */ segment.timestampOffset,
/* appendWindowStart= */ 0,
/* appendWindowEnd= */ Infinity);
reference.discontinuitySequence = segment.discontinuitySequence;
reference.setSegmentData(segment.data);
this.receivedFirstSegment_.add(type);
const timelineLocked = this.presentationTimeline_.isStartTimeLocked();
if (timelineLocked) {
// StreamingEngine walks the index in order from its last appended
// reference, so it can never skip to content whose predecessor has been
// evicted: when the reference it wants next is gone,
// getSegmentReferenceNeeded_ returns null and update_ parks in its
// "segment could not be found" retry, falls further behind, and the
// eviction that caused the stall guarantees the stall persists.
//
// Retaining a couple of seconds is only sane when a reference is a
// multi-second segment. Where one reference is one frame it is a few
// dozen of them, and any hiccup longer than that is unrecoverable.
// References are metadata, so retention is cheap.
const evictTime = Math.min(
reference.startTime - shaka.msf.MSFParser.INDEX_RETENTION_SEC_,
this.presentationTimeline_.getSegmentAvailabilityStart());
stream.segmentIndex.mergeAndEvict([reference], evictTime);
} else {
stream.segmentIndex.merge([reference]);
}
this.presentationTimeline_.notifySegments([reference]);
this.presentationTimeline_.notifyMaxSegmentDuration(segment.duration);
this.updateAvailabilityWindow_();
if (!timelineLocked &&
(!this.audioStreams_.length ||
this.receivedFirstSegment_.has(ContentType.AUDIO)) &&
(!this.videoStreams_.length ||
this.receivedFirstSegment_.has(ContentType.VIDEO))) {
// Only lock once we have first segments from all expected types.
this.presentationTimeline_.lockStartTime();
}
}
/**
* @private
*/
createVariants_() {
const ContentType = shaka.util.ManifestParserUtils.ContentType;
// Create variants for all audio/video combinations.
let nextVariantId = 0;
const variants = [];
if (!this.videoStreams_.length || !this.audioStreams_.length) {
// For audio-only or video-only content, just give each stream its own
// variant.
const streams = this.videoStreams_.length ? this.videoStreams_ :
this.audioStreams_;
for (const stream of streams) {
const id = nextVariantId++;
let bandwidth = stream.bandwidth || 0;
if (stream.dependencyStream) {
bandwidth += stream.dependencyStream.bandwidth || 0;
}
variants.push({
id,
language: stream.language,
disabledUntilTime: 0,
primary: stream.primary,
audio: stream.type == ContentType.AUDIO ? stream : null,
video: stream.type == ContentType.VIDEO ? stream : null,
bandwidth,
drmInfos: stream.drmInfos,
allowedByApplication: true,
allowedByKeySystem: true,
decodingInfos: [],
});
}
} else {
for (const audio of this.audioStreams_) {
for (const video of this.videoStreams_) {
const commonDrmInfos = shaka.drm.DrmUtils.getCommonDrmInfos(
audio.drmInfos, video.drmInfos);
if (audio.drmInfos.length && video.drmInfos.length &&
!commonDrmInfos.length) {
shaka.log.warning(
'Incompatible DRM in audio & video, skipping variant creation.',
audio, video);
continue;
}
let bandwidth = (audio.bandwidth || 0) + (video.bandwidth || 0);
if (audio.dependencyStream) {
bandwidth += audio.dependencyStream.bandwidth || 0;
}
if (video.dependencyStream) {
bandwidth += video.dependencyStream.bandwidth || 0;
}
const id = nextVariantId++;
variants.push({
id,
language: audio.language,
disabledUntilTime: 0,
primary: audio.primary,
audio,
video,
bandwidth,
drmInfos: commonDrmInfos,
allowedByApplication: true,
allowedByKeySystem: true,
decodingInfos: [],
});
}
}
}
this.variants_ = variants;
}
};
/**
* How many seconds of SegmentReferences to keep in the index behind the newest
* one.
*
* Must exceed how far StreamingEngine can lag the live edge, because it walks
* the index in order and cannot skip to content whose predecessor has been
* evicted. Where one reference is one frame this is a large number of
* references but a small amount of memory.
*
* @private @const {number}
*/
shaka.msf.MSFParser.INDEX_RETENTION_SEC_ = 20;
/**
* Floor for the live segment-availability window, in seconds.
*
* Deriving the window from one object's duration gives tens of milliseconds on
* a frame-per-object packaging. That is too narrow on two counts: the
* playhead, which trails the live edge, falls outside its own seek range and
* is re-seeked continuously; and the trailing track of a pair published a few
* hundred milliseconds apart is never inside the window at all, so
* StreamingEngine reports "cannot find segment" for it forever.
*
* Bounded from above as well. This window IS the seek range, so it is what
* the player offers the user as DVR, and it is how far the playhead may drift
* behind the live edge before being pulled back to it. Neither is wanted on a
* transport chosen for low latency, and a window of five seconds or more also
* makes the UI show a seek bar (shaka.ui.SeekBar's minimum seek window), which
* would advertise DVR on a stream that has none.
*
* The quantity it has to cover is the lag between the newest reference and the
* newest sample actually appended, since that is where the playhead can be.
* Measured on a live LOC stream, that lag ran 2.0-4.7 s and the playhead sat
* 1.0-3.8 s behind the live edge; a window of 1.5 s put it outside the range
* again and it thrashed.
*
* @private @const {number}
*/
shaka.msf.MSFParser.MIN_AVAILABILITY_WINDOW_SEC_ = 4;
/**
* Whether a segment index holds a reference covering the given time.
*
* SegmentIndex.find() answers a position, and for a time before everything it
* holds it answers the first position rather than nothing, so its result says
* where to look, not whether the time is there. Seeking behind the index is
* exactly the case that distinction decides.
*
* @param {!shaka.media.SegmentIndex} segmentIndex
* @param {number} time
* @return {boolean}
* @private
*/
shaka.msf.MSFParser.isIndexed_ = (segmentIndex, time) => {
const position = segmentIndex.find(time);
if (position == null) {
return false;
}
const reference = segmentIndex.get(position);
return !!reference &&
time >= reference.startTime && time < reference.endTime;
};
/**
* Converts a catalog track's `namespace` string into the Track Namespace
* tuple that goes on the wire.
*
* A Track Namespace is an ordered set of fields (MoQT Section 2.4.1), but the
* catalog carries it as one string whose fields are joined with "/", as in
* the MSF example "conference.example.com/conference123/alice". Sending that
* string as a single field never matches a namespace announced as several.
*
* When the string is the session's own namespace written out, the session's
* tuple is used as is, so a publisher whose single field contains a "/" keeps
* working. Otherwise the string is split on "/", dropping empty fields
* because a field must carry at least one byte.
*
* @param {string} namespace
* @param {Array<string>} sessionNamespace
* @return {Array<string>}
*/
shaka.msf.MSFParser.namespaceTuple = (namespace, sessionNamespace) => {
if (sessionNamespace && sessionNamespace.length &&
namespace == sessionNamespace.join('/')) {
return sessionNamespace;
}
return namespace.split('/').filter((field) => field.length > 0);
};
/**
* The catalog `packaging` value of a media timeline track.
*
* @private @const {string}
*/
shaka.msf.MSFParser.MEDIA_TIMELINE_PACKAGING_ = 'mediatimeline';
/**
* The catalog `packaging` value of an event timeline track.
*
* @private @const {string}
*/
shaka.msf.MSFParser.EVENT_TIMELINE_PACKAGING_ = 'eventtimeline';
/**
* How close to the live edge a seek target has to land to be treated as a
* return to the live edge rather than as a seek into the DVR window, in
* seconds.
*
* Seeking to "live" lands on the seek range end, which is the newest segment
* end time; by the time the seek is handled the edge may have moved on, so an
* exact comparison would read the seek as one into the past and anchor the
* subscription just behind the edge instead of following it.
*
* @private @const {number}
*/
shaka.msf.MSFParser.LIVE_EDGE_TOLERANCE_SEC_ = 1;
/**
* What the parser keeps for a stream that is currently subscribed. It is
* everything a subscription needs to be started again from somewhere else,
* which is what a seek does.
*
* @typedef {{
* stream: shaka.extern.Stream,
* type: shaka.util.ManifestParserUtils.ContentType,
* track: msfCatalog.Track,
* trackKey: string,
* packaging: shaka.extern.MsfPackaging,
* description: shaka.extern.MsfTrackDescription,
* startLocation: ?shaka.msf.Utils.Location,
* generation: number,
* }}
*
* @property {shaka.extern.Stream} stream
* The stream this track feeds.
* @property {shaka.util.ManifestParserUtils.ContentType} type
* The stream's content type.
* @property {msfCatalog.Track} track
* The catalog track, which is what a subscription is addressed with.
* @property {string} trackKey
* The namespace-qualified track name, used as the subscription key.
* @property {shaka.extern.MsfPackaging} packaging
* The packaging that makes segments out of this track's Objects.
* @property {shaka.extern.MsfTrackDescription} description
* What the packaging derived from the catalog, held for the initialization
* segment reference every segment carries.
* @property {?shaka.msf.Utils.Location} startLocation
* Where the current subscription was asked to start, or null when it
* follows the live edge.
* @property {number} generation
* Counts the subscriptions this stream has had, so that Objects still in
* flight from a previous one can be told apart and dropped.
*/
shaka.msf.MSFParser.StreamState;
shaka.media.ManifestParser.registerParserByMime(
'application/msf', () => new shaka.msf.MSFParser());