use super::ids::Generation;
use media_plane::ids::{PartIndex, SegmentSeq};
use std::collections::{BTreeMap, VecDeque};
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
use std::time::Duration;
use transmux::TrackId;
use broadcast_common::{Parse, Timestamp, Unpackage};
use broadcast_hls::{
ByteRange, ClosedCaptionsRef, DecimalSeconds, FrameRate, LowLatencyConfig, MapTag,
MasterPlaylist, MediaPlaylist, MediaRendition, MediaSegment, MediaType, OpenSegment, PartSpec,
Variant, VideoRange,
};
use bytes::Bytes;
use media_plane::egress::{AwaitPolicy, CachePolicy, EgressResponse, ServedEgress};
use media_plane::trunk::{PartEntry, SegmentCursor, SegmentCursorItem, SegmentEntry, Trunk};
const UNDRAINED: u64 = u64::MAX;
static LAST_INSTANCE: AtomicU64 = AtomicU64::new(0);
fn next_instance_token() -> u64 {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|d| u64::try_from(d.as_millis()).ok())
.unwrap_or(0);
let mut previous = LAST_INSTANCE.load(Ordering::Relaxed);
loop {
let token = now_ms.max(previous.saturating_add(1));
match LAST_INSTANCE.compare_exchange_weak(
previous,
token,
Ordering::AcqRel,
Ordering::Relaxed,
) {
Ok(_) => return token,
Err(actual) => previous = actual,
}
}
}
fn locked<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
m.lock().unwrap_or_else(PoisonError::into_inner)
}
pub const DEFAULT_TRACK_ID: TrackId = TrackId::new(1);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum Container {
Fmp4,
MpegTs,
WebVtt,
}
impl Container {
pub fn name(&self) -> &'static str {
match self {
Container::Fmp4 => "fmp4",
Container::MpegTs => "mpeg-ts",
Container::WebVtt => "webvtt",
}
}
fn segment_extension(self) -> &'static str {
match self {
Container::Fmp4 => "m4s",
Container::MpegTs => "ts",
Container::WebVtt => "vtt",
}
}
}
broadcast_common::impl_spec_display!(Container);
impl Default for Container {
fn default() -> Self {
Container::Fmp4
}
}
const INIT_HISTORY: usize = 8;
const NANOS_PER_SEC: u128 = 1_000_000_000;
const MIN_WINDOW_SEGMENTS: usize = 3;
const PART_RETENTION_TARGET_DURATIONS: f64 = 3.0;
const FUTURE_PART_SEGMENT_BOUND: u32 = 2;
const ADVANCE_PART_LIMIT_NUMERATOR: u64 = 3;
const MILLIS_PER_SEC: u64 = 1_000;
const ABUSE_MSN_FUTURE_BOUND: u64 = 2;
const PART_HOLD_BACK_MULTIPLIER: f64 = 3.0;
const HOLD_BACK_MULTIPLIER: f64 = 3.0;
const SKIP_BOUNDARY_MULTIPLIER: f64 = 6.0;
pub fn master_playlist_m3u8(media_playlist_name: &str, bandwidth_bps: u32) -> String {
format!("#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH={bandwidth_bps}\n{media_playlist_name}\n")
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum SkipDirective {
Yes,
V2,
}
impl SkipDirective {
pub fn name(&self) -> &'static str {
match self {
SkipDirective::Yes => "YES",
SkipDirective::V2 => "v2",
}
}
}
broadcast_common::impl_spec_display!(SkipDirective);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HttpPriority {
pub urgency: u8,
pub incremental: bool,
}
impl HttpPriority {
pub const PLAYLIST: Self = Self {
urgency: 1,
incremental: false,
};
pub const SEGMENT: Self = Self {
urgency: 2,
incremental: true,
};
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct BlockingQuery {
pub hls_msn: Option<u64>,
pub hls_part: Option<u32>,
pub hls_skip: Option<SkipDirective>,
pub hls_primary: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HlsRequest {
Playlist {
track_id: TrackId,
query: BlockingQuery,
},
Resource {
name: String,
},
IFramePlaylist {
track_id: TrackId,
query: BlockingQuery,
media_playlist_name: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HlsBody {
Playlist(String),
Resource(Bytes),
}
#[derive(Clone)]
struct WindowSegment {
sequence_number: SegmentSeq,
init_gen: Generation,
bytes: Bytes,
duration_secs: f64,
discontinuous: bool,
iframe_len: Option<u64>,
start_ns: u64,
video_millihertz: Option<u32>,
}
#[derive(Debug, Clone, Copy, PartialEq)]
#[non_exhaustive]
pub struct ClosedSegment {
pub sequence_number: SegmentSeq,
pub start_ns: u64,
pub duration_secs: f64,
pub discontinuous: bool,
pub init_generation: Generation,
}
impl ClosedSegment {
#[must_use]
pub fn with_init_generation(mut self, generation: Generation) -> Self {
self.init_generation = generation;
self
}
pub fn new(
sequence_number: SegmentSeq,
start_ns: u64,
duration_secs: f64,
discontinuous: bool,
) -> Self {
ClosedSegment {
sequence_number,
start_ns,
duration_secs,
discontinuous,
init_generation: Generation::new(1),
}
}
}
struct Window {
segments: VecDeque<WindowSegment>,
capacity: usize,
max_segment_duration_secs: f64,
peak_bandwidth_bps: u64,
total_bytes: u64,
total_duration_nanos: u128,
iframe_peak_bandwidth_bps: u64,
iframe_total_bytes: u64,
iframe_total_duration_nanos: u128,
discontinuity_sequence: u64,
gap_pending: bool,
}
impl Window {
fn new(capacity: NonZeroUsize) -> Self {
Window {
segments: VecDeque::new(),
capacity: capacity.get().max(MIN_WINDOW_SEGMENTS),
max_segment_duration_secs: 0.0,
peak_bandwidth_bps: 0,
total_bytes: 0,
total_duration_nanos: 0,
iframe_peak_bandwidth_bps: 0,
iframe_total_bytes: 0,
iframe_total_duration_nanos: 0,
discontinuity_sequence: 0,
gap_pending: false,
}
}
fn note_gap(&mut self) {
self.gap_pending = true;
}
fn max_video_millihertz(&self) -> Option<u32> {
self.segments
.iter()
.filter_map(|segment| segment.video_millihertz)
.max()
}
fn clear_rolling_off(&mut self) {
let discontinuities = self.segments.iter().filter(|s| s.discontinuous).count();
self.discontinuity_sequence = self
.discontinuity_sequence
.saturating_add(u64::try_from(discontinuities).unwrap_or(u64::MAX));
self.segments.clear();
}
fn push(
&mut self,
entry: SegmentEntry,
init_gen: Generation,
iframe_len: Option<u64>,
video_millihertz: Option<u32>,
) {
let follows = self
.segments
.back()
.is_none_or(|last| last.sequence_number.checked_next() == Some(entry.sequence_number));
let mut discontinuous = entry.meta.discontinuous;
if !follows || self.gap_pending {
self.clear_rolling_off();
discontinuous = true;
}
self.gap_pending = false;
if self
.segments
.back()
.is_some_and(|last| last.init_gen != init_gen)
{
discontinuous = true;
}
let duration_secs = entry.duration.as_secs_f64();
self.max_segment_duration_secs = self.max_segment_duration_secs.max(duration_secs);
if let Some(bps) = segment_bandwidth_bps(entry.bytes.len(), entry.duration) {
self.peak_bandwidth_bps = self.peak_bandwidth_bps.max(bps);
}
if !entry.duration.is_zero() {
self.total_bytes = self
.total_bytes
.saturating_add(u64::try_from(entry.bytes.len()).unwrap_or(u64::MAX));
self.total_duration_nanos = self
.total_duration_nanos
.saturating_add(entry.duration.as_nanos());
}
if let Some(len) = iframe_len {
if let Some(bps) =
segment_bandwidth_bps(usize::try_from(len).unwrap_or(usize::MAX), entry.duration)
{
self.iframe_peak_bandwidth_bps = self.iframe_peak_bandwidth_bps.max(bps);
}
if !entry.duration.is_zero() {
self.iframe_total_bytes = self.iframe_total_bytes.saturating_add(len);
self.iframe_total_duration_nanos = self
.iframe_total_duration_nanos
.saturating_add(entry.duration.as_nanos());
}
}
if self.segments.len() >= self.capacity
&& let Some(evicted) = self.segments.pop_front()
&& evicted.discontinuous
{
self.discontinuity_sequence = self.discontinuity_sequence.saturating_add(1);
}
self.segments.push_back(WindowSegment {
sequence_number: entry.sequence_number,
init_gen,
bytes: entry.bytes,
duration_secs,
discontinuous,
start_ns: entry.timeline_position.as_nanos(),
iframe_len,
video_millihertz,
});
}
fn bytes_of(&self, sequence_number: SegmentSeq) -> Option<Bytes> {
let front = self.segments.front()?.sequence_number;
let index = usize::try_from(sequence_number.checked_distance_from(front)?).ok()?;
let segment = self.segments.get(index)?;
(segment.sequence_number == sequence_number).then(|| segment.bytes.clone())
}
}
struct WindowView {
segments: VecDeque<WindowSegment>,
max_segment_duration_secs: f64,
discontinuity_sequence: u64,
}
#[derive(Default)]
struct InitStore {
generation: Generation,
current: Option<Bytes>,
history: VecDeque<(Generation, Bytes)>,
starts: VecDeque<(SegmentSeq, Generation)>,
}
impl InitStore {
fn set(&mut self, bytes: Bytes, next_seq: SegmentSeq) {
if self.current.as_ref() == Some(&bytes) {
return;
}
self.generation = Generation::new(self.generation.value().saturating_add(1));
let first_seq = if self.current.is_none() {
SegmentSeq::new(0)
} else {
next_seq
};
self.current = Some(bytes.clone());
self.history.push_back((self.generation, bytes));
self.starts.push_back((first_seq, self.generation));
while self.history.len() > INIT_HISTORY {
self.history.pop_front();
}
while self.starts.len() > INIT_HISTORY {
self.starts.pop_front();
}
}
fn generation_for(&self, seq: SegmentSeq) -> Generation {
self.starts
.iter()
.rev()
.find(|(first, _)| *first <= seq)
.or_else(|| self.starts.front())
.map_or(Generation::new(1), |&(_, generation)| generation)
}
fn bytes_of(&self, generation: Generation) -> Option<Bytes> {
self.history
.iter()
.find(|(g, _)| *g == generation)
.map(|(_, b)| b.clone())
}
}
const DV_BRAND_HDR10: &str = "db1p";
const DV_BRAND_HLG: &str = "db4h";
const HDR10PLUS_BRAND: &str = "cdm4";
const TRANSFER_PQ: u8 = 16;
const MILLIHERTZ_PER_HZ: u128 = 1000;
const MIN_SAMPLES_FOR_RATE: u64 = 2;
const MAX_PLAUSIBLE_FRAME_RATE_MILLIHERTZ: u32 = 300 * 1000;
const SUPPLEMENTAL_CODECS_ATTR: &str = "SUPPLEMENTAL-CODECS";
struct DolbyVisionHls {
codecs: String,
supplemental_codecs: Option<String>,
video_range: VideoRange,
}
fn dolby_vision_hls(spec: &transmux::TrackSpec) -> Option<DolbyVisionHls> {
use transmux::DoviBlSignalCompatibilityId as Ccid;
let record = spec.dovi.as_ref()?;
if !record.is_single_layer() || !record.is_hevc_profile() {
return None;
}
let dv = record.codec_string()?;
match record.bl_signal_compatibility_id {
Ccid::None => Some(DolbyVisionHls {
codecs: dv,
supplemental_codecs: None,
video_range: VideoRange::Pq,
}),
Ccid::Hdr10 => Some(DolbyVisionHls {
codecs: transmux::rfc6381_codec_string(&spec.config).ok()?,
supplemental_codecs: Some(format!("{dv}/{DV_BRAND_HDR10}")),
video_range: VideoRange::Pq,
}),
Ccid::Hlg => Some(DolbyVisionHls {
codecs: transmux::rfc6381_codec_string(&spec.config).ok()?,
supplemental_codecs: Some(format!("{dv}/{DV_BRAND_HLG}")),
video_range: VideoRange::Hlg,
}),
_ => None,
}
}
fn hdr10plus_supplemental_codecs(spec: &transmux::TrackSpec) -> Option<String> {
if !spec.hdr10plus || !matches!(spec.config, transmux::CodecConfig::Hevc { .. }) {
return None;
}
let transfer = spec
.display
.as_ref()?
.video_signal
.as_ref()?
.colour
.as_ref()?
.transfer
.code();
if transfer != TRANSFER_PQ {
return None;
}
let codec = transmux::rfc6381_codec_string(&spec.config).ok()?;
Some(format!("{codec}/{HDR10PLUS_BRAND}"))
}
fn is_video_codec(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 dedupe_rendition_names(mut media: Vec<MediaRendition>) -> Vec<MediaRendition> {
let mut seen: std::collections::BTreeSet<(String, String)> = std::collections::BTreeSet::new();
for m in &mut media {
let base = m.name.clone();
let mut n = 1u32;
while !seen.insert((m.group_id.clone(), m.name.clone())) {
n += 1;
m.name = format!("{base} ({n})");
}
}
media
}
fn channel_count_string(config: &transmux::CodecConfig) -> Option<String> {
let n = match config {
transmux::CodecConfig::Aac { channel_count, .. }
| transmux::CodecConfig::Ac3 { channel_count, .. }
| transmux::CodecConfig::Eac3 { channel_count, .. }
| transmux::CodecConfig::Opus { channel_count, .. }
| transmux::CodecConfig::Flac { channel_count, .. }
| transmux::CodecConfig::Ac4 { channel_count, .. }
| transmux::CodecConfig::MpegH { channel_count, .. }
| transmux::CodecConfig::MpegAudio { channel_count, .. } => *channel_count,
_ => return None,
};
Some(n.to_string())
}
fn first_moof(segment: &[u8]) -> Option<(usize, transmux::movie_fragment::MovieFragmentBox)> {
let mut offset = 0usize;
let mut moof_off: Option<usize> = None;
let mut moof_total: usize = 0;
let mut moof_hdr: usize = 0;
while offset + transmux::box_types::BOX_HEADER_MIN_SIZE <= segment.len() {
let rest = &segment[offset..];
let header = transmux::box_types::BoxHeader::parse(rest).ok()?;
let hdr_sz = header.header_size();
if hdr_sz > rest.len() {
break;
}
let total = if header.size == u64::from(transmux::box_types::SIZE_TO_EOF) {
rest.len()
} else {
match usize::try_from(header.size) {
Ok(t) if t >= hdr_sz => t,
_ => break,
}
};
if header.box_type.is(b"moof") {
moof_off = Some(offset);
moof_total = total.min(rest.len());
moof_hdr = hdr_sz;
break;
}
let step = total.min(rest.len());
if step == 0 {
break;
}
offset += step;
}
let moof_off = moof_off?;
let body = &segment[moof_off + moof_hdr..moof_off + moof_total];
let moof = transmux::movie_fragment::MovieFragmentBox::parse_body(body).ok()?;
Some((moof_off, moof))
}
fn segment_frame_rate_millihertz(samples: u64, ticks: u64, timescale: u32) -> Option<u32> {
if samples < MIN_SAMPLES_FOR_RATE || ticks == 0 || timescale == 0 {
return None;
}
let ticks = u128::from(ticks);
let millihertz =
(u128::from(samples) * u128::from(timescale) * MILLIHERTZ_PER_HZ + ticks / 2) / ticks;
u32::try_from(millihertz)
.ok()
.filter(|mh| *mh <= MAX_PLAUSIBLE_FRAME_RATE_MILLIHERTZ)
}
fn fmp4_video_rate(segment: &[u8], video_track_id: TrackId, timescale: u32) -> Option<u32> {
let (moof_off, moof) = first_moof(segment)?;
let mdats = transmux::frag_offsets::mdat_ranges(segment);
let samples =
transmux::frag_offsets::sample_ranges_in(segment, &mdats, moof_off, &moof, video_track_id)
.ok()?;
let ticks: u64 = samples.iter().map(|r| u64::from(r.duration)).sum();
segment_frame_rate_millihertz(u64::try_from(samples.len()).ok()?, ticks, timescale)
}
fn ts_video_rate(segment: &[u8]) -> Option<u32> {
let media = transmux::ts_demux::TsDemux::new().demux(segment).ok()?;
let track = media
.tracks
.iter()
.find(|track| is_video_codec(&track.spec.config))?;
let ticks: u64 = track
.samples
.iter()
.filter_map(|sample| sample.duration)
.map(u64::from)
.sum();
segment_frame_rate_millihertz(
u64::try_from(track.samples.len()).ok()?,
ticks,
track.spec.timescale,
)
}
fn compute_iframe_len(segment: &[u8], video_track_id: TrackId) -> Option<u64> {
let (moof_off, moof) = first_moof(segment)?;
let mdats = transmux::frag_offsets::mdat_ranges(segment);
let samples =
transmux::frag_offsets::sample_ranges_in(segment, &mdats, moof_off, &moof, video_track_id)
.ok()?;
if samples.is_empty() {
return None;
}
let first = samples[0];
if first.flags & (1 << 16) != 0 {
return None;
}
let mdat_payload_start = mdat_payload_start_for(segment, first.start)?;
if first.start != mdat_payload_start {
return None;
}
u64::try_from(first.end).ok()
}
fn mdat_payload_start_for(segment: &[u8], sample_start: usize) -> Option<usize> {
let mut offset = 0usize;
while offset + transmux::box_types::BOX_HEADER_MIN_SIZE <= segment.len() {
let rest = &segment[offset..];
let header = transmux::box_types::BoxHeader::parse(rest).ok()?;
let hdr_sz = header.header_size();
if hdr_sz > rest.len() {
break;
}
let total = if header.size == u64::from(transmux::box_types::SIZE_TO_EOF) {
rest.len()
} else {
match usize::try_from(header.size) {
Ok(t) if t >= hdr_sz => t,
_ => break,
}
};
if header.box_type.is(b"mdat") {
let payload = offset + hdr_sz;
let end = offset + total;
if sample_start >= payload && sample_start < end {
return Some(payload);
}
}
let step = total.min(rest.len());
if step == 0 {
break;
}
offset += step;
}
None
}
fn average_bandwidth_bps(total_bytes: u64, total_duration_nanos: u128) -> Option<u32> {
if total_duration_nanos == 0 {
return None;
}
let bits_nanos = u128::from(total_bytes)
.checked_mul(8)?
.checked_mul(NANOS_PER_SEC)?;
let mean = bits_nanos.checked_add(total_duration_nanos / 2)? / total_duration_nanos;
u32::try_from(mean).ok()
}
fn segment_bandwidth_bps(len: usize, duration: Duration) -> Option<u64> {
let nanos = duration.as_nanos();
if nanos == 0 {
return None;
}
let bits = u128::try_from(len).ok()?.checked_mul(8)?;
let bps = bits.checked_mul(NANOS_PER_SEC)?.div_ceil(nanos);
u64::try_from(bps).ok()
}
fn parse_part(file: &str, container: Container) -> Option<(Option<u64>, u64, PartIndex)> {
let suffix = format!(".{}", container.segment_extension());
let rest = file.strip_prefix("part-")?.strip_suffix(suffix.as_str())?;
let (names, idx) = rest.rsplit_once('.')?;
let (epoch, seq) = parse_epoch_seq(names)?;
Some((epoch, seq, PartIndex::new(idx.parse().ok()?)))
}
fn parse_epoch_seq(names: &str) -> Option<(Option<u64>, u64)> {
let mut fields = names.split('-');
fields.next()?.parse::<u32>().ok()?;
let first = fields.next()?;
match fields.next() {
None => Some((None, first.parse().ok()?)),
Some(seq) => {
if fields.next().is_some() {
return None;
}
Some((Some(first.parse().ok()?), seq.parse().ok()?))
}
}
}
pub const IFRAME_SEGMENT_PREFIX: &str = "iframe-seg-";
enum ImmediateResource {
CurrentInit,
VersionedInit { epoch: u64, generation: Generation },
Segment { epoch: Option<u64>, seq: u64 },
}
fn parse_immediate(file: &str, container: Container) -> Option<ImmediateResource> {
if container == Container::Fmp4
&& let Some(rest) = file.strip_prefix("init-")
{
let rest = rest.strip_suffix(".mp4")?;
let mut fields = rest.split('-');
fields.next()?.parse::<u32>().ok()?;
return match (fields.next(), fields.next(), fields.next()) {
(None, ..) => Some(ImmediateResource::CurrentInit),
(Some(epoch), Some(generation), None) => Some(ImmediateResource::VersionedInit {
epoch: epoch.parse().ok()?,
generation: Generation::new(generation.parse().ok()?),
}),
_ => None,
};
}
let seg_rest = file
.strip_prefix("seg-")
.or_else(|| file.strip_prefix(IFRAME_SEGMENT_PREFIX));
if let Some(rest) = seg_rest {
let suffix = format!(".{}", container.segment_extension());
let (epoch, seq) = parse_epoch_seq(rest.strip_suffix(suffix.as_str())?)?;
return Some(ImmediateResource::Segment { epoch, seq });
}
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,
#[error("HlsOrigin::builder(...).target_duration_secs(...) must be finite and positive")]
InvalidTargetDuration,
#[error("HlsOrigin::builder(...).low_latency(0): the part target must be positive")]
ZeroPartTarget,
#[error(
"HlsOrigin::builder(...).media_sequence_offset(...) leaves no room for u32 sequence numbers"
)]
MediaSequenceOffsetTooLarge,
#[error("HlsOrigin::builder(...).declared_bandwidth_bps(...) must be > 0 and <= u32::MAX")]
InvalidDeclaredBandwidth,
#[error("HlsOrigin::builder(...).audio_renditions(...): group_id must be non-empty")]
EmptyAudioRenditionGroupId,
#[error("HlsOrigin::builder(...).audio_renditions(...): name must be non-empty")]
EmptyAudioRenditionName,
#[error("HlsOrigin::builder(...).audio_renditions(...): duplicate NAME in a group")]
DuplicateAudioRenditionName,
#[error(
"HlsOrigin::builder(...).audio_renditions(...): a group must have exactly one DEFAULT=YES member"
)]
AudioRenditionDefaultCount,
#[error("HlsOrigin::builder(...).audio_renditions(...): DEFAULT=YES requires AUTOSELECT=YES")]
AudioRenditionDefaultNotAutoselect,
#[error("HlsOrigin::builder(...).audio_renditions(...): language must be non-empty")]
EmptyAudioRenditionLanguage,
#[error("HlsOrigin::builder(...).audio_renditions(...): uri must be non-empty")]
EmptyAudioRenditionUri,
#[error(
"HlsOrigin::builder(...).audio_renditions(...): every group must have the same members, differing only in URI and CHANNELS"
)]
AudioRenditionGroupsDiffer,
#[error("HlsOrigin::builder(...).subtitle_renditions(...): name must be non-empty")]
EmptySubtitleRenditionName,
#[error("HlsOrigin::builder(...).subtitle_renditions(...): language must be non-empty")]
EmptySubtitleRenditionLanguage,
#[error("HlsOrigin::builder(...).subtitle_renditions(...): uri must be non-empty")]
EmptySubtitleRenditionUri,
#[error("HlsOrigin::builder(...).subtitle_renditions(...): duplicate NAME")]
DuplicateSubtitleRenditionName,
#[error("HlsOrigin::builder(...).subtitle_renditions(...): duplicate URI")]
DuplicateSubtitleRenditionUri,
#[error(
"HlsOrigin::builder(...).subtitle_renditions(...): DEFAULT=YES requires AUTOSELECT=YES"
)]
SubtitleRenditionDefaultNotAutoselect,
#[error("HlsOrigin::builder(...).subtitle_renditions(...): at most one DEFAULT=YES member")]
MultipleDefaultSubtitleRenditions,
#[error(
"HlsOrigin::builder(...).subtitle_sibling_origins(...): one origin per declared subtitle rendition"
)]
SubtitleSiblingCountMismatch,
#[error(
"HlsOrigin::builder(...).subtitle_sibling_origins(...): sibling must be a WebVtt origin"
)]
SubtitleSiblingNotWebVtt,
#[error(
"HlsOrigin::builder(...).subtitle_sibling_origins(...): sibling target duration must equal this origin's"
)]
SubtitleSiblingTargetDurationMismatch,
#[error("HlsOrigin::builder(...).closed_caption_overrides(...): invalid INSTREAM-ID")]
InvalidClosedCaptionInstreamId,
#[error("HlsOrigin::builder(...).closed_caption_overrides(...): duplicate INSTREAM-ID")]
DuplicateClosedCaptionInstreamId,
}
#[derive(Debug, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum HlsMasterError {
#[error("BANDWIDTH unknown: no measured segment and no declared bandwidth")]
BandwidthUnknown,
#[error("measured BANDWIDTH {bandwidth} exceeds u32")]
BandwidthOverflow {
bandwidth: u64,
},
#[error("master playlist render failed: {0}")]
Render(broadcast_hls::Error),
#[error("audio rendition uri must not equal the media playlist name")]
AudioRenditionUriEqualsMediaPlaylist,
}
pub fn validate_audio_renditions(decls: &[AudioRenditionDecl]) -> Result<(), HlsOriginBuildError> {
use std::collections::{BTreeMap, BTreeSet};
let mut groups: BTreeMap<&str, Vec<&AudioRenditionDecl>> = BTreeMap::new();
for decl in decls {
if decl.group_id.is_empty() {
return Err(HlsOriginBuildError::EmptyAudioRenditionGroupId);
}
if decl.name.is_empty() {
return Err(HlsOriginBuildError::EmptyAudioRenditionName);
}
if decl.language.is_empty() {
return Err(HlsOriginBuildError::EmptyAudioRenditionLanguage);
}
if decl.uri.is_empty() {
return Err(HlsOriginBuildError::EmptyAudioRenditionUri);
}
if decl.default && !decl.autoselect {
return Err(HlsOriginBuildError::AudioRenditionDefaultNotAutoselect);
}
groups.entry(&decl.group_id).or_default().push(decl);
}
let mut signatures = BTreeSet::new();
for members in groups.values() {
let mut names = BTreeSet::new();
if !members.iter().all(|m| names.insert(m.name.as_str())) {
return Err(HlsOriginBuildError::DuplicateAudioRenditionName);
}
if members.iter().filter(|m| m.default).count() != 1 {
return Err(HlsOriginBuildError::AudioRenditionDefaultCount);
}
let mut signature: Vec<_> = members
.iter()
.map(|m| {
(
m.name.as_str(),
m.language.as_str(),
m.default,
m.autoselect,
m.characteristics.as_slice(),
)
})
.collect();
signature.sort();
signatures.insert(signature);
}
if signatures.len() > 1 {
return Err(HlsOriginBuildError::AudioRenditionGroupsDiffer);
}
Ok(())
}
pub fn validate_audio_renditions_relaxed(
decls: &[AudioRenditionDecl],
) -> Result<(), HlsOriginBuildError> {
use std::collections::BTreeMap;
for decl in decls {
if decl.group_id.is_empty() {
return Err(HlsOriginBuildError::EmptyAudioRenditionGroupId);
}
if decl.uri.is_empty() {
return Err(HlsOriginBuildError::EmptyAudioRenditionUri);
}
if decl.default && !decl.autoselect {
return Err(HlsOriginBuildError::AudioRenditionDefaultNotAutoselect);
}
}
let mut groups: BTreeMap<&str, Vec<&AudioRenditionDecl>> = BTreeMap::new();
for decl in decls {
groups.entry(&decl.group_id).or_default().push(decl);
}
for members in groups.values() {
if members.iter().filter(|m| m.default).count() != 1 {
return Err(HlsOriginBuildError::AudioRenditionDefaultCount);
}
}
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct AudioRenditionDecl {
pub group_id: String,
pub name: String,
pub language: String,
pub default: bool,
pub autoselect: bool,
pub channels: Option<String>,
pub uri: String,
pub characteristics: Vec<String>,
}
pub const SUBTITLES_GROUP_ID: &str = "subs";
const HEARING_IMPAIRED_CHARACTERISTICS: [&str; 2] = [
"public.accessibility.transcribes-spoken-dialog",
"public.accessibility.describes-music-and-sound",
];
const WEBVTT_CODEC: &str = "wvtt";
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct SubtitleRenditionDecl {
pub name: String,
pub language: String,
pub uri: String,
pub default: bool,
pub autoselect: bool,
pub forced: bool,
pub hearing_impaired: bool,
}
impl SubtitleRenditionDecl {
pub fn new(
name: impl Into<String>,
language: impl Into<String>,
uri: impl Into<String>,
) -> Self {
SubtitleRenditionDecl {
name: name.into(),
language: language.into(),
uri: uri.into(),
default: false,
autoselect: true,
forced: false,
hearing_impaired: false,
}
}
fn effective_autoselect(&self) -> bool {
self.autoselect || self.forced || self.hearing_impaired
}
}
pub fn validate_subtitle_renditions(
decls: &[SubtitleRenditionDecl],
) -> Result<(), HlsOriginBuildError> {
use std::collections::BTreeSet;
let mut names = BTreeSet::new();
let mut uris = BTreeSet::new();
let mut defaults = 0usize;
for decl in decls {
if decl.name.is_empty() {
return Err(HlsOriginBuildError::EmptySubtitleRenditionName);
}
if decl.language.is_empty() {
return Err(HlsOriginBuildError::EmptySubtitleRenditionLanguage);
}
if decl.uri.is_empty() {
return Err(HlsOriginBuildError::EmptySubtitleRenditionUri);
}
if decl.default && !decl.effective_autoselect() {
return Err(HlsOriginBuildError::SubtitleRenditionDefaultNotAutoselect);
}
if !names.insert(decl.name.as_str()) {
return Err(HlsOriginBuildError::DuplicateSubtitleRenditionName);
}
if !uris.insert(decl.uri.as_str()) {
return Err(HlsOriginBuildError::DuplicateSubtitleRenditionUri);
}
defaults += usize::from(decl.default);
}
if defaults > 1 {
return Err(HlsOriginBuildError::MultipleDefaultSubtitleRenditions);
}
Ok(())
}
pub const CLOSED_CAPTIONS_GROUP_ID: &str = "cc";
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct ClosedCaptionDecl {
pub instream_id: String,
pub name: String,
pub language: String,
}
pub fn validate_closed_caption_overrides(
decls: &[ClosedCaptionDecl],
) -> Result<(), HlsOriginBuildError> {
let mut seen: Vec<broadcast_hls::InstreamId> = Vec::new();
for decl in decls {
let id = broadcast_hls::InstreamId::parse_closed_captions(&decl.instream_id)
.map_err(|_| HlsOriginBuildError::InvalidClosedCaptionInstreamId)?;
if seen.contains(&id) {
return Err(HlsOriginBuildError::DuplicateClosedCaptionInstreamId);
}
seen.push(id);
}
Ok(())
}
#[derive(Clone)]
pub struct RenditionSibling {
pub origin: Weak<HlsOrigin>,
pub playlist_uri: String,
}
pub struct HlsOriginBuilder {
trunk: Arc<Trunk>,
target_duration_secs: Option<f64>,
window_segments: Option<NonZeroUsize>,
container: Container,
part_target_ms: Option<u32>,
media_sequence_offset: u64,
instance: Option<u64>,
arrival_clock: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
declared_bandwidth_bps: Option<u64>,
audio_renditions: Vec<AudioRenditionDecl>,
audio_sibling_origins: Vec<Arc<HlsOrigin>>,
subtitle_renditions: Vec<SubtitleRenditionDecl>,
subtitle_sibling_origins: Vec<Arc<HlsOrigin>>,
playlist_uri: Option<String>,
rendition_siblings: Vec<RenditionSibling>,
closed_captions_none: bool,
closed_caption_overrides: Vec<ClosedCaptionDecl>,
}
impl HlsOriginBuilder {
fn new(trunk: Arc<Trunk>) -> Self {
HlsOriginBuilder {
trunk,
target_duration_secs: None,
window_segments: None,
container: Container::default(),
part_target_ms: None,
media_sequence_offset: 0,
instance: None,
arrival_clock: None,
declared_bandwidth_bps: None,
audio_renditions: Vec::new(),
audio_sibling_origins: Vec::new(),
subtitle_renditions: Vec::new(),
subtitle_sibling_origins: Vec::new(),
playlist_uri: None,
rendition_siblings: Vec::new(),
closed_captions_none: false,
closed_caption_overrides: Vec::new(),
}
}
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 media_sequence_offset(mut self, offset: u64) -> Self {
self.media_sequence_offset = offset;
self
}
pub fn instance(mut self, token: u64) -> Self {
self.instance = Some(token);
self
}
pub fn program_date_time_arrival(mut self, clock: Arc<dyn Fn() -> i64 + Send + Sync>) -> Self {
self.arrival_clock = Some(clock);
self
}
pub fn declared_bandwidth_bps(mut self, declared_bandwidth_bps: u64) -> Self {
self.declared_bandwidth_bps = Some(declared_bandwidth_bps);
self
}
pub fn audio_renditions(mut self, audio_renditions: Vec<AudioRenditionDecl>) -> Self {
self.audio_renditions = audio_renditions;
self
}
pub fn audio_sibling_origins(mut self, origins: Vec<Arc<HlsOrigin>>) -> Self {
self.audio_sibling_origins = origins;
self
}
pub fn subtitle_renditions(mut self, subtitle_renditions: Vec<SubtitleRenditionDecl>) -> Self {
self.subtitle_renditions = subtitle_renditions;
self
}
pub fn subtitle_sibling_origins(mut self, origins: Vec<Arc<HlsOrigin>>) -> Self {
self.subtitle_sibling_origins = origins;
self
}
pub fn playlist_uri(mut self, playlist_uri: impl Into<String>) -> Self {
self.playlist_uri = Some(playlist_uri.into());
self
}
pub fn rendition_siblings(mut self, siblings: Vec<RenditionSibling>) -> Self {
self.rendition_siblings = siblings;
self
}
pub fn closed_captions_none(mut self, closed_captions_none: bool) -> Self {
self.closed_captions_none = closed_captions_none;
self
}
pub fn closed_caption_overrides(mut self, overrides: Vec<ClosedCaptionDecl>) -> Self {
self.closed_caption_overrides = overrides;
self
}
pub fn build(self) -> Result<HlsOrigin, HlsOriginBuildError> {
let target_duration_secs = self
.target_duration_secs
.ok_or(HlsOriginBuildError::MissingTargetDurationSecs)?;
if !target_duration_secs.is_finite() || target_duration_secs <= 0.0 {
return Err(HlsOriginBuildError::InvalidTargetDuration);
}
let window_segments = self
.window_segments
.ok_or(HlsOriginBuildError::MissingWindowSegments)?;
if self.part_target_ms == Some(0) {
return Err(HlsOriginBuildError::ZeroPartTarget);
}
if self
.media_sequence_offset
.checked_add(u64::from(u32::MAX))
.is_none()
{
return Err(HlsOriginBuildError::MediaSequenceOffsetTooLarge);
}
if let Some(declared) = self.declared_bandwidth_bps
&& (declared == 0 || declared > u64::from(u32::MAX))
{
return Err(HlsOriginBuildError::InvalidDeclaredBandwidth);
}
if self.audio_sibling_origins.is_empty() {
validate_audio_renditions(&self.audio_renditions)?;
} else {
validate_audio_renditions_relaxed(&self.audio_renditions)?;
}
validate_closed_caption_overrides(&self.closed_caption_overrides)?;
validate_subtitle_renditions(&self.subtitle_renditions)?;
if !self.subtitle_sibling_origins.is_empty() {
if self.subtitle_sibling_origins.len() != self.subtitle_renditions.len() {
return Err(HlsOriginBuildError::SubtitleSiblingCountMismatch);
}
let target = target_duration_secs.ceil();
for sibling in &self.subtitle_sibling_origins {
if sibling.container != Container::WebVtt {
return Err(HlsOriginBuildError::SubtitleSiblingNotWebVtt);
}
if sibling.target_duration_secs.ceil() != target {
return Err(HlsOriginBuildError::SubtitleSiblingTargetDurationMismatch);
}
}
}
let cursor = self.trunk.subscribe_segments();
Ok(HlsOrigin {
trunk: self.trunk,
cursor: Mutex::new(cursor),
drained_through: AtomicU64::new(UNDRAINED),
drain_gate: Mutex::new(()),
window: Mutex::new(Window::new(window_segments)),
init: Mutex::new(InitStore::default()),
codecs: Mutex::new(None),
video_codecs: Mutex::new(None),
video_range: Mutex::new(None),
supplemental_codecs: Mutex::new(None),
resolution: Mutex::new(None),
frame_rate: Mutex::new(None),
video_track_id: Mutex::new(None),
target_duration_secs,
container: self.container,
part_target_ms: self.part_target_ms,
part_target_override_ms: AtomicU32::new(0),
media_sequence_offset: self.media_sequence_offset,
instance: self.instance.unwrap_or_else(next_instance_token),
arrival_clock: self.arrival_clock,
arrival_origin: Mutex::new(None),
declared_bandwidth_bps: self.declared_bandwidth_bps,
audio_renditions: self.audio_renditions,
audio_sibling_origins: self.audio_sibling_origins,
subtitle_renditions: Mutex::new(self.subtitle_renditions),
playlist_uri: self
.playlist_uri
.unwrap_or_else(|| "media.m3u8".to_string()),
rendition_siblings: Mutex::new(self.rendition_siblings),
track_specs: Mutex::new(Vec::new()),
closed_captions_none: self.closed_captions_none,
closed_caption_overrides: self.closed_caption_overrides,
detected_closed_captions: Mutex::new(None),
declared_closed_caption_languages: Mutex::new(Vec::new()),
frozen_target_duration_secs: Mutex::new(None),
target_duration_violations: AtomicU64::new(0),
})
}
}
pub struct HlsOrigin {
trunk: Arc<Trunk>,
cursor: Mutex<SegmentCursor>,
drained_through: AtomicU64,
drain_gate: Mutex<()>,
window: Mutex<Window>,
init: Mutex<InitStore>,
codecs: Mutex<Option<String>>,
video_codecs: Mutex<Option<String>>,
video_range: Mutex<Option<VideoRange>>,
supplemental_codecs: Mutex<Option<String>>,
resolution: Mutex<Option<(u32, u32)>>,
frame_rate: Mutex<Option<FrameRate>>,
video_track_id: Mutex<Option<TrackId>>,
target_duration_secs: f64,
container: Container,
part_target_ms: Option<u32>,
part_target_override_ms: AtomicU32,
media_sequence_offset: u64,
instance: u64,
arrival_clock: Option<Arc<dyn Fn() -> i64 + Send + Sync>>,
arrival_origin: Mutex<Option<(u64, i64)>>,
declared_bandwidth_bps: Option<u64>,
audio_renditions: Vec<AudioRenditionDecl>,
audio_sibling_origins: Vec<Arc<HlsOrigin>>,
subtitle_renditions: Mutex<Vec<SubtitleRenditionDecl>>,
playlist_uri: String,
rendition_siblings: Mutex<Vec<RenditionSibling>>,
track_specs: Mutex<Vec<transmux::TrackSpec>>,
closed_captions_none: bool,
closed_caption_overrides: Vec<ClosedCaptionDecl>,
detected_closed_captions: Mutex<Option<Vec<broadcast_hls::InstreamId>>>,
declared_closed_caption_languages: Mutex<Vec<(broadcast_hls::InstreamId, String)>>,
frozen_target_duration_secs: Mutex<Option<f64>>,
target_duration_violations: AtomicU64,
}
impl HlsOrigin {
pub fn builder(trunk: Arc<Trunk>) -> HlsOriginBuilder {
HlsOriginBuilder::new(trunk)
}
pub fn set_init(&self, bytes: impl Into<Bytes>) {
let bytes = bytes.into();
if self.container == Container::Fmp4 {
let derived = transmux::Fmp4Demux::new()
.unpackage(bytes.as_ref())
.ok()
.map(|media| {
media
.tracks
.into_iter()
.map(|t| t.spec)
.collect::<Vec<transmux::TrackSpec>>()
})
.map(|specs| self.with_known_es_info(specs))
.and_then(|specs| self.set_track_specs(&specs).ok());
if derived.is_none() {
*locked(&self.codecs) = None;
*locked(&self.video_codecs) = None;
*locked(&self.video_range) = None;
*locked(&self.supplemental_codecs) = None;
*locked(&self.resolution) = None;
*locked(&self.frame_rate) = None;
*locked(&self.video_track_id) = None;
*locked(&self.track_specs) = Vec::new();
}
}
let next_seq = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0))
.saturating_next();
locked(&self.init).set(bytes, next_seq);
}
pub fn set_track_specs(&self, specs: &[transmux::TrackSpec]) -> Result<(), transmux::Error> {
let mut ordered: Vec<&transmux::TrackSpec> = specs.iter().collect();
ordered.sort_by_key(|spec| spec.track_id);
let stored_track_specs: Vec<transmux::TrackSpec> =
ordered.iter().map(|s| (*s).clone()).collect();
*locked(&self.video_track_id) = ordered
.iter()
.find(|spec| is_video_codec(&spec.config))
.map(|spec| spec.track_id);
let mut codecs: Vec<String> = Vec::new();
let mut video_codecs: Option<String> = None;
let mut video_codecs_decided = false;
let mut video_range: Option<VideoRange> = None;
let mut video_range_decided = false;
let mut resolution: Option<(u32, u32)> = None;
let mut resolution_decided = false;
let mut frame_rate: Option<FrameRate> = None;
let mut supplemental_codecs: Option<String> = None;
for spec in ordered {
let dolby_vision = if !video_codecs_decided && is_video_codec(&spec.config) {
dolby_vision_hls(spec)
} else {
None
};
match transmux::rfc6381_codec_string(&spec.config) {
Ok(codec) => {
codecs.push(dolby_vision.as_ref().map_or(codec, |dv| dv.codecs.clone()))
}
Err(transmux::Error::UnsupportedCodec { .. }) => {}
Err(other) => return Err(other),
}
if !video_codecs_decided && is_video_codec(&spec.config) {
match transmux::rfc6381_codec_string(&spec.config) {
Ok(codec) => {
video_codecs =
Some(dolby_vision.as_ref().map_or(codec, |dv| dv.codecs.clone()))
}
Err(transmux::Error::UnsupportedCodec { .. }) => {}
Err(other) => return Err(other),
}
video_codecs_decided = true;
if let Some(dv) = dolby_vision {
supplemental_codecs = dv.supplemental_codecs;
video_range = Some(dv.video_range);
video_range_decided = true;
} else {
supplemental_codecs = hdr10plus_supplemental_codecs(spec);
}
}
if !video_range_decided
&& let Some(display) = &spec.display
&& let Some(signal) = &display.video_signal
&& let Some(colour) = &signal.colour
{
video_range = VideoRange::from_transfer_characteristics(colour.transfer.code());
video_range_decided = true;
}
if !resolution_decided
&& let Some((w, h)) = match &spec.config {
transmux::CodecConfig::Avc { width, height, .. } => Some((*width, *height)),
transmux::CodecConfig::Hevc { width, height, .. } => Some((*width, *height)),
_ => None,
}
{
resolution = Some(transmux::presentation_size(spec.display.as_ref(), w, h));
resolution_decided = true;
}
if let Some(mh) = transmux::nominal_frame_rate_millihertz(&spec.config) {
let candidate = FrameRate::from_millihertz(mh);
frame_rate = Some(match frame_rate {
Some(existing) if existing.millihertz() >= mh => existing,
_ => candidate,
});
}
}
*locked(&self.codecs) = (!codecs.is_empty()).then(|| codecs.join(","));
*locked(&self.video_codecs) = video_codecs;
*locked(&self.video_range) = video_range;
*locked(&self.supplemental_codecs) = supplemental_codecs;
*locked(&self.resolution) = resolution;
*locked(&self.frame_rate) = frame_rate;
*locked(&self.track_specs) = stored_track_specs;
Ok(())
}
fn with_known_es_info(&self, mut specs: Vec<transmux::TrackSpec>) -> Vec<transmux::TrackSpec> {
let known = locked(&self.track_specs);
for spec in &mut specs {
if let Some(prev) = known.iter().find(|k| k.track_id == spec.track_id) {
spec.hdr10plus |= prev.hdr10plus;
}
if !spec.es_info_descriptors.is_empty() {
continue;
}
if let Some(prev) = known.iter().find(|k| k.track_id == spec.track_id) {
spec.source_pid = prev.source_pid;
spec.es_info_descriptors = prev.es_info_descriptors.clone();
}
}
specs
}
fn window_snapshot(&self) -> (u64, u64, u128) {
let w = locked(&self.window);
(w.peak_bandwidth_bps, w.total_bytes, w.total_duration_nanos)
}
pub fn set_detected_closed_captions(&self, detected: Vec<broadcast_hls::InstreamId>) {
*locked(&self.detected_closed_captions) = Some(detected);
}
pub fn set_declared_closed_caption_languages(
&self,
languages: Vec<(broadcast_hls::InstreamId, String)>,
) {
*locked(&self.declared_closed_caption_languages) = languages;
}
pub fn set_subtitle_renditions(
&self,
decls: Vec<SubtitleRenditionDecl>,
) -> Result<(), HlsOriginBuildError> {
validate_subtitle_renditions(&decls)?;
*locked(&self.subtitle_renditions) = decls;
Ok(())
}
pub fn master_playlist(&self, media_playlist_name: &str) -> Result<String, HlsMasterError> {
self.master_playlist_inner(media_playlist_name, None)
}
pub fn master_playlist_with_iframes(
&self,
media_playlist_name: &str,
iframe_playlist_name: &str,
) -> Result<String, HlsMasterError> {
self.master_playlist_inner(media_playlist_name, Some(iframe_playlist_name))
}
fn master_playlist_inner(
&self,
media_playlist_name: &str,
iframe_playlist_name: Option<&str>,
) -> Result<String, HlsMasterError> {
self.drain();
let (video_peak, video_bytes, video_duration) = {
let window = locked(&self.window);
(
window.peak_bandwidth_bps,
window.total_bytes,
window.total_duration_nanos,
)
};
let declared = self.declared_bandwidth_bps.unwrap_or(0);
let video_part_peak = video_peak.max(declared);
if video_part_peak == 0 {
return Err(HlsMasterError::BandwidthUnknown);
}
let video_avg = average_bandwidth_bps(video_bytes, video_duration);
let has_siblings = !self.audio_sibling_origins.is_empty();
let primary_codecs = locked(&self.codecs).clone();
let mut extra_attrs: Vec<(String, broadcast_hls::AttrValue)> = Vec::new();
if let Some(range) = *locked(&self.video_range)
&& matches!(range, VideoRange::Hlg | VideoRange::Pq)
{
extra_attrs.push(range.attr());
}
if let Some(supplemental) = locked(&self.supplemental_codecs).clone() {
extra_attrs.push((
String::from(SUPPLEMENTAL_CODECS_ATTR),
broadcast_hls::AttrValue::for_attr(SUPPLEMENTAL_CODECS_ATTR, supplemental)
.map_err(HlsMasterError::Render)?,
));
}
for decl in &self.audio_renditions {
if decl.uri == media_playlist_name {
return Err(HlsMasterError::AudioRenditionUriEqualsMediaPlaylist);
}
}
let media: Vec<MediaRendition> = self
.audio_renditions
.iter()
.enumerate()
.map(|(i, decl)| {
let sibling = self.audio_sibling_origins.get(i);
let audio_spec = sibling.and_then(|s| {
locked(&s.track_specs)
.iter()
.find(|t| !is_video_codec(&t.config))
.cloned()
});
let meta = audio_spec.as_ref().map(|t| t.audio_metadata());
let role = meta
.as_ref()
.map(|m| m.role)
.unwrap_or(transmux::AudioRole::Unknown);
let derived_iso = meta
.as_ref()
.and_then(|m| m.languages.first().map(|l| l.language.clone()))
.unwrap_or_else(|| "und".to_string());
let derived_lang =
bcp47_language(&derived_iso).unwrap_or_else(|| "und".to_string());
let language = if decl.language.is_empty() {
derived_lang.clone()
} else {
decl.language.clone()
};
let name_source = if decl.language.is_empty() {
derived_iso.clone()
} else {
decl.language.clone()
};
let name = if decl.name.is_empty() {
if role == transmux::AudioRole::AudioDescription {
format!("{} (AD)", name_source)
} else {
name_source.clone()
}
} else {
decl.name.clone()
};
let characteristics = if decl.characteristics.is_empty() {
if role == transmux::AudioRole::AudioDescription {
vec!["public.accessibility.describes-video".to_string()]
} else {
Vec::new()
}
} else {
decl.characteristics.clone()
};
let channels = decl.channels.clone().or_else(|| {
audio_spec
.as_ref()
.and_then(|t| channel_count_string(&t.config))
});
let mut m = MediaRendition::default();
m.media_type = MediaType::Audio;
m.uri = Some(decl.uri.clone());
m.group_id = decl.group_id.clone();
m.language = Some(language);
m.name = name;
m.default = decl.default;
m.autoselect = decl.autoselect;
m.channels = channels;
m.characteristics = characteristics;
m
})
.collect();
let mut media = dedupe_rendition_names(media);
media.extend(self.subtitle_media_renditions());
let subtitles_group =
(!locked(&self.subtitle_renditions).is_empty()).then(|| SUBTITLES_GROUP_ID.to_string());
let with_wvtt = |codecs: Option<String>| match (&subtitles_group, codecs) {
(Some(_), Some(c)) => Some(format!("{c},{WEBVTT_CODEC}")),
(_, c) => c,
};
let detected_captions = locked(&self.detected_closed_captions).clone();
let scanned_without_captions = detected_captions.as_ref().is_some_and(Vec::is_empty);
let caption_media =
self.closed_caption_renditions(detected_captions.as_deref().unwrap_or_default());
let captions_advertised = !caption_media.is_empty();
media.extend(caption_media);
let audio_groups: Vec<String> = {
let mut g: Vec<String> = Vec::new();
for decl in &self.audio_renditions {
if !g.contains(&decl.group_id) {
g.push(decl.group_id.clone());
}
}
g
};
let closed_captions = if captions_advertised {
Some(ClosedCaptionsRef::Group(
CLOSED_CAPTIONS_GROUP_ID.to_string(),
))
} else if self.closed_captions_none || scanned_without_captions {
Some(ClosedCaptionsRef::None)
} else {
None
};
let mut variants: Vec<Variant> = Vec::new();
if audio_groups.is_empty() {
let bw = video_part_peak;
let bw = u32::try_from(bw)
.map_err(|_| HlsMasterError::BandwidthOverflow { bandwidth: bw })?;
let avg = video_avg.map(|a| a.min(bw));
variants.push(Variant {
bandwidth: bw,
average_bandwidth: avg,
codecs: with_wvtt(primary_codecs.clone()),
uri: media_playlist_name.to_string(),
audio: None,
video: None,
subtitles: subtitles_group.clone(),
closed_captions,
extra_attrs: extra_attrs.clone(),
resolution: *locked(&self.resolution),
frame_rate: self.effective_frame_rate(),
});
} else {
for group in &audio_groups {
let mut audio_peak_max = 0u64;
let mut audio_avg_max = 0u32;
let mut audio_codecs: Vec<String> = Vec::new();
for (i, decl) in self.audio_renditions.iter().enumerate() {
if decl.group_id != *group {
continue;
}
if let Some(sib) = self.audio_sibling_origins.get(i) {
sib.drain();
let (peak, tb, td) = sib.window_snapshot();
audio_peak_max = audio_peak_max.max(peak);
if let Some(avg) = average_bandwidth_bps(tb, td) {
audio_avg_max = audio_avg_max.max(avg);
}
if let Some(codec) = locked(&sib.codecs).clone()
&& !audio_codecs.contains(&codec)
{
audio_codecs.push(codec);
}
}
}
let group_peak = video_part_peak.saturating_add(audio_peak_max);
let group_bw =
u32::try_from(group_peak).map_err(|_| HlsMasterError::BandwidthOverflow {
bandwidth: group_peak,
})?;
let group_avg_candidate = video_avg.unwrap_or(0) as u64 + u64::from(audio_avg_max);
let group_avg = if group_avg_candidate == 0 {
None
} else {
Some(
u32::try_from(group_avg_candidate.min(u64::from(group_bw)))
.unwrap_or(group_bw),
)
};
let group_codecs = if has_siblings {
let mut all: Vec<String> = primary_codecs.iter().cloned().collect();
all.extend(audio_codecs);
if all.is_empty() {
None
} else {
Some(all.join(","))
}
} else {
primary_codecs.clone()
};
variants.push(Variant {
bandwidth: group_bw,
average_bandwidth: group_avg,
codecs: with_wvtt(group_codecs),
uri: media_playlist_name.to_string(),
audio: Some(group.clone()),
video: None,
subtitles: subtitles_group.clone(),
closed_captions: closed_captions.clone(),
extra_attrs: extra_attrs.clone(),
resolution: *locked(&self.resolution),
frame_rate: self.effective_frame_rate(),
});
}
}
let mut iframe_variants: Vec<broadcast_hls::IFrameVariant> = Vec::new();
if let Some(iframe_name) = iframe_playlist_name
&& self.iframe_playlist_available()
{
let (iframe_measured, iframe_total_bytes, iframe_total_duration_nanos) = {
let window = locked(&self.window);
(
window.iframe_peak_bandwidth_bps,
window.iframe_total_bytes,
window.iframe_total_duration_nanos,
)
};
if iframe_measured != 0 {
let iframe_bandwidth = u32::try_from(iframe_measured).map_err(|_| {
HlsMasterError::BandwidthOverflow {
bandwidth: iframe_measured,
}
})?;
let iframe_average =
average_bandwidth_bps(iframe_total_bytes, iframe_total_duration_nanos)
.map(|avg| avg.min(iframe_bandwidth));
let mut iframe_extra: Vec<(String, broadcast_hls::AttrValue)> = Vec::new();
if let Some(range) = *locked(&self.video_range)
&& matches!(range, VideoRange::Hlg | VideoRange::Pq)
{
iframe_extra.push(range.attr());
}
iframe_variants.push(broadcast_hls::IFrameVariant {
bandwidth: iframe_bandwidth,
average_bandwidth: iframe_average,
codecs: locked(&self.video_codecs).clone(),
resolution: *locked(&self.resolution),
uri: iframe_name.to_string(),
extra_attrs: iframe_extra,
});
}
}
let master = MasterPlaylist {
media,
variants,
iframe_variants,
..Default::default()
};
master.to_m3u8().map_err(HlsMasterError::Render)
}
fn subtitle_media_renditions(&self) -> Vec<MediaRendition> {
locked(&self.subtitle_renditions)
.iter()
.map(|decl| {
let mut m = MediaRendition::default();
m.media_type = MediaType::Subtitles;
m.uri = Some(decl.uri.clone());
m.group_id = SUBTITLES_GROUP_ID.to_string();
m.language = Some(decl.language.clone());
m.name = decl.name.clone();
m.default = decl.default;
m.autoselect = decl.effective_autoselect();
m.forced = decl.forced;
if decl.hearing_impaired {
m.characteristics = HEARING_IMPAIRED_CHARACTERISTICS
.iter()
.map(|c| (*c).to_string())
.collect();
}
m
})
.collect()
}
fn closed_caption_renditions(
&self,
detected: &[broadcast_hls::InstreamId],
) -> Vec<MediaRendition> {
let stream_languages = locked(&self.declared_closed_caption_languages).clone();
let renditions = detected
.iter()
.map(|id| {
let rendered = id.render();
let declared = self.closed_caption_overrides.iter().find(|d| {
broadcast_hls::InstreamId::parse_closed_captions(&d.instream_id).as_ref()
== Ok(id)
});
let pick = |value: Option<&String>, default: &str| match value {
Some(v) if !v.is_empty() => v.clone(),
_ => default.to_string(),
};
let mut m = MediaRendition::default();
m.media_type = MediaType::ClosedCaptions;
m.group_id = CLOSED_CAPTIONS_GROUP_ID.to_string();
m.name = pick(declared.map(|d| &d.name), &rendered);
m.language = declared
.map(|d| d.language.clone())
.filter(|l| !l.is_empty())
.or_else(|| {
stream_languages
.iter()
.find(|(declared_id, _)| declared_id == id)
.map(|(_, language)| language.clone())
});
m.autoselect = false;
m.instream_id = Some(id.clone());
m
})
.collect();
dedupe_rendition_names(renditions)
}
fn iframe_playlist_available(&self) -> bool {
if self.container != Container::Fmp4 {
return false;
}
if locked(&self.video_track_id).is_none() {
return false;
}
let window = self.window_view();
self.iframe_entries(&window).is_some()
}
fn iframe_entries(&self, window: &WindowView) -> Option<Vec<(WindowSegment, u64)>> {
let mut entries: Vec<(WindowSegment, u64)> = Vec::new();
let mut seen_entry = false;
for s in window.segments.iter() {
match s.iframe_len {
Some(len) => {
seen_entry = true;
entries.push((s.clone(), len));
}
None => {
if seen_entry {
return None;
}
}
}
}
if entries.is_empty() {
None
} else {
Some(entries)
}
}
pub fn instance(&self) -> u64 {
self.instance
}
pub fn iframe_playlist_name(&self, track_id: TrackId) -> String {
format!("iframe-{}.m3u8", track_id)
}
pub fn init_name(&self, track_id: TrackId, generation: Generation) -> String {
format!("init-{track_id}-{}-{generation}.mp4", self.instance)
}
fn segment_name(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
format!(
"seg-{track_id}-{}-{}.{ext}",
self.instance,
self.public_msn(seq)
)
}
pub fn iframe_segment_name(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
format!(
"iframe-seg-{track_id}-{}-{}.{ext}",
self.instance,
self.public_msn(seq)
)
}
fn part_name(&self, track_id: TrackId, seq: SegmentSeq, idx: PartIndex, ext: &str) -> String {
format!(
"part-{track_id}-{}-{}.{idx}.{ext}",
self.instance,
self.public_msn(seq)
)
}
fn root_relative(&self, name: String) -> String {
let depth = self.playlist_uri.rfind('/').map_or(0, |i| {
self.playlist_uri[..i]
.split('/')
.filter(|s| !s.is_empty())
.count()
});
format!("{}{name}", "../".repeat(depth))
}
fn init_uri(&self, track_id: TrackId, generation: Generation) -> String {
self.root_relative(self.init_name(track_id, generation))
}
fn segment_uri(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
self.root_relative(self.segment_name(track_id, seq, ext))
}
fn iframe_segment_uri(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> String {
self.root_relative(self.iframe_segment_name(track_id, seq, ext))
}
fn part_uri(&self, track_id: TrackId, seq: SegmentSeq, idx: PartIndex, ext: &str) -> String {
self.root_relative(self.part_name(track_id, seq, idx, ext))
}
pub fn media_sequence_offset(&self) -> u64 {
self.media_sequence_offset
}
pub fn next_media_sequence(&self) -> u64 {
self.public_msn(
self.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0))
.saturating_next(),
)
}
pub fn http_priority(&self, request: &HlsRequest) -> HttpPriority {
match request {
HlsRequest::Playlist { .. } | HlsRequest::IFramePlaylist { .. } => {
HttpPriority::PLAYLIST
}
HlsRequest::Resource { name } => {
let _ = parse_part(name, self.container);
HttpPriority::SEGMENT
}
}
}
fn public_msn(&self, seq: SegmentSeq) -> u64 {
self.media_sequence_offset
.saturating_add(u64::from(seq.value()))
}
fn internal_seq(&self, public: u64) -> Option<SegmentSeq> {
u32::try_from(public.checked_sub(self.media_sequence_offset)?)
.ok()
.map(SegmentSeq::new)
}
fn advance_part_limit(&self) -> u64 {
match self.effective_part_target_ms() {
None | Some(0) => 0,
Some(ms) if u64::from(ms) < MILLIS_PER_SEC => {
(ADVANCE_PART_LIMIT_NUMERATOR * MILLIS_PER_SEC).div_ceil(u64::from(ms))
}
Some(_) => ADVANCE_PART_LIMIT_NUMERATOR,
}
}
pub fn target_duration_violations(&self) -> u64 {
self.target_duration_violations.load(Ordering::Relaxed)
}
pub fn published_target_duration(&self) -> Duration {
let frozen = *locked(&self.frozen_target_duration_secs);
let secs = frozen.unwrap_or_else(|| {
let sibling = locked(&self.rendition_siblings)
.iter()
.filter_map(|s| s.origin.upgrade())
.map(|o| o.target_duration_candidate_secs())
.fold(0.0, f64::max);
self.target_duration_candidate_secs().max(sibling)
});
Duration::from_secs(secs.ceil().clamp(0.0, f64::from(u32::MAX)) as u64)
}
pub fn set_part_target_ms(&self, ms: u32) {
self.part_target_override_ms.store(ms, Ordering::Relaxed);
}
fn effective_part_target_ms(&self) -> Option<u32> {
self.part_target_ms
.map(|base| base.max(self.part_target_override_ms.load(Ordering::Relaxed)))
}
fn max_part_target_ms(&self) -> Option<u32> {
let own = self.effective_part_target_ms()?;
let siblings = locked(&self.rendition_siblings).clone();
Some(
siblings
.iter()
.filter_map(|s| s.origin.upgrade())
.filter_map(|o| o.effective_part_target_ms())
.fold(own, u32::max),
)
}
fn part_timing(&self) -> Option<(DecimalSeconds, DecimalSeconds)> {
let part_target = self.effective_part_target_ms()?;
let hold_basis = self.max_part_target_ms().unwrap_or(part_target);
let part_target = DecimalSeconds::new(f64::from(part_target) / 1000.0)
.expect("u32 / 1000.0 is finite, >= 0");
let part_hold_back =
DecimalSeconds::new(f64::from(hold_basis) / 1000.0 * PART_HOLD_BACK_MULTIPLIER)
.expect("finite, >= 0 times a positive constant is finite, >= 0");
Some((part_target, part_hold_back))
}
pub fn set_rendition_siblings(&self, siblings: Vec<RenditionSibling>) {
*locked(&self.rendition_siblings) = siblings;
}
pub fn current_live_edge(&self) -> (u64, Option<u64>) {
self.drain();
let (seq, open_parts) = self.live_edge();
if !open_parts.is_empty() {
let msn = self.public_msn(seq);
let last_part = open_parts.last().map(|p| u64::from(p.part_index.value()));
return (msn, last_part);
}
let last_closed = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0));
let msn = self.public_msn(last_closed);
let closed_parts = self.trunk.parts_in_segment(last_closed);
let last_part = closed_parts.last().map(|p| u64::from(p.part_index.value()));
(msn, last_part)
}
pub fn init_bytes(&self) -> Option<Bytes> {
locked(&self.init).current.clone()
}
fn drain(&self) {
let head = self
.trunk
.last_closed_segment()
.map_or(0, |n| u64::from(n.value()).saturating_add(1));
if self.drained_through.load(Ordering::Acquire) == head {
return;
}
let _gate = locked(&self.drain_gate);
if self.drained_through.load(Ordering::Acquire) == head {
return;
}
let mut cursor = locked(&self.cursor);
let mut polled = Vec::new();
while let Some(item) = cursor.poll() {
polled.push(item);
}
drop(cursor);
enum Prepared {
Segment(SegmentEntry, Generation, Option<u64>, Option<u32>),
Gap,
}
let video_track_id = *locked(&self.video_track_id);
let signalled_rate = locked(&self.frame_rate).is_some();
let video_timescale = locked(&self.track_specs)
.iter()
.find(|spec| is_video_codec(&spec.config))
.map(|spec| spec.timescale);
let mut prepared = Vec::with_capacity(polled.len());
for item in polled {
match item {
SegmentCursorItem::Segment(entry) => {
let iframe_len = video_track_id
.and_then(|track_id| compute_iframe_len(&entry.bytes, track_id));
let video_rate = if signalled_rate {
None
} else {
match self.container {
Container::MpegTs => ts_video_rate(&entry.bytes),
_ => video_track_id.zip(video_timescale).and_then(
|(track_id, timescale)| {
fmp4_video_rate(&entry.bytes, track_id, timescale)
},
),
}
};
let init_gen = locked(&self.init).generation_for(entry.sequence_number);
prepared.push(Prepared::Segment(entry, init_gen, iframe_len, video_rate));
}
_ => prepared.push(Prepared::Gap),
}
}
let mut window = locked(&self.window);
for item in prepared {
match item {
Prepared::Segment(entry, init_gen, iframe_len, video_rate) => {
if self.arrival_clock.is_some() {
let mut origin = locked(&self.arrival_origin);
if origin.is_none()
&& let Some(clock) = &self.arrival_clock
{
let now_ms = clock();
let dur_nanos = entry.duration.as_nanos();
let dur_ms = (dur_nanos + 500_000) / 1_000_000;
let dur_ms = i64::try_from(dur_ms).unwrap_or(i64::MAX);
let epoch_ms = now_ms.saturating_sub(dur_ms);
*origin = Some((entry.timeline_position.as_nanos(), epoch_ms));
}
}
let seg_duration = entry.duration;
window.push(entry, init_gen, iframe_len, video_rate);
if let Some(frozen) = *locked(&self.frozen_target_duration_secs)
&& seg_duration.as_secs_f64().round() > frozen.round()
{
self.target_duration_violations
.fetch_add(1, Ordering::Relaxed);
}
}
Prepared::Gap => window.note_gap(),
}
}
self.drained_through.store(head, Ordering::Release);
}
fn effective_frame_rate(&self) -> Option<FrameRate> {
if let Some(signalled) = *locked(&self.frame_rate) {
return Some(signalled);
}
let millihertz = locked(&self.window).max_video_millihertz()?;
Some(FrameRate::from_millihertz(millihertz))
}
fn window_view(&self) -> WindowView {
let w = locked(&self.window);
WindowView {
segments: w.segments.clone(),
max_segment_duration_secs: w.max_segment_duration_secs,
discontinuity_sequence: w.discontinuity_sequence,
}
}
fn segment_pdt(
&self,
seq: SegmentSeq,
start_ns: u64,
) -> Option<broadcast_hls::ProgramDateTime> {
if let Some(anchor) = self.trunk.time_anchor()
&& let Some(media_start) = self.trunk.segment_start_time(seq)
&& let Ok(ts) = jiff::Timestamp::from_millisecond(anchor.media_to_epoch_ms(media_start))
{
return Some(broadcast_hls::ProgramDateTime::from_timestamp(ts));
}
let origin = *locked(&self.arrival_origin);
if let Some((origin_start_ns, origin_epoch_ms)) = origin {
let delta_ns = i128::from(start_ns) - i128::from(origin_start_ns);
let delta_ms = delta_ns / 1_000_000;
let epoch_ms = i128::from(origin_epoch_ms) + delta_ms;
let epoch_ms: i64 = epoch_ms.try_into().ok()?;
if let Ok(ts) = jiff::Timestamp::from_millisecond(epoch_ms) {
return Some(broadcast_hls::ProgramDateTime::from_timestamp(ts));
}
}
None
}
pub fn closed_segments(&self) -> Vec<ClosedSegment> {
self.drain();
locked(&self.window)
.segments
.iter()
.map(|s| ClosedSegment {
sequence_number: s.sequence_number,
start_ns: s.start_ns,
duration_secs: s.duration_secs,
discontinuous: s.discontinuous,
init_generation: s.init_gen,
})
.collect()
}
fn live_edge(&self) -> (SegmentSeq, Vec<PartEntry>) {
let last_closed = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0));
let candidate = last_closed.saturating_next();
let parts = self.trunk.parts_in_segment(candidate);
if parts.is_empty() {
(last_closed, Vec::new())
} else {
(candidate, parts)
}
}
fn resident_part_specs(&self, track_id: TrackId, seq: SegmentSeq, ext: &str) -> Vec<PartSpec> {
let by_index: BTreeMap<PartIndex, PartEntry> = self
.trunk
.parts_in_segment(seq)
.into_iter()
.map(|p| (p.part_index, p))
.collect();
let contiguous = by_index
.keys()
.copied()
.zip(0u32..)
.all(|(index, expected)| index.value() == expected);
if by_index.is_empty() || !contiguous {
return Vec::new();
}
by_index
.values()
.map(|p| PartSpec {
uri: self.part_uri(track_id, p.segment_number, p.part_index, ext),
duration: DecimalSeconds::new(p.duration.as_secs_f64())
.expect("Duration::as_secs_f64 is finite, >= 0"),
independent: p.independent,
..Default::default()
})
.collect()
}
fn target_duration_candidate_secs(&self) -> f64 {
let own = self
.target_duration_secs
.max(locked(&self.window).max_segment_duration_secs);
(*locked(&self.frozen_target_duration_secs)).unwrap_or(own)
}
fn published_target_duration_secs(&self, window: &WindowView) -> f64 {
let siblings = locked(&self.rendition_siblings).clone();
let sibling_target = siblings
.iter()
.filter_map(|s| s.origin.upgrade())
.map(|o| o.target_duration_candidate_secs())
.fold(0.0, f64::max);
let candidate = self
.target_duration_secs
.max(window.max_segment_duration_secs)
.max(sibling_target);
let mut frozen = locked(&self.frozen_target_duration_secs);
if let Some(existing) = *frozen {
existing
} else if window.segments.is_empty() {
candidate
} else {
*frozen = Some(candidate);
candidate
}
}
fn render_playlist(
&self,
track_id: TrackId,
skip: Option<SkipDirective>,
) -> core::result::Result<String, String> {
self.drain();
let window = self.window_view();
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| self.public_msn(s.sequence_number))
.or_else(|| has_open_parts.then(|| self.public_msn(open_seq)))
.unwrap_or_else(|| self.public_msn(SegmentSeq::new(1)));
let target_duration = self.published_target_duration_secs(&window);
let target_duration = target_duration.ceil().clamp(0.0, f64::from(u32::MAX)) as u32;
let skip_boundary = if low_latency_enabled {
Some(f64::from(target_duration) * SKIP_BOUNDARY_MULTIPLIER)
} else {
None
};
let playlist_has_endlist = false;
let skip_accepted = skip.is_some() && skip_boundary.is_some() && !playlist_has_endlist;
let omitted = if skip_accepted {
let durations: Vec<f64> = window.segments.iter().map(|s| s.duration_secs).collect();
let total: f64 = durations.iter().sum();
let mut prefix = 0.0_f64;
let mut count = 0usize;
for d in &durations {
let distance = total - prefix;
if distance > skip_boundary.unwrap_or(f64::INFINITY) {
count += 1;
prefix += d;
} else {
break;
}
}
count.min(durations.len().saturating_sub(1))
} else {
0
};
let retention_secs = PART_RETENTION_TARGET_DURATIONS * f64::from(target_duration);
let mut behind_secs: f64 = if has_open_parts {
open_parts.iter().map(|p| p.duration.as_secs_f64()).sum()
} else {
0.0
};
let mut closed_parts: Vec<Vec<PartSpec>> = Vec::with_capacity(window.segments.len());
for s in window.segments.iter().rev() {
let keep = low_latency_enabled && behind_secs <= retention_secs;
closed_parts.push(if keep {
self.resident_part_specs(track_id, s.sequence_number, ext)
} else {
Vec::new()
});
behind_secs += s.duration_secs;
}
closed_parts.reverse();
let any_closed_parts = closed_parts.iter().any(|parts| !parts.is_empty());
let fmp4 = self.container == Container::Fmp4;
let map_for = |generation: Generation| {
fmp4.then(|| MapTag {
uri: self.init_uri(track_id, generation),
byte_range: None,
extra_attrs: Vec::new(),
})
};
let segments: Vec<MediaSegment> = window
.segments
.iter()
.zip(closed_parts)
.map(|(s, parts)| {
let program_date_time = self.segment_pdt(s.sequence_number, s.start_ns);
MediaSegment {
map: map_for(s.init_gen),
uri: self.segment_uri(track_id, s.sequence_number, ext),
duration: DecimalSeconds::new(s.duration_secs)
.expect("Duration::as_secs_f64 is finite, >= 0"),
discontinuous: s.discontinuous,
parts,
program_date_time,
gap: s.bytes.is_empty(),
..Default::default()
}
})
.collect::<Vec<_>>();
let segments: Vec<MediaSegment> = if omitted > 0 {
segments.into_iter().skip(omitted).collect()
} else {
segments
};
let open_gen = locked(&self.init).generation_for(open_seq);
let open_segment = has_open_parts.then(|| {
let mut open = OpenSegment::new(
open_parts
.iter()
.map(|p| PartSpec {
uri: self.part_uri(track_id, p.segment_number, p.part_index, ext),
duration: DecimalSeconds::new(p.duration.as_secs_f64())
.expect("Duration::as_secs_f64 is finite, >= 0"),
independent: p.independent,
..Default::default()
})
.collect(),
);
if let Some(map) = map_for(open_gen) {
open = open.with_map(map);
}
let open_start_ns = window.segments.back().map(|s| {
s.start_ns.saturating_add(
u64::try_from((s.duration_secs * 1_000_000_000.0).round() as i64).unwrap_or(0),
)
});
open.program_date_time = match open_start_ns {
Some(sn) => self.segment_pdt(open_seq, sn),
None => {
if self.trunk.time_anchor().is_some() {
self.segment_pdt(open_seq, 0)
} else {
None
}
}
};
open
});
let next_part_hint = if has_open_parts {
let next_idx = open_parts
.iter()
.map(|p| p.part_index)
.max()
.map(PartIndex::saturating_next)
.unwrap_or(PartIndex::new(0));
Some(self.part_uri(track_id, open_seq, next_idx, ext))
} else if low_latency_enabled && any_closed_parts {
Some(self.part_uri(track_id, open_seq.saturating_next(), PartIndex::new(0), ext))
} else {
None
};
let mut segment_dateranges: Vec<Vec<(String, String)>> =
vec![Vec::new(); window.segments.len()];
let mut recently_removed_daterange_ids: Vec<String> = Vec::new();
if let Some(anchor) = self.trunk.time_anchor() {
let timeline = timed_metadata::Timeline::with_anchor(anchor);
for (idx, seg) in window.segments.iter().enumerate() {
for entry in self.trunk.events_in_segment(seg.sequence_number) {
if let Ok(dr) = timeline.to_daterange(&entry.event)
&& let Ok(tag) = dr.to_tag_line()
{
if idx < omitted && skip_accepted && skip == Some(SkipDirective::V2) {
recently_removed_daterange_ids.push(dr.id.clone());
}
segment_dateranges[idx].push((tag, dr.id));
}
}
}
}
let mut extra_tags = Vec::new();
if fmp4 && segments.is_empty() && open_segment.is_none() {
let generation = locked(&self.init).generation.max(Generation::new(1));
extra_tags.push(format!(
"#EXT-X-MAP:URI=\"{}\"",
self.init_uri(track_id, generation)
));
}
for (idx, tags) in segment_dateranges.iter().enumerate() {
if idx < omitted && skip == Some(SkipDirective::V2) {
continue;
}
for (tag, _) in tags {
extra_tags.push(tag.clone());
}
}
let low_latency = self.part_timing().map(|(part_target, part_hold_back)| {
LowLatencyConfig {
part_target: Some(part_target),
part_hold_back: Some(part_hold_back),
hold_back: DecimalSeconds::new(f64::from(target_duration) * HOLD_BACK_MULTIPLIER)
.ok(),
preload_hint_part: next_part_hint,
can_skip_until: skip_boundary.and_then(|b| DecimalSeconds::new(b).ok()),
can_skip_dateranges: true,
can_block_reload: true,
..Default::default()
}
});
let skip_info = if skip_accepted {
Some(broadcast_hls::SkipInfo {
skipped_segments: omitted as u64,
recently_removed_daterange_ids,
extra_attrs: Vec::new(),
})
} else {
None
};
let rendition_reports = if self.part_target_ms.is_some() {
let siblings = locked(&self.rendition_siblings).clone();
let mut reports = Vec::with_capacity(siblings.len());
for sibling in siblings {
let Some(origin) = sibling.origin.upgrade() else {
continue;
};
if sibling.playlist_uri == self.playlist_uri {
continue;
}
let uri = relative_playlist_uri(&self.playlist_uri, &sibling.playlist_uri);
let (last_msn, last_part) = origin.current_live_edge();
reports.push(broadcast_hls::RenditionReport {
uri,
last_msn,
last_part,
extra_attrs: Vec::new(),
});
}
reports
} else {
Vec::new()
};
let playlist = MediaPlaylist {
target_duration,
media_sequence,
discontinuity_sequence: window.discontinuity_sequence,
always_emit_discontinuity_sequence: true,
segments,
open_segment,
endlist: false,
extra_tags,
low_latency,
iframes_only: false,
skip: skip_info,
rendition_reports,
..Default::default()
};
playlist.to_m3u8().map_err(|error| error.to_string())
}
fn resolve_playlist(
&self,
track_id: TrackId,
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();
let in_progress = self.public_msn(in_progress_seg);
if msn > in_progress.saturating_add(ABUSE_MSN_FUTURE_BOUND) {
return EgressResponse::BadRequest {
reason: "_HLS_msn unreasonably far beyond the live edge",
};
}
if let Some(part) = query.hls_part
&& msn >= in_progress
{
let existing = if msn == in_progress {
u64::try_from(live_parts.len()).unwrap_or(u64::MAX)
} else {
0
};
let limit = existing
.saturating_sub(1)
.saturating_add(self.advance_part_limit());
if u64::from(part) > limit {
return EgressResponse::BadRequest {
reason: "_HLS_part beyond the Advance Part Limit",
};
}
}
let satisfied = match query.hls_part {
Some(part) => {
in_progress > msn
|| (in_progress == msn
&& u64::try_from(live_parts.len()).unwrap_or(u64::MAX)
> u64::from(part))
}
None => {
self.public_msn(
self.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0)),
) >= msn
}
};
if !satisfied {
return EgressResponse::pending(await_policy, now, now);
}
}
match self.render_playlist(track_id, query.hls_skip) {
Ok(playlist) => EgressResponse::Ready {
body: HlsBody::Playlist(playlist),
cache: CachePolicy::NoCache,
},
Err(_) => EgressResponse::NotFound,
}
}
fn resolve_iframe_playlist(
&self,
track_id: TrackId,
query: BlockingQuery,
media_playlist_name: String,
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 last_closed = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0));
if msn
> self
.public_msn(last_closed.saturating_next())
.saturating_add(ABUSE_MSN_FUTURE_BOUND)
{
return EgressResponse::BadRequest {
reason: "_HLS_msn unreasonably far beyond the live edge",
};
}
let satisfied = self.public_msn(last_closed) >= msn;
if !satisfied {
return EgressResponse::pending(await_policy, now, now);
}
}
match self.render_iframe_playlist(track_id, &media_playlist_name) {
Ok(playlist) => EgressResponse::Ready {
body: HlsBody::Playlist(playlist),
cache: CachePolicy::NoCache,
},
Err(_) => EgressResponse::NotFound,
}
}
fn render_iframe_playlist(
&self,
track_id: TrackId,
media_playlist_name: &str,
) -> core::result::Result<String, String> {
self.drain();
if self.container != Container::Fmp4 {
return Err("not fmp4".to_string());
}
if locked(&self.video_track_id).is_none() {
return Err("no video track".to_string());
}
let window = self.window_view();
let Some(entries) = self.iframe_entries(&window) else {
return Err("no iframe entry".to_string());
};
let target_duration = self.published_target_duration_secs(&window);
let target_duration = target_duration.ceil().clamp(0.0, f64::from(u32::MAX)) as u32;
let fmp4 = self.container == Container::Fmp4;
let map_for = |generation: Generation| {
fmp4.then(|| MapTag {
uri: self.init_uri(track_id, generation),
byte_range: None,
extra_attrs: Vec::new(),
})
};
let first_entry_seq = entries[0].0.sequence_number;
let media_sequence = self.public_msn(first_entry_seq);
let segments: Vec<MediaSegment> = entries
.iter()
.map(|(s, len)| MediaSegment {
map: map_for(s.init_gen),
uri: self.iframe_segment_uri(
track_id,
s.sequence_number,
self.container.segment_extension(),
),
duration: DecimalSeconds::new(s.duration_secs).unwrap_or_default(),
discontinuous: s.discontinuous,
byte_range: Some(ByteRange {
length: *len,
offset: Some(0),
}),
parts: Vec::new(),
program_date_time: self.segment_pdt(s.sequence_number, s.start_ns),
..Default::default()
})
.collect();
let low_latency = if let Some((part_target, part_hold_back)) = self.part_timing() {
Some(broadcast_hls::LowLatencyConfig {
part_target: Some(part_target),
part_hold_back: Some(part_hold_back),
hold_back: DecimalSeconds::new(f64::from(target_duration) * HOLD_BACK_MULTIPLIER)
.ok(),
can_skip_until: DecimalSeconds::new(
f64::from(target_duration) * SKIP_BOUNDARY_MULTIPLIER,
)
.ok(),
can_skip_dateranges: true,
can_block_reload: true,
..Default::default()
})
} else {
None
};
let rendition_reports = if self.part_target_ms.is_some() {
let (last_msn, last_part) = self.current_live_edge();
let mut reports = vec![broadcast_hls::RenditionReport {
uri: media_playlist_name.to_string(),
last_msn,
last_part,
extra_attrs: Vec::new(),
}];
let siblings = locked(&self.rendition_siblings).clone();
for sibling in siblings {
let Some(origin) = sibling.origin.upgrade() else {
continue;
};
if sibling.playlist_uri == self.playlist_uri {
continue;
}
let uri = relative_playlist_uri(
&self.iframe_playlist_name(track_id),
&sibling.playlist_uri,
);
let (last_msn, last_part) = origin.current_live_edge();
reports.push(broadcast_hls::RenditionReport {
uri,
last_msn,
last_part,
extra_attrs: Vec::new(),
});
}
reports
} else {
Vec::new()
};
let playlist = MediaPlaylist {
target_duration,
media_sequence,
discontinuity_sequence: window.discontinuity_sequence,
always_emit_discontinuity_sequence: true,
segments,
open_segment: None,
endlist: false,
extra_tags: Vec::new(),
low_latency,
rendition_reports,
iframes_only: true,
..Default::default()
};
playlist.to_m3u8().map_err(|error| error.to_string())
}
fn resolve_resource(
&self,
name: &str,
now: Timestamp,
await_policy: AwaitPolicy,
) -> EgressResponse<HlsBody> {
if let Some((epoch, public_seq, idx)) = parse_part(name, self.container) {
if epoch.is_some_and(|e| e != self.instance) {
return EgressResponse::NotFound;
}
let Some(seq) = self.internal_seq(public_seq) else {
return EgressResponse::NotFound;
};
if let Some(bytes) = self.trunk.part_bytes(seq, idx) {
return EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache: if epoch.is_some() {
CachePolicy::Immutable
} else {
CachePolicy::NoCache
},
};
}
if self.part_target_ms.is_none() {
return EgressResponse::NotFound;
}
let never_will = self.trunk.last_closed_segment().is_some_and(|c| c >= seq);
if never_will {
return EgressResponse::NotFound;
}
let last_closed = self
.trunk
.last_closed_segment()
.unwrap_or(SegmentSeq::new(0));
if last_closed
.checked_add(FUTURE_PART_SEGMENT_BOUND)
.is_some_and(|limit| seq > limit)
{
return EgressResponse::NotFound;
}
let next_hinted = self
.trunk
.parts_in_segment(seq)
.iter()
.map(|p| p.part_index)
.max()
.map_or(PartIndex::new(0), PartIndex::saturating_next);
if idx != next_hinted {
return EgressResponse::NotFound;
}
return EgressResponse::pending(await_policy, now, now);
}
self.drain();
let (bytes, cache) = match parse_immediate(name, self.container) {
Some(ImmediateResource::CurrentInit) => (self.init_bytes(), CachePolicy::NoCache),
Some(ImmediateResource::VersionedInit { epoch, generation })
if epoch == self.instance =>
{
(
locked(&self.init).bytes_of(generation),
CachePolicy::Immutable,
)
}
Some(ImmediateResource::Segment { epoch, seq })
if epoch.is_none_or(|e| e == self.instance) =>
{
(
self.internal_seq(seq)
.and_then(|s| locked(&self.window).bytes_of(s)),
if epoch.is_some() {
CachePolicy::Immutable
} else {
CachePolicy::NoCache
},
)
}
_ => (None, CachePolicy::NoCache),
};
match bytes {
Some(bytes) => EgressResponse::Ready {
body: HlsBody::Resource(bytes),
cache,
},
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::IFramePlaylist {
track_id,
query,
media_playlist_name,
} => self.resolve_iframe_playlist(
track_id,
query,
media_playlist_name,
now,
await_policy,
),
HlsRequest::Resource { name } => self.resolve_resource(&name, now, await_policy),
}
}
}
pub fn bcp47_language(iso639_2: &str) -> Option<String> {
let code = iso639_2.trim().to_ascii_lowercase();
if code.len() != 3 || !code.bytes().all(|b| b.is_ascii_lowercase()) {
return None;
}
if code == "und" || code == "mul" || code == "mis" || code == "zxx" {
return Some(code);
}
if code.starts_with('q') {
let bytes = code.as_bytes();
if (b'a'..=b't').contains(&bytes[1]) && bytes[2].is_ascii_lowercase() {
return Some(code);
}
}
let lang = iso_639::Language::from_str(&code)?;
if let Some(a2) = lang.alpha2() {
return Some(a2.to_string());
}
if let Some(t) = lang.alpha3_t() {
return Some(t.to_string());
}
lang.alpha3_b().map(|b| b.to_string())
}
fn relative_playlist_uri(from: &str, to: &str) -> String {
let from_dir: Vec<&str> = match from.rfind('/') {
Some(i) => from[..i].split('/').filter(|s| !s.is_empty()).collect(),
None => Vec::new(),
};
let to_segs: Vec<&str> = to.split('/').filter(|s| !s.is_empty()).collect();
let mut common = 0usize;
while common < from_dir.len() && common < to_segs.len() && from_dir[common] == to_segs[common] {
common += 1;
}
let mut parts: Vec<&str> = Vec::new();
parts.extend(std::iter::repeat_n(
"..",
from_dir.len().saturating_sub(common),
));
parts.extend(&to_segs[common..]);
if parts.is_empty() {
to_segs.last().copied().unwrap_or("").to_string()
} else {
parts.join("/")
}
}
#[cfg(test)]
mod tests {
#[test]
fn a_segment_with_fewer_than_two_samples_has_no_measurable_frame_rate() {
assert_eq!(super::segment_frame_rate_millihertz(1, 6000, 90_000), None);
assert_eq!(
super::segment_frame_rate_millihertz(2, 6000, 90_000),
Some(30_000)
);
assert_eq!(super::segment_frame_rate_millihertz(2, 0, 90_000), None);
}
#[test]
fn a_segment_with_broken_timing_does_not_yield_an_implausible_frame_rate() {
assert_eq!(super::segment_frame_rate_millihertz(2, 1, 90_000), None);
assert_eq!(
super::segment_frame_rate_millihertz(300, 90_000, 90_000),
Some(300_000)
);
assert_eq!(
super::segment_frame_rate_millihertz(301, 90_000, 90_000),
None
);
}
use super::*;
use media_plane::trunk::{EventAnchor, TrunkConfig};
use std::time::{Duration, Instant};
use timed_metadata::{MediaTime, TimeAnchor};
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)
.instance(7)
.build()
.expect("both required fields set");
origin.set_init(vec![0xAAu8; 8]);
(trunk, origin, writer)
}
#[test]
fn resource_names_place_every_distinct_id_in_its_own_slot() {
let (_trunk, origin, _writer) = make_origin();
let track = TrackId::new(3);
assert_eq!(origin.instance(), 7);
assert_eq!(
origin.init_name(track, Generation::new(4)),
"init-3-7-4.mp4"
);
assert_eq!(
origin.segment_name(track, SegmentSeq::new(5), "m4s"),
"seg-3-7-5.m4s"
);
assert_eq!(
origin.part_name(track, SegmentSeq::new(5), PartIndex::new(2), "m4s"),
"part-3-7-5.2.m4s"
);
let trunk = Trunk::new(TrunkConfig::new(nz(64), nz(8), nz(8), nz(8), nz(64)));
let offset = HlsOrigin::builder(trunk)
.target_duration_secs(4.0)
.window_segments(nz(4))
.instance(7)
.media_sequence_offset(100)
.build()
.expect("both required fields set");
assert_eq!(
offset.segment_name(track, SegmentSeq::new(5), "m4s"),
"seg-3-7-105.m4s",
"the Media Sequence Number offset applies to the sequence slot only"
);
assert_eq!(
offset.part_name(track, SegmentSeq::new(5), PartIndex::new(2), "m4s"),
"part-3-7-105.2.m4s"
);
assert_eq!(
offset.init_name(track, Generation::new(4)),
"init-3-7-4.mp4",
"the offset never touches the generation"
);
}
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]),
SegmentSeq::new(seq),
Duration::from_secs_f64(duration_secs),
Timestamp::from_nanos(0),
SegmentMeta { discontinuous },
))
.unwrap();
}
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]),
SegmentSeq::new(seg_no),
PartIndex::new(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 trunk_events_render_as_ext_x_daterange() {
let hex = "FC302100000000000000FFF01005000007D27FEF7F7E0020F580C0000000000088B9661D";
let splice_bytes: Vec<u8> = (0..hex.len())
.step_by(2)
.map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
.collect();
let (trunk, origin, writer) = make_origin();
writer.set_time_anchor(TimeAnchor {
pts_90k: 0,
utc_epoch_ms: 1_000_000_000_000,
});
let mut timeline = timed_metadata::Timeline::new();
let ev = timeline.push_scte35(&splice_bytes).unwrap();
let trunk_writer = trunk.writer().expect("first trunk writer");
trunk_writer.publish_event(
ev.clone(),
EventAnchor::Media(ev.at.unwrap_or(MediaTime(0))),
);
writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
writer.note_segment_start(SegmentSeq::new(2), MediaTime(4_000_000));
seg(&writer, 1, 4.0, false);
let playlist = origin
.render_playlist(DEFAULT_TRACK_ID, None)
.expect("render");
assert!(
playlist.contains("#EXT-X-DATERANGE"),
"playlist should contain EXT-X-DATERANGE, got: {playlist}"
);
assert!(
playlist.contains("SCTE35-OUT="),
"daterange should carry SCTE35-OUT attribute: {playlist}"
);
assert!(
playlist.contains("ID=\"2002\""),
"daterange ID should be the event_id: {playlist}"
);
}
#[test]
fn non_scte35_events_are_silently_skipped() {
use timed_metadata::event::{EventKind, SourcePayload};
use timed_metadata::{MediaDuration, TimedEvent};
let (trunk, origin, writer) = make_origin();
writer.set_time_anchor(TimeAnchor {
pts_90k: 0,
utc_epoch_ms: 1_000_000_000_000,
});
let emsg_ev = TimedEvent {
id: Some(7),
kind: EventKind::Unspecified,
at: Some(MediaTime(2_000_000)),
duration: Some(MediaDuration(2_000_000)),
source: SourcePayload::Emsg {
scheme_id_uri: "urn:example".to_string(),
value: "1".to_string(),
raw: vec![0x01, 0x02],
},
};
let trunk_writer = trunk.writer().expect("first trunk writer");
trunk_writer.publish_event(emsg_ev, EventAnchor::Media(MediaTime(2_000_000)));
writer.note_segment_start(SegmentSeq::new(1), MediaTime(0));
writer.note_segment_start(SegmentSeq::new(2), MediaTime(4_000_000));
seg(&writer, 1, 4.0, false);
let playlist = origin
.render_playlist(DEFAULT_TRACK_ID, None)
.expect("render");
assert!(
!playlist.contains("#EXT-X-DATERANGE"),
"no DATERANGE expected for a non-SCTE-35 event: {playlist}"
);
}
#[test]
fn master_playlist_has_stream_inf() {
let m = master_playlist_m3u8("media.m3u8", 5_000_000);
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", 5_000_000);
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-7-1.mp4\""),
"body: {body}"
);
assert!(body.contains("seg-1-7-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-7-2.2.m4s"),
"preload hint for the next part: {body}"
);
}
#[test]
fn preload_hint_present_between_segments() {
let (_trunk, origin, writer) = make_origin();
part(&writer, 1, 0, true);
part(&writer, 1, 1, false);
seg(&writer, 1, 4.0, false);
let body = render_body(&origin);
assert!(body.contains("#EXT-X-PART:"), "body: {body}");
assert!(
body.contains("#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\"part-1-7-2.0.m4s\""),
"hint names part 0 of the next segment: {body}"
);
let first = origin.resolve(
HlsRequest::Resource {
name: "part-1-7-2.0.m4s".to_string(),
},
Timestamp::from_nanos(0),
AwaitPolicy::new(Timestamp::from_nanos(5_000_000_000)),
);
assert!(
matches!(first, EgressResponse::Await { .. }),
"the hinted resource blocks, got {first:?}"
);
part(&writer, 2, 0, true);
let served = resolve_now(
&origin,
HlsRequest::Resource {
name: "part-1-7-2.0.m4s".to_string(),
},
);
assert!(
matches!(served, EgressResponse::Ready { .. }),
"served once produced, got {served:?}"
);
}
#[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-7-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-7-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-7-1.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-7-1.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-7-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-7-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-7-1.m4s"),
"closed segment rendered whole: {body}"
);
assert_eq!(
body.lines().skip(5).collect::<Vec<_>>(),
vec![
"#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,PART-HOLD-BACK=1.5,HOLD-BACK=12,CAN-SKIP-UNTIL=24,CAN-SKIP-DATERANGES=YES",
"#EXT-X-PART-INF:PART-TARGET=0.5",
"#EXT-X-MAP:URI=\"init-1-7-1.mp4\"",
"#EXT-X-PART:DURATION=0.5,URI=\"part-1-7-1.0.m4s\",INDEPENDENT=YES",
"#EXT-X-PART:DURATION=0.5,URI=\"part-1-7-1.1.m4s\"",
"#EXTINF:4,",
"seg-1-7-1.m4s",
"#EXT-X-PRELOAD-HINT:TYPE=PART,URI=\"part-1-7-2.0.m4s\"",
],
"{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:0"),
"fresh live playlist always carries DISCONTINUITY-SEQUENCE:0 (Apple 8.17): {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:0"),
"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 published_target_duration_matches_the_rendered_playlist() {
let (_trunk, origin, writer) = make_origin(); assert_eq!(origin.published_target_duration(), Duration::from_secs(4));
seg(&writer, 1, 7.5, false);
let _ = resolve_now(
&origin,
HlsRequest::Playlist {
track_id: DEFAULT_TRACK_ID,
query: BlockingQuery::default(),
},
);
assert_eq!(origin.published_target_duration(), Duration::from_secs(8));
seg(&writer, 2, 20.0, false);
assert_eq!(origin.published_target_duration(), Duration::from_secs(8));
}
#[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,
hls_skip: None,
hls_primary: 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,
hls_skip: None,
hls_primary: 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,
hls_skip: None,
hls_primary: 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),
hls_skip: None,
hls_primary: None,
},
},
);
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::NoCache);
}
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
.instance(7)
.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:")
&& let Some(next) = lines.get(i + 1)
&& !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-7-1.ts"), "body: {body}");
assert_version_present_and_matches_derivation(&body);
let uris = segment_uris(&body);
assert_eq!(uris, vec!["seg-1-7-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-7-1.ts"), "body: {body}");
assert!(body.contains("part-1-7-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 (_epoch, 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.value() 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-7-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-7-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-7-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-7-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-7-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-7-1.m4s"), "body: {body}");
assert!(body.contains("part-1-7-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 (_epoch, _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.value() 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))
.instance(7)
.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)
.instance(7)
.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);
}
#[test]
fn closed_segments_snapshot_matches_published_segments_ascending() {
let (_trunk, origin, writer) = make_origin();
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![1u8; 8]),
SegmentSeq::new(1),
Duration::from_secs_f64(4.0),
Timestamp::from_nanos(0),
SegmentMeta {
discontinuous: false,
},
))
.unwrap();
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![2u8; 8]),
SegmentSeq::new(2),
Duration::from_secs_f64(4.0),
Timestamp::from_nanos(4_000_000_000),
SegmentMeta {
discontinuous: true,
},
))
.unwrap();
let snapshot = origin.closed_segments();
assert_eq!(snapshot.len(), 2);
assert_eq!(snapshot[0].sequence_number, SegmentSeq::new(1));
assert_eq!(snapshot[0].start_ns, 0);
assert!(!snapshot[0].discontinuous);
assert_eq!(snapshot[1].sequence_number, SegmentSeq::new(2));
assert_eq!(snapshot[1].start_ns, 4_000_000_000);
assert!(snapshot[1].discontinuous);
}
#[test]
fn closed_segments_reflects_window_eviction() {
let (_trunk, origin, writer) = make_origin(); for seq in 1..=5 {
seg(&writer, seq, 4.0, false);
}
let snapshot = origin.closed_segments();
let seqs: Vec<u32> = snapshot.iter().map(|s| s.sequence_number.value()).collect();
assert_eq!(
seqs,
vec![2, 3, 4, 5],
"oldest segment 1 must have rolled off"
);
}
fn window_entry(seq: u32, tag: u8, discontinuous: bool) -> SegmentEntry {
SegmentEntry::new(
Bytes::from(vec![tag; 4]),
SegmentSeq::new(seq),
Duration::from_secs(4),
Timestamp::from_nanos(0),
SegmentMeta { discontinuous },
)
}
fn window_seqs(w: &Window) -> Vec<(u32, bool)> {
w.segments
.iter()
.map(|s| (s.sequence_number.value(), s.discontinuous))
.collect()
}
#[test]
fn window_restarts_on_a_repeated_or_decreasing_sequence_number() {
let mut w = Window::new(nz(6));
w.push(window_entry(5, 0xA5, false), Generation::new(1), None, None);
w.push(window_entry(6, 0xA6, true), Generation::new(1), None, None);
assert_eq!(window_seqs(&w), vec![(5, false), (6, true)]);
w.push(window_entry(3, 0xB3, false), Generation::new(1), None, None);
assert_eq!(window_seqs(&w), vec![(3, true)]);
assert_eq!(w.discontinuity_sequence, 1);
assert_eq!(
w.bytes_of(SegmentSeq::new(3)),
Some(Bytes::from(vec![0xB3; 4]))
);
assert_eq!(w.bytes_of(SegmentSeq::new(5)), None);
assert_eq!(w.bytes_of(SegmentSeq::new(6)), None);
w.push(window_entry(3, 0xC3, false), Generation::new(1), None, None);
assert_eq!(window_seqs(&w), vec![(3, true)]);
assert_eq!(
w.bytes_of(SegmentSeq::new(3)),
Some(Bytes::from(vec![0xC3; 4]))
);
}
#[test]
fn window_handles_u32_max_sequence_numbers() {
let mut w = Window::new(nz(4));
w.push(
window_entry(u32::MAX - 1, 1, false),
Generation::new(1),
None,
None,
);
w.push(
window_entry(u32::MAX, 2, false),
Generation::new(1),
None,
None,
);
assert_eq!(
window_seqs(&w),
vec![(u32::MAX - 1, false), (u32::MAX, false)]
);
w.push(
window_entry(u32::MAX, 3, false),
Generation::new(1),
None,
None,
);
assert_eq!(window_seqs(&w), vec![(u32::MAX, true)]);
assert_eq!(
w.bytes_of(SegmentSeq::new(u32::MAX)),
Some(Bytes::from(vec![3; 4]))
);
}
#[test]
fn window_capacity_is_never_below_three_segments() {
let mut w = Window::new(nz(1));
for seq in 1..=5 {
w.push(window_entry(seq, 0, false), Generation::new(1), None, None);
}
assert_eq!(window_seqs(&w), vec![(3, false), (4, false), (5, false)]);
}
#[test]
fn master_playlist_reports_typed_bandwidth_overflow() {
let (_trunk, origin, writer) = make_origin();
writer
.publish_segment(SegmentEntry::new(
Bytes::from(vec![1u8; 8]),
SegmentSeq::new(1),
Duration::from_nanos(10),
Timestamp::from_nanos(0),
SegmentMeta {
discontinuous: false,
},
))
.unwrap();
assert_eq!(
origin.master_playlist("media.m3u8"),
Err(HlsMasterError::BandwidthOverflow {
bandwidth: 6_400_000_000
})
);
}
#[test]
fn segment_bandwidth_is_rounded_up_and_rejects_degenerate_input() {
assert_eq!(
segment_bandwidth_bps(1001, Duration::from_secs(3)),
Some(2670)
);
assert_eq!(
segment_bandwidth_bps(1_500_000, Duration::from_secs(1)),
Some(12_000_000)
);
assert_eq!(segment_bandwidth_bps(10, Duration::ZERO), None);
assert_eq!(
segment_bandwidth_bps(usize::MAX, Duration::from_nanos(1)),
None
);
}
#[test]
fn bytes_of_addresses_exactly_the_matching_entry() {
let mut w = Window::new(nz(6));
for (seq, tag) in [(5u32, 0x15u8), (6, 0x16), (7, 0x17)] {
w.push(
window_entry(seq, tag, false),
Generation::new(1),
None,
None,
);
}
assert_eq!(
w.bytes_of(SegmentSeq::new(5)),
Some(Bytes::from(vec![0x15; 4]))
);
assert_eq!(
w.bytes_of(SegmentSeq::new(6)),
Some(Bytes::from(vec![0x16; 4]))
);
assert_eq!(
w.bytes_of(SegmentSeq::new(7)),
Some(Bytes::from(vec![0x17; 4]))
);
assert_eq!(w.bytes_of(SegmentSeq::new(4)), None);
assert_eq!(w.bytes_of(SegmentSeq::new(8)), None);
assert_eq!(w.bytes_of(SegmentSeq::new(u32::MAX)), None);
}
fn segment_request(name: &str) -> HlsRequest {
HlsRequest::Resource {
name: name.to_string(),
}
}
#[test]
fn a_request_with_nothing_new_does_not_take_the_cursor_lock() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
EgressResponse::Ready { .. }
));
let guard = locked(&origin.cursor);
let (tx, rx) = std::sync::mpsc::channel();
std::thread::scope(|s| {
s.spawn(|| {
let _ = tx.send(resolve_now(&origin, segment_request("seg-1-7-1.m4s")));
});
let got = rx.recv_timeout(Duration::from_secs(10));
drop(guard); assert!(
matches!(got, Ok(EgressResponse::Ready { .. })),
"request blocked on the cursor lock: {got:?}"
);
});
}
#[test]
fn a_segment_published_after_a_drain_is_served_by_the_next_request() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
EgressResponse::Ready { .. }
));
assert_eq!(
resolve_now(&origin, segment_request("seg-1-7-2.m4s")),
EgressResponse::NotFound
);
seg(&writer, 2, 4.0, false);
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-2.m4s")),
EgressResponse::Ready { .. }
));
}
#[test]
fn a_poisoned_lock_does_not_cascade_into_later_requests() {
let (_trunk, origin, writer) = make_origin();
seg(&writer, 1, 4.0, false);
std::thread::scope(|s| {
let origin = &origin;
for which in 0..4 {
let handle = s.spawn(move || {
match which {
0 => drop(origin.window.lock().map(|_g| panic!("poison window"))),
1 => drop(origin.cursor.lock().map(|_g| panic!("poison cursor"))),
2 => drop(origin.init.lock().map(|_g| panic!("poison init"))),
_ => drop(origin.codecs.lock().map(|_g| panic!("poison codecs"))),
}
});
assert!(handle.join().is_err(), "setup thread {which} must panic");
}
});
assert!(origin.window.is_poisoned());
assert!(origin.cursor.is_poisoned());
assert!(origin.init.is_poisoned());
assert!(origin.codecs.is_poisoned());
seg(&writer, 2, 4.0, false);
assert_eq!(origin.closed_segments().len(), 2);
assert!(origin.init_bytes().is_some());
assert!(origin.master_playlist("media.m3u8").is_ok());
origin.set_init(vec![0xCC; 8]);
assert!(render_body(&origin).contains("seg-1-7-2.m4s"));
assert!(matches!(
resolve_now(&origin, segment_request("seg-1-7-1.m4s")),
EgressResponse::Ready { .. }
));
}
}