use std::collections::VecDeque;
use std::num::NonZeroUsize;
use std::sync::{Arc, Mutex};
use broadcast_common::Timestamp;
use broadcast_hls::{LowLatencyConfig, MediaPlaylist, MediaSegment, OpenSegment, PartSpec};
use bytes::Bytes;
use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
use media_plane::trunk::{PartEntry, SegmentCursor, SegmentCursorItem, SegmentEntry, Trunk};
pub const DEFAULT_TRACK_ID: u32 = 1;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Container {
Fmp4,
MpegTs,
}
impl Container {
pub fn name(&self) -> &'static str {
match self {
Container::Fmp4 => "fmp4",
Container::MpegTs => "mpeg-ts",
}
}
fn segment_extension(self) -> &'static str {
match self {
Container::Fmp4 => "m4s",
Container::MpegTs => "ts",
}
}
}
broadcast_common::impl_spec_display!(Container);
impl Default for Container {
fn default() -> Self {
Container::Fmp4
}
}
const PLACEHOLDER_BANDWIDTH_BPS: u64 = 5_000_000;
const ABUSE_MSN_FUTURE_BOUND: u64 = 2;
const PART_HOLD_BACK_MULTIPLIER: f64 = 3.0;
pub fn master_playlist_m3u8(media_playlist_name: &str) -> String {
format!(
"#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH={PLACEHOLDER_BANDWIDTH_BPS}\n{media_playlist_name}\n"
)
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct BlockingQuery {
pub hls_msn: Option<u64>,
pub hls_part: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HlsRequest {
Playlist {
track_id: u32,
query: BlockingQuery,
},
Resource {
name: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HlsBody {
Playlist(String),
Resource(Bytes),
}
struct WindowSegment {
sequence_number: u32,
bytes: Bytes,
duration_secs: f64,
discontinuous: bool,
}
struct Window {
segments: VecDeque<WindowSegment>,
capacity: usize,
max_segment_duration_secs: f64,
discontinuity_sequence: u64,
}
impl Window {
fn new(capacity: NonZeroUsize) -> Self {
Window {
segments: VecDeque::new(),
capacity: capacity.get(),
max_segment_duration_secs: 0.0,
discontinuity_sequence: 0,
}
}
fn push(&mut self, entry: SegmentEntry) {
let duration_secs = entry.duration.as_secs_f64();
self.max_segment_duration_secs = self.max_segment_duration_secs.max(duration_secs);
if self.segments.len() == self.capacity {
if let Some(evicted) = self.segments.pop_front() {
if evicted.discontinuous {
self.discontinuity_sequence += 1;
}
}
}
self.segments.push_back(WindowSegment {
sequence_number: entry.sequence_number,
bytes: entry.bytes,
duration_secs,
discontinuous: entry.meta.discontinuous,
});
}
fn bytes_of(&self, sequence_number: u32) -> Option<Bytes> {
self.segments
.iter()
.find(|s| s.sequence_number == sequence_number)
.map(|s| s.bytes.clone())
}
}
fn parse_part(file: &str, container: Container) -> Option<(u32, u32)> {
let suffix = format!(".{}", container.segment_extension());
let rest = file.strip_prefix("part-")?.strip_suffix(suffix.as_str())?;
let (track_seq, idx) = rest.rsplit_once('.')?;
let (track, seq) = track_seq.split_once('-')?;
track.parse::<u32>().ok()?;
Some((seq.parse().ok()?, idx.parse().ok()?))
}
enum ImmediateResource {
Init,
Segment(u32),
}
fn parse_immediate(file: &str, container: Container) -> Option<ImmediateResource> {
if container == Container::Fmp4 {
if let Some(rest) = file.strip_prefix("init-") {
let track = rest.strip_suffix(".mp4")?;
track.parse::<u32>().ok()?;
return Some(ImmediateResource::Init);
}
}
if let Some(rest) = file.strip_prefix("seg-") {
let suffix = format!(".{}", container.segment_extension());
let rest = rest.strip_suffix(suffix.as_str())?;
let (track, seq) = rest.split_once('-')?;
track.parse::<u32>().ok()?;
return Some(ImmediateResource::Segment(seq.parse().ok()?));
}
None
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum HlsOriginBuildError {
#[error("HlsOrigin::builder(...).target_duration_secs(...) is required but was never called")]
MissingTargetDurationSecs,
#[error("HlsOrigin::builder(...).window_segments(...) is required but was never called")]
MissingWindowSegments,
}
pub struct HlsOriginBuilder {
trunk: Arc<Trunk>,
target_duration_secs: Option<f64>,
window_segments: Option<NonZeroUsize>,
container: Container,
part_target_ms: Option<u32>,
}
impl HlsOriginBuilder {
fn new(trunk: Arc<Trunk>) -> Self {
HlsOriginBuilder {
trunk,
target_duration_secs: None,
window_segments: None,
container: Container::default(),
part_target_ms: None,
}
}
pub fn target_duration_secs(mut self, target_duration_secs: f64) -> Self {
self.target_duration_secs = Some(target_duration_secs);
self
}
pub fn window_segments(mut self, window_segments: NonZeroUsize) -> Self {
self.window_segments = Some(window_segments);
self
}
pub fn container(mut self, container: Container) -> Self {
self.container = container;
self
}
pub fn low_latency(mut self, part_target_ms: u32) -> Self {
self.part_target_ms = Some(part_target_ms);
self
}
pub fn build(self) -> Result<HlsOrigin, HlsOriginBuildError> {
let target_duration_secs = self
.target_duration_secs
.ok_or(HlsOriginBuildError::MissingTargetDurationSecs)?;
let window_segments = self
.window_segments
.ok_or(HlsOriginBuildError::MissingWindowSegments)?;
let cursor = self.trunk.subscribe_segments();
Ok(HlsOrigin {
trunk: self.trunk,
cursor: Mutex::new(cursor),
window: Mutex::new(Window::new(window_segments)),
init: Mutex::new(None),
target_duration_secs,
container: self.container,
part_target_ms: self.part_target_ms,
})
}
}
pub struct HlsOrigin {
trunk: Arc<Trunk>,
cursor: Mutex<SegmentCursor>,
window: Mutex<Window>,
init: Mutex<Option<Bytes>>,
target_duration_secs: f64,
container: Container,
part_target_ms: Option<u32>,
}
impl HlsOrigin {
pub fn builder(trunk: Arc<Trunk>) -> HlsOriginBuilder {
HlsOriginBuilder::new(trunk)
}
pub fn set_init(&self, bytes: impl Into<Bytes>) {
*self.init.lock().unwrap() = Some(bytes.into());
}
pub fn init_bytes(&self) -> Option<Bytes> {
self.init.lock().unwrap().clone()
}
fn drain(&self) {
let mut cursor = self.cursor.lock().unwrap();
let mut window = self.window.lock().unwrap();
while let Some(item) = cursor.poll() {
if let SegmentCursorItem::Segment(entry) = item {
window.push(entry);
}
}
}
fn live_edge(&self) -> (u32, Vec<PartEntry>) {
let last_closed = self.trunk.last_closed_segment().unwrap_or(0);
let candidate = last_closed + 1;
let parts = self.trunk.parts_in_segment(candidate);
if parts.is_empty() {
(last_closed, Vec::new())
} else {
(candidate, parts)
}
}
fn render_playlist(&self, track_id: u32) -> String {
self.drain();
let window = self.window.lock().unwrap();
let (open_seq, open_parts) = self.live_edge();
let low_latency_enabled = self.part_target_ms.is_some();
let has_open_parts = low_latency_enabled && !open_parts.is_empty();
let ext = self.container.segment_extension();
let media_sequence = window
.segments
.front()
.map(|s| u64::from(s.sequence_number))
.or_else(|| has_open_parts.then_some(u64::from(open_seq)))
.unwrap_or(1);
let segments: Vec<MediaSegment> = window
.segments
.iter()
.map(|s| MediaSegment {
uri: format!("seg-{track_id}-{}.{ext}", s.sequence_number),
duration: s.duration_secs,
discontinuous: s.discontinuous,
parts: Vec::new(),
..Default::default()
})
.collect();
let open_segment = has_open_parts.then(|| {
OpenSegment::new(
open_parts
.iter()
.map(|p| PartSpec {
uri: format!(
"part-{track_id}-{}.{}.{ext}",
p.segment_number, p.part_index
),
duration: p.duration.as_secs_f64(),
independent: p.independent,
..Default::default()
})
.collect(),
)
});
let next_part_hint = has_open_parts.then(|| {
let next_idx = open_parts
.iter()
.map(|p| p.part_index)
.max()
.map(|idx| idx + 1)
.unwrap_or(0);
format!("part-{track_id}-{open_seq}.{next_idx}.{ext}")
});
let target_duration = self
.target_duration_secs
.max(window.max_segment_duration_secs)
.round() as u32;
let extra_tags = match self.container {
Container::Fmp4 => vec![format!("#EXT-X-MAP:URI=\"init-{track_id}.mp4\"")],
Container::MpegTs => Vec::new(),
};
let low_latency = low_latency_enabled.then(|| {
let part_target_ms = self
.part_target_ms
.expect("low_latency_enabled implies Some");
let part_target = f64::from(part_target_ms) / 1000.0;
LowLatencyConfig {
part_target,
part_hold_back: part_target * PART_HOLD_BACK_MULTIPLIER,
preload_hint_part: next_part_hint,
..Default::default()
}
});
let playlist = MediaPlaylist {
target_duration,
media_sequence,
discontinuity_sequence: window.discontinuity_sequence,
segments,
open_segment,
endlist: false,
extra_tags,
low_latency,
iframes_only: false,
..Default::default()
};
playlist.to_m3u8()
}
fn resolve_playlist(
&self,
track_id: u32,
query: BlockingQuery,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
if query.hls_part.is_some() && query.hls_msn.is_none() {
return EgressResponse::BadRequest {
reason: "_HLS_part without _HLS_msn is meaningless",
};
}
if let Some(msn) = query.hls_msn {
let (in_progress_seg, live_parts) = self.live_edge();
if msn > u64::from(in_progress_seg) + ABUSE_MSN_FUTURE_BOUND {
return EgressResponse::BadRequest {
reason: "_HLS_msn unreasonably far beyond the live edge",
};
}
let satisfied = match query.hls_part {
Some(part) => {
u64::from(in_progress_seg) > msn
|| (u64::from(in_progress_seg) == msn
&& live_parts.len() as u64 > u64::from(part))
}
None => self.trunk.last_closed_segment().unwrap_or(0) as u64 >= msn,
};
if !satisfied {
return EgressResponse::pending(await_policy, now, now);
}
}
EgressResponse::Ready {
body: HlsBody::Playlist(self.render_playlist(track_id)),
cache: CachePolicy::NoCache,
}
}
fn resolve_resource(
&self,
name: &str,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
if let Some((seq, idx)) = parse_part(name, self.container) {
if let Some(bytes) = self.trunk.part_bytes(seq, idx) {
return EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache: CachePolicy::Immutable,
};
}
let never_will = self.trunk.last_closed_segment().is_some_and(|c| c >= seq);
return if never_will {
EgressResponse::NotFound
} else {
EgressResponse::pending(await_policy, now, now)
};
}
self.drain();
let bytes = match parse_immediate(name, self.container) {
Some(ImmediateResource::Init) => self.init_bytes(),
Some(ImmediateResource::Segment(seq)) => self.window.lock().unwrap().bytes_of(seq),
None => None,
};
match bytes {
Some(bytes) => EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache: CachePolicy::Immutable,
},
None => EgressResponse::NotFound,
}
}
}
impl ServedEgress for HlsOrigin {
type Request = HlsRequest;
type Body = HlsBody;
fn resolve(
&self,
request: HlsRequest,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
match request {
HlsRequest::Playlist { track_id, query } => {
self.resolve_playlist(track_id, query, now, await_policy)
}
HlsRequest::Resource { name } => self.resolve_resource(&name, now, await_policy),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use media_plane::trunk::TrunkConfig;
use std::time::{Duration, Instant};
use transmux::SegmentMeta;
fn nz(n: usize) -> NonZeroUsize {
NonZeroUsize::new(n).expect("test capacity must be non-zero")
}
fn make_origin() -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
let writer = trunk.segment_writer().expect("first segment writer");
let origin = HlsOrigin::builder(Arc::clone(&trunk))
.target_duration_secs(4.0)
.window_segments(nz(4))
.low_latency(500)
.build()
.expect("both required fields set");
origin.set_init(vec![0xAAu8; 8]);
(trunk, origin, writer)
}
fn seg(
writer: &media_plane::trunk::SegmentWriter,
seq: u32,
duration_secs: f64,
discontinuous: bool,
) {
writer.publish_segment(SegmentEntry::new(
Bytes::from(vec![seq as u8; 8]),
seq,
Duration::from_secs_f64(duration_secs),
Timestamp::from_nanos(0),
SegmentMeta { discontinuous },
));
}
fn part(writer: &media_plane::trunk::SegmentWriter, seg_no: u32, idx: u32, independent: bool) {
writer.publish_part(PartEntry::new(
Bytes::from(vec![idx as u8; 4]),
seg_no,
idx,
Duration::from_millis(500),
independent,
));
}
fn resolve_now(origin: &HlsOrigin, request: HlsRequest) -> EgressResponse<HlsBody> {
origin.resolve(
request,
Timestamp::from_nanos(0),
AwaitPolicy::new(Timestamp::from_nanos(0)),
)
}
#[test]
fn master_playlist_has_stream_inf() {
let m = master_playlist_m3u8("media.m3u8");
assert!(m.contains("#EXTM3U"));
assert!(m.contains("#EXT-X-STREAM-INF"));
assert!(m.contains("media.m3u8"));
}
#[test]
fn master_playlist_points_at_configured_playlist_name() {
let m = master_playlist_m3u8("index.m3u8");
assert!(m.contains("index.m3u8"));
assert!(!m.contains("media.m3u8"));
}
#[test]
fn playlist_rendered_from_populated_trunk_matches_expected_shape() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
part(&writer, 2, 0, true);
part(&writer, 2, 1, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
cache,
} => {
assert_eq!(cache, CachePolicy::NoCache);
m
}
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-VERSION:6"), "body: {body}");
assert!(!body.contains("#EXT-X-VERSION:9"), "body: {body}");
assert!(body.contains("#EXT-X-TARGETDURATION:4"), "body: {body}");
assert!(
body.contains("#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5"),
"body: {body}"
);
assert!(
body.contains("#EXT-X-PART-INF:PART-TARGET=0.5"),
"body: {body}"
);
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
"body: {body}"
);
assert!(body.contains("seg-1-1.m4s"), "body: {body}");
assert!(
body.contains("#EXT-X-PART:DURATION=0.5") && body.contains("INDEPENDENT=YES"),
"body: {body}"
);
assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(
body.contains("part-1-2.2.m4s"),
"preload hint for the next part: {body}"
);
}
#[test]
fn preload_hinted_part_blocks_until_produced_then_serves() {
let (trunk, origin, writer) = make_origin();
let origin = Arc::new(origin);
let deadline = Timestamp::from_nanos(5_000_000_000);
let policy = AwaitPolicy::new(deadline);
let first = origin.resolve(
HlsRequest::Resource {
name: "part-1-1.0.m4s".to_string(),
},
Timestamp::from_nanos(0),
policy,
);
assert!(
matches!(first, EgressResponse::Await { .. }),
"expected Await before the part exists, got {first:?}"
);
let listener = trunk.listen().expect("listener slot available");
let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let woken2 = std::sync::Arc::clone(&woken);
let waiter = std::thread::spawn(move || {
let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
});
part(&writer, 1, 0, true);
waiter.join().expect("waiter thread must not panic");
assert!(
woken.load(std::sync::atomic::Ordering::SeqCst),
"Trunk::listen() must wake once publish_part lands"
);
match origin.resolve(
HlsRequest::Resource {
name: "part-1-1.0.m4s".to_string(),
},
Timestamp::from_nanos(1),
policy,
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache,
} => {
assert_eq!(bytes, Bytes::from(vec![0u8; 4]));
assert_eq!(cache, CachePolicy::Immutable);
}
other => panic!("expected Ready once produced, got {other:?}"),
}
}
#[test]
fn awaiting_part_is_bounded_by_await_policy_deadline() {
let (_trunk, origin, _writer) = make_origin();
let deadline = Timestamp::from_nanos(1_000_000_000);
let policy = AwaitPolicy::new(deadline);
let still_waiting = origin.resolve(
HlsRequest::Resource {
name: "part-1-9.0.m4s".to_string(),
},
Timestamp::from_nanos(999_999_999),
policy,
);
assert!(matches!(still_waiting, EgressResponse::Await { .. }));
let expired = origin.resolve(
HlsRequest::Resource {
name: "part-1-9.0.m4s".to_string(),
},
deadline,
policy,
);
assert!(
matches!(expired, EgressResponse::NotFound),
"expected NotFound once the deadline passed, got {expired:?}"
);
}
#[test]
fn just_closed_segment_final_part_still_serves() {
let (_trunk, origin, writer) = make_origin();
part(&writer, 1, 0, true);
part(&writer, 1, 1, false); seg(&writer, 1, 4.0, false);
match resolve_now(
&origin,
HlsRequest::Resource {
name: "part-1-1.1.m4s".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![1u8; 4])),
other => panic!("the just-closed segment's final part must still serve, got {other:?}"),
}
assert_eq!(
resolve_now(
&origin,
HlsRequest::Resource {
name: "part-1-1.9.m4s".to_string(),
}
),
EgressResponse::NotFound
);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(
body.contains("seg-1-1.m4s"),
"closed segment rendered whole: {body}"
);
assert!(
!body.contains("part-1-1."),
"closed parts not rendered as open: {body}"
);
}
#[test]
fn media_sequence_and_discontinuity_sequence_advance_as_window_rolls() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
seg(&writer, 2, 4.0, true); seg(&writer, 3, 4.0, false);
seg(&writer, 4, 4.0, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:1"), "body: {body}");
assert!(
!body.contains("#EXT-X-DISCONTINUITY-SEQUENCE"),
"nothing has rolled off the window yet: {body}"
);
assert!(body.contains("#EXT-X-DISCONTINUITY\n"), "body: {body}");
seg(&writer, 5, 4.0, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:2"), "body: {body}");
assert!(
!body.contains("#EXT-X-DISCONTINUITY-SEQUENCE"),
"evicted segment 1 was not discontinuous: {body}"
);
seg(&writer, 6, 4.0, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(body.contains("#EXT-X-MEDIA-SEQUENCE:3"), "body: {body}");
assert!(
body.contains("#EXT-X-DISCONTINUITY-SEQUENCE:1"),
"segment 2 (discontinuous) has now rolled off the window: {body}"
);
}
#[test]
fn target_duration_is_max_of_configured_and_actual_segment_duration() {
let (_trunk, origin, writer) = make_origin(); seg(&writer, 1, 7.5, false);
let body = match resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
};
assert!(
body.contains("#EXT-X-TARGETDURATION:8"),
"TARGETDURATION must be round(7.5)=8, not the configured target: {body}"
);
}
#[test]
fn far_future_msn_rejected() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: Some(1002),
hls_part: None,
},
},
);
assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
}
#[test]
fn msn_at_spec_bound_is_accepted() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: Some(3),
hls_part: None,
},
},
);
assert!(
!matches!(outcome, EgressResponse::BadRequest { .. }),
"msn at spec bound (last_closed+2) must be accepted, not rejected"
);
}
#[test]
fn msn_one_beyond_spec_bound_is_rejected() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: Some(4),
hls_part: None,
},
},
);
assert!(
matches!(outcome, EgressResponse::BadRequest { .. }),
"msn at spec bound + 1 (last_closed+3) must be rejected"
);
}
#[test]
fn part_without_msn_rejected() {
let (_trunk, origin, _writer) = make_origin();
let outcome = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery {
hls_msn: None,
hls_part: Some(0),
},
},
);
assert!(matches!(outcome, EgressResponse::BadRequest { .. }));
}
#[test]
fn resolve_resource_init_present() {
let (_trunk, origin, _writer) = make_origin();
match resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache,
} => {
assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8]));
assert_eq!(cache, CachePolicy::Immutable);
}
other => panic!("expected Ready, got {other:?}"),
}
}
#[test]
fn resolve_resource_unmatched_filename_not_found() {
let (_trunk, origin, _writer) = make_origin();
assert_eq!(
resolve_now(
&origin,
HlsRequest::Resource {
name: "not-a-thing.txt".to_string(),
}
),
EgressResponse::NotFound
);
}
fn make_origin_with(
container: Container,
low_latency_ms: Option<u32>,
) -> (Arc<Trunk>, HlsOrigin, media_plane::trunk::SegmentWriter) {
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
let writer = trunk.segment_writer().expect("first segment writer");
let mut builder = HlsOrigin::builder(Arc::clone(&trunk))
.target_duration_secs(4.0)
.window_segments(nz(4))
.container(container);
if let Some(ms) = low_latency_ms {
builder = builder.low_latency(ms);
}
let origin = builder.build().expect("both required fields set");
(trunk, origin, writer)
}
fn render_body(origin: &HlsOrigin) -> String {
match resolve_now(
origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
) {
EgressResponse::Ready {
body: HlsBody::Playlist(m),
..
} => m,
other => panic!("expected Ready(Playlist), got {other:?}"),
}
}
fn extract_version_tag(body: &str) -> Option<u8> {
body.lines()
.find_map(|l| l.strip_prefix("#EXT-X-VERSION:")?.parse::<u8>().ok())
}
fn segment_uris(body: &str) -> Vec<String> {
let lines: Vec<&str> = body.lines().collect();
let mut out = Vec::new();
for i in 0..lines.len() {
if lines[i].starts_with("#EXTINF:") {
if let Some(next) = lines.get(i + 1) {
if !next.starts_with('#') {
out.push((*next).to_string());
}
}
}
}
out
}
fn part_uris(body: &str) -> Vec<String> {
body.lines()
.filter(|l| l.starts_with("#EXT-X-PART:"))
.filter_map(|l| {
let start = l.find("URI=\"")? + "URI=\"".len();
let rest = &l[start..];
let end = rest.find('"')?;
Some(rest[..end].to_string())
})
.collect()
}
fn assert_version_matches_broadcast_hls_derivation(body: &str) -> Option<u8> {
let parsed = MediaPlaylist::parse(body).expect("rendered body must round-trip parse");
let rendered = extract_version_tag(body);
assert_eq!(
rendered,
parsed.computed_version(),
"rendered #EXT-X-VERSION must equal broadcast_hls's own derivation, body: {body}"
);
rendered
}
fn assert_version_present_and_matches_derivation(body: &str) -> u8 {
assert_version_matches_broadcast_hls_derivation(body).unwrap_or_else(|| {
panic!(
"this cell must trigger a real RFC 8216bis §8 version rule -- a \
missing #EXT-X-VERSION makes the derivation check vacuous, body: {body}"
)
})
}
#[test]
fn mpegts_classic_no_map_ts_uris_no_ll_tags() {
let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
seg(&writer, 1, 4.004, false);
let body = render_body(&origin);
assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
assert!(!body.contains("#EXT-X-PART"), "body: {body}");
assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(!body.contains(".m4s"), "body: {body}");
assert!(body.contains("seg-1-1.ts"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
let uris = segment_uris(&body);
assert_eq!(uris, vec!["seg-1-1.ts".to_string()]);
for uri in uris {
let ImmediateResource::Segment(seq) = parse_immediate(&uri, Container::MpegTs)
.expect("advertised segment URI must parse under MpegTs")
else {
panic!("expected a Segment resource for {uri}");
};
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
}
#[test]
fn mpegts_low_latency_part_ts_uris_blocking_part_requests_resolve() {
let (trunk, origin, writer) = make_origin_with(Container::MpegTs, Some(500));
let origin = Arc::new(origin);
seg(&writer, 1, 4.004, false);
part(&writer, 2, 0, true);
part(&writer, 2, 1, false);
let body = render_body(&origin);
assert!(!body.contains("#EXT-X-MAP"), "body: {body}");
assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
assert!(body.contains("#EXT-X-PART:"), "body: {body}");
assert!(body.contains("seg-1-1.ts"), "body: {body}");
assert!(body.contains("part-1-2.0.ts"), "body: {body}");
assert!(!body.contains(".m4s"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
let parts = part_uris(&body);
assert!(!parts.is_empty(), "body: {body}");
for uri in &parts {
let (seq, idx) = parse_part(uri, Container::MpegTs)
.unwrap_or_else(|| panic!("advertised part URI {uri} must parse under MpegTs"));
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => {
assert_eq!(bytes, Bytes::from(vec![idx as u8; 4]));
assert_eq!(seq, 2);
}
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
let deadline = Timestamp::from_nanos(5_000_000_000);
let policy = AwaitPolicy::new(deadline);
let pending = origin.resolve(
HlsRequest::Resource {
name: "part-1-2.2.ts".to_string(),
},
Timestamp::from_nanos(0),
policy,
);
assert!(
matches!(pending, EgressResponse::Await { .. }),
"expected Await before the part exists, got {pending:?}"
);
let listener = trunk.listen().expect("listener slot available");
let woken = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let woken2 = std::sync::Arc::clone(&woken);
let waiter = std::thread::spawn(move || {
let ok = listener.wait_deadline(Instant::now() + Duration::from_secs(60));
woken2.store(ok, std::sync::atomic::Ordering::SeqCst);
});
part(&writer, 2, 2, false);
waiter.join().expect("waiter thread must not panic");
assert!(
woken.load(std::sync::atomic::Ordering::SeqCst),
"Trunk::listen() must wake once publish_part lands"
);
match origin.resolve(
HlsRequest::Resource {
name: "part-1-2.2.ts".to_string(),
},
Timestamp::from_nanos(1),
policy,
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![2u8; 4])),
other => panic!("expected Ready once produced, got {other:?}"),
}
}
#[test]
fn mpegts_classic_integral_extinf_emits_no_version_tag() {
let (_trunk, origin, writer) = make_origin_with(Container::MpegTs, None);
seg(&writer, 1, 4.0, false);
let body = render_body(&origin);
assert!(body.contains("#EXTINF:4,"), "body: {body}");
assert!(!body.contains("4.000"), "body: {body}");
assert_eq!(
assert_version_matches_broadcast_hls_derivation(&body),
None,
"nothing in this playlist triggers an RFC 8216bis §8 row: {body}"
);
}
#[test]
fn low_latency_does_not_raise_the_derived_version() {
let (_t1, classic, w1) = make_origin_with(Container::MpegTs, None);
seg(&w1, 1, 4.004, false);
let classic_version = assert_version_present_and_matches_derivation(&render_body(&classic));
let (_t2, low_latency, w2) = make_origin_with(Container::MpegTs, Some(500));
seg(&w2, 1, 4.004, false);
part(&w2, 2, 0, true);
let ll_body = render_body(&low_latency);
assert!(ll_body.contains("#EXT-X-PART:"), "body: {ll_body}");
assert_eq!(
assert_version_present_and_matches_derivation(&ll_body),
classic_version,
"enabling low latency must not raise the derived version"
);
}
#[test]
fn fmp4_classic_map_present_no_ll_tags() {
let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, None);
origin.set_init(vec![0xBBu8; 8]);
seg(&writer, 1, 4.004, false);
let body = render_body(&origin);
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
"body: {body}"
);
assert!(!body.contains("#EXT-X-PART"), "body: {body}");
assert!(!body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
assert!(!body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(body.contains("seg-1-1.m4s"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
match resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![0xBBu8; 8])),
other => panic!("expected Ready(init), got {other:?}"),
}
for uri in segment_uris(&body) {
let ImmediateResource::Segment(seq) = parse_immediate(&uri, Container::Fmp4)
.expect("advertised segment URI must parse under Fmp4")
else {
panic!("expected a Segment resource for {uri}");
};
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
}
#[test]
fn fmp4_low_latency_existing_behaviour_preserved() {
let (_trunk, origin, writer) = make_origin_with(Container::Fmp4, Some(500));
origin.set_init(vec![0xAAu8; 8]);
seg(&writer, 1, 4.004, false);
part(&writer, 2, 0, true);
part(&writer, 2, 1, false);
let body = render_body(&origin);
assert!(
body.contains("#EXT-X-MAP:URI=\"init-1.mp4\""),
"body: {body}"
);
assert!(body.contains("#EXT-X-PART-INF"), "body: {body}");
assert!(body.contains("#EXT-X-SERVER-CONTROL"), "body: {body}");
assert!(body.contains("#EXT-X-PRELOAD-HINT"), "body: {body}");
assert!(body.contains("seg-1-1.m4s"), "body: {body}");
assert!(body.contains("part-1-2.0.m4s"), "body: {body}");
assert!(!body.contains(".ts\""), "body: {body}");
assert_version_present_and_matches_derivation(&body);
match resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
},
) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![0xAAu8; 8])),
other => panic!("expected Ready(init), got {other:?}"),
}
for uri in segment_uris(&body) {
let ImmediateResource::Segment(seq) = parse_immediate(&uri, Container::Fmp4)
.expect("advertised segment URI must parse under Fmp4")
else {
panic!("expected a Segment resource for {uri}");
};
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![seq as u8; 8])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
for uri in part_uris(&body) {
let (_seq, idx) = parse_part(&uri, Container::Fmp4)
.unwrap_or_else(|| panic!("advertised part URI {uri} must parse under Fmp4"));
match resolve_now(&origin, HlsRequest::Resource { name: uri.clone() }) {
EgressResponse::Ready {
body: HlsBody::Resource(bytes),
..
} => assert_eq!(bytes, Bytes::from(vec![idx as u8; 4])),
other => panic!("expected Ready for {uri}, got {other:?}"),
}
}
}
#[test]
fn cross_container_refusal_mp4_init_under_mpegts_not_found() {
let (_trunk, origin, _writer) = make_origin_with(Container::MpegTs, None);
origin.set_init(vec![0xCCu8; 8]);
assert_eq!(
resolve_now(
&origin,
HlsRequest::Resource {
name: "init-1.mp4".to_string(),
}
),
EgressResponse::NotFound
);
}
#[test]
fn builder_errors_on_missing_required_fields() {
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
match HlsOrigin::builder(Arc::clone(&trunk))
.window_segments(nz(4))
.build()
{
Err(e) => assert_eq!(e, HlsOriginBuildError::MissingTargetDurationSecs),
Ok(_) => panic!("expected an error: target_duration_secs was never set"),
}
match HlsOrigin::builder(Arc::clone(&trunk))
.target_duration_secs(4.0)
.build()
{
Err(e) => assert_eq!(e, HlsOriginBuildError::MissingWindowSegments),
Ok(_) => panic!("expected an error: window_segments was never set"),
}
}
#[test]
fn container_label_and_display() {
assert_eq!(Container::Fmp4.name(), "fmp4");
assert_eq!(Container::MpegTs.name(), "mpeg-ts");
assert_eq!(Container::Fmp4.to_string(), "fmp4");
assert_eq!(Container::default(), Container::Fmp4);
}
}