mod live;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::sync::Arc;
use container_probe::segment;
use transmux::{TrackId, TrackSpec};
use url::Url;
use super::action::{Action, RepId, ResourceId};
use super::config::{DashConfig, PullLimits};
use super::demux::{self, Staged};
use super::error::{Error, Failure, FailureClass, FailureKind, FatalKind, Rejection, Result};
use super::meta::ResponseMeta;
use super::output::{AudioRendition, DropCause, ManifestSelection, Notice, Output, VariantInfo};
use crate::address::base_index;
use crate::address::{AddressError, PlanError, PlanOptions, SegmentPlan, plan_representation};
use crate::error::Unsupported;
use crate::mpd::urls::{BaseCandidate, append_query, representation_candidates, segment_url_query};
use crate::selection::{RepRef, Selection, SetClass, select};
use crate::timing::clock::UtcClock;
use crate::timing::period::period_timings;
use broadcast_dash::descriptors::SCHEME_PERIOD_CONTINUITY;
use broadcast_dash::inherit::{SegmentInfo, effective_common, resolve_segment_info};
use broadcast_dash::{ByteRange, Descriptor, Mpd, PresentationType};
const MISSING_TAIL_MAX_US: i128 = 50_000;
const ROTATE_AFTER_FAILURES: u32 = 2;
const STATUS_NOT_FOUND: u16 = 404;
const STATUS_GONE: u16 = 410;
const STATUS_TOO_EARLY: u16 = 425;
const MAX_REMEMBERED_DONE: usize = 4096;
const START_WITH_SAP_1: u8 = 1;
const START_WITH_SAP_2: u8 = 2;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Phase {
NeedMpd,
Running,
Ended,
Fatal,
}
enum Candidate {
Ready(i128),
Wait(i128),
Idle,
Unaddressable(&'static str),
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct InitKey {
url: String,
range: Option<(u64, u64)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum InitState {
Wanted,
InFlight,
Ready,
}
struct InitEntry {
state: InitState,
bytes: Option<Arc<Vec<u8>>>,
specs: Vec<TrackSpec>,
skipped_reported: BTreeSet<(Option<u32>, String)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReqKind {
Init,
Index,
Segment,
}
struct ReqInfo {
rep: RepId,
reference: String,
range: Option<(u64, u64)>,
kind: ReqKind,
avail_from_us: Option<i128>,
avail_until_us: Option<i128>,
}
struct IndexReq {
range: ByteRange,
}
struct RepState {
id: RepId,
at: RepRef,
candidates: Vec<BaseCandidate>,
cand: usize,
cand_failures: u32,
query: Option<String>,
plan: Option<SegmentPlan>,
index: Option<IndexReq>,
init_key: InitKey,
next_issue: u64,
start_time_us: Option<i128>,
dead: bool,
set_id: Option<u64>,
set_pos: usize,
rep_key: String,
joined: bool,
last_time: Option<u64>,
}
enum SlotState {
Issued,
Held(Vec<u8>),
Staged(Staged),
Skipped { marker: bool },
Marker,
}
struct Slot {
id: Option<ResourceId>,
rep: usize,
plan_index: u64,
state: SlotState,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct TrackIdentity {
kind: SetClass,
set_id: Option<u64>,
group: Option<u64>,
lang: Option<String>,
roles: Vec<(String, Option<String>)>,
}
struct Lane {
kind: SetClass,
set_id: Option<u64>,
identity: TrackIdentity,
reps: Vec<RepState>,
queue: VecDeque<Slot>,
base: u32,
count: u32,
announced_rep: Option<usize>,
announced_init: Option<InitKey>,
expired_run: u32,
}
impl Lane {
fn current(&self) -> Option<&RepState> {
self.reps.last().filter(|r| !r.dead)
}
fn exhausted(&self) -> bool {
match self.current() {
None => true,
Some(rep) => rep
.plan
.as_ref()
.is_some_and(|p| p.len().is_some_and(|n| rep.next_issue >= n)),
}
}
}
struct PeriodRun {
index: usize,
id: Option<String>,
start_us: i128,
lanes: Vec<Lane>,
selection: Selection,
notices: Vec<Output>,
started: bool,
issue_done: bool,
finished: bool,
}
struct PrevPeriod {
id: Option<String>,
lanes: Vec<(SetClass, Option<u64>, InitKey)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub struct StateSize {
pub periods: usize,
pub requests: usize,
pub inflight: usize,
pub parked: usize,
pub failures: usize,
pub done: usize,
pub inits: usize,
pub reps: usize,
pub queued: usize,
pub outputs: usize,
pub live: usize,
pub track_identities: usize,
}
pub struct DashClient {
cfg: PullLimits,
dash: DashConfig,
mpd_url: String,
doc_url: Option<Url>,
mpd_query: Option<String>,
phase: Phase,
want_mpd: bool,
mpd_inflight: bool,
mpd: Option<Mpd>,
periods: Vec<PeriodRun>,
issue: usize,
deliver: usize,
next_rep: u32,
next_track_id: u32,
track_ids: BTreeMap<TrackIdentity, (u32, u32)>,
previous_identities: Vec<TrackIdentity>,
reqs: BTreeMap<ResourceId, ReqInfo>,
queue: VecDeque<ResourceId>,
inflight: BTreeSet<ResourceId>,
parked: BTreeSet<ResourceId>,
failures: BTreeMap<ResourceId, u32>,
done: BTreeSet<ResourceId>,
inits: BTreeMap<InitKey, InitEntry>,
rep_loc: BTreeMap<RepId, (usize, usize, usize)>,
outputs: VecDeque<Output>,
announced_tracks: Option<String>,
prev: Option<PrevPeriod>,
any_started: bool,
outstanding: usize,
notices_sent: Vec<Notice>,
clock: UtcClock,
mpd_failures: u32,
live: Option<live::Live>,
next_utc_id: u32,
last_was_discontinuity: bool,
remembered_done: VecDeque<ResourceId>,
}
fn without_query(url: &str) -> String {
match Url::parse(url) {
Ok(mut u) => {
u.set_query(None);
u.set_fragment(None);
u.to_string()
}
Err(_) => url.to_owned(),
}
}
fn same_origin(a: &Url, b: &Url) -> bool {
a.scheme() == b.scheme()
&& a.host_str() == b.host_str()
&& a.port_or_known_default() == b.port_or_known_default()
}
pub(crate) fn filter_candidates(
candidates: Vec<BaseCandidate>,
doc_url: Option<&Url>,
cross_origin_urls: bool,
) -> Vec<BaseCandidate> {
if cross_origin_urls {
return candidates;
}
let Some(doc) = doc_url else {
return candidates;
};
candidates
.into_iter()
.filter(|c| {
c.resolve(Some(doc), "")
.and_then(|u| Url::parse(&u).ok())
.map(|p| same_origin(doc, &p))
.unwrap_or(false)
})
.collect()
}
fn address_name(e: &AddressError) -> &'static str {
match e {
AddressError::InvalidTemplate { .. } => "invalid_template",
AddressError::MissingTemplateValue { .. } => "missing_template_value",
AddressError::Unsupported(u) => u.name(),
AddressError::MissingMediaTemplate => "missing_media_template",
AddressError::MissingDuration => "missing_duration",
AddressError::TimelineNotIncreasing { .. } => "timeline_not_increasing",
AddressError::NumberNotIncreasing { .. } => "number_not_increasing",
AddressError::TimelineTooLong { .. } => "timeline_too_long",
AddressError::ListTimelineMismatch { .. } => "list_timeline_mismatch",
AddressError::Sidx(_) => "bad_sidx",
AddressError::Overflow => "overflow",
AddressError::IndexRequired => "index_required",
AddressError::EmptySegmentUrl => "empty_segment_url",
AddressError::ZeroSegmentDuration => "zero_segment_duration",
_ => "unsupported_feature",
}
}
fn plan_error_name(e: &PlanError) -> &'static str {
match e {
PlanError::NoSuchRepresentation => "no_such_representation",
PlanError::Inherit(_) => "inherit",
PlanError::Address(a) => address_name(a),
PlanError::PeriodUnresolved => "period_unresolved",
PlanError::MissingAvailabilityStartTime => "missing_availability_start_time",
}
}
fn range_pair(r: Option<ByteRange>) -> std::result::Result<Option<(u64, u64)>, DropCause> {
r.map(|r| r.offset_len().map_err(|u| DropCause::Addressing(u.name())))
.transpose()
}
fn continuity_signals(set: &broadcast_dash::AdaptationSet, previous: &str) -> bool {
let named = |d: &Descriptor| {
d.scheme_id_uri == SCHEME_PERIOD_CONTINUITY && d.value.as_deref() == Some(previous)
};
set.supplemental_properties.iter().any(named) || set.essential_properties.iter().any(named)
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct DashClientConfig {
pub mpd_url: String,
pub dash: DashConfig,
pub limits: PullLimits,
}
impl DashClientConfig {
pub fn new(mpd_url: impl Into<String>) -> Self {
DashClientConfig {
mpd_url: mpd_url.into(),
dash: DashConfig::default(),
limits: PullLimits::default(),
}
}
#[must_use]
pub fn with_dash(mut self, dash: DashConfig) -> Self {
self.dash = dash;
self
}
#[must_use]
pub fn with_limits(mut self, limits: PullLimits) -> Self {
self.limits = limits;
self
}
}
impl DashClient {
#[must_use]
pub fn new(config: DashClientConfig) -> Self {
let DashClientConfig {
mpd_url,
dash,
limits,
} = config;
let doc_url = Url::parse(&mpd_url).ok();
let mpd_query = doc_url.as_ref().and_then(|u| u.query()).map(str::to_owned);
Self {
cfg: limits,
dash,
mpd_url,
doc_url,
mpd_query,
phase: Phase::NeedMpd,
want_mpd: true,
mpd_inflight: false,
mpd: None,
periods: Vec::new(),
issue: 0,
deliver: 0,
next_rep: 0,
next_track_id: 1,
track_ids: BTreeMap::new(),
previous_identities: Vec::new(),
reqs: BTreeMap::new(),
queue: VecDeque::new(),
inflight: BTreeSet::new(),
parked: BTreeSet::new(),
failures: BTreeMap::new(),
done: BTreeSet::new(),
inits: BTreeMap::new(),
rep_loc: BTreeMap::new(),
outputs: VecDeque::new(),
announced_tracks: None,
prev: None,
any_started: false,
outstanding: 0,
notices_sent: Vec::new(),
clock: UtcClock::new(),
mpd_failures: 0,
live: None,
next_utc_id: 0,
last_was_discontinuity: false,
remembered_done: VecDeque::new(),
}
}
#[must_use]
pub fn mpd_url(&self) -> &str {
&self.mpd_url
}
pub fn set_mpd_base(&mut self, base: impl Into<String>) {
self.doc_url = Url::parse(&base.into()).ok();
}
#[must_use]
pub fn is_static(&self) -> Option<bool> {
self.mpd
.as_ref()
.map(|m| m.kind == PresentationType::Static)
}
#[must_use]
pub fn presentation_ended(&self) -> bool {
self.phase == Phase::Ended
}
#[must_use]
pub fn max_segment_duration_secs(&self) -> Option<u64> {
let d = self.mpd.as_ref()?.max_segment_duration?;
Some(d.as_secs() + u64::from(d.subsec_nanos() > 0))
}
#[must_use]
pub fn state_size(&self) -> StateSize {
StateSize {
periods: self.periods.len(),
requests: self.reqs.len(),
inflight: self.inflight.len(),
parked: self.parked.len(),
failures: self.failures.len(),
done: self.done.len(),
inits: self.inits.len(),
reps: self.rep_loc.len(),
queued: self.queue.len(),
outputs: self.outputs.len(),
live: self.live.as_ref().map_or(0, live::Live::bookkeeping),
track_identities: self.track_ids.len(),
}
}
#[must_use]
pub fn awaits(&self, id: ResourceId) -> bool {
self.inflight.contains(&id)
}
pub fn next_output(&mut self) -> Option<Output> {
self.outputs.pop_front()
}
fn stopped(&self) -> bool {
matches!(self.phase, Phase::Ended | Phase::Fatal)
}
fn ensure_live(&self) -> Result<()> {
if self.stopped() {
Err(Error::Stopped)
} else {
Ok(())
}
}
fn go_fatal(&mut self, kind: FatalKind) {
if self.phase == Phase::Fatal {
return;
}
self.phase = Phase::Fatal;
self.queue.clear();
self.live = None;
self.outputs.push_back(Output::Fatal(kind));
}
fn discontinuity(&mut self) {
if !self.last_was_discontinuity {
self.last_was_discontinuity = true;
self.outputs.push_back(Output::Discontinuity);
}
}
fn mark_done(&mut self, id: ResourceId) {
if self.done.insert(id) {
self.remembered_done.push_back(id);
while self.remembered_done.len() > MAX_REMEMBERED_DONE {
if let Some(old) = self.remembered_done.pop_front() {
self.done.remove(&old);
}
}
}
}
fn notice_once(&mut self, n: Notice) {
if !self.notices_sent.contains(&n) {
self.notices_sent.push(n);
self.outputs.push_back(Output::Notice(n));
}
}
pub fn on_mpd(&mut self, now_us: u64, bytes: &[u8]) -> Result<()> {
self.on_mpd_with(now_us, bytes, &ResponseMeta::at(now_us))
}
pub fn on_mpd_with(
&mut self,
now_us: u64,
bytes: &[u8],
meta: &ResponseMeta<'_>,
) -> Result<()> {
self.ensure_live()?;
if !self.mpd_inflight {
return Err(Error::MpdNotAwaited);
}
self.mpd_inflight = false;
let mpd = match Mpd::parse_with(bytes, &self.cfg.limits) {
Ok(m) => m,
Err(e) => {
self.mpd_failed(now_us);
return Err(Error::Mpd(e));
}
};
if self.mpd.is_some() {
self.refresh_mpd(now_us, mpd, bytes, meta);
return Ok(());
}
if mpd.kind == PresentationType::List {
self.go_fatal(FatalKind::Unsupported(Unsupported::PresentationTypeList));
return Ok(());
}
if mpd.periods.is_empty() {
self.go_fatal(FatalKind::MpdInvalid {
reason: "no_period",
});
return Ok(());
}
let dynamic = mpd.kind == PresentationType::Dynamic;
if dynamic && mpd.availability_start_time_us.is_none() {
self.go_fatal(FatalKind::MpdInvalid {
reason: "missing_availability_start_time",
});
return Ok(());
}
let features = mpd.features;
let event_streams = mpd.has_event_streams();
let preselections = mpd.has_preselections();
let patch_location = mpd.has_patch_location();
self.mpd = Some(mpd);
self.phase = Phase::Running;
if event_streams {
self.notice_once(Notice::EventStreamIgnored);
}
if preselections {
self.notice_once(Notice::PreselectionIgnored);
}
if patch_location {
self.notice_once(Notice::PatchIgnored);
}
if dynamic && features.low_latency {
self.notice_once(Notice::LowLatencyIgnored);
}
if dynamic {
self.start_live(now_us, bytes, meta);
} else {
if let Some(u) = meta.final_url.and_then(|f| self.accept_final_url(f)) {
self.doc_url = Some(u);
}
self.activate(0);
self.issue = 0;
}
Ok(())
}
pub fn on_mpd_failure(&mut self, now_us: u64, failure: Failure) -> FailureClass {
self.mpd_inflight = false;
self.want_mpd = false;
let attempt = self.mpd_failed(now_us);
let kind = match failure {
Failure::Status(STATUS_NOT_FOUND | STATUS_GONE) => FailureKind::Missing,
Failure::Status(STATUS_TOO_EARLY) => FailureKind::TooEarly,
Failure::Status(_) => FailureKind::Server,
Failure::Transport | Failure::Timeout => FailureKind::Transport,
Failure::TooLarge => FailureKind::Invalid,
};
FailureClass::Retry { attempt, kind }
}
pub fn retry_mpd(&mut self, _now_us: u64) {
if self.mpd_inflight || self.stopped() {
return;
}
let waiting = self.phase == Phase::NeedMpd
|| self.live.as_ref().is_some_and(|l| l.refresh.awaiting_retry);
if waiting {
self.want_mpd = true;
}
}
pub fn poll(&mut self, now_us: u64) -> Option<Action> {
if self.stopped() {
return None;
}
self.tick(now_us);
if self.stopped() {
return None;
}
if let Some(a) = self.mpd_action(now_us) {
return Some(a);
}
if self.phase != Phase::Running {
return None;
}
if let Some(a) = self.utc_action(now_us) {
return Some(a);
}
loop {
if self.queue.is_empty() {
self.fill_queue(now_us);
if self.queue.is_empty() {
self.drain();
}
}
let id = self.queue.pop_front()?;
if let Some(action) = self.emit(id) {
return Some(action);
}
if self.stopped() {
return None;
}
}
}
fn window_open(&self) -> bool {
self.outstanding < self.cfg.window()
}
fn fill_queue(&mut self, now_us: u64) {
self.reset_wake();
loop {
let Some(pi) = (self.issue < self.periods.len()).then_some(self.issue) else {
return;
};
if self.live_blocked() {
return;
}
let exhausted = self.periods[pi].lanes.iter().all(Lane::exhausted);
if self.periods[pi].issue_done || (exhausted && !self.period_open(pi, now_us)) {
self.periods[pi].issue_done = true;
let next = self.periods[pi].index + 1;
if self.can_activate(next) {
self.activate(next);
self.issue = self.periods.len() - 1;
continue;
}
return;
}
if exhausted {
self.note_exhausted(now_us);
return;
}
if !self.window_open() {
return;
}
let utc = self.utc_now(now_us);
let mut pick: Option<(usize, i128)> = None;
for li in 0..self.periods[pi].lanes.len() {
match self.lane_candidate(pi, li, utc, now_us) {
Candidate::Ready(start_us) => {
if pick.is_none_or(|(_, best)| start_us < best) {
pick = Some((li, start_us));
}
}
Candidate::Wait(at_utc_us) => self.note_wake(at_utc_us),
Candidate::Idle => {}
Candidate::Unaddressable(reason) => {
self.rep_segment_error(pi, li, reason);
if self.stopped() {
return;
}
match self.lane_candidate(pi, li, utc, now_us) {
Candidate::Ready(start_us) => pick = Some((li, start_us)),
Candidate::Wait(at_utc_us) => self.note_wake(at_utc_us),
Candidate::Idle | Candidate::Unaddressable(_) => {}
}
}
}
}
let Some((li, _)) = pick else {
return;
};
if !self.issue_segment(pi, li) {
if self.stopped() || self.periods[pi].lanes[li].current().is_some() {
return;
}
continue;
}
}
}
fn lane_candidate(
&mut self,
pi: usize,
li: usize,
utc: Option<i128>,
now_us: u64,
) -> Candidate {
match utc {
Some(utc) if self.is_live_mpd() => self.live_lane_candidate(pi, li, utc, now_us),
_ => {
let lane = &self.periods[pi].lanes[li];
let Some(rep) = lane.current() else {
return Candidate::Idle;
};
let Some(plan) = rep.plan.as_ref() else {
return Candidate::Idle;
};
match plan.get(rep.next_issue) {
Ok(Some(s)) => Candidate::Ready(s.start_us),
Ok(None) => Candidate::Idle,
Err(e) => Candidate::Unaddressable(address_name(&e)),
}
}
}
}
fn issue_segment(&mut self, pi: usize, li: usize) -> bool {
let lane = &mut self.periods[pi].lanes[li];
let ri = lane.reps.len() - 1;
let rep = &mut lane.reps[ri];
let seg = match rep.plan.as_ref().map(|p| p.get(rep.next_issue)) {
Some(Ok(Some(seg))) => seg,
Some(Err(e)) => {
let reason = address_name(&e);
self.rep_segment_error(pi, li, reason);
return false;
}
_ => return false,
};
let range = match range_pair(seg.byte_range) {
Ok(r) => r,
Err(cause) => {
let reason = match cause {
DropCause::Addressing(r) => r,
_ => "unusable_url",
};
self.go_fatal(FatalKind::UnsupportedAddressing { reason });
return false;
}
};
let id = ResourceId::Segment {
rep: rep.id,
number: seg.number,
};
let plan_index = rep.next_issue;
rep.next_issue += 1;
rep.last_time = Some(seg.time_ticks);
let rep_id = rep.id;
lane.queue.push_back(Slot {
id: Some(id),
rep: ri,
plan_index,
state: SlotState::Issued,
});
self.outstanding += 1;
self.reqs.insert(
id,
ReqInfo {
rep: rep_id,
reference: seg.reference,
range,
kind: ReqKind::Segment,
avail_from_us: seg.available_from_us,
avail_until_us: seg.available_until_us,
},
);
self.queue.push_back(id);
true
}
fn rep(&self, id: RepId) -> Option<&RepState> {
let (p, l, r) = *self.rep_loc.get(&id)?;
self.periods.get(p)?.lanes.get(l)?.reps.get(r)
}
fn rep_mut(&mut self, id: RepId) -> Option<&mut RepState> {
let (p, l, r) = *self.rep_loc.get(&id)?;
self.periods.get_mut(p)?.lanes.get_mut(l)?.reps.get_mut(r)
}
fn rep_segment_error(&mut self, pi: usize, li: usize, reason: &'static str) -> bool {
if self.periods[pi].lanes[li].current().is_none() {
return false;
}
self.drop_rep(pi, li, DropCause::Addressing(reason));
if self
.periods
.iter()
.all(|p| p.lanes.iter().all(|l| l.current().is_none()))
{
self.go_fatal(FatalKind::UnsupportedAddressing { reason });
}
true
}
fn drop_rep(&mut self, pi: usize, li: usize, cause: DropCause) {
if let Some(ri) = self.periods[pi].lanes[li].reps.len().checked_sub(1) {
self.drop_rep_at(pi, li, ri, cause);
}
self.note_wake(i128::MIN);
}
fn drop_rep_in(&mut self, rep: RepId, cause: DropCause) {
if let Some((pi, li, _)) = self.rep_loc.get(&rep).copied() {
self.drop_rep(pi, li, cause);
}
}
fn drop_rep_at(&mut self, pi: usize, li: usize, ri: usize, cause: DropCause) {
self.periods[pi].lanes[li].reps[ri].dead = true;
let at = self.periods[pi].lanes[li].reps[ri].at;
self.outputs.push_back(Output::TrackDropped {
period: self.periods[pi].index,
set: at.set,
rep: Some(at.rep),
cause,
});
}
fn resolve(
&self,
rep: &RepState,
reference: &str,
media: bool,
) -> std::result::Result<String, DropCause> {
let cand = rep
.candidates
.get(rep.cand % rep.candidates.len().max(1))
.ok_or(DropCause::UnusableUrl)?;
let mut url = cand
.resolve(self.doc_url.as_ref(), reference)
.ok_or(DropCause::UnusableUrl)?;
if media && let Some(q) = &rep.query {
url = append_query(&url, q);
}
let parsed = Url::parse(&url).map_err(|_| DropCause::UnusableUrl)?;
if parsed.scheme() == "data" {
return Err(DropCause::Unsupported(Unsupported::DataUrl));
}
if !matches!(parsed.scheme(), "http" | "https") {
return Err(DropCause::UnusableUrl);
}
if !self.dash.cross_origin_urls
&& let Some(doc) = &self.doc_url
&& !same_origin(doc, &parsed)
{
return Err(DropCause::CrossOrigin);
}
Ok(url)
}
pub(super) fn accept_final_url(&self, final_url: &str) -> Option<Url> {
let parsed = Url::parse(final_url).ok()?;
if !matches!(parsed.scheme(), "http" | "https") {
return None;
}
let pinned_elsewhere = !self.dash.cross_origin_urls
&& self
.doc_url
.as_ref()
.is_some_and(|d| !same_origin(d, &parsed));
(!pinned_elsewhere).then_some(parsed)
}
fn emit(&mut self, id: ResourceId) -> Option<Action> {
let (rep_id, reference, range, kind) = {
let info = self.reqs.get(&id)?;
(info.rep, info.reference.clone(), info.range, info.kind)
};
let url = {
let rep = self.rep(rep_id)?;
match self.resolve(rep, &reference, kind == ReqKind::Segment) {
Ok(u) => u,
Err(cause) => {
let reason = match cause {
DropCause::CrossOrigin => "cross_origin",
DropCause::Unsupported(u) => u.name(),
_ => "unusable_url",
};
self.go_fatal(FatalKind::UnsupportedAddressing { reason });
return None;
}
}
};
self.parked.remove(&id);
self.inflight.insert(id);
match kind {
ReqKind::Init => {
let key = self.rep(rep_id)?.init_key.clone();
if let Some(e) = self.inits.get_mut(&key) {
e.state = InitState::InFlight;
}
Some(Action::FetchInit {
id,
url,
byte_range: range,
})
}
ReqKind::Index => Some(Action::FetchIndex {
id,
url,
byte_range: range,
}),
ReqKind::Segment => Some(Action::FetchSegment {
id,
url,
byte_range: range,
}),
}
}
fn activate(&mut self, index: usize) {
let Some(mpd) = self.mpd.as_ref() else { return };
let timing = period_timings(mpd).get(index).copied();
let start_us = timing.and_then(|t| t.start_us).unwrap_or(0);
let selection = select(mpd, index, &self.cfg.policy);
let period_id = mpd.periods.get(index).and_then(|p| p.id.clone());
let mut notices = Vec::new();
let mut encryption = false;
for d in &selection.dropped {
if d.reason == crate::selection::DropReason::UnsupportedEncryption {
encryption = true;
}
notices.push(Output::TrackDropped {
period: index,
set: d.set,
rep: d.rep,
cause: DropCause::Selection(d.reason),
});
}
if encryption {
notices.push(Output::Notice(Notice::UnsupportedEncryption));
}
if selection
.dropped
.iter()
.any(|d| d.reason == crate::selection::DropReason::UnresolvedXlink)
{
notices.push(Output::Notice(Notice::XlinkSkipped));
}
let video = selection
.chosen_variant
.and_then(|i| selection.variants.get(i).copied());
let audio = selection.chosen_audio;
let mut run = PeriodRun {
index,
id: period_id,
start_us,
lanes: Vec::new(),
selection,
notices,
started: false,
issue_done: false,
finished: false,
};
let pi = self.periods.len();
for (kind, at) in [(SetClass::Video, video), (SetClass::Audio, audio)] {
let Some(at) = at else { continue };
match self.build_rep(at) {
Ok(rep) => {
let resolved = at.resolve(self.mpd.as_ref().expect("mpd present"));
let set_id = resolved.and_then(|(_, s, _)| s.id);
let identity = resolved.map_or_else(
|| TrackIdentity {
kind,
set_id: None,
group: None,
lang: None,
roles: Vec::new(),
},
|(_, s, _)| TrackIdentity {
kind,
set_id: s.id,
group: s.group,
lang: s.lang.clone(),
roles: s
.roles
.iter()
.map(|d| (d.scheme_id_uri.clone(), d.value.clone()))
.collect(),
},
);
run.lanes.push(Lane {
kind,
set_id,
identity,
reps: vec![rep],
queue: VecDeque::new(),
base: 0,
count: 0,
announced_rep: None,
announced_init: None,
expired_run: 0,
});
}
Err(cause) => run.notices.push(Output::TrackDropped {
period: index,
set: at.set,
rep: Some(at.rep),
cause,
}),
}
}
self.periods.push(run);
for li in 0..self.periods[pi].lanes.len() {
self.register_rep(pi, li, 0);
}
}
fn init_reference(
mpd: &Mpd,
at: RepRef,
planned: &std::result::Result<crate::address::PlannedRepresentation, PlanError>,
) -> std::result::Result<Option<(String, Option<ByteRange>)>, DropCause> {
match planned {
Ok(p) => Ok(p.plan.init().map(|r| (r.reference, r.byte_range))),
Err(PlanError::Address(AddressError::IndexRequired)) => {
let Some((period, set, rep)) = at.resolve(mpd) else {
return Ok(None);
};
Ok(match resolve_segment_info(period, set, rep).ok() {
Some(SegmentInfo::Base(b)) => b
.initialization
.as_ref()
.map(|u| (u.source_url.clone().unwrap_or_default(), u.range)),
_ => None,
})
}
Err(e) => Err(DropCause::Addressing(plan_error_name(e))),
}
}
fn build_rep(&mut self, at: RepRef) -> std::result::Result<RepState, DropCause> {
let mpd = self.mpd.as_ref().expect("mpd present");
let (period, set, rep) = at
.resolve(mpd)
.ok_or(DropCause::Addressing("no_such_representation"))?;
let candidates = filter_candidates(
representation_candidates(mpd, period, set, rep),
self.doc_url.as_ref(),
self.dash.cross_origin_urls,
);
if candidates.is_empty() && !self.dash.cross_origin_urls {
return Err(DropCause::CrossOrigin);
}
let query = segment_url_query(mpd, period, set, rep, self.mpd_query.as_deref())
.map_err(|_| DropCause::UnusableUrl)?;
let planned = plan_representation(mpd, at, &self.plan_options());
let init = Self::init_reference(mpd, at, &planned)?.ok_or(DropCause::NoInitialization)?;
let (plan, index) = match planned {
Ok(p) => {
if p.plan.len().is_none() && mpd.kind != PresentationType::Dynamic {
return Err(DropCause::UnboundedPlan);
}
(Some(p.plan), None)
}
Err(PlanError::Address(AddressError::IndexRequired)) => {
let info = resolve_segment_info(period, set, rep)
.map_err(|_| DropCause::Addressing("inherit"))?;
let range = match info {
SegmentInfo::Base(b) => b.index_range,
_ => None,
};
let range = range.ok_or(DropCause::Addressing("index_required"))?;
(None, Some(IndexReq { range }))
}
Err(e) => return Err(DropCause::Addressing(plan_error_name(&e))),
};
let init_range = range_pair(init.1)?;
let id = RepId(self.next_rep);
self.next_rep = self.next_rep.saturating_add(1);
let mut state = RepState {
id,
at,
candidates,
cand: 0,
cand_failures: 0,
query,
plan,
index,
init_key: InitKey {
url: String::new(),
range: init_range,
},
next_issue: 0,
start_time_us: None,
dead: false,
set_id: set.id,
set_pos: at.set,
rep_key: rep.id.clone(),
joined: mpd.kind != PresentationType::Dynamic,
last_time: None,
};
let init_url = self.resolve(&state, &init.0, false)?;
state.init_key.url = without_query(&init_url);
if let Some(ix) = &state.index {
range_pair(Some(ix.range))?;
self.resolve(&state, "", false)?;
}
let first = state
.plan
.as_ref()
.map(|p| p.get(0))
.transpose()
.map_err(|e| DropCause::Addressing(address_name(&e)))?
.flatten();
if let Some(seg) = first {
range_pair(seg.byte_range)?;
self.resolve(&state, &seg.reference, true)?;
}
Ok(state)
}
fn register_rep(&mut self, pi: usize, li: usize, ri: usize) {
let rep = &self.periods[pi].lanes[li].reps[ri];
let rep_id = rep.id;
let key = rep.init_key.clone();
let has_index = rep.index.is_some();
let index_range = rep.index.as_ref().map(|i| i.range);
let init_reference = {
let mpd = self.mpd.as_ref().expect("mpd present");
let planned = rep.plan.as_ref().map(|p| p.init());
match planned {
Some(init) => init.map(|r| (r.reference, r.byte_range)),
None => Self::init_reference(
mpd,
rep.at,
&Err(PlanError::Address(AddressError::IndexRequired)),
)
.ok()
.flatten(),
}
};
self.rep_loc.insert(rep_id, (pi, li, ri));
if let std::collections::btree_map::Entry::Vacant(v) = self.inits.entry(key) {
v.insert(InitEntry {
state: InitState::Wanted,
bytes: None,
specs: Vec::new(),
skipped_reported: BTreeSet::new(),
});
let (reference, range) = init_reference.unwrap_or_default();
let init_id = ResourceId::Init { rep: rep_id };
self.reqs.insert(
init_id,
ReqInfo {
rep: rep_id,
reference,
range: range.and_then(|r| r.offset_len().ok()),
kind: ReqKind::Init,
avail_from_us: None,
avail_until_us: None,
},
);
self.queue.push_back(init_id);
}
if has_index && let Some(r) = index_range {
let index_id = ResourceId::Index { rep: rep_id };
self.reqs.insert(
index_id,
ReqInfo {
rep: rep_id,
reference: String::new(),
range: r.offset_len().ok(),
kind: ReqKind::Index,
avail_from_us: None,
avail_until_us: None,
},
);
self.queue.push_back(index_id);
}
}
fn requested(&self, id: ResourceId) -> Result<()> {
if self.done.contains(&id) {
return Err(Error::DuplicateResource { id });
}
if !self.inflight.contains(&id) {
return Err(Error::UnrequestedResource { id });
}
Ok(())
}
fn body_rejection(&self, id: ResourceId, bytes: &[u8]) -> Option<Rejection> {
if bytes.len() > self.cfg.max_segment_bytes {
return Some(Rejection::TooLarge);
}
let range = self.reqs.get(&id).and_then(|r| r.range)?;
(u64::try_from(bytes.len()).ok() != Some(range.1)).then_some(Rejection::WrongLength)
}
fn max_failures(&self, id: ResourceId) -> u32 {
match id {
ResourceId::Segment { .. } => self.cfg.max_segment_failures,
ResourceId::Init { .. } | ResourceId::Index { .. } => self.cfg.max_init_failures,
}
}
fn count_failure(
&mut self,
now_us: u64,
id: ResourceId,
kind: FailureKind,
status: Option<u16>,
) -> FailureClass {
self.inflight.remove(&id);
let mut rotated = false;
let count = {
let n = self.failures.entry(id).or_insert(0);
*n = n.saturating_add(1);
*n
};
if count >= self.max_failures(id) {
if self.is_live_mpd() && matches!(id, ResourceId::Segment { .. }) {
return self.give_up_live_segment(id, status);
}
let fatal = match id {
ResourceId::Segment { .. } => FatalKind::SegmentFetchFailed { id, status },
ResourceId::Init { .. } | ResourceId::Index { .. } => {
FatalKind::InitFetchFailed { id, status }
}
};
self.go_fatal(fatal);
return FailureClass::GiveUp;
}
let doc = self.doc_url.clone();
let cross = self.dash.cross_origin_urls;
if let Some(rep) = self.rep_mut(id.rep()) {
rep.cand_failures += 1;
if rep.cand_failures >= ROTATE_AFTER_FAILURES && rep.candidates.len() > 1 {
let len = rep.candidates.len();
let mut next = (rep.cand + 1) % len;
let mut found = false;
for _ in 0..len {
let ok = cross
|| doc.as_ref().is_none_or(|d| {
rep.candidates[next]
.resolve(Some(d), "")
.and_then(|u| Url::parse(&u).ok())
.map(|p| same_origin(d, &p))
.unwrap_or(false)
});
if ok {
found = true;
break;
}
next = (next + 1) % len;
}
if found && next != rep.cand {
rep.cand = next;
rep.cand_failures = 0;
rotated = true;
}
}
}
if rotated {
self.after_rotation(now_us);
}
self.parked.insert(id);
FailureClass::Retry {
attempt: count,
kind,
}
}
fn reject(&mut self, now_us: u64, id: ResourceId, reason: Rejection) -> Error {
let class = self.count_failure(now_us, id, FailureKind::Invalid, None);
Error::Rejected { id, reason, class }
}
fn clear_success(&mut self, id: ResourceId) {
self.inflight.remove(&id);
self.parked.remove(&id);
self.failures.remove(&id);
self.mark_done(id);
}
pub fn on_init(&mut self, now_us: u64, id: ResourceId, bytes: &[u8]) -> Result<()> {
self.ensure_live()?;
let ResourceId::Init { rep } = id else {
return Err(Error::UnrequestedResource { id });
};
self.requested(id)?;
if let Some(r) = self.body_rejection(id, bytes) {
return Err(self.reject(now_us, id, r));
}
if !segment::progressive_movie(bytes) {
return Err(self.reject(now_us, id, Rejection::InitUndecodable));
}
let info = match demux::init_info(bytes) {
Ok(i) => i,
Err(_) => return Err(self.reject(now_us, id, Rejection::DemuxFailed)),
};
let key = self
.rep(rep)
.ok_or(Error::UnrequestedResource { id })?
.init_key
.clone();
if let Some(e) = self.inits.get_mut(&key) {
e.state = InitState::Ready;
e.bytes = Some(Arc::new(bytes.to_vec()));
e.specs = info.specs;
for s in info.skipped {
let k = (s.track_id.map(TrackId::value), s.fourcc.clone());
if e.skipped_reported.insert(k) {
self.outputs.push_back(Output::TrackSkipped {
track_id: None,
fourcc: s.fourcc,
reason: s.reason,
});
}
}
}
self.clear_success(id);
self.reqs.remove(&id);
let held = self.demux_held(now_us, &key);
self.drain();
held
}
fn demux_held(&mut self, now_us: u64, key: &InitKey) -> Result<()> {
let Some(init) = self.inits.get(key).and_then(|e| e.bytes.clone()) else {
return Ok(());
};
let mut first_error = None;
let from = self.deliver.min(self.periods.len());
for pi in from..self.periods.len() {
for li in 0..self.periods[pi].lanes.len() {
let n = self.periods[pi].lanes[li].queue.len();
for qi in 0..n {
let (id, matches) = {
let lane = &self.periods[pi].lanes[li];
let slot = &lane.queue[qi];
let m = matches!(slot.state, SlotState::Held(_))
&& lane.reps.get(slot.rep).is_some_and(|r| &r.init_key == key);
(slot.id, m)
};
if !matches {
continue;
}
let slot = &mut self.periods[pi].lanes[li].queue[qi];
let SlotState::Held(bytes) =
std::mem::replace(&mut slot.state, SlotState::Issued)
else {
continue;
};
match demux::demux_segment(&init, &bytes) {
Ok(staged) => {
self.periods[pi].lanes[li].queue[qi].state = SlotState::Staged(staged);
}
Err(_) => {
if let Some(id) = id {
self.inflight.insert(id);
let e = self.reject(now_us, id, Rejection::DemuxFailed);
first_error.get_or_insert(e);
}
}
}
}
}
}
match first_error {
Some(e) => Err(e),
None => Ok(()),
}
}
pub fn on_index(&mut self, now_us: u64, id: ResourceId, bytes: &[u8]) -> Result<()> {
self.ensure_live()?;
let ResourceId::Index { rep } = id else {
return Err(Error::UnrequestedResource { id });
};
self.requested(id)?;
if let Some(r) = self.body_rejection(id, bytes) {
return Err(self.reject(now_us, id, r));
}
let sidx = match base_index::parse_index(bytes) {
Ok(s) => s,
Err(_) => return Err(self.reject(now_us, id, Rejection::BadIndex)),
};
let (at, range) = {
let r = self.rep(rep).ok_or(Error::UnrequestedResource { id })?;
let range = r.index.as_ref().map(|i| i.range);
(r.at, range)
};
let Some(range) = range else {
return Err(Error::UnrequestedResource { id });
};
let anchor = match base_index::anchor_after(&range) {
Ok(a) => a,
Err(_) => return Err(self.reject(now_us, id, Rejection::BadIndex)),
};
let planned = {
let mpd = self.mpd.as_ref().expect("mpd present");
plan_representation(mpd, at, &self.plan_options().with_index(&sidx, anchor))
};
self.clear_success(id);
self.reqs.remove(&id);
match planned {
Ok(p) => {
let start = self.rep_mut(rep).and_then(|r| r.start_time_us.take());
match start.map_or(Ok(None), |t| p.plan.index_at_or_before(t)) {
Ok(index) => {
if let Some(r) = self.rep_mut(rep) {
r.next_issue = index.unwrap_or(0);
r.plan = Some(p.plan);
}
}
Err(e) => self.drop_rep_in(rep, DropCause::Addressing(address_name(&e))),
}
}
Err(e) => self.drop_rep_in(rep, DropCause::Addressing(plan_error_name(&e))),
}
self.drain();
Ok(())
}
pub fn on_segment(&mut self, now_us: u64, id: ResourceId, bytes: &[u8]) -> Result<()> {
self.ensure_live()?;
let ResourceId::Segment { rep, .. } = id else {
return Err(Error::UnrequestedResource { id });
};
self.requested(id)?;
if let Some(r) = self.body_rejection(id, bytes) {
return Err(self.reject(now_us, id, r));
}
if segment::progressive_movie(bytes) {
self.inflight.remove(&id);
self.go_fatal(FatalKind::NonFragmentedMp4 { id });
return Err(Error::Rejected {
id,
reason: Rejection::NonFragmentedMp4,
class: FailureClass::GiveUp,
});
}
if !segment::media_fragment(bytes) {
return Err(self.reject(now_us, id, Rejection::NotMedia));
}
let key = self
.rep(rep)
.ok_or(Error::UnrequestedResource { id })?
.init_key
.clone();
let init = self.inits.get(&key).and_then(|e| e.bytes.clone());
let state = match init {
Some(init) => match demux::demux_segment(&init, bytes) {
Ok(staged) => SlotState::Staged(staged),
Err(_) => return Err(self.reject(now_us, id, Rejection::DemuxFailed)),
},
None => SlotState::Held(bytes.to_vec()),
};
let Some(slot) = self.slot_mut(id) else {
return Err(Error::UnrequestedResource { id });
};
slot.state = state;
self.inflight.remove(&id);
self.parked.remove(&id);
self.failures.remove(&id);
self.mark_done(id);
self.segment_delivered(id);
self.drain();
Ok(())
}
fn slot_mut(&mut self, id: ResourceId) -> Option<&mut Slot> {
let (p, l, _) = *self.rep_loc.get(&id.rep())?;
self.periods
.get_mut(p)?
.lanes
.get_mut(l)?
.queue
.iter_mut()
.find(|s| s.id == Some(id))
}
pub fn on_failure(
&mut self,
now_us: u64,
id: ResourceId,
failure: Failure,
) -> Result<FailureClass> {
self.ensure_live()?;
if !self.inflight.contains(&id) {
return Err(Error::UnrequestedResource { id });
}
if self.is_live_mpd() && matches!(id, ResourceId::Segment { .. }) {
return self.live_segment_failure(now_us, id, failure);
}
let status = match failure {
Failure::Status(s) => Some(s),
Failure::Transport | Failure::Timeout | Failure::TooLarge => None,
};
if matches!(status, Some(STATUS_NOT_FOUND | STATUS_GONE)) && self.is_missing_tail(id) {
self.inflight.remove(&id);
self.failures.remove(&id);
if let Some(live) = self.live.as_mut() {
live.early.remove(&id);
}
self.mark_done(id);
if let Some(slot) = self.slot_mut(id) {
slot.state = SlotState::Skipped { marker: false };
}
self.drain();
return Ok(FailureClass::EndTail);
}
Ok(self.count_failure(now_us, id, live::failure_kind(failure), status))
}
fn is_missing_tail(&self, id: ResourceId) -> bool {
let ResourceId::Segment { rep, .. } = id else {
return false;
};
let Some((pi, li, _)) = self.rep_loc.get(&rep).copied() else {
return false;
};
let lane = &self.periods[pi].lanes[li];
let Some(slot) = lane.queue.iter().find(|s| s.id == Some(id)) else {
return false;
};
let Some(r) = lane.reps.get(slot.rep) else {
return false;
};
let Some(plan) = r.plan.as_ref() else {
return false;
};
let last = plan.len().and_then(|n| n.checked_sub(1));
last == Some(slot.plan_index)
&& plan.len().is_some_and(|n| n > 1)
&& plan
.get(slot.plan_index)
.is_ok_and(|s| s.is_some_and(|s| s.duration_us < MISSING_TAIL_MAX_US))
}
pub fn retry(&mut self, _now_us: u64, id: ResourceId) -> bool {
if self.stopped() || !self.parked.remove(&id) {
return false;
}
self.queue.push_front(id);
true
}
pub fn skip(&mut self, _now_us: u64, id: ResourceId) -> bool {
if self.stopped() || !matches!(id, ResourceId::Segment { .. }) || !self.parked.remove(&id) {
return false;
}
self.failures.remove(&id);
if let Some(live) = self.live.as_mut() {
live.early.remove(&id);
}
self.mark_done(id);
if let Some(slot) = self.slot_mut(id) {
slot.state = SlotState::Skipped { marker: true };
}
self.drain();
true
}
fn drain(&mut self) {
loop {
if self.phase != Phase::Running {
return;
}
let Some(pi) = (self.deliver < self.periods.len()).then_some(self.deliver) else {
return;
};
if !self.periods[pi].started {
if !self.ready_to_start(pi) {
return;
}
self.start_period(pi);
}
self.deliver_slots(pi);
let p = &self.periods[pi];
let complete = p.issue_done && p.lanes.iter().all(|l| l.queue.is_empty());
if !complete {
return;
}
self.periods[pi].finished = true;
self.periods[pi].lanes.clear();
let total = self.mpd.as_ref().map_or(0, |m| m.periods.len());
let last_of_mpd = self.periods[pi].index + 1 >= total;
self.deliver += 1;
self.prune_finished();
if last_of_mpd {
if self.live_open() {
return;
}
if self.any_started {
self.outputs.push_back(Output::EndOfStream);
self.phase = Phase::Ended;
self.live = None;
} else {
self.go_fatal(FatalKind::MpdInvalid {
reason: "no_playable_representation",
});
}
return;
}
}
}
fn prune_finished(&mut self) {
while self.deliver > 0 && self.issue > 0 && self.periods.first().is_some_and(|p| p.finished)
{
self.periods.remove(0);
self.deliver -= 1;
self.issue -= 1;
self.rep_loc.retain(|_, loc| loc.0 != 0);
for loc in self.rep_loc.values_mut() {
loc.0 -= 1;
}
}
let mut keep: BTreeSet<InitKey> = self
.periods
.iter()
.flat_map(|p| p.lanes.iter())
.flat_map(|l| l.reps.iter())
.map(|r| r.init_key.clone())
.collect();
if let Some(prev) = &self.prev {
keep.extend(prev.lanes.iter().map(|(_, _, k)| k.clone()));
}
self.inits.retain(|k, _| keep.contains(k));
}
fn ready_to_start(&self, pi: usize) -> bool {
!self.live_blocked()
&& self.periods[pi].lanes.iter().all(|lane| {
lane.current().is_none_or(|rep| {
self.inits
.get(&rep.init_key)
.is_some_and(|e| e.state == InitState::Ready)
})
})
}
fn start_period(&mut self, pi: usize) {
self.periods[pi].started = true;
let mut assignments: Vec<(usize, u32, u32)> = Vec::new();
let mut used_ranges: Vec<(u32, u32)> = Vec::new();
for li in 0..self.periods[pi].lanes.len() {
let n = self.periods[pi].lanes[li]
.current()
.and_then(|r| self.inits.get(&r.init_key))
.map_or(0, |e| u32::try_from(e.specs.len()).unwrap_or(u32::MAX));
let identity = self.periods[pi].lanes[li].identity.clone();
let base = allocate_track_base(
&mut self.track_ids,
&mut self.next_track_id,
&mut used_ranges,
identity,
n,
);
assignments.push((li, base, n));
}
for (li, base, n) in assignments {
let lane = &mut self.periods[pi].lanes[li];
lane.base = base;
lane.count = n;
}
let current: Vec<TrackIdentity> = self.periods[pi]
.lanes
.iter()
.map(|lane| lane.identity.clone())
.collect();
prune_track_ids(
&mut self.track_ids,
current.iter().chain(self.previous_identities.iter()),
);
self.previous_identities = current;
let notices = std::mem::take(&mut self.periods[pi].notices);
self.outputs.extend(notices);
let live: Vec<usize> = (0..self.periods[pi].lanes.len())
.filter(|&li| self.periods[pi].lanes[li].current().is_some())
.collect();
if live.is_empty() {
self.periods[pi].issue_done = true;
return;
}
let first = !self.any_started;
self.any_started = true;
let continuous = !first && self.is_continuous(pi);
let manifest = self.manifest(pi);
let p = &self.periods[pi];
self.outputs.push_back(Output::Manifest(manifest));
self.outputs.push_back(Output::PeriodChanged {
index: p.index,
id: p.id.clone(),
start_us: p.start_us,
continuous,
});
if !first && !continuous {
self.discontinuity();
}
let mut summary = Vec::new();
for li in live {
let lane = &mut self.periods[pi].lanes[li];
let ri = lane.reps.len() - 1;
let key = lane.reps[ri].init_key.clone();
summary.push((lane.kind, lane.set_id, key.clone()));
if !continuous && let Some(bytes) = self.inits.get(&key).and_then(|e| e.bytes.as_ref())
{
self.outputs.push_back(Output::Init(bytes.as_ref().clone()));
}
lane.announced_init = Some(key);
lane.announced_rep = Some(ri);
}
self.announce_tracks(pi);
self.prev = Some(PrevPeriod {
id: self.periods[pi].id.clone(),
lanes: summary,
});
}
fn is_continuous(&self, pi: usize) -> bool {
let Some(prev) = &self.prev else { return false };
let Some(prev_id) = prev.id.as_deref() else {
return false;
};
let Some(mpd) = self.mpd.as_ref() else {
return false;
};
let p = &self.periods[pi];
let live: Vec<&Lane> = p.lanes.iter().filter(|l| l.current().is_some()).collect();
if live.len() != prev.lanes.len() {
return false;
}
live.iter()
.zip(&prev.lanes)
.all(|(lane, (kind, set_id, key))| {
let Some(rep) = lane.current() else {
return false;
};
let Some((_, set, _)) = rep.at.resolve(mpd) else {
return false;
};
lane.kind == *kind
&& lane.set_id.is_some()
&& lane.set_id == *set_id
&& &rep.init_key == key
&& continuity_signals(set, prev_id)
})
}
fn announce_tracks(&mut self, pi: usize) {
let mut specs: Vec<TrackSpec> = Vec::new();
for lane in &self.periods[pi].lanes {
let Some(rep) = lane.current() else { continue };
let Some(entry) = self.inits.get(&rep.init_key) else {
continue;
};
for (pos, spec) in entry.specs.iter().enumerate() {
let pos = u32::try_from(pos).unwrap_or(u32::MAX);
if pos >= lane.count {
break;
}
let mut s = spec.clone();
s.track_id = TrackId::new(lane.base.saturating_add(pos));
specs.push(s);
}
}
let fingerprint = format!("{specs:?}");
if self.announced_tracks.as_deref() != Some(fingerprint.as_str()) {
self.announced_tracks = Some(fingerprint);
self.outputs.push_back(Output::Tracks(specs));
}
}
fn deliver_slots(&mut self, pi: usize) {
for li in 0..self.periods[pi].lanes.len() {
while self.periods[pi].lanes[li]
.queue
.front()
.is_some_and(|f| !matches!(f.state, SlotState::Issued | SlotState::Held(_)))
{
let Some(slot) = self.periods[pi].lanes[li].queue.pop_front() else {
break;
};
if !matches!(slot.state, SlotState::Marker) {
self.outstanding = self.outstanding.saturating_sub(1);
}
if let Some(id) = slot.id {
self.reqs.remove(&id);
}
match slot.state {
SlotState::Marker => self.discontinuity(),
SlotState::Skipped { marker } => {
if marker {
self.discontinuity();
}
}
SlotState::Staged(staged) => {
self.emit_staged(pi, li, slot.rep, slot.id, staged);
}
SlotState::Issued | SlotState::Held(_) => {}
}
}
}
}
fn emit_staged(
&mut self,
pi: usize,
li: usize,
rep_idx: usize,
id: Option<ResourceId>,
staged: Staged,
) {
if self.periods[pi].lanes[li].announced_rep != Some(rep_idx) {
let key = self.periods[pi].lanes[li].reps[rep_idx].init_key.clone();
if self.periods[pi].lanes[li].announced_init.as_ref() != Some(&key) {
if let Some(bytes) = self.inits.get(&key).and_then(|e| e.bytes.as_ref()) {
self.outputs.push_back(Output::Init(bytes.as_ref().clone()));
}
self.periods[pi].lanes[li].announced_init = Some(key);
}
self.periods[pi].lanes[li].announced_rep = Some(rep_idx);
self.announce_tracks(pi);
}
let lane = &self.periods[pi].lanes[li];
let key = &lane.reps[rep_idx].init_key;
let locals: Vec<u32> = self
.inits
.get(key)
.map(|e| e.specs.iter().map(|s| s.track_id.value()).collect())
.unwrap_or_default();
let (base, count) = (lane.base, lane.count);
let map = |local: u32| -> Option<TrackId> {
let pos = locals.iter().position(|l| *l == local)?;
let pos = u32::try_from(pos).ok()?;
(pos < count).then(|| TrackId::new(base.saturating_add(pos)))
};
if staged.damaged > 0
&& let Some(id) = id
{
self.outputs.push_back(Output::FragmentDamaged {
id,
skipped: staged.damaged,
});
}
for s in staged.skipped {
let k = (s.track_id.map(TrackId::value), s.fourcc.clone());
let first = self
.inits
.get_mut(&self.periods[pi].lanes[li].reps[rep_idx].init_key.clone())
.is_some_and(|e| e.skipped_reported.insert(k));
if first {
self.outputs.push_back(Output::TrackSkipped {
track_id: s.track_id.and_then(|t| map(t.value())),
fourcc: s.fourcc,
reason: s.reason,
});
}
}
for (local, samples) in staged.tracks {
if samples.is_empty() {
continue;
}
if let Some(track_id) = map(local) {
self.last_was_discontinuity = false;
self.outputs
.push_back(Output::Samples { track_id, samples });
}
}
}
fn init_url_of(&self, at: RepRef) -> std::result::Result<Option<String>, DropCause> {
let Some(mpd) = self.mpd.as_ref() else {
return Ok(None);
};
let planned = plan_representation(mpd, at, &PlanOptions::default());
let Some((reference, _)) = Self::init_reference(mpd, at, &planned)? else {
return Ok(None);
};
let Some((period, set, rep)) = at.resolve(mpd) else {
return Ok(None);
};
let cands = filter_candidates(
representation_candidates(mpd, period, set, rep),
self.doc_url.as_ref(),
self.dash.cross_origin_urls,
);
Ok(cands
.first()
.and_then(|c| c.resolve(self.doc_url.as_ref(), &reference)))
}
fn manifest(&self, pi: usize) -> ManifestSelection {
let mpd = self.mpd.as_ref().expect("mpd present");
let p = &self.periods[pi];
let period = &mpd.periods[p.index];
let audio_group = p
.selection
.chosen_audio
.and_then(|a| period.adaptation_sets.get(a.set))
.and_then(|s| s.group.or(s.id))
.map(|g| g.to_string());
let variants = p
.selection
.variants
.iter()
.filter_map(|&at| {
let (_, set, rep) = at.resolve(mpd)?;
let common = effective_common(set, rep);
let init_url = self.init_url_of(at).ok().flatten();
Some(VariantInfo {
rep: at,
init_url,
bandwidth: rep.bandwidth,
resolution: common.width.zip(common.height),
codecs: common.codecs,
audio_group: audio_group.clone(),
})
})
.collect();
let audio: Vec<AudioRendition> = p
.selection
.audio_sets
.iter()
.filter_map(|a| {
let set = period.adaptation_sets.get(a.set)?;
let first = a.reps.first()?;
let (_, _, first_rep) = first.resolve(mpd)?;
let chosen = p
.selection
.chosen_audio
.filter(|c| c.set == a.set)
.unwrap_or(*first);
Some(AudioRendition {
set: a.set,
group: set.group.or(set.id).map(|g| g.to_string()),
name: set
.labels
.first()
.map(|l| l.text.clone())
.or_else(|| set.lang.clone())
.unwrap_or_else(|| first_rep.id.clone()),
language: set.lang.clone(),
default: a.is_main,
autoselect: true,
init_url: self.init_url_of(chosen).ok().flatten(),
reps: a.reps.clone(),
})
})
.collect();
let chosen_audio = p
.selection
.chosen_audio
.and_then(|c| p.selection.audio_sets.iter().position(|a| a.set == c.set));
ManifestSelection {
period: p.index,
variants,
chosen: p.selection.chosen_variant,
audio,
chosen_audio,
}
}
pub fn select_representation(&mut self, index: usize) -> Result<()> {
self.ensure_live()?;
let pi = self.switchable_period()?;
let at = *self.periods[pi]
.selection
.variants
.get(index)
.ok_or(Error::NoSuchVariant)?;
self.switch_lane(pi, SetClass::Video, at)?;
self.periods[pi].selection.chosen_variant = Some(index);
Ok(())
}
pub fn select_audio(&mut self, index: Option<usize>) -> Result<()> {
self.ensure_live()?;
let pi = self.switchable_period()?;
let Some(index) = index else {
let lane = self.periods[pi]
.lanes
.iter_mut()
.find(|l| l.kind == SetClass::Audio)
.ok_or(Error::NotSwitchable)?;
if let Some(rep) = lane.reps.last_mut() {
rep.dead = true;
}
self.periods[pi].selection.chosen_audio = None;
return Ok(());
};
let at = self.best_audio_rep(pi, index)?;
self.switch_lane(pi, SetClass::Audio, at)?;
self.periods[pi].selection.chosen_audio = Some(at);
Ok(())
}
fn switchable_period(&self) -> Result<usize> {
let pi = self.deliver;
if pi != self.issue || self.periods.get(pi).is_none_or(|p| !p.started) {
return Err(Error::NotSwitchable);
}
Ok(pi)
}
fn best_audio_rep(&self, pi: usize, index: usize) -> Result<RepRef> {
let mpd = self.mpd.as_ref().expect("mpd present");
let set = self.periods[pi]
.selection
.audio_sets
.get(index)
.ok_or(Error::NoSuchAudio)?;
let bandwidth = |r: &RepRef| r.resolve(mpd).map_or(0, |(_, _, rep)| rep.bandwidth);
set.reps
.iter()
.max_by_key(|r| bandwidth(r))
.copied()
.ok_or(Error::NoSuchAudio)
}
fn switch_lane(&mut self, pi: usize, kind: SetClass, at: RepRef) -> Result<()> {
let revive = kind == SetClass::Audio;
let li = self.periods[pi]
.lanes
.iter()
.position(|l| l.kind == kind && (revive || l.current().is_some()))
.ok_or(Error::NotSwitchable)?;
let read = {
let lane = &self.periods[pi].lanes[li];
let cur = lane.reps.last().ok_or(Error::NotSwitchable)?;
let plan = cur.plan.as_ref().ok_or(Error::NotSwitchable)?;
let end = plan
.len()
.and_then(|n| n.checked_sub(1))
.map_or(Ok(None), |last| plan.get(last));
(cur.at, cur.dead, plan.get(cur.next_issue), end)
};
let (current_at, was_dead, current, end) = read;
let switch_us = match current {
Ok(Some(seg)) => seg.start_us,
found => {
let end = end.map_err(|_| Error::NotSwitchable)?;
let end = end.ok_or(Error::NotSwitchable)?;
if let Err(e) = found {
self.rep_segment_error(pi, li, address_name(&e));
return Err(Error::NotSwitchable);
}
end.start_us.saturating_add(end.duration_us)
}
};
if current_at == at && !was_dead {
return Ok(());
}
let aligned = {
let mpd = self.mpd.as_ref().expect("mpd present");
at.resolve(mpd).is_some_and(|(_, set, _)| {
set.segment_alignment
&& matches!(
set.start_with_sap,
Some(START_WITH_SAP_1 | START_WITH_SAP_2)
)
})
};
let mut rep = match self.build_rep(at) {
Ok(r) => r,
Err(cause) => {
self.drop_rep(pi, li, cause);
return Err(Error::NotSwitchable);
}
};
match rep.plan.as_ref() {
Some(plan) => {
rep.next_issue = match plan.index_at_or_before(switch_us) {
Ok(Some(i)) => i,
_ => 0,
};
rep.joined = true;
let last = rep
.next_issue
.checked_sub(1)
.map_or(Ok(None), |i| plan.get(i))
.map_err(|_| Error::NotSwitchable)?;
rep.last_time = last.map(|s| s.time_ticks);
}
None => rep.start_time_us = Some(switch_us),
}
if !aligned {
self.periods[pi].lanes[li].queue.push_back(Slot {
id: None,
rep: 0,
plan_index: 0,
state: SlotState::Marker,
});
}
let ri = self.periods[pi].lanes[li].reps.len();
self.periods[pi].lanes[li].reps.push(rep);
self.periods[pi].issue_done = false;
self.register_rep(pi, li, ri);
Ok(())
}
}
fn ranges_overlap(a_base: u32, a_len: u32, b_base: u32, b_len: u32) -> bool {
let a_end = a_base.saturating_add(a_len);
let b_end = b_base.saturating_add(b_len);
a_base < b_end && b_base < a_end
}
fn allocate_track_base(
track_ids: &mut BTreeMap<TrackIdentity, (u32, u32)>,
next_track_id: &mut u32,
used: &mut Vec<(u32, u32)>,
identity: TrackIdentity,
n: u32,
) -> u32 {
if n == 0 {
return 0;
}
if let Some((existing, len)) = track_ids.get(&identity).copied()
&& len >= n
&& !used.iter().any(|&(b, l)| ranges_overlap(existing, n, b, l))
{
used.push((existing, n));
return existing;
}
let base = *next_track_id;
*next_track_id = next_track_id.saturating_add(n);
track_ids.insert(identity, (base, n));
used.push((base, n));
base
}
fn prune_track_ids<'a>(
track_ids: &mut BTreeMap<TrackIdentity, (u32, u32)>,
keep: impl Iterator<Item = &'a TrackIdentity>,
) {
let keep: BTreeSet<&TrackIdentity> = keep.collect();
track_ids.retain(|identity, _| keep.contains(identity));
}
#[cfg(test)]
mod track_id_allocation_tests {
use super::*;
#[test]
fn two_same_identity_lanes_get_unique_bases() {
let mut track_ids: BTreeMap<TrackIdentity, (u32, u32)> = BTreeMap::new();
let mut next = 1u32;
let mut used = Vec::new();
let id = TrackIdentity {
kind: SetClass::Audio,
set_id: None,
group: None,
lang: Some("en".to_string()),
roles: Vec::new(),
};
let a = allocate_track_base(&mut track_ids, &mut next, &mut used, id.clone(), 1);
let b = allocate_track_base(&mut track_ids, &mut next, &mut used, id, 1);
assert_ne!(a, b, "duplicate base for two lanes with equal identity");
}
#[test]
fn growing_identity_does_not_overlap_neighbor() {
let mut track_ids: BTreeMap<TrackIdentity, (u32, u32)> = BTreeMap::new();
let mut next = 1u32;
let mut used = Vec::new();
let a = TrackIdentity {
kind: SetClass::Video,
set_id: None,
group: None,
lang: None,
roles: Vec::new(),
};
let b = TrackIdentity {
kind: SetClass::Audio,
set_id: None,
group: None,
lang: Some("en".to_string()),
roles: Vec::new(),
};
let a1 = allocate_track_base(&mut track_ids, &mut next, &mut used, a.clone(), 1);
let b1 = allocate_track_base(&mut track_ids, &mut next, &mut used, b.clone(), 1);
let mut used2 = Vec::new();
let a2 = allocate_track_base(&mut track_ids, &mut next, &mut used2, a.clone(), 2);
let mut ids = std::collections::BTreeSet::new();
ids.insert(a1);
ids.insert(b1);
ids.insert(a2);
ids.insert(a2 + 1);
assert_eq!(ids.len(), 4, "range overlap or duplicate ids");
}
fn video(set_id: u64) -> TrackIdentity {
TrackIdentity {
kind: SetClass::Video,
set_id: Some(set_id),
group: None,
lang: None,
roles: Vec::new(),
}
}
fn audio(set_id: u64) -> TrackIdentity {
TrackIdentity {
kind: SetClass::Audio,
set_id: Some(set_id),
group: None,
lang: None,
roles: Vec::new(),
}
}
#[test]
fn renumbered_periods_keep_the_identity_map_bounded_and_ids_unique_per_kind() {
let mut track_ids: BTreeMap<TrackIdentity, (u32, u32)> = BTreeMap::new();
let mut next = 1u32;
let mut owner: BTreeMap<u32, SetClass> = BTreeMap::new();
let mut previous: Vec<TrackIdentity> = Vec::new();
let mut peak = 0usize;
for period in 0..200u64 {
let current = vec![video(period), audio(period)];
let mut used = Vec::new();
for identity in ¤t {
let base =
allocate_track_base(&mut track_ids, &mut next, &mut used, identity.clone(), 1);
let kind = owner.entry(base).or_insert(identity.kind);
assert_eq!(*kind, identity.kind, "id {base} announced for two kinds");
}
prune_track_ids(&mut track_ids, current.iter().chain(previous.iter()));
peak = peak.max(track_ids.len());
previous = current;
}
assert!(peak <= 4, "identity map grew to {peak}");
assert_eq!(next, 401, "ids must stay monotonic, never recycled");
}
}