use std::collections::{HashMap, HashSet};
use crate::address::template::match_url;
use crate::mpd::{Mpd, SegmentTemplate};
use broadcast_common::impl_spec_display;
use broadcast_common::units::MICROS_PER_SEC;
use http::StatusCode;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum IngestMethod {
Post,
Put,
}
impl IngestMethod {
pub fn name(&self) -> &'static str {
match self {
IngestMethod::Post => "POST",
IngestMethod::Put => "PUT",
}
}
}
impl_spec_display!(IngestMethod);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ManifestFormat {
Dash,
Hls,
}
impl ManifestFormat {
pub fn name(&self) -> &'static str {
match self {
ManifestFormat::Dash => "dash",
ManifestFormat::Hls => "hls",
}
}
}
impl_spec_display!(ManifestFormat);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Addressing {
Number,
Time,
}
impl Addressing {
pub fn name(&self) -> &'static str {
match self {
Addressing::Number => "number",
Addressing::Time => "time",
}
}
}
impl_spec_display!(Addressing);
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ObjectKind {
Probe,
Manifest(ManifestFormat),
Header {
representation_id: String,
},
Segment {
representation_id: String,
number: Option<u64>,
time: Option<u64>,
},
}
impl ObjectKind {
pub fn name(&self) -> &'static str {
match self {
ObjectKind::Probe => "probe",
ObjectKind::Manifest(_) => "manifest",
ObjectKind::Header { .. } => "header",
ObjectKind::Segment { .. } => "segment",
}
}
}
impl_spec_display!(ObjectKind);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum SegmentDisposition {
Accepted,
Duplicate,
Regression,
}
impl SegmentDisposition {
pub fn name(&self) -> &'static str {
match self {
SegmentDisposition::Accepted => "accepted",
SegmentDisposition::Duplicate => "duplicate",
SegmentDisposition::Regression => "regression",
}
}
}
impl_spec_display!(SegmentDisposition);
#[derive(Debug, Clone, Copy)]
pub struct RequestMeta<'a> {
pub method: IngestMethod,
pub path: &'a str,
pub content_length: Option<u64>,
pub chunked: bool,
pub body_empty: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ManifestGrouping {
pub init_template: String,
pub media_template: String,
pub addressing: Addressing,
pub switching_sets: Vec<Vec<String>>,
pub availability_start_time: Option<String>,
pub period_start: Option<std::time::Duration>,
}
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
#[non_exhaustive]
pub enum IngestError {
#[error("segment without a usable header (§5.3.5c)")]
SegmentWithoutHeader,
#[error("receiver-side fault (§5.3.5d)")]
Internal,
#[error("malformed request (§5.3.5e)")]
BadRequest,
#[error("no manifest received yet")]
NoManifest,
#[error("MPD has multiple Period elements (§6.2.16h)")]
MpdMultiplePeriods,
#[error("MPD SegmentTemplate {attr} template mismatch (§6.2.16a/b/c)")]
MpdTemplateMismatch { attr: &'static str },
#[error("MPD contains a BaseURL element (§6.2.16d)")]
MpdBaseUrlPresent,
#[error("MPD AdaptationSet/Representation lacks a SegmentTemplate (§6.2.16 intro)")]
MpdMissingSegmentTemplate,
#[error("header changed mid-session (§6.2.19)")]
HeaderChanged,
#[error("segment already in progress for this representation (§5.3 catch-all)")]
SegmentInProgress,
}
impl IngestError {
pub fn status(&self) -> StatusCode {
match self {
IngestError::SegmentWithoutHeader => StatusCode::PRECONDITION_FAILED,
IngestError::Internal => StatusCode::INTERNAL_SERVER_ERROR,
IngestError::SegmentInProgress => StatusCode::CONFLICT,
_ => StatusCode::BAD_REQUEST,
}
}
}
impl ManifestGrouping {
pub fn from_mpd(bytes: &[u8]) -> Result<Self, IngestError> {
let mpd = Mpd::parse(bytes).map_err(|_| IngestError::BadRequest)?;
if mpd.periods.len() != 1 {
return Err(IngestError::MpdMultiplePeriods);
}
let period = &mpd.periods[0];
if !mpd.base_urls.is_empty()
|| !period.base_urls.is_empty()
|| mpd
.periods
.iter()
.flat_map(|p| &p.adaptation_sets)
.any(|a| {
!a.base_urls.is_empty()
|| a.representations.iter().any(|r| !r.base_urls.is_empty())
})
{
return Err(IngestError::MpdBaseUrlPresent);
}
let mut templates: Vec<(String, &SegmentTemplate)> = Vec::new();
let mut switching_sets: Vec<Vec<String>> = Vec::new();
for aset in &period.adaptation_sets {
let mut set_ids: Vec<String> = Vec::new();
for rep in &aset.representations {
let tmpl = rep
.segment
.template
.as_ref()
.or(aset.segment.template.as_ref());
let tmpl = match tmpl {
Some(t) => t,
None => return Err(IngestError::MpdMissingSegmentTemplate),
};
templates.push((rep.id.clone(), tmpl));
set_ids.push(rep.id.clone());
}
if !set_ids.is_empty() {
switching_sets.push(set_ids);
}
}
if templates.is_empty() {
return Err(IngestError::MpdMissingSegmentTemplate);
}
let first_init = templates[0].1.initialization.clone();
let first_media = templates[0].1.media.clone();
for (_, t) in &templates {
if t.initialization != first_init {
return Err(IngestError::MpdTemplateMismatch {
attr: "initialization",
});
}
if t.media != first_media {
return Err(IngestError::MpdTemplateMismatch { attr: "media" });
}
}
let init_template = first_init.unwrap_or_default();
let media_template = first_media.unwrap_or_default();
if init_template.matches("$RepresentationID$").count() != 1 {
return Err(IngestError::MpdTemplateMismatch {
attr: "initialization",
});
}
if media_template.matches("$RepresentationID$").count() != 1 {
return Err(IngestError::MpdTemplateMismatch { attr: "media" });
}
let number_count = count_ident(&media_template, "Number");
let time_count = count_ident(&media_template, "Time");
let has_number = number_count > 0;
let has_time = time_count > 0;
if !(has_number ^ has_time) {
return Err(IngestError::MpdTemplateMismatch { attr: "media" });
}
if has_number && number_count != 1 {
return Err(IngestError::MpdTemplateMismatch { attr: "media" });
}
if has_time && time_count != 1 {
return Err(IngestError::MpdTemplateMismatch { attr: "media" });
}
let addressing = if has_number {
Addressing::Number
} else {
Addressing::Time
};
Ok(ManifestGrouping {
init_template,
media_template,
addressing,
switching_sets,
availability_start_time: mpd
.availability_start_time_us
.and_then(|us| us.checked_div(MICROS_PER_SEC as i64))
.and_then(|s| jiff::Timestamp::from_second(s).ok())
.map(|t| t.to_string()),
period_start: period.start,
})
}
}
fn count_ident(template: &str, name: &str) -> usize {
let mut count = 0;
let mut rest = template;
while let Some(start) = rest.find('$') {
let tail = &rest[start + 1..];
if let Some(end) = tail.find('$') {
let ident = &tail[..end];
let id_name = ident.split('%').next().unwrap_or(ident);
if id_name == name {
count += 1;
}
rest = &tail[end + 1..];
} else {
break;
}
}
count
}
pub fn classify(
req: &RequestMeta<'_>,
grouping: Option<&ManifestGrouping>,
) -> Result<ObjectKind, IngestError> {
if req.body_empty {
return Ok(ObjectKind::Probe);
}
let lower = req.path.to_ascii_lowercase();
if lower.ends_with(".mpd") {
return Ok(ObjectKind::Manifest(ManifestFormat::Dash));
}
if lower.ends_with(".m3u8") {
return Ok(ObjectKind::Manifest(ManifestFormat::Hls));
}
let g = grouping.ok_or(IngestError::NoManifest)?;
let known: Vec<&str> = g
.switching_sets
.iter()
.flatten()
.map(|s| s.as_str())
.collect();
if let Some(m) = match_url(&g.init_template, req.path, &known) {
return Ok(ObjectKind::Header {
representation_id: m.representation_id,
});
}
if let Some(m) = match_url(&g.media_template, req.path, &known) {
return Ok(ObjectKind::Segment {
representation_id: m.representation_id,
number: m.number,
time: m.time,
});
}
Err(IngestError::BadRequest)
}
#[derive(Debug, Default)]
pub struct IngestState {
headers: HashSet<String>,
last_number: HashMap<String, u64>,
last_time: HashMap<String, u64>,
last_segments: HashSet<String>,
}
impl IngestState {
pub fn new() -> Self {
Self::default()
}
pub fn on_header(&mut self, id: &str) {
self.headers.insert(id.to_string());
self.last_number.remove(id);
self.last_time.remove(id);
self.last_segments.remove(id);
}
pub fn has_header(&self, id: &str) -> bool {
self.headers.contains(id)
}
pub fn on_segment(
&mut self,
id: &str,
number: Option<u64>,
time: Option<u64>,
) -> Result<SegmentDisposition, IngestError> {
if !self.headers.contains(id) {
return Err(IngestError::SegmentWithoutHeader);
}
let mut regression = false;
let mut duplicate = true;
let mut has_comparable = false;
let mut any = false;
if let Some(n) = number {
any = true;
if let Some(ln) = self.last_number.get(id) {
has_comparable = true;
if n < *ln {
regression = true;
}
if n != *ln {
duplicate = false;
}
} else {
duplicate = false;
}
}
if let Some(t) = time {
any = true;
if let Some(lt) = self.last_time.get(id) {
has_comparable = true;
if t < *lt {
regression = true;
}
if t != *lt {
duplicate = false;
}
} else {
duplicate = false;
}
}
if !any {
duplicate = false;
}
let disposition = if regression {
SegmentDisposition::Regression
} else if has_comparable && duplicate {
SegmentDisposition::Duplicate
} else {
SegmentDisposition::Accepted
};
if let Some(n) = number {
self.last_number.insert(id.to_string(), n);
}
if let Some(t) = time {
self.last_time.insert(id.to_string(), t);
}
Ok(disposition)
}
pub fn mark_last(&mut self, id: &str) {
self.last_segments.insert(id.to_string());
}
pub fn is_last(&self, id: &str) -> bool {
self.last_segments.contains(id)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn status_mapping_exhaustive() {
fn expected(e: &IngestError) -> u16 {
match e {
IngestError::SegmentWithoutHeader => 412,
IngestError::Internal => 500,
IngestError::BadRequest => 400,
IngestError::NoManifest => 400,
IngestError::MpdMultiplePeriods => 400,
IngestError::MpdTemplateMismatch { .. } => 400,
IngestError::MpdBaseUrlPresent => 400,
IngestError::MpdMissingSegmentTemplate => 400,
IngestError::HeaderChanged => 400,
IngestError::SegmentInProgress => 409,
}
}
let variants = [
IngestError::SegmentWithoutHeader,
IngestError::Internal,
IngestError::BadRequest,
IngestError::NoManifest,
IngestError::MpdMultiplePeriods,
IngestError::MpdTemplateMismatch { attr: "media" },
IngestError::MpdBaseUrlPresent,
IngestError::MpdMissingSegmentTemplate,
IngestError::HeaderChanged,
IngestError::SegmentInProgress,
];
for v in &variants {
assert_eq!(v.status().as_u16(), expected(v));
}
}
}
pub mod session;
pub use session::{
DEFAULT_BUFFERED_BUDGET_BYTES, DEFAULT_MAX_BUFFERED_OBJECT_BYTES, DEFAULT_MAX_OBJECT_BYTES,
DashifIngestConfig, DashifIngestSession, IngestObject, SegmentHandle, SessionError,
};