use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use axum::Router;
use axum::extract::State;
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use broadcast_common::Package;
use hls_runtime::server::DEFAULT_TRACK_ID;
use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
use transmux::{Addressing, DashPackager, Media, Track, TrackSegments, TrackSpec};
use crate::http::{self, BLOCKING_RELOAD_TIMEOUT};
use crate::origin::resource::cors_preflight;
use crate::output::{Output, OutputKind};
use crate::route::RouteHandle;
pub(crate) const DASH_MANIFEST_CONTENT_TYPE: &str = "application/dash+xml";
pub struct DashOutput;
impl Output for DashOutput {
fn kind(&self) -> OutputKind {
OutputKind::Dash
}
fn manifest_routes(&self, route: Arc<RouteHandle>) -> Router {
Router::new()
.route("/manifest.mpd", get(manifest).options(cors_preflight))
.with_state(route)
}
}
async fn manifest(State(route): State<Arc<RouteHandle>>) -> Response {
let serving = match http::resolve_route_program(&route) {
Ok(serving) => serving,
Err(resp) => return *resp,
};
let trunk = serving.trunk();
let origin = DashOrigin { route };
let resp = http::resolve_blocking(&trunk, &origin, (), BLOCKING_RELOAD_TIMEOUT, || ()).await;
http::into_response(resp, StatusCode::SERVICE_UNAVAILABLE, |body| {
([(header::CONTENT_TYPE, DASH_MANIFEST_CONTENT_TYPE)], body).into_response()
})
}
struct DashOrigin {
route: Arc<RouteHandle>,
}
impl ServedEgress for DashOrigin {
type Request = ();
type Body = String;
fn resolve(
&self,
_request: (),
_now: broadcast_common::Timestamp,
_await_policy: AwaitPolicy,
) -> EgressResponse<String> {
match render_mpd(&self.route) {
Some(body) => EgressResponse::Ready {
body,
cache: CachePolicy::NoCache,
},
None => EgressResponse::NotFound,
}
}
}
pub(crate) fn select_representable_track(specs: &[TrackSpec]) -> Option<TrackSpec> {
specs
.iter()
.filter(|s| is_video_like(&s.config))
.find(|s| track_is_representable(s))
.or_else(|| specs.iter().find(|s| track_is_representable(s)))
.cloned()
}
fn is_video_like(config: &transmux::CodecConfig) -> bool {
matches!(
config,
transmux::CodecConfig::Avc { .. }
| transmux::CodecConfig::Hevc { .. }
| transmux::CodecConfig::Vvc { .. }
| transmux::CodecConfig::Av1 { .. }
| transmux::CodecConfig::Vp9 { .. }
| transmux::CodecConfig::Vp8 { .. }
| transmux::CodecConfig::Mpeg2Video { .. }
)
}
fn track_is_representable(spec: &TrackSpec) -> bool {
let media = Media::new(
vec![Track::new(spec.clone(), Vec::new())],
spec.timescale.max(1),
);
DashPackager::default().package(&media).is_ok()
}
fn render_mpd(route: &RouteHandle) -> Option<String> {
let specs = route.track_specs(crate::route::SPTS_PROGRAM_ID);
let mut spec = select_representable_track(&specs)?;
spec.track_id = DEFAULT_TRACK_ID;
let timescale = spec.timescale.max(1);
let window = route.window_segments(crate::route::SPTS_PROGRAM_ID);
let start_number = window
.first()
.map(|s| u64::from(s.segment_seq))
.unwrap_or(1);
let duration_ticks: Vec<u64> = window
.iter()
.map(|s| (s.duration_secs * f64::from(timescale)).round() as u64)
.collect();
let segments = if duration_ticks.is_empty() {
Vec::new()
} else {
vec![TrackSegments {
track_id: DEFAULT_TRACK_ID,
durations: duration_ticks,
}]
};
let target_duration_secs = route.target_duration_secs();
let time_shift_buffer_depth_secs = target_duration_secs * (window.len().max(1) as f64);
let media = Media::new(vec![Track::new(spec, Vec::new())], timescale);
let mut packager = DashPackager {
dynamic: true,
addressing: Addressing::Number,
start_number,
init_template: "init-$RepresentationID$.mp4".to_string(),
media_template: "seg-$RepresentationID$-$Number$.m4s".to_string(),
availability_start_time: Some(format_iso8601(route.created_at())),
minimum_update_period: Some(format!("PT{target_duration_secs}S")),
time_shift_buffer_depth: Some(format!("PT{time_shift_buffer_depth_secs}S")),
segments,
..DashPackager::default()
};
packager.package(&media).ok()
}
pub(crate) fn format_iso8601(t: SystemTime) -> String {
let secs = t
.duration_since(UNIX_EPOCH)
.unwrap_or(Duration::ZERO)
.as_secs();
let days = (secs / 86_400) as i64;
let time_of_day = secs % 86_400;
let (h, m, s) = (
time_of_day / 3600,
(time_of_day / 60) % 60,
time_of_day % 60,
);
let (y, mo, d) = civil_from_days(days);
format!("{y:04}-{mo:02}-{d:02}T{h:02}:{m:02}:{s:02}Z")
}
fn civil_from_days(days: i64) -> (i64, u32, u32) {
let z = days + 719_468;
let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
let doe = (z - era * 146_097) as u64; let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; let y = yoe as i64 + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); let mp = (5 * doy + 2) / 153; let d = (doy - (153 * mp + 2) / 5 + 1) as u32; let m = (if mp < 10 { mp + 3 } else { mp - 9 }) as u32; (y + i64::from(m <= 2), m, d)
}
#[cfg(test)]
mod tests {
use super::*;
use transmux::CodecConfig;
use transmux::ll_hls::SegmentInfo;
fn video_spec(track_id: u32) -> TrackSpec {
TrackSpec::new(
track_id,
90_000,
CodecConfig::Vp8 {
width: 1280,
height: 720,
},
)
}
fn teletext_spec(track_id: u32) -> TrackSpec {
TrackSpec::new(
track_id,
90_000,
CodecConfig::Data {
stream_type: 0x06,
descriptors: Vec::new(),
carriage: transmux::ir::DataCarriage::Pes,
},
)
}
fn seg(seq: u32, duration: f64) -> SegmentInfo {
SegmentInfo {
bytes: vec![seq as u8; 8],
duration,
segment_seq: seq,
part_count: 1,
}
}
#[test]
fn civil_from_days_matches_known_dates() {
assert_eq!(civil_from_days(0), (1970, 1, 1));
let days_2024_01_01 = 19_723;
assert_eq!(civil_from_days(days_2024_01_01), (2024, 1, 1));
}
#[test]
fn format_iso8601_renders_utc_z_suffix() {
let t = UNIX_EPOCH + Duration::from_secs(0);
assert_eq!(format_iso8601(t), "1970-01-01T00:00:00Z");
}
#[test]
fn render_mpd_none_without_track_specs() {
let route = RouteHandle::new(4.0, 500, 4);
assert!(
render_mpd(&route).is_none(),
"no track specs recorded yet -> nothing to describe"
);
}
#[test]
fn render_mpd_valid_before_any_segment_closes() {
let route = RouteHandle::new(4.0, 500, 4);
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(crate::route::SPTS_PROGRAM_ID, vec![video_spec(7)]);
let mpd = render_mpd(&route).expect("must render even with an empty window");
assert!(mpd.contains("<MPD"));
assert!(mpd.contains("type=\"dynamic\""));
}
#[test]
fn render_mpd_forces_representation_id_to_default_track() {
let route = RouteHandle::new(4.0, 500, 4);
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(crate::route::SPTS_PROGRAM_ID, vec![video_spec(7)]);
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(1, 4.0));
let mpd = render_mpd(&route).unwrap();
assert!(
mpd.contains(&format!("id=\"{DEFAULT_TRACK_ID}\"")),
"Representation @id must be the DEFAULT_TRACK_ID, not the source's own \
track_id (7): {mpd}"
);
assert!(
!mpd.contains("id=\"7\""),
"source track_id must not leak into the MPD: {mpd}"
);
}
#[test]
fn render_mpd_number_addressing_and_start_number_track_window() {
let route = RouteHandle::new(4.0, 500, 2);
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(crate::route::SPTS_PROGRAM_ID, vec![video_spec(1)]);
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(1, 4.0));
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(2, 4.0));
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(3, 4.0));
let mpd = render_mpd(&route).unwrap();
assert!(
mpd.contains("startNumber=\"2\""),
"startNumber must track the window's oldest retained segment_seq (2, \
since seq 1 was evicted): {mpd}"
);
assert!(
mpd.contains("$Number$"),
"media template must use literal $Number$ substitution: {mpd}"
);
assert!(
!mpd.contains("$Time$"),
"must not use $Time$ addressing -- store filenames are seq-numbered, \
not time-addressed: {mpd}"
);
assert!(mpd.contains("seg-$RepresentationID$-$Number$.m4s"));
assert!(mpd.contains("init-$RepresentationID$.mp4"));
}
#[test]
fn render_mpd_carries_live_attributes() {
let route = RouteHandle::new(2.0, 500, 4);
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(crate::route::SPTS_PROGRAM_ID, vec![video_spec(1)]);
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(1, 2.0));
let mpd = render_mpd(&route).unwrap();
assert!(mpd.contains("availabilityStartTime="), "{mpd}");
assert!(mpd.contains("minimumUpdatePeriod=\"PT2S\""), "{mpd}");
assert!(mpd.contains("timeShiftBufferDepth=\"PT2S\""), "{mpd}");
}
#[test]
fn select_representable_track_skips_leading_opaque_track() {
let specs = vec![teletext_spec(1), video_spec(2)];
let selected =
select_representable_track(&specs).expect("the video track must be selected");
assert_eq!(selected.track_id, 2);
}
#[test]
fn render_mpd_skips_leading_opaque_track_instead_of_503ing() {
let route = RouteHandle::new(4.0, 500, 4);
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(
crate::route::SPTS_PROGRAM_ID,
vec![teletext_spec(1), video_spec(2)],
);
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(1, 4.0));
let mpd = render_mpd(&route)
.expect("a representable track behind an opaque one must still render");
assert!(
mpd.contains(&format!("id=\"{DEFAULT_TRACK_ID}\"")),
"the selected (video) track's @id must still be forced to DEFAULT_TRACK_ID: {mpd}"
);
}
#[test]
fn render_mpd_none_when_every_track_is_opaque() {
let route = RouteHandle::new(4.0, 500, 4);
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(
crate::route::SPTS_PROGRAM_ID,
vec![teletext_spec(1), teletext_spec(2)],
);
assert!(
render_mpd(&route).is_none(),
"a track set with no representable track is still a genuine 503"
);
}
#[tokio::test]
async fn manifest_handler_503_before_track_specs_known() {
let route = Arc::new(RouteHandle::new(4.0, 500, 4));
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
let resp = manifest(State(route)).await;
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn manifest_handler_200_with_dash_content_type() {
let route = Arc::new(RouteHandle::new(4.0, 500, 4));
route.publish_new_program(crate::route::SPTS_PROGRAM_ID);
route.set_track_specs(crate::route::SPTS_PROGRAM_ID, vec![video_spec(1)]);
route.add_segment(crate::route::SPTS_PROGRAM_ID, seg(1, 4.0));
let resp = manifest(State(route)).await;
assert_eq!(resp.status(), StatusCode::OK);
assert_eq!(
resp.headers().get(header::CONTENT_TYPE).unwrap(),
DASH_MANIFEST_CONTENT_TYPE
);
}
#[tokio::test]
async fn manifest_not_yet_announced_is_503_not_404() {
let route = Arc::new(RouteHandle::new(4.0, 500, 4));
let resp = manifest(State(route)).await;
assert_eq!(
resp.status(),
StatusCode::SERVICE_UNAVAILABLE,
"a route with no program announced yet must be 503 (not ready), not 404 (gone)"
);
}
}