use alloc::collections::btree_map::Entry;
use alloc::collections::{BTreeMap, BTreeSet, VecDeque};
use alloc::vec::Vec;
use core::marker::PhantomData;
use broadcast_common::{Demand, Serialize, Stage, Timestamp, Unpackage};
use mpeg_pes::{PesAssembler, PesPacket};
use mpeg_ts::resync::TsResync;
use mpeg_ts::ts::{SectionReassembler, TS_PACKET_SIZE, TsPacket};
use crate::aac_asc::{AdtsHeader, AudioSpecificConfig, SamplingFrequencyIndex, parse_adts_header};
use crate::ac3::{
AC3_SAMPLES_PER_SYNCFRAME, Ac3SyncframeInfo, Ec3SpecificBox, Ec3SyncframeInfo,
split_ac3_syncframes, split_eac3_syncframes,
};
use crate::annexb::{annexb_to_length_prefixed, iter_annexb_nals};
use crate::avc_config::{AVCConfigurationBox, AVCDecoderConfigurationRecord};
use crate::dts::{DtsCoreFrameInfo, split_dts_core_frames};
use crate::error::{Error, Result};
use crate::hevc_config::{HEVCConfigurationBox, HEVCDecoderConfigurationRecord};
use crate::media::{Media, PcrSample, Track};
use crate::mp4esds::{
DecoderConfigDescriptor, DecoderSpecificInfo, ESDescriptor, EsdsBox, SLConfigDescriptor,
};
use crate::mpeg_legacy::{Mpeg2SeqHeader, MpegAudioFrameHeader};
use crate::mpegh::{MHADecoderConfigurationRecord, find_mpegh3da_config};
use crate::nal::{NalCodec, access_unit_is_rap, is_keyframe_nal, nal_unit_type};
use crate::nalu_types::{AvcPps, AvcSps, HevcNalArray, HevcNalUnit};
use crate::pipeline::{CodecConfig, DataCarriage, Provenance, Sample, SampleFlags, TrackSpec};
const PAT_PID: u16 = 0x0000;
const TABLE_ID_PAT: u8 = 0x00;
const TABLE_ID_PMT: u8 = 0x02;
const SECTION_HEADER_LEN: usize = 8;
const VERSION_NUMBER_MASK: u8 = 0x1F;
const CURRENT_NEXT_INDICATOR_BIT: u8 = 0x01;
const CRC32_LEN: usize = 4;
const SECTION_SYNTAX_INDICATOR_BIT: u8 = 0x80;
const SECTION_LENGTH_HI_MASK: u8 = 0x0F;
const PID_HI_MASK: u8 = 0x1F;
const PAT_ENTRY_LEN: usize = 4;
const INFO_LENGTH_HI_MASK: u8 = 0x0F;
const NETWORK_PROGRAM_NUMBER: u16 = 0x0000;
const NULL_PACKET_PID: u16 = 0x1FFF;
const MAX_UNATTRIBUTED_BYTES: usize = 4 * 1024 * 1024;
const TS_MAX_PAYLOAD_BYTES: usize = TS_PACKET_SIZE - 4;
const PES_LENGTH_FIELD_END: usize = 6;
const MAX_PES_BUFFER_BYTES: usize = 4 * 1024 * 1024;
const MAX_PROBE_BACKLOG_BYTES: usize = 4 * 1024 * 1024;
const STREAM_TYPE_MPEG2_VIDEO: u8 = 0x02;
const STREAM_TYPE_MPEG1_AUDIO: u8 = 0x03;
const STREAM_TYPE_MPEG2_AUDIO: u8 = 0x04;
const STREAM_TYPE_AVC: u8 = 0x1B;
const STREAM_TYPE_HEVC: u8 = 0x24;
const STREAM_TYPE_AAC_ADTS: u8 = 0x0F;
const STREAM_TYPE_AC3: u8 = 0x81;
const STREAM_TYPE_EAC3: u8 = 0x87;
const STREAM_TYPE_DTS_82: u8 = 0x82;
const STREAM_TYPE_DTS_85: u8 = 0x85;
const STREAM_TYPE_DTS_8A: u8 = 0x8A;
const STREAM_TYPE_MPEGH: u8 = 0x2D;
const STREAM_TYPE_PES_PRIVATE: u8 = 0x06;
const STREAM_TYPE_METADATA_PES: u8 = 0x15;
const DESC_TAG_AC3: u8 = 0x6A;
const DESC_TAG_ENHANCED_AC3: u8 = 0x7A;
const DESC_TAG_DTS: u8 = 0x7B;
const STREAM_TYPE_PRIVATE_SECTIONS: u8 = 0x05;
const STREAM_TYPE_DSMCC_TYPE_A: u8 = 0x0A;
const STREAM_TYPE_DSMCC_TYPE_B: u8 = 0x0B;
const STREAM_TYPE_DSMCC_TYPE_C: u8 = 0x0C;
const STREAM_TYPE_DSMCC_TYPE_D: u8 = 0x0D;
const STREAM_TYPE_DSMCC_SYNC_DOWNLOAD: u8 = 0x14;
const STREAM_TYPE_SCTE35: u8 = 0x86;
const NAL_LENGTH_SIZE_MINUS_ONE: u8 = 3;
const H264_NAL_SPS: u8 = 7;
const H264_NAL_PPS: u8 = 8;
const H264_NAL_TYPE_MASK: u8 = 0x1F;
const H265_NAL_VPS: u8 = 32;
const H265_NAL_SPS: u8 = 33;
const H265_NAL_PPS: u8 = 34;
const HVCC_CONFIGURATION_VERSION: u8 = 1;
const HVCC_CONSTANT_FRAME_RATE_UNSPEC: u8 = 0;
const HVCC_NUM_TEMPORAL_LAYERS: u8 = 1;
const HVCC_PARALLELISM_TYPE_UNKNOWN: u8 = 0;
const HVCC_AVG_FRAME_RATE_UNSPEC: u16 = 0;
const HVCC_MIN_SPATIAL_SEGMENTATION_UNSPEC: u16 = 0;
const OTI_MPEG4_AUDIO: u8 = 0x40;
pub(crate) const OTI_MPEG2_VIDEO_MAIN: u8 = 0x61;
const OTI_MPEG1_AUDIO: u8 = 0x6B;
const OTI_MPEG2_AUDIO: u8 = 0x69;
const STREAM_TYPE_AUDIO: u8 = 0x05;
pub(crate) const STREAM_TYPE_VISUAL: u8 = 0x04;
const ESDS_ES_ID: u16 = 1;
pub(crate) const ESDS_VIDEO_ES_ID: u16 = 2;
const AUDIO_SAMPLE_SIZE_BITS: u16 = 16;
const MPEGH_REFERENCE_CHANNEL_LAYOUT_UNSPECIFIED: u8 = 0;
const MPEGH_CHANNEL_COUNT_UNSPECIFIED: u16 = 0;
const MPEGH_SAMPLE_RATE_UNSPECIFIED: u32 = 0;
const VIDEO_TIMESCALE: u32 = 90_000;
const AAC_SAMPLES_PER_FRAME: u32 = 1024;
const ADTS_HEADER_SIZE: usize = 7;
pub(crate) const MPEG2_PICTURE_START_CODE: u8 = 0x00;
pub(crate) const MPEG2_PICTURE_CODING_TYPE_I: u8 = 0x01;
const TS_WRAP: u64 = broadcast_common::clock33::WRAP_33BIT;
const TS_WRAP_MASK: u64 = TS_WRAP - 1;
const MAX_FRAME_PERIOD_TICKS: u64 = VIDEO_TIMESCALE as u64;
const FRAME_PERIOD_WINDOW: usize = 5;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Codec {
H264,
Hevc,
Mpeg2Video,
MpegAudio(bool),
Aac,
Ac3,
Eac3,
Dts,
MpegH,
Data(u8),
}
impl Codec {
fn from_stream_type(stream_type: u8) -> Self {
match stream_type {
STREAM_TYPE_MPEG2_VIDEO => Codec::Mpeg2Video,
STREAM_TYPE_MPEG1_AUDIO => Codec::MpegAudio(false),
STREAM_TYPE_MPEG2_AUDIO => Codec::MpegAudio(true),
STREAM_TYPE_AVC => Codec::H264,
STREAM_TYPE_HEVC => Codec::Hevc,
STREAM_TYPE_AAC_ADTS => Codec::Aac,
STREAM_TYPE_AC3 => Codec::Ac3,
STREAM_TYPE_EAC3 => Codec::Eac3,
STREAM_TYPE_DTS_82 | STREAM_TYPE_DTS_85 | STREAM_TYPE_DTS_8A => Codec::Dts,
STREAM_TYPE_MPEGH => Codec::MpegH,
_ => Codec::Data(stream_type),
}
}
fn refine_with_descriptors(self, stream_type: u8, descriptors: &[u8]) -> Self {
if !matches!(
stream_type,
STREAM_TYPE_PES_PRIVATE | STREAM_TYPE_METADATA_PES
) {
return self;
}
let mut off = 0usize;
while off + 2 <= descriptors.len() {
let tag = descriptors[off];
let len = descriptors[off + 1] as usize;
match tag {
DESC_TAG_AC3 => return Codec::Ac3,
DESC_TAG_ENHANCED_AC3 => return Codec::Eac3,
DESC_TAG_DTS => return Codec::Dts,
_ => {}
}
off += 2 + len;
}
self
}
}
fn data_carriage(stream_type: u8) -> DataCarriage {
match stream_type {
STREAM_TYPE_PRIVATE_SECTIONS
| STREAM_TYPE_DSMCC_TYPE_A
| STREAM_TYPE_DSMCC_TYPE_B
| STREAM_TYPE_DSMCC_TYPE_C
| STREAM_TYPE_DSMCC_TYPE_D
| STREAM_TYPE_DSMCC_SYNC_DOWNLOAD
| STREAM_TYPE_SCTE35 => DataCarriage::Sections,
_ => DataCarriage::Pes,
}
}
fn unwrap_ts(prev_unwrapped: i128, prev_raw: u64, raw: u64) -> i128 {
broadcast_common::clock33::unwrap_delta(prev_unwrapped, prev_raw, raw)
}
fn rescale_to_track(anchor_90k: i128, timescale: u32) -> i64 {
let ts = timescale.max(1) as i128;
let scaled = if ts == VIDEO_TIMESCALE as i128 {
anchor_90k
} else {
(anchor_90k * ts).div_euclid(VIDEO_TIMESCALE as i128)
};
to_ticks(scaled)
}
pub(crate) fn mpeg2_is_sync(au: &[u8]) -> bool {
let mut i = 0usize;
while i + 4 <= au.len() {
if au[i] == 0x00 && au[i + 1] == 0x00 && au[i + 2] == 0x01 {
let code = au[i + 3];
if code == crate::mpeg_legacy::SEQUENCE_HEADER_CODE[3] {
return true;
}
if code == MPEG2_PICTURE_START_CODE && i + 6 <= au.len() {
let pct = (au[i + 5] >> 3) & 0x07;
return pct == MPEG2_PICTURE_CODING_TYPE_I;
}
}
i += 1;
}
false
}
#[cfg(test)]
mod probe_counter {
thread_local! {
static SYNC_PROBES: core::cell::Cell<u64> = const { core::cell::Cell::new(0) };
}
pub(super) fn record() {
SYNC_PROBES.with(|c| c.set(c.get().wrapping_add(1)));
}
pub(super) fn read() -> u64 {
SYNC_PROBES.with(|c| c.get())
}
}
#[cfg(test)]
fn record_sync_probe() {
probe_counter::record();
}
#[cfg(test)]
fn record_probe_pass() {
probe_counter::record();
}
#[cfg(not(test))]
#[inline(always)]
fn record_probe_pass() {}
#[cfg(not(test))]
#[inline(always)]
fn record_sync_probe() {}
#[cfg(test)]
fn sync_probes() -> u64 {
probe_counter::read()
}
fn find_mpeg_audio_sync(data: &[u8]) -> Option<(usize, MpegAudioFrameHeader)> {
let mut off = 0usize;
while off + 4 <= data.len() {
record_sync_probe();
if let Ok(hdr) = MpegAudioFrameHeader::parse(&data[off..]) {
return Some((off, hdr));
}
off += 1;
}
None
}
fn split_mpeg_audio_frames(payload: &[u8]) -> Vec<&[u8]> {
let mut frames = Vec::new();
let mut off = 0usize;
while off + 4 <= payload.len() {
let Some((sync_off, hdr)) = find_mpeg_audio_sync(&payload[off..]) else {
break;
};
off += sync_off;
let flen = hdr.frame_length;
if flen < 4 || off + flen > payload.len() {
break;
}
frames.push(&payload[off..off + flen]);
off += flen;
}
frames
}
fn find_adts_sync(data: &[u8]) -> Option<(usize, AdtsHeader)> {
let mut off = 0usize;
while off + ADTS_HEADER_SIZE <= data.len() {
record_sync_probe();
if let Ok(hdr) = parse_adts_header(&data[off..]) {
return Some((off, hdr));
}
off += 1;
}
None
}
fn split_adts_frames(payload: &[u8]) -> Vec<(&[u8], AdtsHeader)> {
let mut frames = Vec::new();
let mut off = 0usize;
while off + ADTS_HEADER_SIZE <= payload.len() {
let Some((sync_off, hdr)) = find_adts_sync(&payload[off..]) else {
break;
};
off += sync_off;
let frame_len = hdr.frame_length as usize;
if frame_len < ADTS_HEADER_SIZE || off + frame_len > payload.len() {
break;
}
frames.push((&payload[off..off + frame_len], hdr));
off += frame_len;
}
frames
}
const ADTS_CRC_SIZE: usize = 2;
fn adts_header_len(hdr: &AdtsHeader) -> usize {
if hdr.protection_absent {
ADTS_HEADER_SIZE
} else {
ADTS_HEADER_SIZE + ADTS_CRC_SIZE
}
}
fn parse_pat(section: &[u8]) -> Result<Vec<(u16, u16)>> {
if section.first().copied() != Some(TABLE_ID_PAT) {
return Ok(Vec::new());
}
let body = section_body(section, "PAT")?;
let mut programs = Vec::new();
let mut off = 0usize;
while off + PAT_ENTRY_LEN <= body.len() {
let program_number = u16::from_be_bytes([body[off], body[off + 1]]);
let pid = (((body[off + 2] & PID_HI_MASK) as u16) << 8) | body[off + 3] as u16;
if program_number != NETWORK_PROGRAM_NUMBER {
programs.push((program_number, pid));
}
off += PAT_ENTRY_LEN;
}
Ok(programs)
}
struct PmtSectionHeader {
program_number: u16,
version: u8,
current_next: bool,
section_number: u8,
last_section_number: u8,
pcr_pid: u16,
}
fn parse_pmt_section_header(section: &[u8]) -> Result<PmtSectionHeader> {
if section.first().copied() != Some(TABLE_ID_PMT) {
return Err(Error::InvalidValue {
field: "table_id",
value: section.first().copied().unwrap_or(0) as u64,
reason: "not a PMT section",
});
}
if section.len() < SECTION_HEADER_LEN {
return Err(Error::BufferTooShort {
need: SECTION_HEADER_LEN,
have: section.len(),
what: "PMT section header",
});
}
Ok(PmtSectionHeader {
program_number: u16::from_be_bytes([section[3], section[4]]),
version: (section[5] >> 1) & VERSION_NUMBER_MASK,
current_next: section[5] & CURRENT_NEXT_INDICATOR_BIT != 0,
section_number: section[6],
last_section_number: section[7],
pcr_pid: (((section[8] & PID_HI_MASK) as u16) << 8) | section[9] as u16,
})
}
fn parse_pmt(section: &[u8]) -> Result<Vec<(u16, Codec, Vec<u8>)>> {
if section.first().copied() != Some(TABLE_ID_PMT) {
return Ok(Vec::new());
}
let body = section_body(section, "PMT")?;
if body.len() < 4 {
return Err(Error::BufferTooShort {
need: 4,
have: body.len(),
what: "PMT program-info prefix",
});
}
let program_info_length = (((body[2] & INFO_LENGTH_HI_MASK) as usize) << 8) | body[3] as usize;
let mut off = 4 + program_info_length;
let mut out = Vec::new();
while off + 5 <= body.len() {
let stream_type = body[off];
let es_pid = (((body[off + 1] & PID_HI_MASK) as u16) << 8) | body[off + 2] as u16;
let es_info_length =
(((body[off + 3] & INFO_LENGTH_HI_MASK) as usize) << 8) | body[off + 4] as usize;
let desc_start = off + 5;
let desc_end = (desc_start + es_info_length).min(body.len());
let descriptors = body[desc_start..desc_end].to_vec();
let codec =
Codec::from_stream_type(stream_type).refine_with_descriptors(stream_type, &descriptors);
out.push((es_pid, codec, descriptors));
off += 5 + es_info_length;
}
Ok(out)
}
fn section_body<'a>(section: &'a [u8], what: &'static str) -> Result<&'a [u8]> {
if section.len() < SECTION_HEADER_LEN + CRC32_LEN {
return Err(Error::BufferTooShort {
need: SECTION_HEADER_LEN + CRC32_LEN,
have: section.len(),
what,
});
}
let section_length =
(((section[1] & SECTION_LENGTH_HI_MASK) as usize) << 8) | section[2] as usize;
let total = 3 + section_length;
let end = total.min(section.len());
if end < SECTION_HEADER_LEN + CRC32_LEN {
return Err(Error::BufferTooShort {
need: SECTION_HEADER_LEN + CRC32_LEN,
have: end,
what,
});
}
Ok(§ion[SECTION_HEADER_LEN..end - CRC32_LEN])
}
fn psi_section_crc_ok(section: &[u8]) -> bool {
if section.len() < SECTION_HEADER_LEN + CRC32_LEN {
return false;
}
if section[1] & SECTION_SYNTAX_INDICATOR_BIT == 0 {
return false;
}
let section_length =
(((section[1] & SECTION_LENGTH_HI_MASK) as usize) << 8) | section[2] as usize;
let total = 3 + section_length;
if total > section.len() || total < SECTION_HEADER_LEN + CRC32_LEN {
return false;
}
let (covered, trailer) = section[..total].split_at(total - CRC32_LEN);
let declared = u32::from_be_bytes([trailer[0], trailer[1], trailer[2], trailer[3]]);
broadcast_common::crc32_mpeg2::compute(covered) == declared
}
fn section_current_next(section: &[u8]) -> bool {
section
.get(5)
.is_some_and(|b| b & CURRENT_NEXT_INDICATOR_BIT != 0)
}
struct BufferedAu {
data: Vec<u8>,
pts_uw: i128,
dts_uw: i128,
}
enum ConfigProbe {
H264 {
sps: Option<Vec<u8>>,
pps: Option<Vec<u8>>,
},
Hevc {
vps: Option<Vec<u8>>,
sps: Option<Vec<u8>>,
pps: Option<Vec<u8>>,
},
Mpeg2Video,
MpegAudio {
is_mpeg2: bool,
},
Aac,
Ac3,
Eac3,
Dts,
MpegH,
Data,
}
fn initial_probe(codec: Codec) -> ConfigProbe {
match codec {
Codec::H264 => ConfigProbe::H264 {
sps: None,
pps: None,
},
Codec::Hevc => ConfigProbe::Hevc {
vps: None,
sps: None,
pps: None,
},
Codec::Mpeg2Video => ConfigProbe::Mpeg2Video,
Codec::MpegAudio(is_mpeg2) => ConfigProbe::MpegAudio { is_mpeg2 },
Codec::Aac => ConfigProbe::Aac,
Codec::Ac3 => ConfigProbe::Ac3,
Codec::Eac3 => ConfigProbe::Eac3,
Codec::Dts => ConfigProbe::Dts,
Codec::MpegH => ConfigProbe::MpegH,
Codec::Data(_) => ConfigProbe::Data,
}
}
#[derive(Clone, Copy)]
enum VideoCodec {
H264,
Hevc,
Mpeg2,
}
enum AudioKind {
Aac,
Ac3,
Eac3,
Dts,
MpegAudio { samples_per_frame: u32 },
}
#[derive(Default)]
struct AudioAnchor {
seed: Option<AudioAnchorSeed>,
}
#[derive(Clone, Copy)]
struct AudioAnchorSeed {
next_dts: i64,
next_pts: i64,
anchor_dts_uw: i128,
anchor_pts_uw: i128,
ticks_since_anchor: i64,
}
const AUDIO_REANCHOR_THRESHOLD_MS: i128 = 20;
pub(crate) fn audio_discontinuity_threshold_90k(sample_rate: u32) -> i128 {
let one_sample_period = (VIDEO_TIMESCALE as u128).div_ceil(sample_rate.max(1) as u128) as i128;
let ms_bound = (VIDEO_TIMESCALE as i128 * AUDIO_REANCHOR_THRESHOLD_MS) / 1000;
ms_bound.max(one_sample_period * 2)
}
struct PendingOneBehind {
data: Vec<u8>,
is_sync: bool,
pts_uw: i128,
dts_uw: i128,
}
fn to_ticks(uw: i128) -> i64 {
uw.clamp(i64::MIN as i128, i64::MAX as i128) as i64
}
fn ts_provenance(dts: i64, pts: i64) -> Provenance {
Provenance {
wire_dts: Some((dts as u64) % TS_WRAP),
wire_pts: Some((pts as u64) % TS_WRAP),
}
}
enum LiveKind {
Video {
pending: Option<PendingOneBehind>,
last_duration: u32,
codec: VideoCodec,
},
Audio {
sample_rate: u32,
kind: AudioKind,
anchor: AudioAnchor,
},
Data {
pending: Option<PendingOneBehind>,
last_duration: u32,
},
MpegH {
pending: Option<PendingOneBehind>,
last_duration: u32,
},
Section,
}
struct LiveTrack {
track_id: u32,
kind: LiveKind,
config: CodecConfig,
timescale: u32,
parameter_sets: ParameterSets,
config_header: Option<Vec<u8>>,
}
enum TrackState {
Probing {
probe: ConfigProbe,
backlog: Vec<BufferedAu>,
},
Parked {
config: CodecConfig,
timescale: u32,
kind: LiveKind,
backlog: Vec<BufferedAu>,
},
Live(LiveTrack),
Abandoned,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DiscontinuityVerdict {
Forward,
Lifted,
}
#[derive(Default)]
struct WrapState {
initialized: bool,
dts_seen_real: bool,
pts_seen_real: bool,
prev_dts_raw: u64,
prev_dts_uw: i128,
prev_pts_raw: u64,
prev_pts_uw: i128,
discontinuity_offset: i128,
}
impl WrapState {
fn push(&mut self, raw_pts: u64, raw_dts: u64) -> (i128, i128) {
if !self.initialized {
self.initialized = true;
self.dts_seen_real = raw_dts != 0;
self.pts_seen_real = raw_pts != 0;
self.prev_dts_raw = raw_dts;
self.prev_dts_uw = raw_dts as i128;
self.prev_pts_raw = raw_pts;
self.prev_pts_uw = raw_pts as i128;
return (self.prev_pts_uw, self.prev_dts_uw);
}
let dts_uw = if self.dts_seen_real {
unwrap_ts(self.prev_dts_uw, self.prev_dts_raw, raw_dts)
} else {
self.dts_seen_real = raw_dts != 0;
raw_dts as i128
};
let pts_uw = if self.pts_seen_real {
unwrap_ts(self.prev_pts_uw, self.prev_pts_raw, raw_pts)
} else {
self.pts_seen_real = raw_pts != 0;
raw_pts as i128
};
self.prev_dts_raw = raw_dts;
self.prev_dts_uw = dts_uw;
self.prev_pts_raw = raw_pts;
self.prev_pts_uw = pts_uw;
(
pts_uw.saturating_add(self.discontinuity_offset),
dts_uw.saturating_add(self.discontinuity_offset),
)
}
fn rebase_at_discontinuity(&mut self, raw_dts: u64, step: i128) -> DiscontinuityVerdict {
if !self.initialized {
return DiscontinuityVerdict::Forward;
}
let next_uw = if self.dts_seen_real {
unwrap_ts(self.prev_dts_uw, self.prev_dts_raw, raw_dts)
} else {
raw_dts as i128
};
let last_emitted = self.prev_dts_uw.saturating_add(self.discontinuity_offset);
let next_emitted = next_uw.saturating_add(self.discontinuity_offset);
let slack = step.max(1);
if next_emitted + slack > last_emitted {
return DiscontinuityVerdict::Forward;
}
let lift = last_emitted
.saturating_sub(next_emitted)
.saturating_add(slack);
self.discontinuity_offset = self.discontinuity_offset.saturating_add(lift);
self.prev_dts_raw = raw_dts;
self.prev_dts_uw = next_uw;
self.prev_pts_raw = raw_dts;
self.prev_pts_uw = next_uw;
self.dts_seen_real = true;
self.pts_seen_real = true;
DiscontinuityVerdict::Lifted
}
}
enum Carrier {
Pes(PesAssembler),
Section(SectionReassembler),
}
fn initial_carrier(codec: Codec) -> Carrier {
match codec {
Codec::Data(stream_type) if data_carriage(stream_type) == DataCarriage::Sections => {
Carrier::Section(SectionReassembler::default())
}
_ => Carrier::Pes(PesAssembler::new()),
}
}
struct StreamState {
codec: Codec,
descriptors: Vec<u8>,
carrier: Carrier,
pes_bytes: usize,
fallback: (u64, u64),
frame_period: Option<u64>,
last_stamped_dts: Option<u64>,
units_since_stamped: u64,
has_any: bool,
wrap: WrapState,
track: Option<TrackState>,
last_frame_period: i128,
recent_steps: Vec<u64>,
pending_rebase: bool,
current_pes_intact: bool,
pes_declared_len: u16,
pes_received: usize,
backlog_bytes: usize,
}
#[allow(clippy::too_many_arguments)]
fn advance_one_behind(
pending: &mut Option<PendingOneBehind>,
last_duration: &mut u32,
data: Vec<u8>,
is_sync: bool,
pts_uw: i128,
dts_uw: i128,
duration_from_pts: bool,
track_id: u32,
events: &mut VecDeque<DemuxEvent>,
) {
if let Some(prev) = pending.take() {
let duration = if duration_from_pts {
(pts_uw - prev.pts_uw).max(0) as u32
} else {
(dts_uw - prev.dts_uw).max(0) as u32
};
*last_duration = duration;
let dts = to_ticks(prev.dts_uw);
let pts = to_ticks(prev.pts_uw);
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample {
data: prev.data.into(),
dts: Some(dts),
pts: Some(pts),
duration: Some(duration),
flags: SampleFlags::new(prev.is_sync),
provenance: Some(ts_provenance(dts, pts)),
},
});
}
*pending = Some(PendingOneBehind {
data,
is_sync,
pts_uw,
dts_uw,
});
}
fn flush_one_behind(
pending: &mut Option<PendingOneBehind>,
last_duration: u32,
track_id: u32,
events: &mut VecDeque<DemuxEvent>,
) {
if let Some(p) = pending.take() {
let dts = to_ticks(p.dts_uw);
let pts = to_ticks(p.pts_uw);
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample {
data: p.data.into(),
dts: Some(dts),
pts: Some(pts),
duration: Some(last_duration),
flags: SampleFlags::new(p.is_sync),
provenance: Some(ts_provenance(dts, pts)),
},
});
}
}
fn video_sample_bytes(codec: VideoCodec, au_data: &[u8]) -> (Vec<u8>, bool) {
match codec {
VideoCodec::H264 => {
let is_rap = access_unit_is_rap(NalCodec::Avc, au_data, false);
(annexb_to_length_prefixed(au_data), is_rap)
}
VideoCodec::Hevc => {
let mut irap = false;
for nal in iter_annexb_nals(au_data) {
if is_keyframe_nal(NalCodec::Hevc, nal) {
irap = true;
}
}
(annexb_to_length_prefixed(au_data), irap)
}
VideoCodec::Mpeg2 => (au_data.to_vec(), mpeg2_is_sync(au_data)),
}
}
#[allow(clippy::too_many_arguments)]
fn emit_audio_au(
kind: &AudioKind,
sample_rate: u32,
anchor: &mut AudioAnchor,
au_data: &[u8],
pts_uw: i128,
dts_uw: i128,
track_id: u32,
events: &mut VecDeque<DemuxEvent>,
) {
let seed = anchor.seed;
let fresh_anchor = match &seed {
None => true,
Some(seed) => {
let expected_dts_uw = seed.anchor_dts_uw
+ (seed.ticks_since_anchor as i128 * VIDEO_TIMESCALE as i128)
/ sample_rate.max(1) as i128;
(dts_uw - expected_dts_uw).abs() > audio_discontinuity_threshold_90k(sample_rate)
}
};
if fresh_anchor && seed.is_some() {
events.push_back(DemuxEvent::Discontinuity {
track: Some(track_id),
kind: DiscontinuityKind::TimelineReanchored,
provenance: EventProvenance::default(),
});
}
let (dts0, pts0) = match (fresh_anchor, &seed) {
(false, Some(seed)) => (seed.next_dts, seed.next_pts),
_ => (
rescale_to_track(dts_uw, sample_rate),
rescale_to_track(pts_uw, sample_rate),
),
};
let mut elapsed = 0u64;
let au_provenance = ts_provenance(to_ticks(dts_uw), to_ticks(pts_uw));
let audio_sample = |data: Vec<u8>, duration: u32, elapsed: u64| -> Sample {
let dts = dts0 + elapsed as i64;
let pts = pts0 + elapsed as i64;
Sample::from_raw(data, Some(dts), Some(pts), Some(duration)).with_provenance(au_provenance)
};
match kind {
AudioKind::Aac => {
for (frame, hdr) in split_adts_frames(au_data) {
let header_len = adts_header_len(&hdr);
let duration = AAC_SAMPLES_PER_FRAME * (hdr.num_raw_data_blocks as u32 + 1);
if frame.len() > header_len {
events.push_back(DemuxEvent::Sample {
track_id,
sample: audio_sample(frame[header_len..].to_vec(), duration, elapsed),
});
}
elapsed += duration as u64;
}
}
AudioKind::Ac3 => {
for frame in split_ac3_syncframes(au_data) {
events.push_back(DemuxEvent::Sample {
track_id,
sample: audio_sample(frame.to_vec(), AC3_SAMPLES_PER_SYNCFRAME, elapsed),
});
elapsed += AC3_SAMPLES_PER_SYNCFRAME as u64;
}
}
AudioKind::Eac3 => {
for split in split_eac3_syncframes(au_data) {
let duration = split.info.samples_per_frame();
events.push_back(DemuxEvent::Sample {
track_id,
sample: audio_sample(split.data, duration, elapsed),
});
elapsed += duration as u64;
}
}
AudioKind::Dts => {
for frame in split_dts_core_frames(au_data) {
events.push_back(DemuxEvent::Sample {
track_id,
sample: audio_sample(frame.data.to_vec(), frame.samples, elapsed),
});
elapsed += frame.samples as u64;
}
}
AudioKind::MpegAudio { samples_per_frame } => {
for frame in split_mpeg_audio_frames(au_data) {
events.push_back(DemuxEvent::Sample {
track_id,
sample: audio_sample(frame.to_vec(), *samples_per_frame, elapsed),
});
elapsed += *samples_per_frame as u64;
}
}
}
let (anchor_dts_uw, anchor_pts_uw, ticks_since_anchor) = match (fresh_anchor, &seed) {
(false, Some(seed)) => (
seed.anchor_dts_uw,
seed.anchor_pts_uw,
seed.ticks_since_anchor,
),
_ => (dts_uw, pts_uw, 0i64),
};
anchor.seed = Some(AudioAnchorSeed {
next_dts: dts0 + elapsed as i64,
next_pts: pts0 + elapsed as i64,
anchor_dts_uw,
anchor_pts_uw,
ticks_since_anchor: ticks_since_anchor + elapsed as i64,
});
}
fn push_live_au(
live: &mut LiveTrack,
data: &[u8],
pts_uw: i128,
dts_uw: i128,
events: &mut VecDeque<DemuxEvent>,
) {
let track_id = live.track_id;
match &mut live.kind {
LiveKind::Video {
pending,
last_duration,
codec,
} => {
let (bytes, is_sync) = video_sample_bytes(*codec, data);
advance_one_behind(
pending,
last_duration,
bytes,
is_sync,
pts_uw,
dts_uw,
false,
track_id,
events,
);
}
LiveKind::Data {
pending,
last_duration,
} => {
advance_one_behind(
pending,
last_duration,
data.to_vec(),
true,
pts_uw,
dts_uw,
true,
track_id,
events,
);
}
LiveKind::Audio {
sample_rate,
kind,
anchor,
} => {
emit_audio_au(
kind,
*sample_rate,
anchor,
data,
pts_uw,
dts_uw,
track_id,
events,
);
}
LiveKind::MpegH {
pending,
last_duration,
} => {
let is_sync = find_mpegh3da_config(data).is_some();
advance_one_behind(
pending,
last_duration,
data.to_vec(),
is_sync,
pts_uw,
dts_uw,
true,
track_id,
events,
);
}
LiveKind::Section => {
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample::from_raw(data.to_vec(), None, None, None),
});
}
}
}
fn reprobe_if_config_changed(
stream: &mut StreamState,
live: &mut LiveTrack,
data: &[u8],
) -> Option<(CodecConfig, u32)> {
let changed = match stream.codec {
Codec::H264 | Codec::Hevc => {
observe_parameter_sets(stream.codec, data, &mut live.parameter_sets)
}
Codec::Aac => {
let (_, hdr) = find_adts_sync(data)?;
let asc = AudioSpecificConfig::from_adts_header(&hdr).to_bytes();
if live.config_header.as_deref() == Some(asc.as_slice()) {
false
} else {
live.config_header = Some(asc);
true
}
}
_ => return None,
};
if !changed {
return None;
}
let descriptors = stream.descriptors.clone();
let mut probe = initial_probe(stream.codec);
seed_probe_parameter_sets(&mut probe, &live.parameter_sets);
let recovered = finalize_probe(stream.codec, &descriptors, &mut probe, data);
let (config, timescale, _kind) = recovered?;
Some((config, timescale))
}
#[derive(Default, Clone)]
struct ParameterSets {
vps: Option<Vec<u8>>,
sps: Option<Vec<u8>>,
pps: Option<Vec<u8>>,
}
impl ParameterSets {
fn observe(&mut self, codec: Codec, data: &[u8]) -> bool {
let mut changed = false;
let mut record = |slot: &mut Option<Vec<u8>>, bytes: &[u8]| {
if slot.as_deref() != Some(bytes) {
*slot = Some(bytes.to_vec());
changed = true;
}
};
match codec {
Codec::H264 => {
for nal in iter_annexb_nals(data) {
match nal[0] & H264_NAL_TYPE_MASK {
H264_NAL_SPS => record(&mut self.sps, nal),
H264_NAL_PPS => record(&mut self.pps, nal),
_ => {}
}
}
}
Codec::Hevc => {
for nal in iter_annexb_nals(data) {
match nal_unit_type(NalCodec::Hevc, nal) {
Some(H265_NAL_VPS) => record(&mut self.vps, nal),
Some(H265_NAL_SPS) => record(&mut self.sps, nal),
Some(H265_NAL_PPS) => record(&mut self.pps, nal),
_ => {}
}
}
}
_ => {}
}
changed
}
}
fn observe_parameter_sets(codec: Codec, data: &[u8], sets: &mut ParameterSets) -> bool {
sets.observe(codec, data)
}
fn seed_probe_parameter_sets(probe: &mut ConfigProbe, sets: &ParameterSets) {
match probe {
ConfigProbe::H264 { sps, pps } => {
sps.clone_from(&sets.sps);
pps.clone_from(&sets.pps);
}
ConfigProbe::Hevc { vps, sps, pps } => {
vps.clone_from(&sets.vps);
sps.clone_from(&sets.sps);
pps.clone_from(&sets.pps);
}
_ => {}
}
}
fn finalize_probe(
codec: Codec,
descriptors: &[u8],
probe: &mut ConfigProbe,
latest: &[u8],
) -> Option<(CodecConfig, u32, LiveKind)> {
record_probe_pass();
match probe {
ConfigProbe::Data => {
let Codec::Data(stream_type) = codec else {
return None;
};
let carriage = data_carriage(stream_type);
let kind = match carriage {
DataCarriage::Pes => LiveKind::Data {
pending: None,
last_duration: 0,
},
DataCarriage::Sections => LiveKind::Section,
};
Some((
CodecConfig::Data {
stream_type,
descriptors: descriptors.to_vec(),
carriage,
},
VIDEO_TIMESCALE,
kind,
))
}
ConfigProbe::H264 { sps, pps } => {
for nal in iter_annexb_nals(latest) {
match nal[0] & H264_NAL_TYPE_MASK {
H264_NAL_SPS if sps.is_none() => *sps = Some(nal.to_vec()),
H264_NAL_PPS if pps.is_none() => *pps = Some(nal.to_vec()),
_ => {}
}
}
let (sps_bytes, pps_bytes) = (sps.as_ref()?, pps.as_ref()?);
if sps_bytes.len() < 4 {
return None;
}
let info = crate::sps::decode_avc_sps(sps_bytes).ok();
let (width, height) = info
.as_ref()
.map(|i| {
(
i.width.min(u16::MAX as u32) as u16,
i.height.min(u16::MAX as u32) as u16,
)
})
.unwrap_or((0, 0));
let ext = info
.as_ref()
.filter(|i| crate::sps::is_high_profile(i.profile_idc));
let record = AVCDecoderConfigurationRecord {
configuration_version: 1,
profile_indication: sps_bytes[1],
profile_compatibility: sps_bytes[2],
level_indication: sps_bytes[3],
length_size_minus_one: NAL_LENGTH_SIZE_MINUS_ONE,
sps: alloc::vec![AvcSps(sps_bytes.clone())],
pps: alloc::vec![AvcPps(pps_bytes.clone())],
chroma_format: ext.map(|i| i.chroma_format_idc),
bit_depth_luma_minus8: ext.map(|i| i.bit_depth_luma.saturating_sub(8)),
bit_depth_chroma_minus8: ext.map(|i| i.bit_depth_chroma.saturating_sub(8)),
sps_ext: alloc::vec![],
};
Some((
CodecConfig::Avc {
config: AVCConfigurationBox::new(record),
width,
height,
},
VIDEO_TIMESCALE,
LiveKind::Video {
pending: None,
last_duration: 0,
codec: VideoCodec::H264,
},
))
}
ConfigProbe::Hevc { vps, sps, pps } => {
for nal in iter_annexb_nals(latest) {
match nal_unit_type(NalCodec::Hevc, nal) {
Some(H265_NAL_VPS) if vps.is_none() => *vps = Some(nal.to_vec()),
Some(H265_NAL_SPS) if sps.is_none() => *sps = Some(nal.to_vec()),
Some(H265_NAL_PPS) if pps.is_none() => *pps = Some(nal.to_vec()),
_ => {}
}
}
let sps_bytes = sps.as_ref()?;
let info = crate::sps::decode_hevc_sps(sps_bytes).ok()?;
let width = info.width.min(u16::MAX as u32) as u16;
let height = info.height.min(u16::MAX as u32) as u16;
let mut arrays: Vec<HevcNalArray> = Vec::new();
if let Some(vps_nal) = vps.clone() {
arrays.push(HevcNalArray::new(
true,
H265_NAL_VPS,
alloc::vec![HevcNalUnit::new(vps_nal)],
));
}
arrays.push(HevcNalArray::new(
true,
H265_NAL_SPS,
alloc::vec![HevcNalUnit::new(sps_bytes.clone())],
));
if let Some(pps_nal) = pps.clone() {
arrays.push(HevcNalArray::new(
true,
H265_NAL_PPS,
alloc::vec![HevcNalUnit::new(pps_nal)],
));
}
let record = HEVCDecoderConfigurationRecord {
configuration_version: HVCC_CONFIGURATION_VERSION,
general_profile_space: info.general_profile_space,
general_tier_flag: info.general_tier_flag,
general_profile_idc: info.general_profile_idc,
general_profile_compatibility_flags: info.general_profile_compatibility_flags,
general_constraint_indicator_flags: info.general_constraint_indicator_flags,
general_level_idc: info.general_level_idc,
min_spatial_segmentation_idc: HVCC_MIN_SPATIAL_SEGMENTATION_UNSPEC,
parallelism_type: HVCC_PARALLELISM_TYPE_UNKNOWN,
chroma_format_idc: info.chroma_format_idc,
bit_depth_luma_minus8: info.bit_depth_luma.saturating_sub(8),
bit_depth_chroma_minus8: info.bit_depth_chroma.saturating_sub(8),
avg_frame_rate: HVCC_AVG_FRAME_RATE_UNSPEC,
constant_frame_rate: HVCC_CONSTANT_FRAME_RATE_UNSPEC,
num_temporal_layers: HVCC_NUM_TEMPORAL_LAYERS,
temporal_id_nested: false,
length_size_minus_one: NAL_LENGTH_SIZE_MINUS_ONE,
arrays,
};
Some((
CodecConfig::Hevc {
config: HEVCConfigurationBox::new(record),
width,
height,
},
VIDEO_TIMESCALE,
LiveKind::Video {
pending: None,
last_duration: 0,
codec: VideoCodec::Hevc,
},
))
}
ConfigProbe::Mpeg2Video => {
let seq = Mpeg2SeqHeader::find(latest).ok()?;
let esds = EsdsBox::new(ESDescriptor::new(
ESDS_VIDEO_ES_ID,
0,
Some(DecoderConfigDescriptor::new(
OTI_MPEG2_VIDEO_MAIN,
STREAM_TYPE_VISUAL,
false,
0,
0,
0,
None,
)),
Some(SLConfigDescriptor::predefined_two()),
));
Some((
CodecConfig::Mpeg2Video {
esds,
width: seq.width,
height: seq.height,
},
VIDEO_TIMESCALE,
LiveKind::Video {
pending: None,
last_duration: 0,
codec: VideoCodec::Mpeg2,
},
))
}
ConfigProbe::MpegAudio { is_mpeg2 } => {
let (_, first) = find_mpeg_audio_sync(latest)?;
let sample_rate = first.sample_rate;
let channel_count = first.channels;
let samples_per_frame = first.samples_per_frame;
let oti = if *is_mpeg2 {
OTI_MPEG2_AUDIO
} else {
OTI_MPEG1_AUDIO
};
let esds = EsdsBox::new(ESDescriptor::new(
ESDS_ES_ID,
0,
Some(DecoderConfigDescriptor::new(
oti,
STREAM_TYPE_AUDIO,
false,
0,
0,
0,
None,
)),
Some(SLConfigDescriptor::predefined_two()),
));
Some((
CodecConfig::MpegAudio {
esds,
layer: first.layer,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::MpegAudio { samples_per_frame },
anchor: AudioAnchor::default(),
},
))
}
ConfigProbe::Aac => {
let (_, first_hdr) = find_adts_sync(latest)?;
let asc = AudioSpecificConfig::from_adts_header(&first_hdr);
let sample_rate =
SamplingFrequencyIndex::from(first_hdr.sampling_frequency_index).table_hz()?;
let channel_count = crate::flv::aac_channel_count(&asc);
let esds = EsdsBox::new(ESDescriptor::new(
ESDS_ES_ID,
0,
Some(DecoderConfigDescriptor::new(
OTI_MPEG4_AUDIO,
STREAM_TYPE_AUDIO,
false,
0,
0,
0,
Some(DecoderSpecificInfo::new(asc.to_bytes())),
)),
Some(SLConfigDescriptor::predefined_two()),
));
Some((
CodecConfig::Aac {
esds,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::Aac,
anchor: AudioAnchor::default(),
},
))
}
ConfigProbe::Ac3 => {
let info = Ac3SyncframeInfo::from_es(latest).ok()?;
let sample_rate = info.sample_rate;
let channel_count = info.channel_count() as u16;
let config = info.into_dac3();
Some((
CodecConfig::Ac3 {
config,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::Ac3,
anchor: AudioAnchor::default(),
},
))
}
ConfigProbe::Eac3 => {
let frames: alloc::vec::Vec<Ec3SyncframeInfo> =
Ec3SyncframeInfo::from_es_first_au(latest);
let info = *frames.first()?;
let sample_rate = info.sample_rate;
let channel_count = info.channel_count() as u16;
let config = Ec3SpecificBox::from_syncframes(&frames).ok()?;
Some((
CodecConfig::Eac3 {
config,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::Eac3,
anchor: AudioAnchor::default(),
},
))
}
ConfigProbe::Dts => {
let info = DtsCoreFrameInfo::from_es(latest).ok()?;
let sample_rate = info.sample_rate;
let channel_count = info.channels as u16;
let config = info.into_ddts();
Some((
CodecConfig::Dts {
config,
codec_fourcc: crate::dts::DTSC_FOURCC,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::Dts,
anchor: AudioAnchor::default(),
},
))
}
ConfigProbe::MpegH => {
let config_bytes = find_mpegh3da_config(latest)?;
let profile_level_indication = *config_bytes.first()?;
let config = MHADecoderConfigurationRecord::new(
profile_level_indication,
MPEGH_REFERENCE_CHANNEL_LAYOUT_UNSPECIFIED,
config_bytes.to_vec(),
);
Some((
CodecConfig::MpegH {
config,
channel_count: MPEGH_CHANNEL_COUNT_UNSPECIFIED,
sample_rate: MPEGH_SAMPLE_RATE_UNSPECIFIED,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
VIDEO_TIMESCALE,
LiveKind::MpegH {
pending: None,
last_duration: 0,
},
))
}
}
}
fn abandon_backlog(
stream: &mut StreamState,
pid: u16,
events: &mut VecDeque<DemuxEvent>,
) -> TrackState {
stream.backlog_bytes = 0;
events.push_back(DemuxEvent::TrackAbandoned {
track_id: None,
reason: AbandonReason::BudgetExceeded,
provenance: EventProvenance {
pid: Some(pid),
packet_index: None,
},
});
TrackState::Abandoned
}
fn advance_track(
stream: &mut StreamState,
pid: u16,
data: &[u8],
pts_uw: i128,
dts_uw: i128,
events: &mut VecDeque<DemuxEvent>,
) {
let Some(track) = stream.track.take() else {
return;
};
let new_track = match track {
TrackState::Live(mut live) => {
if let Some((config, timescale)) = reprobe_if_config_changed(stream, &mut live, data) {
live.config = config;
live.timescale = timescale;
events.push_back(DemuxEvent::TrackUpdated(
TrackSpec::new(live.track_id, live.timescale, live.config.clone())
.with_source(pid, stream.descriptors.clone()),
));
}
push_live_au(&mut live, data, pts_uw, dts_uw, events);
TrackState::Live(live)
}
TrackState::Abandoned => TrackState::Abandoned,
TrackState::Parked {
config,
timescale,
kind,
mut backlog,
} => {
stream.backlog_bytes = stream.backlog_bytes.saturating_add(data.len());
backlog.push(BufferedAu {
data: data.to_vec(),
pts_uw,
dts_uw,
});
if stream.backlog_bytes > MAX_PROBE_BACKLOG_BYTES {
abandon_backlog(stream, pid, events)
} else {
TrackState::Parked {
config,
timescale,
kind,
backlog,
}
}
}
TrackState::Probing {
mut probe,
mut backlog,
} => {
stream.backlog_bytes = stream.backlog_bytes.saturating_add(data.len());
backlog.push(BufferedAu {
data: data.to_vec(),
pts_uw,
dts_uw,
});
let Some(latest) = backlog.last() else {
return;
};
match finalize_probe(stream.codec, &stream.descriptors, &mut probe, &latest.data) {
Some((config, timescale, kind)) => TrackState::Parked {
config,
timescale,
kind,
backlog,
},
None if stream.backlog_bytes > MAX_PROBE_BACKLOG_BYTES => {
abandon_backlog(stream, pid, events)
}
None => TrackState::Probing { probe, backlog },
}
}
};
stream.track = Some(new_track);
}
fn feed_pes_bounded(
stream: &mut StreamState,
pid: u16,
pusi: bool,
payload: &[u8],
events: &mut VecDeque<DemuxEvent>,
) -> Option<Vec<u8>> {
let Carrier::Pes(assembler) = &mut stream.carrier else {
return None;
};
if pusi {
stream.pes_received = payload.len();
stream.pes_declared_len = if payload.len() >= PES_LENGTH_FIELD_END {
u16::from_be_bytes([payload[4], payload[5]])
} else {
0
};
} else {
stream.pes_received = stream.pes_received.saturating_add(payload.len());
}
if pusi {
stream.pes_bytes = payload.len();
} else if stream.pes_bytes > 0 {
stream.pes_bytes = stream.pes_bytes.saturating_add(payload.len());
}
let completed = assembler.feed(pusi, payload);
if stream.pes_bytes > MAX_PES_BUFFER_BYTES {
let dropped_bytes = stream.pes_bytes as u64;
let _ = assembler.flush();
stream.pes_bytes = 0;
let track = match stream.track.as_ref() {
Some(TrackState::Live(live)) => Some(live.track_id),
_ => None,
};
events.push_back(DemuxEvent::Discontinuity {
track,
kind: DiscontinuityKind::BudgetExceeded {
bytes: dropped_bytes,
},
provenance: EventProvenance {
pid: Some(pid),
packet_index: None,
},
});
}
completed
}
fn on_completed_pes(
stream: &mut StreamState,
pid: u16,
pes_bytes: &[u8],
events: &mut VecDeque<DemuxEvent>,
) {
let Ok(pes) = PesPacket::parse(pes_bytes) else {
return;
};
if pes.payload.is_empty() {
return;
}
let (pts, dts, stamped) = match pes.header.as_ref() {
Some(h) => {
let hp = h.pts.map(|p| p.0);
let hd = h.dts.map(|d| d.0);
match (hp, hd) {
(Some(p), Some(d)) => (p, d, true),
(Some(p), None) => (p, p, true),
(None, Some(d)) => (d, d, true),
(None, None) => (0, 0, false),
}
}
None => (0, 0, false),
};
let (pts, dts) = if stamped {
if let Some(prev) = stream.last_stamped_dts
&& !stream.pending_rebase
{
let span = dts.wrapping_sub(prev) & TS_WRAP_MASK;
let periods = stream.units_since_stamped.saturating_add(1);
let period = span / periods;
if period > 0 && period <= MAX_FRAME_PERIOD_TICKS {
if stream.recent_steps.len() == FRAME_PERIOD_WINDOW {
stream.recent_steps.remove(0);
}
stream.recent_steps.push(period);
let mut sorted = stream.recent_steps.clone();
sorted.sort_unstable();
let median = sorted[sorted.len() / 2];
stream.frame_period = Some(median);
stream.last_frame_period = median as i128;
}
}
stream.last_stamped_dts = Some(dts);
stream.units_since_stamped = 0;
stream.fallback = (pts, dts);
stream.has_any = true;
(pts, dts)
} else if stream.has_any {
let step = stream.frame_period.unwrap_or(0);
stream.units_since_stamped = stream.units_since_stamped.saturating_add(1);
let pts = stream.fallback.0.wrapping_add(step) & TS_WRAP_MASK;
let dts = stream.fallback.1.wrapping_add(step) & TS_WRAP_MASK;
stream.fallback = (pts, dts);
(pts, dts)
} else {
stream.units_since_stamped = stream.units_since_stamped.saturating_add(1);
(0, 0)
};
if stream.pending_rebase {
stream.pending_rebase = stream
.wrap
.rebase_at_discontinuity(dts, stream.last_frame_period)
== DiscontinuityVerdict::Forward;
}
let (pts_uw, dts_uw) = stream.wrap.push(pts, dts);
advance_track(stream, pid, pes.payload, pts_uw, dts_uw, events);
}
fn on_completed_section(
stream: &mut StreamState,
pid: u16,
section: &[u8],
events: &mut VecDeque<DemuxEvent>,
) {
if section.is_empty() {
return;
}
advance_track(stream, pid, section, 0, 0, events);
}
pub use crate::ir::{
AbandonReason, DemuxEvent, DiscontinuityKind, EventProvenance, InputDegradation,
};
const PCR_CLOCK_HZ: u32 = 27_000_000;
#[derive(Debug, Clone)]
struct CcState {
initialized: bool,
last_cc: u8,
last_packet: Option<[u8; TS_PACKET_SIZE]>,
dup_used_for_last: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CcVerdict {
Deliver,
Duplicate,
Gap,
}
pub struct StreamingTsDemux {
resync: TsResync,
packet_index: u64,
pat_reasm: SectionReassembler,
pmt_reasm: BTreeMap<u16, PmtState>,
es_seen: BTreeSet<u16>,
streams: BTreeMap<u16, StreamState>,
unattributed: BTreeMap<u16, VecDeque<(bool, Vec<u8>)>>,
removed_pids: BTreeSet<u16>,
es_declarers: BTreeMap<u16, BTreeSet<u16>>,
unattributed_order: VecDeque<u16>,
unattributed_bytes: usize,
codec_order: Vec<u16>,
data_order: Vec<u16>,
resolved: BTreeSet<u16>,
next_track_id: u32,
events: VecDeque<DemuxEvent>,
generation: u32,
tracks_resolved_signalled_at: Option<u32>,
cc_states: BTreeMap<u16, CcState>,
}
struct PmtState {
reasm: SectionReassembler,
program_number: u16,
pcr_pid: Option<u16>,
last_applied_version: Option<u8>,
applied_es: BTreeSet<u16>,
}
impl Default for StreamingTsDemux {
fn default() -> Self {
Self::new()
}
}
impl StreamingTsDemux {
pub fn new() -> Self {
Self {
resync: TsResync::new(),
packet_index: 0,
pat_reasm: SectionReassembler::default(),
pmt_reasm: BTreeMap::new(),
es_seen: BTreeSet::new(),
streams: BTreeMap::new(),
unattributed: BTreeMap::new(),
removed_pids: BTreeSet::new(),
es_declarers: BTreeMap::new(),
unattributed_order: VecDeque::new(),
unattributed_bytes: 0,
codec_order: Vec::new(),
data_order: Vec::new(),
resolved: BTreeSet::new(),
next_track_id: 1,
events: VecDeque::new(),
generation: 0,
tracks_resolved_signalled_at: None,
cc_states: BTreeMap::new(),
}
}
pub fn feed(&mut self, data: &[u8]) {
let packets = self.resync.feed(data);
for raw in &packets {
self.process_packet(raw);
}
}
fn live_track_id(&self, pid: u16) -> Option<u32> {
match self.streams.get(&pid)?.track.as_ref()? {
TrackState::Live(live) => Some(live.track_id),
_ => None,
}
}
fn process_packet(&mut self, raw: &[u8; TS_PACKET_SIZE]) {
let idx = self.packet_index;
self.packet_index += 1;
let Ok(pkt) = TsPacket::parse(raw) else {
return;
};
if pkt.header.tei {
let provenance = EventProvenance {
pid: Some(pkt.header.pid),
packet_index: Some(idx),
};
self.events.push_back(DemuxEvent::InputDegraded {
track: self.live_track_id(pkt.header.pid),
kind: InputDegradation::TransportError,
provenance,
});
}
let discontinuity_signalled = pkt.header.has_adaptation
&& raw
.get(4)
.is_some_and(|af_len| *af_len > 0 && raw.get(5).is_some_and(|b| b & 0x80 != 0));
if let Some(Ok(af)) = pkt.adaptation_field() {
let provenance = EventProvenance {
pid: Some(pkt.header.pid),
packet_index: Some(idx),
};
if af.discontinuity_indicator {
let pid = pkt.header.pid;
let program_ok = self.pmt_reasm.values().any(|pmt| pmt.pcr_pid == Some(pid));
if program_ok {
let members: Vec<u16> = self
.es_declarers
.iter()
.filter(|(_, pmt_pids)| {
pmt_pids.iter().any(|pmt_pid| {
self.pmt_reasm
.get(pmt_pid)
.is_some_and(|pmt| pmt.pcr_pid == Some(pid))
})
})
.map(|(&es_pid, _)| es_pid)
.collect();
for es_pid in members {
if let Some(stream) = self.streams.get_mut(&es_pid) {
stream.pending_rebase = true;
}
}
}
self.events.push_back(DemuxEvent::Discontinuity {
track: self.live_track_id(pid),
kind: DiscontinuityKind::Signalled,
provenance,
});
}
if let Some(pcr) = af.pcr {
self.events.push_back(DemuxEvent::ClockReference {
ticks: pcr.as_27mhz(),
clock_hz: PCR_CLOCK_HZ,
discontinuous: af.discontinuity_indicator,
provenance,
});
}
}
let mut cc_verdict = CcVerdict::Deliver;
if pkt.header.has_payload && pkt.header.pid != NULL_PACKET_PID {
cc_verdict = self.check_cc(
pkt.header.pid,
pkt.header.continuity_counter,
raw,
discontinuity_signalled,
idx,
);
}
let pid = pkt.header.pid;
let pusi = pkt.header.pusi;
let Some(payload) = pkt.payload else {
return;
};
if cc_verdict == CcVerdict::Duplicate {
self.try_promote_ready();
return;
}
if pid == PAT_PID {
self.pat_reasm.feed(payload, pusi);
while let Some(section) = self.pat_reasm.pop_section() {
if !psi_section_crc_ok(§ion) {
continue;
}
if !section_current_next(§ion) {
continue;
}
if let Ok(programs) = parse_pat(§ion) {
for (program_number, pmt_pid) in programs {
self.learn_pmt_pid(pmt_pid, program_number);
}
}
}
return;
}
if let Some(pmt_state) = self.pmt_reasm.get_mut(&pid) {
pmt_state.reasm.feed(payload, pusi);
let mut sections: Vec<Vec<u8>> = Vec::new();
while let Some(section) = pmt_state.reasm.pop_section() {
sections.push(section.to_vec());
}
let program_number = pmt_state.program_number;
let mut to_apply: Option<Vec<(u16, Codec, Vec<u8>)>> = None;
for section in §ions {
if !psi_section_crc_ok(section) {
continue;
}
let Ok(header) = parse_pmt_section_header(section) else {
continue;
};
if header.program_number != program_number {
continue;
}
if header.section_number != 0 || header.last_section_number != 0 {
continue;
}
if !header.current_next {
continue;
}
if pmt_state.last_applied_version == Some(header.version) {
continue;
}
pmt_state.last_applied_version = Some(header.version);
pmt_state.pcr_pid = Some(header.pcr_pid);
if let Ok(es_list) = parse_pmt(section) {
to_apply = Some(es_list);
}
}
if let Some(es_list) = to_apply {
let old_applied_es = pmt_state.applied_es.clone();
let new_applied_es: BTreeSet<u16> = es_list.iter().map(|(p, _, _)| *p).collect();
self.apply_pmt_diff(pid, &old_applied_es, es_list);
if let Some(pmt_state) = self.pmt_reasm.get_mut(&pid) {
pmt_state.applied_es = new_applied_es;
}
}
self.try_promote_ready();
return;
}
if let Some(stream) = self.streams.get_mut(&pid) {
let mut sections: Vec<Vec<u8>> = Vec::new();
let gap_here = cc_verdict == CcVerdict::Gap;
let short_of_declared = stream.pes_declared_len > 0
&& stream.pes_received < PES_LENGTH_FIELD_END + stream.pes_declared_len as usize;
let completed_intact = stream.current_pes_intact && !short_of_declared;
let completed_pes = if matches!(stream.carrier, Carrier::Pes(_)) {
feed_pes_bounded(stream, pid, pusi, payload, &mut self.events)
} else if let Carrier::Section(reasm) = &mut stream.carrier {
reasm.feed(payload, pusi);
while let Some(s) = reasm.pop_section() {
sections.push(s.to_vec());
}
None
} else {
None
};
if completed_intact && let Some(completed) = completed_pes {
on_completed_pes(stream, pid, &completed, &mut self.events);
}
if matches!(stream.carrier, Carrier::Pes(_)) {
if pusi {
stream.current_pes_intact = true;
} else if gap_here {
stream.current_pes_intact = false;
if let Carrier::Pes(assembler) = &mut stream.carrier {
let _ = assembler.flush();
}
}
}
for s in sections {
on_completed_section(stream, pid, &s, &mut self.events);
}
} else if pid != NULL_PACKET_PID && !self.removed_pids.contains(&pid) {
self.unattributed
.entry(pid)
.or_default()
.push_back((pusi, payload.to_vec()));
self.unattributed_order.push_back(pid);
self.unattributed_bytes += payload.len();
self.evict_unattributed();
}
self.try_promote_ready();
}
fn check_cc(
&mut self,
pid: u16,
cc: u8,
raw: &[u8; TS_PACKET_SIZE],
discontinuity_signalled: bool,
packet_index: u64,
) -> CcVerdict {
let track = self.live_track_id(pid);
match self.cc_states.entry(pid) {
Entry::Occupied(mut e) => {
let state = e.get_mut();
if !state.initialized {
state.initialized = true;
state.last_cc = cc;
state.last_packet = Some(*raw);
return CcVerdict::Deliver;
}
if cc == state.last_cc
&& let Some(prev) = state.last_packet
&& broadcast_common::ts_dup::check_duplicate(
&prev,
raw,
state.dup_used_for_last,
) == broadcast_common::ts_dup::DuplicateVerdict::Legal
{
state.dup_used_for_last = true;
state.last_packet = Some(*raw);
return CcVerdict::Duplicate;
}
let expected = (state.last_cc + 1) & 0x0F;
let mut verdict = CcVerdict::Deliver;
if !discontinuity_signalled && cc != expected {
self.events.push_back(DemuxEvent::InputDegraded {
track,
kind: InputDegradation::ContinuityGap { expected, got: cc },
provenance: EventProvenance {
pid: Some(pid),
packet_index: Some(packet_index),
},
});
verdict = CcVerdict::Gap;
}
state.last_cc = cc;
state.last_packet = Some(*raw);
state.dup_used_for_last = false;
verdict
}
Entry::Vacant(e) => {
e.insert(CcState {
initialized: true,
last_cc: cc,
last_packet: Some(*raw),
dup_used_for_last: false,
});
CcVerdict::Deliver
}
}
}
fn learn_pmt_pid(&mut self, pmt_pid: u16, program_number: u16) {
match self.pmt_reasm.get_mut(&pmt_pid) {
Some(state) if state.program_number != program_number => {
state.program_number = program_number;
state.last_applied_version = None;
}
Some(_) => {}
None => {
self.pmt_reasm.insert(
pmt_pid,
PmtState {
reasm: SectionReassembler::default(),
program_number,
pcr_pid: None,
last_applied_version: None,
applied_es: BTreeSet::new(),
},
);
}
}
}
fn evict_unattributed(&mut self) {
while self.unattributed_bytes > MAX_UNATTRIBUTED_BYTES {
let Some(pid) = self.unattributed_order.pop_front() else {
break;
};
if let Some(buf) = self.unattributed.get_mut(&pid) {
if let Some((_, payload)) = buf.pop_front() {
self.unattributed_bytes = self.unattributed_bytes.saturating_sub(payload.len());
}
if buf.is_empty() {
self.unattributed.remove(&pid);
}
}
}
}
fn register_new_es(&mut self, es_pid: u16, codec: Codec, descriptors: Vec<u8>) {
self.register_new_es_at(es_pid, codec, descriptors, None);
}
fn register_new_es_at(
&mut self,
es_pid: u16,
codec: Codec,
descriptors: Vec<u8>,
reinsert_at: Option<usize>,
) {
self.removed_pids.remove(&es_pid);
let order = if matches!(codec, Codec::Data(_)) {
&mut self.data_order
} else {
&mut self.codec_order
};
match reinsert_at {
Some(idx) if idx <= order.len() => order.insert(idx, es_pid),
_ => order.push(es_pid),
}
let mut stream = StreamState {
codec,
descriptors,
carrier: initial_carrier(codec),
last_frame_period: 0,
recent_steps: Vec::new(),
pending_rebase: false,
current_pes_intact: true,
pes_declared_len: 0,
pes_received: 0,
pes_bytes: 0,
fallback: (0, 0),
frame_period: None,
last_stamped_dts: None,
units_since_stamped: 0,
has_any: false,
wrap: WrapState::default(),
track: Some(TrackState::Probing {
probe: initial_probe(codec),
backlog: Vec::new(),
}),
backlog_bytes: 0,
};
if let Some(buffered) = self.unattributed.remove(&es_pid) {
for (buf_pusi, buf_payload) in buffered {
self.unattributed_bytes = self.unattributed_bytes.saturating_sub(buf_payload.len());
let mut sections: Vec<Vec<u8>> = Vec::new();
let completed_pes = if matches!(stream.carrier, Carrier::Pes(_)) {
feed_pes_bounded(
&mut stream,
es_pid,
buf_pusi,
&buf_payload,
&mut self.events,
)
} else if let Carrier::Section(reasm) = &mut stream.carrier {
reasm.feed(&buf_payload, buf_pusi);
while let Some(s) = reasm.pop_section() {
sections.push(s.to_vec());
}
None
} else {
None
};
if let Some(completed) = completed_pes {
on_completed_pes(&mut stream, es_pid, &completed, &mut self.events);
}
for s in sections {
on_completed_section(&mut stream, es_pid, &s, &mut self.events);
}
}
}
self.streams.insert(es_pid, stream);
}
fn remove_track(&mut self, pid: u16) {
self.es_seen.remove(&pid);
self.codec_order.retain(|&p| p != pid);
self.data_order.retain(|&p| p != pid);
self.resolved.remove(&pid);
if let Some(buffered) = self.unattributed.remove(&pid) {
for (_, payload) in &buffered {
self.unattributed_bytes = self.unattributed_bytes.saturating_sub(payload.len());
}
}
self.removed_pids.insert(pid);
self.cc_states.remove(&pid);
if let Some(stream) = self.streams.remove(&pid)
&& let Some(TrackState::Live(live)) = stream.track
{
self.events.push_back(DemuxEvent::TrackRemoved {
track_id: live.track_id,
provenance: EventProvenance {
pid: Some(pid),
packet_index: None,
},
});
}
}
fn apply_pmt_diff(
&mut self,
pmt_pid: u16,
old_applied_es: &BTreeSet<u16>,
es_list: Vec<(u16, Codec, Vec<u8>)>,
) {
let new_pids: BTreeSet<u16> = es_list.iter().map(|(p, _, _)| *p).collect();
let removed: Vec<u16> = old_applied_es
.iter()
.copied()
.filter(|p| !new_pids.contains(p))
.collect();
for pid in removed {
let declarers = self.es_declarers.get_mut(&pid);
let still_declared = match declarers {
Some(declarers) => {
declarers.remove(&pmt_pid);
let empty = declarers.is_empty();
if empty {
self.es_declarers.remove(&pid);
}
!empty
}
None => false,
};
if !still_declared {
self.remove_track(pid);
}
}
for (es_pid, codec, descriptors) in es_list {
self.es_declarers.entry(es_pid).or_default().insert(pmt_pid);
if self.es_seen.insert(es_pid) {
self.register_new_es(es_pid, codec, descriptors);
continue;
}
let Some(existing) = self.streams.get(&es_pid) else {
continue;
};
let old_codec = existing.codec;
let codec_changed = old_codec != codec;
let descriptors_changed = existing.descriptors != descriptors;
if !codec_changed && !descriptors_changed {
continue;
}
if codec_changed {
let other_declarers_remain = self
.es_declarers
.get(&es_pid)
.is_some_and(|declarers| declarers.iter().any(|&p| p != pmt_pid));
if other_declarers_remain {
continue;
}
let old_is_data = matches!(old_codec, Codec::Data(_));
let new_is_data = matches!(codec, Codec::Data(_));
let order_slot = if old_is_data == new_is_data {
let order = if old_is_data {
&self.data_order
} else {
&self.codec_order
};
order.iter().position(|&p| p == es_pid)
} else {
None
};
self.remove_track(es_pid);
self.es_seen.insert(es_pid);
self.register_new_es_at(es_pid, codec, descriptors, order_slot);
continue;
}
let Some(stream) = self.streams.get_mut(&es_pid) else {
continue;
};
stream.descriptors = descriptors.clone();
if let Some(TrackState::Live(live)) = stream.track.as_ref() {
let spec = TrackSpec::new(live.track_id, live.timescale, live.config.clone())
.with_source(es_pid, descriptors);
self.events.push_back(DemuxEvent::TrackUpdated(spec));
}
}
self.generation = self.generation.wrapping_add(1);
}
fn try_promote_ready(&mut self) {
while let Some(&next_pid) = self
.codec_order
.iter()
.chain(self.data_order.iter())
.find(|p| !self.resolved.contains(p))
{
let Some(stream) = self.streams.get_mut(&next_pid) else {
break;
};
let Some(track) = stream.track.take() else {
break;
};
match track {
TrackState::Parked {
config,
timescale,
kind,
backlog,
} => {
let track_id = self.next_track_id;
self.next_track_id += 1;
let spec = TrackSpec::new(track_id, timescale, config.clone())
.with_source(next_pid, stream.descriptors.clone());
let program_number = self
.es_declarers
.get(&next_pid)
.and_then(|pmt_pids| pmt_pids.iter().next())
.and_then(|&pmt_pid| self.pmt_reasm.get(&pmt_pid))
.map(|state| state.program_number);
let spec = if let Some(pn) = program_number {
spec.with_program(pn)
} else {
spec
};
self.events.push_back(DemuxEvent::TrackAdded(spec));
let mut live = LiveTrack {
track_id,
kind,
config,
timescale,
parameter_sets: ParameterSets::default(),
config_header: None,
};
for au in &backlog {
let _ = observe_parameter_sets(
stream.codec,
&au.data,
&mut live.parameter_sets,
);
}
if stream.codec == Codec::Aac
&& let Some(au) = backlog.last()
&& let Some((_, hdr)) = find_adts_sync(&au.data)
{
live.config_header =
Some(AudioSpecificConfig::from_adts_header(&hdr).to_bytes());
}
for au in backlog {
push_live_au(&mut live, &au.data, au.pts_uw, au.dts_uw, &mut self.events);
}
stream.track = Some(TrackState::Live(live));
stream.backlog_bytes = 0;
self.resolved.insert(next_pid);
}
other @ TrackState::Probing { .. } => {
stream.track = Some(other);
break; }
other @ TrackState::Live(_) => {
stream.track = Some(other);
self.resolved.insert(next_pid);
}
TrackState::Abandoned => {
stream.track = Some(TrackState::Abandoned);
self.resolved.insert(next_pid);
}
}
}
self.maybe_signal_tracks_resolved();
}
fn maybe_signal_tracks_resolved(&mut self) {
let known = self.codec_order.len() + self.data_order.len();
if known == 0 {
return;
}
if self.resolved.len() == known
&& self.tracks_resolved_signalled_at != Some(self.generation)
{
self.tracks_resolved_signalled_at = Some(self.generation);
self.events.push_back(DemuxEvent::TracksResolved {
generation: self.generation,
});
}
}
pub fn poll_event(&mut self) -> Option<DemuxEvent> {
self.events.pop_front()
}
pub fn finish(&mut self) {
for (&pid, stream) in self.streams.iter_mut() {
let completed = match &mut stream.carrier {
Carrier::Pes(assembler) => assembler.flush(),
Carrier::Section(_) => None,
};
if let Some(completed) = completed {
let short_of_declared = stream.pes_declared_len > 0
&& completed.len() < PES_LENGTH_FIELD_END + stream.pes_declared_len as usize;
if stream.current_pes_intact && !short_of_declared {
on_completed_pes(stream, pid, &completed, &mut self.events);
}
}
}
self.try_promote_ready();
while let Some(&next_pid) = self
.codec_order
.iter()
.chain(self.data_order.iter())
.find(|p| !self.resolved.contains(p))
{
match self.streams.get(&next_pid).and_then(|s| s.track.as_ref()) {
Some(TrackState::Probing { .. }) => {
self.resolved.insert(next_pid);
self.events.push_back(DemuxEvent::TrackAbandoned {
track_id: None,
reason: AbandonReason::ConfigUnrecoverable,
provenance: EventProvenance {
pid: Some(next_pid),
packet_index: None,
},
});
self.try_promote_ready();
}
_ => break,
}
}
for stream in self.streams.values_mut() {
if let Some(TrackState::Live(live)) = &mut stream.track {
match &mut live.kind {
LiveKind::Video {
pending,
last_duration,
..
} => {
flush_one_behind(pending, *last_duration, live.track_id, &mut self.events);
}
LiveKind::Data {
pending,
last_duration,
}
| LiveKind::MpegH {
pending,
last_duration,
} => {
flush_one_behind(pending, *last_duration, live.track_id, &mut self.events);
}
LiveKind::Audio { .. } => {}
LiveKind::Section => {}
}
}
}
}
}
impl Stage for StreamingTsDemux {
type In<'a> = &'a [u8];
type Out = DemuxEvent;
type Error = core::convert::Infallible;
fn feed(&mut self, input: &[u8], _now: Timestamp) -> core::result::Result<(), Self::Error> {
self.feed(input);
Ok(())
}
fn poll(&mut self) -> Option<Self::Out> {
self.poll_event()
}
fn finish(&mut self) -> core::result::Result<(), Self::Error> {
self.finish();
Ok(())
}
fn next_deadline(&self) -> Option<Timestamp> {
None
}
fn on_deadline(&mut self, _now: Timestamp) {}
fn demand(&self) -> Demand {
if self.unattributed_bytes.saturating_add(TS_MAX_PAYLOAD_BYTES) > MAX_UNATTRIBUTED_BYTES {
Demand::saturated()
} else {
Demand::new(TS_PACKET_SIZE)
}
}
}
#[derive(Debug, Default, Clone)]
pub struct TsDemux<'a> {
_marker: PhantomData<&'a [u8]>,
}
impl<'a> TsDemux<'a> {
pub fn new() -> Self {
Self {
_marker: PhantomData,
}
}
pub fn demux(&mut self, input: &'a [u8]) -> Result<Media> {
let mut demux = StreamingTsDemux::new();
demux.feed(input);
demux.finish();
let mut tracks: Vec<Track> = Vec::new();
let mut index_by_id: BTreeMap<u32, usize> = BTreeMap::new();
let mut pcr: Vec<PcrSample> = Vec::new();
while let Some(event) = demux.poll_event() {
match event {
DemuxEvent::TrackAdded(spec) => {
index_by_id.insert(spec.track_id, tracks.len());
tracks.push(Track::new(spec, Vec::new()));
}
DemuxEvent::TrackUpdated(spec) => {
if let Some(&i) = index_by_id.get(&spec.track_id) {
tracks[i].spec = spec;
}
}
DemuxEvent::TrackRemoved { .. } => {}
DemuxEvent::TrackAbandoned { .. } => {}
DemuxEvent::Sample { track_id, sample }
if let Some(&i) = index_by_id.get(&track_id) =>
{
let track = &mut tracks[i];
if track.samples.is_empty()
&& let Some(dts) = sample.dts
{
track.start_decode_time = dts as u64;
}
track.samples.push(sample);
}
DemuxEvent::Sample { .. } => {}
DemuxEvent::ClockReference {
ticks,
discontinuous,
provenance,
..
} => pcr.push(PcrSample {
pcr_27mhz: ticks,
pid: provenance.pid.unwrap_or(0),
packet_index: provenance.packet_index.unwrap_or(0),
discontinuity: discontinuous,
}),
DemuxEvent::Discontinuity { .. } => {}
DemuxEvent::InputDegraded { .. } => {}
DemuxEvent::TracksResolved { .. } => {}
}
}
Ok(Media::new(tracks, VIDEO_TIMESCALE).with_pcr(pcr))
}
}
impl<'a> Unpackage for TsDemux<'a> {
type Input = &'a [u8];
type Media = Media;
type Error = Error;
fn unpackage(&mut self, input: &'a [u8]) -> Result<Media> {
self.demux(input)
}
}
#[cfg(test)]
mod tests {
use super::*;
use broadcast_common::Parse;
#[test]
fn data_carriage_classifies_every_known_stream_type() {
assert_eq!(data_carriage(0x06), DataCarriage::Pes, "PES private data");
assert_eq!(data_carriage(0x15), DataCarriage::Pes, "metadata in PES");
assert_eq!(data_carriage(0x7F), DataCarriage::Pes, "unrecognised → PES");
for &st in &[
STREAM_TYPE_PRIVATE_SECTIONS,
STREAM_TYPE_DSMCC_TYPE_A,
STREAM_TYPE_DSMCC_TYPE_B,
STREAM_TYPE_DSMCC_TYPE_C,
STREAM_TYPE_DSMCC_TYPE_D,
STREAM_TYPE_DSMCC_SYNC_DOWNLOAD,
STREAM_TYPE_SCTE35,
] {
assert_eq!(
data_carriage(st),
DataCarriage::Sections,
"stream_type {st:#04X} must be section-carried"
);
}
}
#[test]
fn from_stream_type_unknown_becomes_opaque_data() {
assert_eq!(Codec::from_stream_type(STREAM_TYPE_AVC), Codec::H264);
assert_eq!(Codec::from_stream_type(0x7F), Codec::Data(0x7F));
assert_eq!(
Codec::from_stream_type(STREAM_TYPE_SCTE35),
Codec::Data(STREAM_TYPE_SCTE35)
);
}
const PACKET_PAYLOAD_LEN: usize = TS_PACKET_SIZE - 4;
fn payload_only_packet(pid: u16, cc: u8) -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47; p[1] = ((pid >> 8) as u8) & PID_HI_MASK; p[2] = (pid & 0xFF) as u8; p[3] = 0x10 | (cc & 0x0F); p
}
#[test]
fn unattributed_buffer_is_bounded_for_never_claimed_pid() {
let target_bytes = MAX_UNATTRIBUTED_BYTES * 3;
let packet_count = target_bytes / PACKET_PAYLOAD_LEN + 1;
let unclaimed_pid: u16 = 0x0123;
let mut demux = StreamingTsDemux::new();
for i in 0..packet_count {
demux.feed(&payload_only_packet(unclaimed_pid, i as u8));
}
assert!(
demux.unattributed_bytes <= MAX_UNATTRIBUTED_BYTES,
"unattributed_bytes {} exceeded cap {}",
demux.unattributed_bytes,
MAX_UNATTRIBUTED_BYTES
);
assert!(
demux.unattributed_bytes > 0,
"expected the never-claimed PID's payload to be buffered"
);
let actual: usize = demux
.unattributed
.values()
.flat_map(|q| q.iter())
.map(|(_, payload)| payload.len())
.sum();
assert_eq!(
actual, demux.unattributed_bytes,
"unattributed_bytes drifted from the real retained size"
);
}
#[test]
fn runaway_pes_without_payload_unit_start_is_bounded_not_unbounded() {
use crate::TsMux;
use crate::media::{Media, Track};
use crate::pipeline::{CodecConfig, Sample, TrackSpec};
use crate::rtp_sdp::avc_config_from_sprop;
use broadcast_common::Package;
const ES_PID: u16 = 0x0100;
let avc = avc_config_from_sprop("Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==").unwrap();
let spec = TrackSpec::new(
1,
VIDEO_TIMESCALE,
CodecConfig::Avc {
config: avc,
width: 0,
height: 0,
},
);
let frame_dur = VIDEO_TIMESCALE / 30;
let samples: Vec<Sample> = (0..3u32)
.map(|i| {
let nal = [0x65u8, 0xAA, i as u8];
let mut data = (nal.len() as u32).to_be_bytes().to_vec();
data.extend_from_slice(&nal);
let dts = i64::from(i) * i64::from(frame_dur);
Sample::new(data, Some(dts), Some(dts), Some(frame_dur), i == 0)
})
.collect();
let track = Track::new(spec, samples);
let media = Media::new(vec![track], VIDEO_TIMESCALE);
let ts_bytes = TsMux::default().package(&media).expect("mux to TS");
let mut demux = StreamingTsDemux::new();
demux.feed(&ts_bytes);
while demux.poll_event().is_some() {}
assert!(
demux.streams.contains_key(&ES_PID),
"expected the muxed video ES at PID {ES_PID:#06X} — TsMux's ES_PID_BASE convention"
);
let mut cc: u8 = 0;
let mut hit_cap = false;
for _ in 0..30_000u32 {
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = ((ES_PID >> 8) as u8) & PID_HI_MASK; pkt[2] = (ES_PID & 0xFF) as u8;
pkt[3] = 0x10 | (cc & 0x0F);
cc = cc.wrapping_add(1);
demux.feed(&pkt);
if matches!(
demux.poll_event(),
Some(DemuxEvent::Discontinuity { provenance, .. }) if provenance.pid == Some(ES_PID)
) {
hit_cap = true;
break;
}
}
assert!(
hit_cap,
"expected MAX_PES_BUFFER_BYTES to trip well within 30000 continuation \
packets (never grow unbounded)"
);
assert_eq!(
demux.streams.get(&ES_PID).unwrap().pes_bytes,
0,
"pes_bytes must reset to 0 once the cap trips"
);
const PUSI_BIT: u8 = 0x40; let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = PUSI_BIT | (((ES_PID >> 8) as u8) & PID_HI_MASK);
pkt[2] = (ES_PID & 0xFF) as u8;
pkt[3] = 0x10 | (cc & 0x0F);
pkt[4] = 0; demux.feed(&pkt);
assert_eq!(
demux.streams.get(&ES_PID).unwrap().pes_bytes,
PACKET_PAYLOAD_LEN,
"a fresh payload_unit_start must be accepted and start a new count"
);
}
#[test]
fn probe_backlog_is_bounded_for_both_the_never_resolving_pid_and_its_collateral_pid() {
use crate::TsMux;
use crate::media::{Media, Track};
use crate::pipeline::{CodecConfig, DataCarriage, Sample, TrackSpec};
use crate::rtp_sdp::avc_config_from_sprop;
use broadcast_common::Package;
const SAMPLE_BYTES: usize = 4096;
const SAMPLE_COUNT: u32 = 1200;
let frame_dur = VIDEO_TIMESCALE / 30;
let avc = avc_config_from_sprop("Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==").unwrap();
let video_spec = TrackSpec::new(
1,
VIDEO_TIMESCALE,
CodecConfig::Avc {
config: avc,
width: 0,
height: 0,
},
);
let video_samples: Vec<Sample> = (0..SAMPLE_COUNT)
.map(|i| {
let mut nal = alloc::vec![0x41u8]; nal.resize(SAMPLE_BYTES, 0xAA);
let mut data = (nal.len() as u32).to_be_bytes().to_vec();
data.extend_from_slice(&nal);
let dts = i64::from(i) * i64::from(frame_dur);
Sample::new(data, Some(dts), Some(dts), Some(frame_dur), false)
})
.collect();
let video_track = Track::new(video_spec, video_samples);
let data_spec = TrackSpec::new(
2,
VIDEO_TIMESCALE,
CodecConfig::Data {
stream_type: 0x7F,
descriptors: Vec::new(),
carriage: DataCarriage::Pes,
},
);
let data_samples: Vec<Sample> = (0..SAMPLE_COUNT)
.map(|i| {
let payload = alloc::vec![0xBBu8; SAMPLE_BYTES];
let dts = i64::from(i) * i64::from(frame_dur);
Sample::new(payload, Some(dts), Some(dts), Some(frame_dur), true)
})
.collect();
let data_track = Track::new(data_spec, data_samples);
let media = Media::new(vec![video_track, data_track], VIDEO_TIMESCALE);
let ts_bytes = TsMux::default().package(&media).expect("mux to TS");
let mut demux = StreamingTsDemux::new();
demux.feed(&ts_bytes);
let mut abandoned_pids: Vec<u16> = Vec::new();
while let Some(ev) = demux.poll_event() {
if let DemuxEvent::TrackAbandoned {
reason: AbandonReason::BudgetExceeded,
provenance,
..
} = ev
&& let Some(pid) = provenance.pid
{
abandoned_pids.push(pid);
}
}
assert!(
!abandoned_pids.is_empty(),
"expected at least one TrackAbandoned{{BudgetExceeded}} from an abandoned probe backlog \
(issue #774: this replaced the mis-typed Discontinuity this path used to emit)"
);
let mut demux2 = StreamingTsDemux::new();
demux2.feed(&ts_bytes);
let saw_discontinuity_for_abandoned_pid = std::iter::from_fn(|| demux2.poll_event())
.any(|ev| matches!(ev, DemuxEvent::Discontinuity { provenance, .. } if provenance.pid == Some(PID_A)));
assert!(
!saw_discontinuity_for_abandoned_pid,
"a probe-backlog-budget abandonment (issue #774) must be a TrackAbandoned, \
never a Discontinuity"
);
for (&pid, stream) in demux.streams.iter() {
assert!(
stream.backlog_bytes <= MAX_PROBE_BACKLOG_BYTES,
"PID {pid:#06X} backlog_bytes {} exceeded cap {MAX_PROBE_BACKLOG_BYTES}",
stream.backlog_bytes
);
}
const PID_A: u16 = 0x0100; let abandoned = matches!(
demux.streams.get(&PID_A).and_then(|s| s.track.as_ref()),
Some(TrackState::Abandoned)
);
assert!(abandoned, "PID A must be Abandoned, never Live");
const PID_B: u16 = 0x0101;
let pid_b_resolved = matches!(
demux.streams.get(&PID_B).and_then(|s| s.track.as_ref()),
Some(TrackState::Live(_)) | Some(TrackState::Abandoned)
);
assert!(
pid_b_resolved,
"PID B must reach a final disposition (Live or Abandoned), not stay wedged"
);
}
#[test]
fn track_abandoned_config_unrecoverable_fires_at_finish() {
use crate::TsMux;
use crate::media::{Media, Track};
use crate::pipeline::{CodecConfig, Sample, TrackSpec};
use crate::rtp_sdp::avc_config_from_sprop;
use broadcast_common::Package;
const PID_A: u16 = 0x0100; let frame_dur = VIDEO_TIMESCALE / 30;
let avc = avc_config_from_sprop("Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==").unwrap();
let video_spec = TrackSpec::new(
1,
VIDEO_TIMESCALE,
CodecConfig::Avc {
config: avc,
width: 0,
height: 0,
},
);
let video_samples: Vec<Sample> = (0..5u32)
.map(|i| {
let nal = alloc::vec![0x41u8, 0xAA, 0xBB]; let mut data = (nal.len() as u32).to_be_bytes().to_vec();
data.extend_from_slice(&nal);
let dts = i64::from(i) * i64::from(frame_dur);
Sample::new(data, Some(dts), Some(dts), Some(frame_dur), false)
})
.collect();
let video_track = Track::new(video_spec, video_samples);
let media = Media::new(vec![video_track], VIDEO_TIMESCALE);
let ts_bytes = TsMux::default().package(&media).expect("mux to TS");
let mut demux = StreamingTsDemux::new();
demux.feed(&ts_bytes);
assert!(
!matches!(
demux.streams.get(&PID_A).and_then(|s| s.track.as_ref()),
Some(TrackState::Abandoned)
),
"sanity: PID A must still be Probing before finish() — the byte cap must not \
have tripped (this test is about the end-of-input path, not the budget one)"
);
while demux.poll_event().is_some() {}
demux.finish();
let mut saw_config_unrecoverable = false;
while let Some(ev) = demux.poll_event() {
if let DemuxEvent::TrackAbandoned {
track_id,
reason: AbandonReason::ConfigUnrecoverable,
provenance,
} = ev
{
assert_eq!(
track_id, None,
"a track abandoned before ever resolving has no track_id to report"
);
assert_eq!(provenance.pid, Some(PID_A));
saw_config_unrecoverable = true;
}
}
assert!(
saw_config_unrecoverable,
"expected TrackAbandoned{{ConfigUnrecoverable}} once finish() concludes PID A's \
config will never resolve"
);
}
const SECTION_BODY_PREFIX_LEN: usize = 5;
const ES_STREAM_ID_VIDEO: u8 = 0xE0;
fn pes_packet_untimed(stream_id: u8, au: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
out.extend_from_slice(&[0x00, 0x00, 0x01, stream_id]);
let payload_len = au.len() + 3;
out.extend_from_slice(&(payload_len as u16).to_be_bytes());
out.push(0x80); out.push(0x00); out.push(0x00); out.extend_from_slice(au);
out
}
const ES_STREAM_ID_PRIVATE_DATA: u8 = 0xBD;
fn data_au(seed: usize) -> Vec<u8> {
(0..300).map(|i| ((i + seed) & 0xFF) as u8).collect()
}
fn long_data_au() -> Vec<u8> {
data_au(0)
}
fn long_data_au_alt() -> Vec<u8> {
data_au(0x55)
}
fn unbounded_pes(stream_id: u8, au: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
out.extend_from_slice(&[0x00, 0x00, 0x01, stream_id]);
out.extend_from_slice(&0u16.to_be_bytes()); out.push(0x80); out.push(0x00); out.push(0x00); out.extend_from_slice(au);
out
}
fn untimed_pes(stream_id: u8, au: &[u8]) -> Vec<u8> {
pes_packet_untimed(stream_id, au)
}
const ES_STREAM_ID_AUDIO: u8 = 0xBD;
fn encode_ts_field(prefix: u8, value: i64) -> [u8; 5] {
let v = (value as u64) & TS_WRAP_MASK;
[
(prefix << 4) | ((((v >> 30) & 0x07) as u8) << 1) | 0x01,
((v >> 22) & 0xFF) as u8,
((((v >> 15) & 0x7F) as u8) << 1) | 0x01,
((v >> 7) & 0xFF) as u8,
(((v & 0x7F) as u8) << 1) | 0x01,
]
}
fn pes_packet(stream_id: u8, pts: i64, dts: Option<i64>, au: &[u8]) -> Vec<u8> {
let header_data_len = if dts.is_some() { 10 } else { 5 };
let mut out = Vec::new();
out.extend_from_slice(&[0x00, 0x00, 0x01, stream_id]);
let payload_len = au.len() + 3 + header_data_len;
out.extend_from_slice(&(payload_len as u16).to_be_bytes());
out.push(0x80); out.push(if dts.is_some() { 0xC0 } else { 0x80 });
out.push(header_data_len as u8);
if let Some(d) = dts {
out.extend_from_slice(&encode_ts_field(0b0011, pts));
out.extend_from_slice(&encode_ts_field(0b0001, d));
} else {
out.extend_from_slice(&encode_ts_field(0b0010, pts));
}
out.extend_from_slice(au);
out
}
const PAT_PID_UNDER_TEST: u16 = 0x0000;
const PMT_PID_UNDER_TEST: u16 = 0x1000;
const ES_PID_UNDER_TEST: u16 = 0x1100;
const PROGRAM_NUMBER_UNDER_TEST: u16 = 1;
fn pat_body() -> Vec<u8> {
let mut body = Vec::new();
body.extend_from_slice(&PROGRAM_NUMBER_UNDER_TEST.to_be_bytes());
body.extend_from_slice(&(0xE000 | PMT_PID_UNDER_TEST).to_be_bytes());
body
}
fn pmt_body(es_pid: u16, stream_type: u8) -> Vec<u8> {
pmt_body_with_pcr_pid(es_pid, stream_type, 0)
}
fn pmt_body_two_es_with_pcr_pid(
video_pid: u16,
video_type: u8,
audio_pid: u16,
audio_type: u8,
pcr_pid: u16,
) -> Vec<u8> {
let mut body = Vec::new();
body.extend_from_slice(&(0xE000 | pcr_pid).to_be_bytes());
body.extend_from_slice(&0xF000u16.to_be_bytes()); for (pid, stream_type) in [(video_pid, video_type), (audio_pid, audio_type)] {
body.push(stream_type);
body.extend_from_slice(&(0xE000 | pid).to_be_bytes());
body.extend_from_slice(&0xF000u16.to_be_bytes()); }
body
}
fn pmt_body_with_pcr_pid(es_pid: u16, stream_type: u8, pcr_pid: u16) -> Vec<u8> {
let mut body = Vec::new();
body.extend_from_slice(&(0xE000 | pcr_pid).to_be_bytes());
body.extend_from_slice(&0xF000u16.to_be_bytes()); body.push(stream_type);
body.extend_from_slice(&(0xE000 | es_pid).to_be_bytes());
body.extend_from_slice(&0xF000u16.to_be_bytes()); body
}
fn psi_section_packets(pid: u16, cc0: u8, table_id: u8, body: &[u8]) -> Vec<u8> {
psi_section_packets_for(pid, cc0, table_id, PROGRAM_NUMBER_UNDER_TEST, body)
}
fn psi_section_packets_for(
pid: u16,
cc0: u8,
table_id: u8,
table_id_extension: u16,
body: &[u8],
) -> Vec<u8> {
let mut section = Vec::new();
let section_length = (SECTION_BODY_PREFIX_LEN + body.len() + CRC32_LEN) as u16;
section.push(table_id);
section.push(SECTION_SYNTAX_INDICATOR_BIT | (section_length >> 8) as u8);
section.push(section_length as u8);
section.extend_from_slice(&table_id_extension.to_be_bytes());
section.push(0xC1); section.push(0x00); section.push(0x00); section.extend_from_slice(body);
let crc = broadcast_common::crc32_mpeg2::compute(§ion);
section.extend_from_slice(&crc.to_be_bytes());
let mut packetised = Vec::with_capacity(section.len() + 1);
packetised.push(0x00);
packetised.extend_from_slice(§ion);
pes_packets(pid, cc0, &packetised)
}
fn pes_packets_with_discontinuity(pid: u16, cc0: u8, payload: &[u8]) -> Vec<u8> {
const AF_LEN: usize = 8;
const PAYLOAD_START: usize = 4 + 1 + AF_LEN;
let first_take = (TS_PACKET_SIZE - PAYLOAD_START).min(payload.len());
let mut out = Vec::new();
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = ((pid >> 8) as u8 & PID_HI_MASK) | 0x40; p[2] = (pid & 0xFF) as u8;
p[3] = 0x30 | (cc0 & 0x0F); p[4] = AF_LEN as u8;
p[5] = 0x80; p[6..PAYLOAD_START].fill(0xFF); p[PAYLOAD_START..PAYLOAD_START + first_take].copy_from_slice(&payload[..first_take]);
out.extend_from_slice(&p);
out.extend_from_slice(&continuation_packets(
pid,
cc0.wrapping_add(1),
&payload[first_take..],
));
out
}
fn continuation_packets(pid: u16, cc0: u8, payload: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
let mut off = 0usize;
let mut cc = cc0;
while off < payload.len() {
let take = (payload.len() - off).min(TS_PACKET_SIZE - 4);
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = (pid >> 8) as u8 & PID_HI_MASK;
p[2] = (pid & 0xFF) as u8;
p[3] = 0x10 | (cc & 0x0F);
p[4..4 + take].copy_from_slice(&payload[off..off + take]);
out.extend_from_slice(&p);
off += take;
cc = cc.wrapping_add(1);
}
out
}
fn pes_packets(pid: u16, cc0: u8, payload: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
let mut off = 0usize;
let mut first = true;
let mut cc = cc0;
while off < payload.len() {
let take = (payload.len() - off).min(TS_PACKET_SIZE - 4);
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = ((pid >> 8) as u8 & PID_HI_MASK) | if first { 0x40 } else { 0x00 };
p[2] = (pid & 0xFF) as u8;
p[3] = 0x10 | (cc & 0x0F);
p[4..4 + take].copy_from_slice(&payload[off..off + take]);
out.extend_from_slice(&p);
off += take;
first = false;
cc = cc.wrapping_add(1);
}
out
}
#[test]
fn aac_frame_length_variation_is_not_a_config_change() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("aac-5_1-640k.ts");
let source =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let count_updates = |bytes: &[u8]| -> (usize, Vec<u16>) {
let mut demux = StreamingTsDemux::new();
demux.feed(bytes);
demux.finish();
let mut n = 0usize;
let mut channels = Vec::new();
while let Some(event) = demux.poll_event() {
if let DemuxEvent::TrackUpdated(spec) = event {
n += 1;
if let CodecConfig::Aac { channel_count, .. } = spec.config {
channels.push(channel_count);
}
}
}
(n, channels)
};
let (updates, _) = count_updates(&source);
assert_eq!(
updates, 0,
"a varying ADTS frame_length must not read as a config change; got {updates} TrackUpdated events on a real AAC capture"
);
let mut demux = TsDemux::new();
let media = demux.demux(&source).expect("demux aac fixture");
let audio = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Aac { .. }))
.expect("the fixture has an AAC track");
assert!(!audio.samples.is_empty(), "fixture must carry AAC frames");
let (sfi, seeded_channels) = match audio.config() {
CodecConfig::Aac {
esds,
channel_count,
..
} => {
let dsi = esds
.es_descriptor
.decoder_config
.as_ref()
.and_then(|dc| dc.decoder_specific_info.as_ref())
.expect("AAC esds carries a DecoderSpecificInfo");
let asc = AudioSpecificConfig::parse(&dsi.data).expect("parse ASC");
(asc.sampling_frequency_index.raw(), *channel_count)
}
_ => unreachable!(),
};
const SEEDED_CHANNELS: u16 = 6;
assert_eq!(
seeded_channels, SEEDED_CHANNELS,
"the fixture is a real 5.1 capture, so its config is Table 1.19 configuration 6"
);
const CONFIG_REBUILT: u8 = 5;
const PES_PAYLOAD_BYTES: usize = 1800;
let aac_pid = audio
.spec
.source_pid
.expect("a TS-demuxed track carries its source PID");
let mut input = source.clone();
let mut cc = 0u8;
let mut buf: Vec<u8> = Vec::new();
let flush = |buf: &mut Vec<u8>, cc: &mut u8, input: &mut Vec<u8>| {
if buf.is_empty() {
return;
}
let pes = pes_packet(ES_STREAM_ID_AUDIO, 0, None, buf);
let packets = pes_packets(aac_pid, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
buf.clear();
};
for sample in &audio.samples {
let frame_len = (ADTS_HEADER_SIZE + sample.data.len()) as u16;
let hdr = crate::aac_asc::build_adts_header(1, sfi, CONFIG_REBUILT, frame_len);
if buf.len() + frame_len as usize > PES_PAYLOAD_BYTES {
flush(&mut buf, &mut cc, &mut input);
}
buf.extend_from_slice(&hdr);
buf.extend_from_slice(&sample.data);
}
flush(&mut buf, &mut cc, &mut input);
let (updates, channels) = count_updates(&input);
assert_eq!(
updates, 1,
"a real channel_configuration change must raise exactly one TrackUpdated, got {updates} (channels: {channels:?})"
);
assert_eq!(
channels,
vec![5],
"the event must carry the new Table 1.19 count"
);
}
#[test]
fn probe_sync_scan_work_is_linear_in_the_input_not_quadratic() {
use crate::aac_asc::build_adts_header;
const REJECTED_SFI: u8 = 13;
const FRAME_BODY: usize = 1400;
const FRAMES_PER_AU: usize = 3;
const ACCESS_UNITS: usize = 500;
let frame_len = (ADTS_HEADER_SIZE + FRAME_BODY) as u16;
let mut frame = alloc::vec![0u8; frame_len as usize];
frame[..ADTS_HEADER_SIZE].copy_from_slice(&build_adts_header(
1,
REJECTED_SFI,
2,
frame_len,
));
let mut au = Vec::with_capacity(frame.len() * FRAMES_PER_AU);
for _ in 0..FRAMES_PER_AU {
au.extend_from_slice(&frame);
}
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_AAC_ADTS),
));
let mut cc = 0u8;
for i in 0..ACCESS_UNITS {
let pes = pes_packet(ES_STREAM_ID_AUDIO, i as i64 * 1024, None, &au);
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
}
let before = sync_probes();
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let probes = sync_probes() - before;
assert!(
matches!(
demux
.streams
.get(&ES_PID_UNDER_TEST)
.and_then(|s| s.track.as_ref()),
Some(TrackState::Probing { .. })
),
"the reserved-SFI AAC PID must never resolve its config, so it stays Probing"
);
let expected = ACCESS_UNITS as u64 * 2;
assert_eq!(
probes, expected,
"the AAC probe must scan only the newest access unit ({ACCESS_UNITS} accesses => {expected} probes: one pass + one sync scan each); {probes} recorded"
);
}
#[test]
fn video_and_mpeg_audio_probe_work_is_linear_in_the_input() {
const ACCESS_UNITS: u32 = 400;
const AU_BYTES: usize = 1024;
let (sps, pps, _) = real_avc_parameter_sets();
let broken_sps = &sps[..sps.len().min(3)];
for (stream_type, au) in [
(
STREAM_TYPE_AVC,
annexb_au(&[broken_sps, &pps, &[0x41u8; 8]]),
),
(
STREAM_TYPE_MPEG1_AUDIO,
alloc::vec![0xE0u8; AU_BYTES],
),
] {
let mut au = au;
au.resize(AU_BYTES, 0x33);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, stream_type),
));
let mut cc = 0u8;
for i in 0..ACCESS_UNITS {
let pes = pes_packet(ES_STREAM_ID_AUDIO, i as i64 * 1024, None, &au);
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
}
let before = sync_probes();
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let probes = sync_probes() - before;
assert!(
matches!(
demux
.streams
.get(&ES_PID_UNDER_TEST)
.and_then(|s| s.track.as_ref()),
Some(TrackState::Probing { .. }) | Some(TrackState::Abandoned)
),
"stream_type {stream_type:#04X} must never resolve its config"
);
if stream_type == STREAM_TYPE_AVC {
assert_eq!(
probes, ACCESS_UNITS as u64,
"the H.264 probe must make exactly one pass per access unit, never a backlog rescan: {probes} probes over {ACCESS_UNITS} access units"
);
} else {
let bound = ACCESS_UNITS as u64 * (1 + AU_BYTES as u64);
assert!(
probes <= bound,
"stream_type {stream_type:#04X}: {probes} probes over {ACCESS_UNITS} access units of {AU_BYTES} bytes — bound is {bound}; a quadratic rescan is ~{}",
(ACCESS_UNITS as u64).pow(2) / 2
);
}
}
}
fn avc_ts_over(aus: &[Vec<u8>]) -> Vec<u8> {
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_AVC),
));
for (i, au) in aus.iter().enumerate() {
let pes = pes_packet(
ES_STREAM_ID_VIDEO,
i as i64 * 3600,
Some(i as i64 * 3600),
au,
);
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
}
input
}
fn updated_avc_width(bytes: &[u8]) -> Option<u16> {
let mut demux = StreamingTsDemux::new();
demux.feed(bytes);
demux.finish();
std::iter::from_fn(|| demux.poll_event()).find_map(|e| match e {
DemuxEvent::TrackUpdated(spec) => match spec.config {
CodecConfig::Avc { width, .. } => Some(width),
_ => None,
},
_ => None,
})
}
#[test]
fn signalled_discontinuity_is_scoped_to_the_signalling_program() {
const A_PMT: u16 = 0x1000;
const A_ES: u16 = 0x1100;
const A_ES2: u16 = 0x1101;
const B_PMT: u16 = 0x1001;
const B_ES: u16 = 0x1200;
const A_PROGRAM: u16 = 1;
const B_PROGRAM: u16 = 2;
const FRAME_PERIOD: i64 = 3600;
const FRAMES: i64 = 8;
const A_BASE_B: i64 = 40_000;
const BASE_A: i64 = 900_000;
let annexb = real_annexb_frames();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
let mut pat = Vec::new();
for (program, pmt_pid) in [(A_PROGRAM, A_PMT), (B_PROGRAM, B_PMT)] {
pat.extend_from_slice(&program.to_be_bytes());
pat.extend_from_slice(&(0xE000 | pmt_pid).to_be_bytes());
}
input.extend_from_slice(&psi_section_packets_for(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
0,
&pat,
));
input.extend_from_slice(&psi_section_packets_for(
A_PMT,
0,
TABLE_ID_PMT,
A_PROGRAM,
&pmt_body_two_es_with_pcr_pid(A_ES, STREAM_TYPE_AVC, A_ES2, STREAM_TYPE_AVC, A_ES),
));
input.extend_from_slice(&psi_section_packets_for(
B_PMT,
0,
TABLE_ID_PMT,
B_PROGRAM,
&pmt_body_with_pcr_pid(B_ES, STREAM_TYPE_AVC, B_ES),
));
let push_au = |input: &mut Vec<u8>, cc: &mut u8, pid: u16, au: &[u8], dts: i64| {
let wire = dts.rem_euclid(1i64 << 33) as u64;
let pes = pes_packet(ES_STREAM_ID_VIDEO, wire as i64, Some(wire as i64), au);
let packets = pes_packets(pid, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
for i in 0..FRAMES {
let au = annexb[(i as usize) % annexb.len()].clone();
push_au(&mut input, &mut cc, A_ES, &au, BASE_A + i * FRAME_PERIOD);
}
{
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = (A_ES >> 8) as u8 & PID_HI_MASK;
pkt[2] = (A_ES & 0xFF) as u8;
pkt[3] = 0x20 | (cc & 0x0F); pkt[4] = 8;
pkt[5] = 0x80; input.extend_from_slice(&pkt);
cc = cc.wrapping_add(1);
}
for i in 0..FRAMES {
let au = annexb[(i as usize) % annexb.len()].clone();
push_au(&mut input, &mut cc, A_ES, &au, A_BASE_B + i * FRAME_PERIOD);
}
for i in 0..FRAMES * 2 {
let au = annexb[(i as usize) % annexb.len()].clone();
let dts = if i < FRAMES {
BASE_A + i * FRAME_PERIOD
} else {
A_BASE_B + (i - FRAMES) * FRAME_PERIOD
};
push_au(&mut input, &mut cc, A_ES2, &au, dts);
}
for i in 0..FRAMES * 2 {
let au = annexb[(i as usize) % annexb.len()].clone();
push_au(&mut input, &mut cc, B_ES, &au, BASE_A + i * FRAME_PERIOD);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let events: Vec<DemuxEvent> = std::iter::from_fn(|| demux.poll_event()).collect();
let ids: BTreeMap<u32, u16> = events
.iter()
.filter_map(|e| match e {
DemuxEvent::TrackAdded(spec) => spec.source_pid.map(|pid| (spec.track_id, pid)),
_ => None,
})
.collect();
let mut samples_by_pid: BTreeMap<u16, Vec<i64>> = BTreeMap::new();
for event in &events {
if let DemuxEvent::Sample { track_id, sample } = event
&& let Some(pid) = ids.get(track_id)
&& let Some(dts) = sample.dts
{
samples_by_pid.entry(*pid).or_default().push(dts);
}
}
let a_dts = samples_by_pid
.get(&A_ES)
.cloned()
.expect("program A's video PID must deliver samples");
let b_dts = samples_by_pid
.get(&B_ES)
.cloned()
.expect("program B must deliver samples");
assert!(
b_dts.len() >= FRAMES as usize,
"program B must deliver its samples, got {}",
b_dts.len()
);
let expected: Vec<i64> = (0..FRAMES * 2).map(|i| BASE_A + i * FRAME_PERIOD).collect();
let got: Vec<i64> = b_dts.iter().copied().take(expected.len()).collect();
assert_eq!(
got, expected,
"program B's dts must be untouched by program A's signalled discontinuity"
);
let a_steps: Vec<i64> = a_dts.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
!a_steps.is_empty() && a_steps.iter().all(|&d| d > 0),
"program A's video timeline must be strictly increasing across its discontinuity: {a_dts:?}"
);
let a2 = samples_by_pid
.get(&A_ES2)
.expect("program A's second elementary stream must deliver samples");
let a2_steps: Vec<i64> = a2.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
!a2_steps.is_empty() && a2_steps.iter().all(|&d| d > 0),
"program A's *second* PID must be rebased by the discontinuity on its PCR_PID too: {a2:?}"
);
}
fn real_annexb_frames() -> Vec<Vec<u8>> {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let source =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let media = TsDemux::new().demux(&source).expect("demux h264_aac.ts");
let video = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Avc { .. }))
.expect("h264_aac.ts has an AVC track");
video
.samples
.iter()
.map(|sample| {
let mut au = Vec::new();
for nal in crate::annexb::iter_length_prefixed_nals(&sample.data)
.expect("TsMux writes valid NAL prefixes")
{
au.extend_from_slice(&[0x00, 0x00, 0x00, 0x01]);
au.extend_from_slice(nal);
}
au
})
.collect()
}
#[test]
fn jittered_cadence_backward_discontinuity_rebases() {
const PERIODS: [i64; 2] = [3754, 3753];
const BASE_A: i64 = 900_000;
const BASE_B: i64 = 120_000;
const FRAMES: i64 = 30;
let annexb = real_annexb_frames();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body_with_pcr_pid(ES_PID_UNDER_TEST, STREAM_TYPE_AVC, ES_PID_UNDER_TEST),
));
let push = |input: &mut Vec<u8>, cc: &mut u8, au: &[u8], dts: i64| {
let wire = dts.rem_euclid(1i64 << 33);
let pes = pes_packet(ES_STREAM_ID_VIDEO, wire, Some(wire), au);
let packets = pes_packets(ES_PID_UNDER_TEST, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
let mut dts = BASE_A;
for i in 0..FRAMES * 2 {
if i == FRAMES {
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = (ES_PID_UNDER_TEST >> 8) as u8 & PID_HI_MASK;
pkt[2] = (ES_PID_UNDER_TEST & 0xFF) as u8;
pkt[3] = 0x20 | (cc & 0x0F);
pkt[4] = 8;
pkt[5] = 0x80; input.extend_from_slice(&pkt);
cc = cc.wrapping_add(1);
dts = BASE_B;
}
let au = annexb[(i as usize) % annexb.len()].clone();
push(&mut input, &mut cc, &au, dts);
dts += PERIODS[(i as usize) % PERIODS.len()];
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let got: Vec<i64> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => sample.dts,
_ => None,
})
.collect();
assert!(
got.len() >= FRAMES as usize,
"expected the access units of both bases, got {}",
got.len()
);
let steps: Vec<i64> = got.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
steps.iter().all(|&d| d > 0),
"dts must be strictly increasing across a jittered discontinuity; dts={got:?}"
);
let seam = steps
.iter()
.copied()
.filter(|&d| d > PERIODS[1])
.max()
.unwrap_or(0);
assert!(
seam <= PERIODS[0] + 1,
"the seam must continue the cadence (one period), not jump: largest step {seam}, steps={steps:?}"
);
}
#[test]
fn an_eight_second_splice_does_not_become_the_frame_period() {
const FRAME_PERIOD: i64 = 3600;
const SPLICE: i64 = 8 * 90_000;
const STAMPED: i64 = 8;
const UNSTAMPED: usize = 4;
let annexb = real_annexb_frames();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_AVC),
));
let push = |input: &mut Vec<u8>, cc: &mut u8, au: &[u8], dts: Option<i64>| {
let pes = match dts {
Some(d) => {
let wire = d.rem_euclid(1i64 << 33);
pes_packet(ES_STREAM_ID_VIDEO, wire, Some(wire), au)
}
None => pes_packet_untimed(ES_STREAM_ID_VIDEO, au),
};
let packets = pes_packets(ES_PID_UNDER_TEST, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
for i in 0..STAMPED {
let au = annexb[(i as usize) % annexb.len()].clone();
push(&mut input, &mut cc, &au, Some(90_000 + i * FRAME_PERIOD));
}
let mut at = 90_000 + (STAMPED - 1) * FRAME_PERIOD + SPLICE;
for i in 0..6usize {
let au = annexb[i % annexb.len()].clone();
push(&mut input, &mut cc, &au, Some(at));
at += SPLICE;
}
let after_splice = at - SPLICE;
for i in 1..=UNSTAMPED {
let au = annexb[i % annexb.len()].clone();
push(&mut input, &mut cc, &au, None);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let dts: Vec<i64> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => sample.dts,
_ => None,
})
.collect();
let tail: Vec<i64> = dts.iter().copied().filter(|&d| d > after_splice).collect();
assert!(
tail.len() >= UNSTAMPED - 1,
"expected the unstamped frames to be delivered, got {dts:?}"
);
let steps: Vec<i64> = tail.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
steps.iter().all(|&d| d == FRAME_PERIOD),
"the splice must not become the frame period: steps={steps:?} (an adopted 8-second period would give {SPLICE})"
);
}
#[test]
fn truncated_final_bounded_pes_is_dropped_at_flush() {
const BODY: usize = 300;
let au: Vec<u8> = (0..BODY).map(|i| (i & 0xFF) as u8).collect();
let pes = untimed_pes(ES_STREAM_ID_PRIVATE_DATA, &au);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let packets = pes_packets(ES_PID_UNDER_TEST, 0, &pes);
assert_eq!(
packets.len(),
2 * TS_PACKET_SIZE,
"the unit spans two packets"
);
input.extend_from_slice(&packets[..TS_PACKET_SIZE]);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let samples = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::Sample { .. }))
.count();
assert_eq!(
samples, 0,
"a bounded PES that never reached its declared length must not be delivered by finish()"
);
}
#[test]
fn wrap_coincident_discontinuity_keeps_one_frame_period_steps() {
const FRAME_PERIOD: i64 = 3600;
const FRAMES_A: i64 = 3;
const WRAP: i64 = 1i64 << 33;
const BASE_A: i64 = WRAP - (FRAMES_A + 1) * FRAME_PERIOD;
const A_LAST: i64 = BASE_A + (FRAMES_A - 1) * FRAME_PERIOD;
const BASE_B: i64 = BASE_A + FRAMES_A * FRAME_PERIOD - WRAP;
let annexb = real_annexb_frames();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body_with_pcr_pid(ES_PID_UNDER_TEST, STREAM_TYPE_AVC, ES_PID_UNDER_TEST),
));
let push = |input: &mut Vec<u8>, cc: &mut u8, au: &[u8], raw: i64| {
let wire = raw.rem_euclid(WRAP);
let pes = pes_packet(ES_STREAM_ID_VIDEO, wire, Some(wire), au);
let packets = pes_packets(ES_PID_UNDER_TEST, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
for i in 0..FRAMES_A {
let au = annexb[(i as usize) % annexb.len()].clone();
push(&mut input, &mut cc, &au, BASE_A + i * FRAME_PERIOD);
}
{
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = (ES_PID_UNDER_TEST >> 8) as u8 & PID_HI_MASK;
pkt[2] = (ES_PID_UNDER_TEST & 0xFF) as u8;
pkt[3] = 0x20 | (cc & 0x0F);
pkt[4] = 8;
pkt[5] = 0x80; input.extend_from_slice(&pkt);
cc = cc.wrapping_add(1);
}
for i in 0..FRAMES_A + 2 {
let au = annexb[((FRAMES_A + i) as usize) % annexb.len()].clone();
push(&mut input, &mut cc, &au, BASE_B + i * FRAME_PERIOD);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let dts: Vec<i64> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => sample.dts,
_ => None,
})
.collect();
assert!(
dts.len() >= (FRAMES_A + 2) as usize,
"expected the access units of both bases, got {}",
dts.len()
);
let steps: Vec<i64> = dts.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
steps.iter().all(|&d| d == FRAME_PERIOD),
"every step must stay exactly one frame period across the wrap and the seam; dts={dts:?} steps={steps:?}"
);
let first_of_b = dts[dts.len() - (FRAMES_A as usize + 2)];
assert!(
first_of_b > A_LAST && first_of_b - A_LAST < 2 * FRAME_PERIOD,
"base B must continue base A across the wrap, not restart near zero: first_of_b={first_of_b} last_of_a={A_LAST}"
);
}
#[test]
fn forward_jump_at_a_signalled_discontinuity_is_not_collapsed() {
const FRAME_PERIOD: i64 = 3600;
const FRAMES: i64 = 6;
const BASE_A: i64 = 900_000;
const FORWARD_JUMP: i64 = 8 * 90_000;
let annexb = real_annexb_frames();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body_with_pcr_pid(ES_PID_UNDER_TEST, STREAM_TYPE_AVC, ES_PID_UNDER_TEST),
));
let push = |input: &mut Vec<u8>, cc: &mut u8, au: &[u8], dts: i64| {
let wire = dts.rem_euclid(1i64 << 33);
let pes = pes_packet(ES_STREAM_ID_VIDEO, wire, Some(wire), au);
let packets = pes_packets(ES_PID_UNDER_TEST, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
for i in 0..FRAMES {
let au = annexb[(i as usize) % annexb.len()].clone();
push(&mut input, &mut cc, &au, BASE_A + i * FRAME_PERIOD);
}
let base_b = BASE_A + (FRAMES - 1) * FRAME_PERIOD + FORWARD_JUMP;
{
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = (ES_PID_UNDER_TEST >> 8) as u8 & PID_HI_MASK;
pkt[2] = (ES_PID_UNDER_TEST & 0xFF) as u8;
pkt[3] = 0x20 | (cc & 0x0F);
pkt[4] = 8;
pkt[5] = 0x80;
input.extend_from_slice(&pkt);
cc = cc.wrapping_add(1);
}
for i in 0..FRAMES {
let au = annexb[(i as usize) % annexb.len()].clone();
push(&mut input, &mut cc, &au, base_b + i * FRAME_PERIOD);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let dts: Vec<i64> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => sample.dts,
_ => None,
})
.collect();
assert!(dts.len() >= 2, "expected samples from both bases");
let steps: Vec<i64> = dts.windows(2).map(|w| w[1] - w[0]).collect();
let seam = steps
.iter()
.copied()
.find(|&d| d > FRAME_PERIOD)
.unwrap_or_else(|| {
panic!("the forward jump must survive as a large step, got {steps:?}")
});
assert!(
seam >= FORWARD_JUMP,
"a forward jump of {FORWARD_JUMP} ticks must not be collapsed to {seam}; the offset must never be reduced"
);
assert!(
steps.iter().all(|&d| d > 0),
"and the timeline must still be strictly increasing: {steps:?}"
);
}
fn real_avc_parameter_sets() -> (Vec<u8>, Vec<u8>, u16) {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let source =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let media = TsDemux::new().demux(&source).expect("demux h264_aac.ts");
let video = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Avc { .. }))
.expect("h264_aac.ts has an AVC track");
let width = match video.config() {
CodecConfig::Avc { width, .. } => *width,
_ => unreachable!(),
};
let first = &video.samples[0].data;
let nals = crate::annexb::iter_length_prefixed_nals(first)
.expect("TsMux writes valid NAL prefixes");
let sps = nals
.iter()
.find(|n| (n[0] & H264_NAL_TYPE_MASK) == H264_NAL_SPS)
.expect("the first access unit carries an SPS");
let pps = nals
.iter()
.find(|n| (n[0] & H264_NAL_TYPE_MASK) == H264_NAL_PPS)
.expect("the first access unit carries a PPS");
(sps.to_vec(), pps.to_vec(), width)
}
fn annexb_au(nals: &[&[u8]]) -> Vec<u8> {
let mut au = Vec::new();
for nal in nals {
au.extend_from_slice(&[0x00, 0x00, 0x00, 0x01]);
au.extend_from_slice(nal);
}
au
}
fn slice_only_au(tag: u8) -> Vec<u8> {
annexb_au(&[&[0x65, 0x88, 0x84, tag]])
}
#[test]
fn parameter_sets_in_one_access_unit_then_slices_emit_no_update() {
let (sps, pps, width) = real_avc_parameter_sets();
let mut aus = vec![annexb_au(&[&sps, &pps, &[0x65, 0x88, 0x84, 0x01]])];
for i in 0..4u8 {
aus.push(slice_only_au(i));
}
let bytes = avc_ts_over(&aus);
let mut demux = StreamingTsDemux::new();
demux.feed(&bytes);
demux.finish();
let added_width = std::iter::from_fn(|| demux.poll_event()).find_map(|e| match e {
DemuxEvent::TrackAdded(spec) => match spec.config {
CodecConfig::Avc { width, .. } => Some(width),
_ => None,
},
_ => None,
});
assert_eq!(
added_width,
Some(width),
"the track must resolve from the parameter sets in the first access unit"
);
assert_eq!(
updated_avc_width(&bytes),
None,
"slice-only access units carry no parameter sets, so nothing changed; a per-access-unit comparison raised a spurious TrackUpdated here"
);
}
#[test]
fn parameter_sets_split_across_access_units_emit_no_update() {
let (sps, pps, width) = real_avc_parameter_sets();
let aus = vec![
annexb_au(&[&sps, &[0x65, 0x88, 0x84, 0x01]]),
annexb_au(&[&pps, &[0x65, 0x88, 0x84, 0x02]]),
slice_only_au(0x03),
];
let bytes = avc_ts_over(&aus);
assert_eq!(
updated_avc_width(&bytes),
None,
"the PPS arriving one access unit after the SPS is not a change; a one-access-unit comparison reported one (and would have reported the SPS's access unit too). Track added at width {width}"
);
}
#[test]
fn a_changed_sps_raises_exactly_one_update() {
let (sps, pps, _) = real_avc_parameter_sets();
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("tests");
path.push("fixtures");
path.push("ts");
path.push("h264-two-resolutions.ts");
let other =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let media = TsDemux::new()
.demux(&other)
.expect("demux two-resolution fixture");
let second_sps = media
.tracks
.iter()
.filter(|t| matches!(t.config(), CodecConfig::Avc { .. }))
.flat_map(|t| t.samples.iter())
.flat_map(|s| {
crate::annexb::iter_length_prefixed_nals(&s.data)
.unwrap_or_default()
.into_iter()
.filter(|n| (n[0] & H264_NAL_TYPE_MASK) == H264_NAL_SPS)
.map(|n| n.to_vec())
.collect::<Vec<_>>()
})
.find(|s| s.as_slice() != sps.as_slice())
.expect("the two-resolution fixture carries a second, different SPS");
assert_ne!(second_sps, sps, "the two SPS really differ");
let aus = vec![
annexb_au(&[&sps, &pps, &[0x65, 0x88, 0x84, 0x01]]),
slice_only_au(0x02),
annexb_au(&[&second_sps, &pps, &[0x65, 0x88, 0x84, 0x03]]),
slice_only_au(0x04),
];
let bytes = avc_ts_over(&aus);
let mut demux = StreamingTsDemux::new();
demux.feed(&bytes);
demux.finish();
let updates: Vec<(u16, u16)> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::TrackUpdated(spec) => match spec.config {
CodecConfig::Avc { width, height, .. } => Some((width, height)),
_ => None,
},
_ => None,
})
.collect();
assert_eq!(
updates.len(),
1,
"exactly one SPS change, so exactly one TrackUpdated: {updates:?}"
);
assert_ne!(
updates[0],
(0, 0),
"the event must carry the newly-decoded geometry, not zeros"
);
}
#[test]
fn parameter_sets_split_across_access_units_raise_exactly_one_update() {
let (sps, pps, _) = real_avc_parameter_sets();
let aus = vec![
annexb_au(&[&sps, &[0x65, 0x88, 0x84, 0x01]]),
annexb_au(&[&[0x65, 0x88, 0x84, 0x02]]),
annexb_au(&[&pps, &[0x65, 0x88, 0x84, 0x03]]),
slice_only_au(0x04),
slice_only_au(0x05),
];
let bytes = avc_ts_over(&aus);
let mut demux = StreamingTsDemux::new();
demux.feed(&bytes);
demux.finish();
let updates = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::TrackUpdated(_)))
.count();
assert_eq!(
updates, 0,
"no parameter set ever changes value here — the SPS and the PPS just arrive in different access units — so nothing may be reported"
);
}
#[test]
fn alternating_parameter_set_carriage_emits_no_update() {
let (sps, pps, _) = real_avc_parameter_sets();
let mut aus = vec![annexb_au(&[&sps, &pps, &[0x65, 0x88, 0x84, 0x01]])];
for i in 0..4u8 {
aus.push(annexb_au(&[&pps, &[0x65, 0x88, 0x84, i + 2]]));
aus.push(annexb_au(&[&sps, &pps, &[0x65, 0x88, 0x84, i + 10]]));
}
let bytes = avc_ts_over(&aus);
let mut demux = StreamingTsDemux::new();
demux.feed(&bytes);
demux.finish();
let updates = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::TrackUpdated(_)))
.count();
assert_eq!(
updates, 0,
"the configuration never changes, so no TrackUpdated may be raised; a per-access-unit snapshot loses the SPS during the PPS-only units and reports each following full set as a change"
);
}
#[test]
fn a_changed_pps_raises_exactly_one_update() {
let (sps, pps, _) = real_avc_parameter_sets();
let mut other_pps = pps.clone();
let last = other_pps.len() - 1;
other_pps[last] ^= 0x55;
assert_ne!(other_pps, pps, "the two PPS really differ");
let aus = vec![
annexb_au(&[&sps, &pps, &[0x65, 0x88, 0x84, 0x01]]),
annexb_au(&[&sps, &other_pps, &[0x65, 0x88, 0x84, 0x02]]),
slice_only_au(0x03),
];
let bytes = avc_ts_over(&aus);
let mut demux = StreamingTsDemux::new();
demux.feed(&bytes);
demux.finish();
let updates = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::TrackUpdated(_)))
.count();
assert_eq!(
updates, 1,
"a changed PPS is a config change, so exactly one TrackUpdated"
);
}
#[test]
fn mid_stream_config_change_emits_track_updated_once_and_only_when_it_changes() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("tests");
path.push("fixtures");
path.push("ts");
path.push("h264-two-resolutions.ts");
let bytes =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let mut demux = StreamingTsDemux::new();
demux.feed(&bytes);
demux.finish();
let mut added: Option<(u32, u16, u16)> = None;
let mut updates: Vec<(u32, u16, u16)> = Vec::new();
while let Some(event) = demux.poll_event() {
match event {
DemuxEvent::TrackAdded(spec) => {
if let CodecConfig::Avc { width, height, .. } = spec.config {
added = Some((spec.track_id, width, height));
}
}
DemuxEvent::TrackUpdated(spec) => {
if let CodecConfig::Avc { width, height, .. } = spec.config {
updates.push((spec.track_id, width, height));
}
}
_ => {}
}
}
let (track_id, w0, h0) = added.expect("the fixture has one AVC track");
assert_eq!(
(w0, h0),
(320, 240),
"the first half of the fixture is 320x240 (ffprobe oracle)"
);
assert_eq!(
updates.len(),
1,
"the config changes exactly once in this fixture (320x240 -> 640x480); the encoder's repeated, unchanged SPS must not raise more. Got {updates:?}"
);
assert_eq!(
updates[0],
(track_id, 640, 480),
"TrackUpdated must carry the *new* config, on the same track id"
);
}
#[test]
fn adts_channel_configuration_is_mapped_through_table_1_19() {
const SFI_48000: u8 = 3;
const ADTS_PROFILE_AAC_LC: u8 = 1;
const CONFIG_7_1: u8 = 7;
const CONFIG_IN_BAND: u8 = 0;
for (config, expected) in [(CONFIG_7_1, 8u16), (CONFIG_IN_BAND, 0u16)] {
let frame_len = (ADTS_HEADER_SIZE + 64) as u16;
let mut au = crate::aac_asc::build_adts_header(
ADTS_PROFILE_AAC_LC,
SFI_48000,
config,
frame_len,
)
.to_vec();
au.resize(frame_len as usize, 0x21);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_AAC_ADTS),
));
let mut cc = 0u8;
for i in 0..4 {
let pes = pes_packet(ES_STREAM_ID_AUDIO, i * 1024, None, &au);
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let channel_count = std::iter::from_fn(|| demux.poll_event())
.find_map(|e| match e {
DemuxEvent::TrackAdded(spec) => match spec.config {
CodecConfig::Aac { channel_count, .. } => Some(channel_count),
_ => None,
},
_ => None,
})
.expect("the AAC PID must resolve a track");
assert_eq!(
channel_count, expected,
"ADTS channel_configuration {config} must map through Table 1.19"
);
}
}
#[test]
fn signalled_discontinuity_keeps_dts_monotonic() {
const BASE_A_DTS: i64 = 90_000;
const BASE_B_DTS: i64 = 9_000;
const FRAME_PERIOD: i64 = 3600;
const FRAMES: usize = 12;
const WRAP: i64 = 1i64 << 33;
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let source =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let media = TsDemux::new().demux(&source).expect("demux h264_aac.ts");
let video = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Avc { .. }))
.expect("h264_aac.ts has an AVC track");
assert!(
video.samples.len() >= FRAMES * 2,
"fixture too small for two time bases"
);
let annexb: Vec<Vec<u8>> = video
.samples
.iter()
.map(|sample| {
let mut au = Vec::new();
for nal in crate::annexb::iter_length_prefixed_nals(&sample.data)
.expect("TsMux writes valid NAL prefixes")
{
au.extend_from_slice(&[0x00, 0x00, 0x00, 0x01]);
au.extend_from_slice(nal);
}
au
})
.collect();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body_with_pcr_pid(ES_PID_UNDER_TEST, STREAM_TYPE_AVC, ES_PID_UNDER_TEST),
));
let push_au = |input: &mut Vec<u8>, cc: &mut u8, au: &[u8], dts: i64| {
let wire = (dts.rem_euclid(WRAP)) as u64;
let pes = pes_packet(ES_STREAM_ID_VIDEO, wire as i64, Some(wire as i64), au);
let packets = pes_packets(ES_PID_UNDER_TEST, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
for i in 0..FRAMES {
let au = annexb[i % annexb.len()].clone();
push_au(
&mut input,
&mut cc,
&au,
BASE_A_DTS + i as i64 * FRAME_PERIOD,
);
}
{
let af = [
0x80u8, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, ];
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = (ES_PID_UNDER_TEST >> 8) as u8 & PID_HI_MASK;
pkt[2] = (ES_PID_UNDER_TEST & 0xFF) as u8;
pkt[3] = 0x20 | (cc & 0x0F); pkt[4] = af.len() as u8;
pkt[5..5 + af.len()].copy_from_slice(&af);
input.extend_from_slice(&pkt);
cc = cc.wrapping_add(1);
}
for i in 0..FRAMES {
let au = annexb[i % annexb.len()].clone();
push_au(
&mut input,
&mut cc,
&au,
BASE_B_DTS + i as i64 * FRAME_PERIOD,
);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let mut saw_signalled = false;
let samples: Vec<Sample> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Discontinuity {
kind: DiscontinuityKind::Signalled,
..
} => {
saw_signalled = true;
None
}
DemuxEvent::Sample { sample, .. } => Some(sample),
_ => None,
})
.collect();
assert!(
saw_signalled,
"the adaptation-field discontinuity_indicator must surface as a Signalled Discontinuity"
);
let dts: Vec<i64> = samples
.iter()
.map(|s| s.dts.expect("stamped dts"))
.collect();
assert!(
dts.len() >= FRAMES,
"expected one sample per access unit, got {}",
dts.len()
);
assert!(
dts.windows(2).all(|w| w[1] >= w[0]),
"dts must stay monotonic across a signalled discontinuity; got {dts:?}"
);
let steps: Vec<i64> = dts.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
steps.iter().all(|&d| d == FRAME_PERIOD),
"every decode step must remain one frame period; got {steps:?}"
);
}
#[test]
fn three_signalled_discontinuities_keep_dts_monotonic() {
const FRAME_PERIOD: i64 = 3600;
const FRAMES: i64 = 6;
const BASES: [i64; 3] = [90_000, 30_000, 5_000];
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let source =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let media = TsDemux::new().demux(&source).expect("demux h264_aac.ts");
let video = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Avc { .. }))
.expect("h264_aac.ts has an AVC track");
let annexb: Vec<Vec<u8>> = video
.samples
.iter()
.map(|sample| {
let mut au = Vec::new();
for nal in crate::annexb::iter_length_prefixed_nals(&sample.data)
.expect("TsMux writes valid NAL prefixes")
{
au.extend_from_slice(&[0x00, 0x00, 0x00, 0x01]);
au.extend_from_slice(nal);
}
au
})
.collect();
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body_with_pcr_pid(ES_PID_UNDER_TEST, STREAM_TYPE_AVC, ES_PID_UNDER_TEST),
));
let push_au = |input: &mut Vec<u8>, cc: &mut u8, au: &[u8], dts: i64| {
let wire = dts.rem_euclid(1i64 << 33) as u64;
let pes = pes_packet(ES_STREAM_ID_VIDEO, wire as i64, Some(wire as i64), au);
let packets = pes_packets(ES_PID_UNDER_TEST, *cc, &pes);
*cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
};
for base in BASES {
for i in 0..FRAMES {
let au = annexb[(i as usize) % annexb.len()].clone();
push_au(&mut input, &mut cc, &au, base + i * FRAME_PERIOD);
}
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = (ES_PID_UNDER_TEST >> 8) as u8 & PID_HI_MASK;
pkt[2] = (ES_PID_UNDER_TEST & 0xFF) as u8;
pkt[3] = 0x20 | (cc & 0x0F); pkt[4] = 8;
pkt[5] = 0x80; input.extend_from_slice(&pkt);
cc = cc.wrapping_add(1);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let dts: Vec<i64> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => sample.dts,
_ => None,
})
.collect();
assert!(
dts.len() >= (BASES.len() as i64 * FRAMES) as usize - 3,
"expected the access units of all three time bases, got {} (first few: {:?})",
dts.len(),
&dts[..dts.len().min(6)]
);
let steps: Vec<i64> = dts.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
steps.iter().all(|&d| d > 0),
"dts must be strictly increasing across every discontinuity; dts={dts:?} steps={steps:?}"
);
}
#[test]
fn cc_gap_on_a_continuation_drops_only_that_access_unit() {
const PACKETS_PER_AU: usize = 4;
const PER_PACKET: usize = TS_PACKET_SIZE - 4;
const PES_HEADER_BYTES: usize = 9;
let body_len = PACKETS_PER_AU * PER_PACKET - PES_HEADER_BYTES;
let aus: Vec<Vec<u8>> = (0..3)
.map(|i| (0..body_len).map(|b| ((b + i * 53) & 0xFF) as u8).collect())
.collect();
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let mut cc = 0u8;
let mut expected = Vec::new();
for (i, au) in aus.iter().enumerate() {
let pes = untimed_pes(ES_STREAM_ID_PRIVATE_DATA, au);
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
let n = packets.len() / TS_PACKET_SIZE;
assert_eq!(
n, PACKETS_PER_AU,
"AU {i} must span {PACKETS_PER_AU} packets"
);
if i == 1 {
input.extend_from_slice(&packets[..TS_PACKET_SIZE]);
input.extend_from_slice(&packets[2 * TS_PACKET_SIZE..]);
} else {
input.extend_from_slice(&packets);
expected.push(au.clone());
}
cc = cc.wrapping_add(n as u8);
}
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let got: Vec<Vec<u8>> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample.data.to_vec()),
_ => None,
})
.collect();
assert_eq!(
got, expected,
"only the unit whose middle packet was lost may be dropped"
);
}
#[test]
fn cc_jump_at_a_pes_start_keeps_the_previous_complete_access_unit() {
let au1 = data_au(0x00);
let au2 = data_au(0x80);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let pes1 = untimed_pes(ES_STREAM_ID_PRIVATE_DATA, &au1);
let pes2 = untimed_pes(ES_STREAM_ID_PRIVATE_DATA, &au2);
let pkt1 = pes_packets(ES_PID_UNDER_TEST, 0, &pes1);
assert_eq!(pkt1.len(), 2 * TS_PACKET_SIZE, "unit 1 spans two packets");
let pkt2 = pes_packets(ES_PID_UNDER_TEST, 0, &pes2);
input.extend_from_slice(&pkt1);
input.extend_from_slice(&pkt2);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let got: Vec<Vec<u8>> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample.data.to_vec()),
_ => None,
})
.collect();
assert_eq!(
got,
vec![au1.clone(), au2.clone()],
"a counter jump at a `payload_unit_start` must not drop the previous unit when its declared length was fully received"
);
}
#[test]
fn cc_restart_at_a_pes_start_keeps_an_unbounded_unit() {
let au1 = data_au(0x10);
let au2 = data_au(0x90);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let pes1 = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, &au1);
let pes2 = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, &au2);
let pkt1 = pes_packets(ES_PID_UNDER_TEST, 0, &pes1);
assert_eq!(pkt1.len(), 2 * TS_PACKET_SIZE, "unit 1 spans two packets");
let pkt2 = pes_packets(ES_PID_UNDER_TEST, 0, &pes2);
input.extend_from_slice(&pkt1);
input.extend_from_slice(&pkt2);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let got: Vec<Vec<u8>> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample.data.to_vec()),
_ => None,
})
.collect();
assert_eq!(
got.len(),
2,
"both units are whole on the wire, so a counter restart between them must drop neither: {got:?}"
);
assert!(got[0].starts_with(&au1), "first unit preserved");
assert!(got[1].starts_with(&au2), "second unit preserved");
}
#[test]
fn concatenated_segments_seam_keeps_every_access_unit() {
use crate::TsMux;
use crate::media::{Media, Track};
use crate::pipeline::{CodecConfig, Sample, TrackSpec};
use crate::rtp_sdp::avc_config_from_sprop;
use broadcast_common::Package;
let frame_dur = VIDEO_TIMESCALE / 25;
let avc = avc_config_from_sprop("Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==").unwrap();
let spec = TrackSpec::new(
1,
VIDEO_TIMESCALE,
CodecConfig::Avc {
config: avc,
width: 0,
height: 0,
},
);
let samples: Vec<Sample> = (0..6u32)
.map(|i| {
let nal = [0x65u8, 0xAA, i as u8];
let mut data = (nal.len() as u32).to_be_bytes().to_vec();
data.extend_from_slice(&nal);
let dts = i64::from(i) * i64::from(frame_dur);
Sample::new(data, Some(dts), Some(dts), Some(frame_dur), true)
})
.collect();
let media = Media::new(vec![Track::new(spec, samples)], VIDEO_TIMESCALE);
let segment = TsMux::default().package(&media).expect("mux to TS");
let mut concatenated = segment.clone();
concatenated.extend_from_slice(&segment);
let mut demux = StreamingTsDemux::new();
demux.feed(&concatenated);
demux.finish();
let video_samples = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::Sample { .. }))
.count();
assert_eq!(
video_samples, 12,
"a segment seam restarts the counter but loses nothing, so all 12 access units must survive"
);
}
#[test]
fn cc_gap_inside_an_unbounded_unit_drops_it() {
const PER_PACKET: usize = TS_PACKET_SIZE - 4;
const PACKETS: usize = 4;
const PES_HEADER_BYTES: usize = 9;
let au: Vec<u8> = (0..PACKETS * PER_PACKET - PES_HEADER_BYTES)
.map(|i| (i & 0xFF) as u8)
.collect();
let pes = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, &au);
let packets = pes_packets(ES_PID_UNDER_TEST, 0, &pes);
assert_eq!(packets.len(), PACKETS * TS_PACKET_SIZE);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
input.extend_from_slice(&packets[..TS_PACKET_SIZE]);
input.extend_from_slice(&packets[2 * TS_PACKET_SIZE..]);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let samples = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::Sample { .. }))
.count();
assert_eq!(
samples, 0,
"a unit missing one of its own packets must not be delivered"
);
}
#[test]
fn signalled_discontinuity_does_not_drop_the_unit_it_ends() {
const SIGNALLED_CC: u8 = 3;
let au1 = data_au(0x20);
let au2 = data_au(0xA0);
let au3 = data_au(0xF0);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let mut cc = 0u8;
for au in [&au1, &au2] {
let pes = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, au);
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
assert_eq!(
packets.len(),
2 * TS_PACKET_SIZE,
"each unit spans two packets"
);
input.extend_from_slice(&packets);
cc = cc.wrapping_add(2);
}
let pes3 = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, &au3);
input.extend_from_slice(&pes_packets_with_discontinuity(
ES_PID_UNDER_TEST,
SIGNALLED_CC,
&pes3,
));
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let got: Vec<Vec<u8>> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample.data.to_vec()),
_ => None,
})
.collect();
assert_eq!(
got.len(),
3,
"every unit here is whole on the wire, so the indicator on the packet starting unit 3 must not drop anything: {got:?}"
);
for (i, (got_au, want)) in got.iter().zip([&au1, &au2, &au3]).enumerate() {
assert!(got_au.starts_with(want), "unit {i} must be delivered whole");
}
}
#[test]
fn unbounded_unit_ended_by_a_counter_restart_is_delivered_by_design() {
let au1 = data_au(0x20);
let au2 = data_au(0xA0);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let pes1 = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, &au1);
let pkt1 = pes_packets(ES_PID_UNDER_TEST, 0, &pes1);
input.extend_from_slice(&pkt1[..TS_PACKET_SIZE]);
let pes2 = unbounded_pes(ES_STREAM_ID_PRIVATE_DATA, &au2);
input.extend_from_slice(&pes_packets(ES_PID_UNDER_TEST, 2, &pes2));
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let got: Vec<Vec<u8>> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample.data.to_vec()),
_ => None,
})
.collect();
assert_eq!(
got.len(),
2,
"both the truncated-tail unit and the whole one are delivered: an unbounded PES gives no way to tell them apart at a seam"
);
}
#[test]
fn unstamped_pes_access_unit_is_interpolated_not_restamped() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let source =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let media = TsDemux::new().demux(&source).expect("demux h264_aac.ts");
const FRAME_PERIOD: i64 = 3600;
const UNSTAMPED_EVERY: usize = 2;
let mut cc = 0u8;
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_AVC),
));
let video = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Avc { .. }))
.expect("h264_aac.ts has an AVC track");
let source_dts0 = video.samples[0]
.dts
.expect("the fixture's AVC track is stamped");
let mut headers = 0usize;
for (i, sample) in video.samples.iter().enumerate() {
let mut au = Vec::new();
let nals = crate::annexb::iter_length_prefixed_nals(&sample.data)
.expect("TsMux writes valid 4-byte-length NAL prefixes");
for nal in nals {
au.extend_from_slice(&[0x00, 0x00, 0x00, 0x01]);
au.extend_from_slice(nal);
}
let stamped = i < UNSTAMPED_EVERY || i % UNSTAMPED_EVERY == 0;
let dts = source_dts0 + i as i64 * FRAME_PERIOD;
let pes = if stamped {
pes_packet(ES_STREAM_ID_VIDEO, dts, Some(dts), &au)
} else {
pes_packet_untimed(ES_STREAM_ID_VIDEO, &au)
};
headers += 1;
let packets = pes_packets(ES_PID_UNDER_TEST, cc, &pes);
cc = cc.wrapping_add((packets.len() / TS_PACKET_SIZE) as u8);
input.extend_from_slice(&packets);
}
assert!(
video.samples.len() >= 20,
"fixture too small for a meaningful interpolation test"
);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let samples: Vec<Sample> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample),
_ => None,
})
.collect();
assert_eq!(
samples.len(),
headers,
"one sample per fed access unit (video is one AU per PES here)"
);
let dts: Vec<i64> = samples.iter().map(|s| s.dts.expect("video dts")).collect();
let steps: Vec<i64> = dts.windows(2).map(|w| w[1] - w[0]).collect();
assert!(
steps.iter().all(|&d| d == FRAME_PERIOD),
"every decode step must be one frame period ({FRAME_PERIOD} ticks); got {steps:?}"
);
assert!(
samples
.iter()
.all(|s| s.duration == Some(FRAME_PERIOD as u32)),
"no zero-duration sample may survive interpolation; durations: {:?}",
samples.iter().map(|s| s.duration).collect::<Vec<_>>()
);
}
fn aac_44100_access_unit() -> Vec<u8> {
const ADTS_PROFILE_AAC_LC: u8 = 1;
const SFI_44100: u8 = 4;
const CHANNELS_STEREO: u8 = 2;
const PAYLOAD_BYTES: usize = 128;
let frame_len = (ADTS_HEADER_SIZE + PAYLOAD_BYTES) as u16;
let header = crate::aac_asc::build_adts_header(
ADTS_PROFILE_AAC_LC,
SFI_44100,
CHANNELS_STEREO,
frame_len,
);
let mut au = header.to_vec();
au.resize(ADTS_HEADER_SIZE + PAYLOAD_BYTES, 0x21);
au
}
#[test]
fn constant_increment_44100_aac_emits_no_timeline_reanchor() {
const PES_INCREMENT_TICKS: i128 = 2090;
const SAMPLE_RATE: u32 = 44_100;
const FRAMES: i128 = 500;
let au = aac_44100_access_unit();
let mut anchor = AudioAnchor::default();
let mut events: VecDeque<DemuxEvent> = VecDeque::new();
for n in 0..FRAMES {
let ts = n * PES_INCREMENT_TICKS;
emit_audio_au(
&AudioKind::Aac,
SAMPLE_RATE,
&mut anchor,
&au,
ts,
ts,
1,
&mut events,
);
}
assert!(
events
.iter()
.any(|e| matches!(e, DemuxEvent::Sample { .. })),
"sanity: the synthesised ADTS frames must actually split into samples"
);
let reanchors = events
.iter()
.filter(|e| {
matches!(
e,
DemuxEvent::Discontinuity {
kind: DiscontinuityKind::TimelineReanchored,
..
}
)
})
.count();
assert_eq!(
reanchors, 0,
"a constant-increment 44.1 kHz stream is continuous — the \
rounding drift a real muxer accrues by construction must never be \
reported as a discontinuity"
);
}
#[test]
fn a_real_timeline_gap_emits_exactly_one_reanchor() {
const PES_INCREMENT_TICKS: i128 = 2090;
const SAMPLE_RATE: u32 = 44_100;
const FRAMES_BEFORE: i128 = 50;
const FRAMES_AFTER: i128 = 50;
const GAP_TICKS: i128 = 180_000;
let au = aac_44100_access_unit();
let mut anchor = AudioAnchor::default();
let mut events: VecDeque<DemuxEvent> = VecDeque::new();
let emit = |ts: i128, anchor: &mut AudioAnchor, events: &mut VecDeque<DemuxEvent>| {
emit_audio_au(&AudioKind::Aac, SAMPLE_RATE, anchor, &au, ts, ts, 1, events);
};
for n in 0..FRAMES_BEFORE {
emit(n * PES_INCREMENT_TICKS, &mut anchor, &mut events);
}
let resume = FRAMES_BEFORE * PES_INCREMENT_TICKS + GAP_TICKS;
for n in 0..FRAMES_AFTER {
emit(resume + n * PES_INCREMENT_TICKS, &mut anchor, &mut events);
}
let reanchors = events
.iter()
.filter(|e| {
matches!(
e,
DemuxEvent::Discontinuity {
kind: DiscontinuityKind::TimelineReanchored,
..
}
)
})
.count();
assert_eq!(
reanchors, 1,
"one genuine gap must produce exactly one TimelineReanchored — not \
zero (the anchor silently absorbing a real splice) and not one \
per following access unit"
);
}
#[test]
fn adts_resyncs_across_pes_boundaries() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let ts_bytes =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let mut demux = TsDemux::new();
let media = demux.demux(&ts_bytes).expect("demux h264_aac.ts");
let audio = media
.tracks
.iter()
.find(|t| matches!(t.config(), CodecConfig::Aac { .. }))
.expect("h264_aac.ts has an AAC track");
let esds = match audio.config() {
CodecConfig::Aac { esds, .. } => esds,
_ => unreachable!(),
};
let dsi = esds
.es_descriptor
.decoder_config
.as_ref()
.and_then(|dc| dc.decoder_specific_info.as_ref())
.expect("AAC esds carries a DecoderSpecificInfo");
let asc = AudioSpecificConfig::parse(&dsi.data).expect("parse real AudioSpecificConfig");
let mut es: Vec<u8> = Vec::new();
for s in &audio.samples {
let frame_len = (ADTS_HEADER_SIZE + s.data.len()) as u16;
let hdr = asc
.to_adts_header(frame_len)
.expect("build real ADTS header");
es.extend_from_slice(&hdr);
es.extend_from_slice(&s.data);
}
assert!(
audio.samples.len() >= 10,
"fixture too small to be a meaningful resync test"
);
const CHUNK: usize = 2000;
let mut recovered = 0usize;
let mut off = 0usize;
while off < es.len() {
let end = (off + CHUNK).min(es.len());
recovered += split_adts_frames(&es[off..end]).len();
off = end;
}
assert!(
recovered * 2 >= audio.samples.len(),
"resync must recover most real AAC frames across misaligned \
chunks (got {recovered} of {} real frames)",
audio.samples.len()
);
}
fn real_aac_asc_and_samples() -> (AudioSpecificConfig, Vec<Vec<u8>>) {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let ts_bytes =
std::fs::read(&path).unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
let mut demux = TsDemux::new();
let media = demux.demux(&ts_bytes).expect("demux h264_aac.ts");
let audio = media
.tracks
.into_iter()
.find(|t| matches!(t.config(), CodecConfig::Aac { .. }))
.expect("h264_aac.ts has an AAC track");
let esds = match audio.config() {
CodecConfig::Aac { esds, .. } => esds.clone(),
_ => unreachable!(),
};
let dsi = esds
.es_descriptor
.decoder_config
.as_ref()
.and_then(|dc| dc.decoder_specific_info.as_ref())
.expect("AAC esds carries a DecoderSpecificInfo")
.clone();
let asc = AudioSpecificConfig::parse(&dsi.data).expect("parse real AudioSpecificConfig");
let samples: Vec<Vec<u8>> = audio.samples.iter().map(|s| s.data.to_vec()).collect();
(asc, samples)
}
#[test]
fn adts_crc_protected_frame_strips_crc_not_sample_data() {
let (asc, samples) = real_aac_asc_and_samples();
assert!(samples.len() >= 2, "fixture must carry several AAC frames");
let mut events: VecDeque<DemuxEvent> = VecDeque::new();
let mut anchor = AudioAnchor::default();
let mut dts_uw: i128 = 0;
for payload in samples.iter().take(5) {
let frame_len = (ADTS_HEADER_SIZE + ADTS_CRC_SIZE + payload.len()) as u16;
let mut header = asc.to_adts_header(frame_len).expect("build ADTS header");
header[1] &= !0x01; let mut au = header.to_vec();
au.extend_from_slice(&[0x00, 0x00]); au.extend_from_slice(payload);
emit_audio_au(
&AudioKind::Aac,
44_100,
&mut anchor,
&au,
dts_uw as u64 as i128,
dts_uw,
1,
&mut events,
);
dts_uw += AAC_SAMPLES_PER_FRAME as i128;
}
let emitted: Vec<Vec<u8>> = events
.iter()
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample.data.to_vec()),
_ => None,
})
.collect();
assert_eq!(emitted.len(), 5, "one sample per CRC-protected ADTS frame");
for (i, (got, want)) in emitted.iter().zip(samples.iter().take(5)).enumerate() {
assert_eq!(
got, want,
"sample {i}: CRC bytes must not leak into (or truncate) the real payload"
);
}
}
#[test]
fn adts_multi_raw_data_block_duration_is_multiplied() {
let (asc, samples) = real_aac_asc_and_samples();
assert!(samples.len() >= 2, "fixture must carry several AAC frames");
let mut combined = samples[0].clone();
combined.extend_from_slice(&samples[1]);
let frame_len = (ADTS_HEADER_SIZE + combined.len()) as u16;
let mut header = asc.to_adts_header(frame_len).expect("build ADTS header");
header[6] |= 0x01; let mut au = header.to_vec();
au.extend_from_slice(&combined);
let mut events: VecDeque<DemuxEvent> = VecDeque::new();
let mut anchor = AudioAnchor::default();
emit_audio_au(
&AudioKind::Aac,
44_100,
&mut anchor,
&au,
0,
0,
1,
&mut events,
);
let emitted: Vec<(Vec<u8>, Option<u32>)> = events
.iter()
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some((sample.data.to_vec(), sample.duration)),
_ => None,
})
.collect();
assert_eq!(
emitted.len(),
1,
"the whole 2-raw-data-block frame is one emitted sample"
);
assert_eq!(
emitted[0].0, combined,
"both raw data blocks' real bytes must be present, undamaged"
);
assert_eq!(
emitted[0].1,
Some(AAC_SAMPLES_PER_FRAME * 2),
"duration must be (num_raw_data_blocks + 1) * AAC_SAMPLES_PER_FRAME, not just 1024"
);
}
#[test]
fn refine_with_descriptors_recognises_every_dolby_dts_tag() {
let ac3_desc = [DESC_TAG_AC3, 1, 0x40];
assert_eq!(
Codec::Data(STREAM_TYPE_PES_PRIVATE)
.refine_with_descriptors(STREAM_TYPE_PES_PRIVATE, &ac3_desc),
Codec::Ac3
);
let dts_desc = [DESC_TAG_DTS, 1, 0x00];
assert_eq!(
Codec::Data(STREAM_TYPE_METADATA_PES)
.refine_with_descriptors(STREAM_TYPE_METADATA_PES, &dts_desc),
Codec::Dts,
"0x15 (metadata in PES) is also descriptor-disambiguated"
);
let eac3_desc = [DESC_TAG_ENHANCED_AC3, 1, 0x00];
assert_eq!(
Codec::Data(STREAM_TYPE_PES_PRIVATE)
.refine_with_descriptors(STREAM_TYPE_PES_PRIVATE, &eac3_desc),
Codec::Eac3
);
}
#[test]
fn refine_with_descriptors_leaves_non_audio_0x06_as_data() {
const DESC_TAG_SUBTITLING: u8 = 0x59;
let subtitle_desc = [DESC_TAG_SUBTITLING, 3, 0x65, 0x6E, 0x67];
assert_eq!(
Codec::Data(STREAM_TYPE_PES_PRIVATE)
.refine_with_descriptors(STREAM_TYPE_PES_PRIVATE, &subtitle_desc),
Codec::Data(STREAM_TYPE_PES_PRIVATE)
);
}
#[test]
fn refine_with_descriptors_ignores_other_stream_types() {
let eac3_desc = [DESC_TAG_ENHANCED_AC3, 1, 0x00];
assert_eq!(
Codec::H264.refine_with_descriptors(STREAM_TYPE_AVC, &eac3_desc),
Codec::H264
);
}
#[test]
fn codec_change_on_shared_pid_does_not_tear_down_other_program_track() {
const PMT_A: u16 = 0x1000;
const PMT_B: u16 = 0x1001;
const SHARED_PID: u16 = 0x0050;
const STREAM_TYPE: u8 = 0x7F;
let mut demux = StreamingTsDemux::new();
demux.apply_pmt_diff(
PMT_A,
&BTreeSet::new(),
alloc::vec![(SHARED_PID, Codec::Data(STREAM_TYPE), Vec::new())],
);
demux.apply_pmt_diff(
PMT_B,
&BTreeSet::new(),
alloc::vec![(SHARED_PID, Codec::Data(STREAM_TYPE), Vec::new())],
);
assert_eq!(
demux.es_declarers.get(&SHARED_PID).map(|d| d.len()),
Some(2),
"both PMTs must be recorded as declarers of the shared PID"
);
let carriage = data_carriage(STREAM_TYPE);
assert_eq!(carriage, DataCarriage::Pes);
demux.streams.get_mut(&SHARED_PID).unwrap().track = Some(TrackState::Parked {
config: CodecConfig::Data {
stream_type: STREAM_TYPE,
descriptors: Vec::new(),
carriage,
},
timescale: VIDEO_TIMESCALE,
kind: LiveKind::Data {
pending: None,
last_duration: 0,
},
backlog: Vec::new(),
});
demux.try_promote_ready();
let track_id_before = match demux.streams.get(&SHARED_PID).unwrap().track.as_ref() {
Some(TrackState::Live(live)) => live.track_id,
_ => panic!("expected the shared PID to be Live after promotion"),
};
while demux.poll_event().is_some() {}
let mut pmt_a_applied = BTreeSet::new();
pmt_a_applied.insert(SHARED_PID);
demux.apply_pmt_diff(
PMT_A,
&pmt_a_applied,
alloc::vec![(SHARED_PID, Codec::Ac3, Vec::new())],
);
match demux
.streams
.get(&SHARED_PID)
.and_then(|s| s.track.as_ref())
{
Some(TrackState::Live(live)) => assert_eq!(
live.track_id, track_id_before,
"shared track must keep its original track_id"
),
_ => {
panic!("expected the shared track to survive PMT A's reclassification, still Live")
}
}
assert!(
!demux
.events
.iter()
.any(|ev| matches!(ev, DemuxEvent::TrackRemoved { .. })),
"PMT A's reclassification must not remove a track PMT B still declares"
);
assert_eq!(
demux.streams.get(&SHARED_PID).unwrap().codec,
Codec::Data(STREAM_TYPE),
"the existing classification wins while another declarer disagrees"
);
let mut pmt_b_applied = BTreeSet::new();
pmt_b_applied.insert(SHARED_PID);
demux.apply_pmt_diff(PMT_B, &pmt_b_applied, Vec::new());
assert_eq!(
demux.es_declarers.get(&SHARED_PID).map(|d| d.len()),
Some(1),
"PMT A must be the sole remaining declarer"
);
demux.apply_pmt_diff(
PMT_A,
&pmt_a_applied,
alloc::vec![(SHARED_PID, Codec::Aac, Vec::new())],
);
assert!(
demux
.events
.iter()
.any(|ev| matches!(ev, DemuxEvent::TrackRemoved { .. })),
"once PMT A is the last declarer, its reclassification must actually tear down \
the old track"
);
assert_eq!(demux.streams.get(&SHARED_PID).unwrap().codec, Codec::Aac);
}
#[test]
fn codec_change_reregistration_preserves_declaration_order_slot() {
const PMT: u16 = 0x1000;
const PID_X: u16 = 0x0050;
const PID_Y: u16 = 0x0051;
let mut demux = StreamingTsDemux::new();
demux.apply_pmt_diff(
PMT,
&BTreeSet::new(),
alloc::vec![
(PID_X, Codec::Data(0x06), Vec::new()),
(PID_Y, Codec::Data(0x07), Vec::new()),
],
);
assert_eq!(demux.data_order, alloc::vec![PID_X, PID_Y]);
let mut old_applied = BTreeSet::new();
old_applied.insert(PID_X);
old_applied.insert(PID_Y);
demux.apply_pmt_diff(
PMT,
&old_applied,
alloc::vec![
(PID_X, Codec::Data(0x08), Vec::new()),
(PID_Y, Codec::Data(0x07), Vec::new()),
],
);
assert_eq!(
demux.data_order,
alloc::vec![PID_X, PID_Y],
"PID X must keep its original (first) declaration-order slot, not move to the back"
);
assert_eq!(demux.streams.get(&PID_X).unwrap().codec, Codec::Data(0x08));
}
fn null_packet() -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = 0x1F;
p[2] = 0xFF;
p[3] = 0x10;
p
}
fn ts_packet_degradation(
tei: bool,
pid: u16,
cc: u8,
afc: u8,
adaptation: Option<&[u8]>,
payload: &[u8],
) -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = (if tei { 0x80 } else { 0x00 }) | ((pid >> 8) as u8 & PID_HI_MASK);
p[2] = (pid & 0xFF) as u8;
p[3] = afc | (cc & 0x0F);
let mut off = 4usize;
if let Some(af) = adaptation {
let af_len = af.len() as u8;
p[off] = af_len;
off += 1;
p[off..off + af.len()].copy_from_slice(af);
off += af.len();
}
let payload_end = off + payload.len().min(TS_PACKET_SIZE - off);
p[off..payload_end].copy_from_slice(&payload[..payload_end - off]);
p
}
#[test]
fn tei_set_packet_emits_transport_error() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let pkt = ts_packet_degradation(
true, 0x0100,
0,
0x10, None,
b"some payload",
);
demux.feed(&pkt);
let events: Vec<_> = std::iter::from_fn(|| demux.poll_event()).collect();
let degraded: Vec<_> = events
.iter()
.filter_map(|e| match e {
DemuxEvent::InputDegraded {
kind, provenance, ..
} => Some((*kind, *provenance)),
_ => None,
})
.collect();
assert_eq!(degraded.len(), 1, "expected exactly one InputDegraded");
assert_eq!(degraded[0].0, InputDegradation::TransportError);
assert_eq!(degraded[0].1.pid, Some(0x0100));
assert_eq!(
degraded[0].1.packet_index,
Some(mpeg_ts::resync::LOCK_CONFIRMATIONS as u64 + 1)
);
}
#[test]
fn h264_aac_clean_fixture_produces_zero_input_degraded() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("h264_aac.ts");
let data = std::fs::read(&path).unwrap_or_else(|e| panic!("{}: {e}", path.display()));
let mut demux = StreamingTsDemux::new();
demux.feed(&data);
demux.finish();
let degraded: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::InputDegraded { .. }))
.collect();
assert!(
degraded.is_empty(),
"h264_aac.ts must produce zero InputDegraded events, got {degraded:?}"
);
}
#[test]
fn m6_discontinuity_fixture_gap_count_matches_media_doctor() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("m6-discontinuity.ts");
let data = std::fs::read(&path).unwrap_or_else(|e| panic!("{}: {e}", path.display()));
let mut demux = StreamingTsDemux::new();
demux.feed(&data);
demux.finish();
let gap_count = std::iter::from_fn(|| demux.poll_event())
.filter(|e| {
matches!(
e,
DemuxEvent::InputDegraded {
kind: InputDegradation::ContinuityGap { .. },
..
}
)
})
.count();
assert_eq!(
gap_count, 877,
"m6-discontinuity.ts must produce exactly 877 ContinuityGap events \
(matching media-doctor CcAnomalyCheck); any divergence means the \
exclusion rules disagree with the two in-repo reference \
implementations"
);
}
#[test]
fn m6_discontinuity_fixture_plays_through() {
let mut path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
path.push("..");
path.push("fixtures");
path.push("ts");
path.push("m6-discontinuity.ts");
let data = std::fs::read(&path).unwrap_or_else(|e| panic!("{}: {e}", path.display()));
let mut demux = StreamingTsDemux::new();
demux.feed(&data);
demux.finish();
let mut saw_discontinuity = false;
while let Some(event) = demux.poll_event() {
if matches!(event, DemuxEvent::Discontinuity { .. }) {
saw_discontinuity = true;
}
}
assert!(
saw_discontinuity,
"m6-discontinuity.ts must produce at least one Discontinuity event"
);
}
#[test]
fn legal_duplicate_same_cc_and_identical_payload_emits_nothing() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let payload = b"duplicate payload bytes";
let pkt = |cc: u8| -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = 0x00; p[2] = 0x31; p[3] = 0x10 | (cc & 0x0F); let end = 4 + payload.len().min(TS_PACKET_SIZE - 4);
p[4..end].copy_from_slice(&payload[..end - 4]);
p
};
demux.feed(&pkt(0));
demux.feed(&pkt(0));
let degraded: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::InputDegraded { .. }))
.collect();
assert!(
degraded.is_empty(),
"legal duplicate must not emit InputDegraded, got {degraded:?}"
);
}
#[test]
fn legal_duplicate_packets_are_not_reassembled_twice() {
const DATA_BYTES: usize = 1000;
let body: Vec<u8> = (0..DATA_BYTES).map(|i| (i & 0xFF) as u8).collect();
let pes_len = (3 + body.len()) as u16;
let mut pes = Vec::new();
pes.extend_from_slice(&[0x00, 0x00, 0x01, 0xBD]);
pes.extend_from_slice(&pes_len.to_be_bytes());
pes.push(0x80); pes.push(0x00); pes.push(0x00); pes.extend_from_slice(&body);
assert!(
pes.len() > TS_MAX_PAYLOAD_BYTES,
"the PES must span more than one TS packet for this test to mean anything"
);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let mut cc = 0u8;
let mut sent = 0usize;
let mut dups = 0usize;
while sent < pes.len() {
let take = (pes.len() - sent).min(TS_MAX_PAYLOAD_BYTES);
let mut pkt = [0xFFu8; TS_PACKET_SIZE];
pkt[0] = 0x47;
pkt[1] = ((ES_PID_UNDER_TEST >> 8) as u8 & PID_HI_MASK)
| if sent == 0 { 0x40 } else { 0x00 };
pkt[2] = (ES_PID_UNDER_TEST & 0xFF) as u8;
pkt[3] = 0x10 | (cc & 0x0F);
pkt[4..4 + take].copy_from_slice(&pes[sent..sent + take]);
input.extend_from_slice(&pkt);
input.extend_from_slice(&pkt);
dups += 1;
sent += take;
cc = cc.wrapping_add(1);
}
assert!(
dups >= 3,
"the PES must span several packets for a duplicate append to be observable ({dups} packets)"
);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.finish();
let samples: Vec<Sample> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample),
_ => None,
})
.collect();
assert!(
dups >= 2,
"the fixture must really contain duplicates ({dups} emitted)"
);
assert_eq!(samples.len(), 1, "one PES => one Data sample");
assert_eq!(
samples[0].data.as_ref(),
body.as_slice(),
"the delivered sample must be the PES *payload* exactly once — a reassembled duplicate appends its 184 bytes a second time"
);
}
#[test]
fn cc_gap_drops_the_access_unit_it_truncated_but_not_the_next_one() {
let au1 = long_data_au();
let au2 = long_data_au_alt();
let pes1 = untimed_pes(ES_STREAM_ID_PRIVATE_DATA, &au1);
let pes2 = untimed_pes(ES_STREAM_ID_PRIVATE_DATA, &au2);
assert!(
pes1.len() > TS_MAX_PAYLOAD_BYTES && pes2.len() > TS_MAX_PAYLOAD_BYTES,
"both access units must span more than one TS packet"
);
let mut input = Vec::new();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
input.extend_from_slice(&null_packet());
}
input.extend_from_slice(&psi_section_packets(
PAT_PID_UNDER_TEST,
0,
TABLE_ID_PAT,
&pat_body(),
));
input.extend_from_slice(&psi_section_packets(
PMT_PID_UNDER_TEST,
0,
TABLE_ID_PMT,
&pmt_body(ES_PID_UNDER_TEST, STREAM_TYPE_PES_PRIVATE),
));
let pkt1 = pes_packets(ES_PID_UNDER_TEST, 0, &pes1);
assert_eq!(pkt1.len(), 2 * TS_PACKET_SIZE, "AU 1 must span two packets");
input.extend_from_slice(&pkt1[..TS_PACKET_SIZE]);
let pkt2 = pes_packets(ES_PID_UNDER_TEST, 2, &pes2);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.feed(&pkt2);
demux.finish();
let degraded = std::iter::from_fn(|| demux.poll_event()).any(|e| {
matches!(
e,
DemuxEvent::InputDegraded {
kind: InputDegradation::ContinuityGap { .. },
..
}
)
});
assert!(
degraded,
"the lost continuation must be reported as a CC gap"
);
let mut demux = StreamingTsDemux::new();
demux.feed(&input);
demux.feed(&pkt2);
demux.finish();
let samples: Vec<Sample> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::Sample { sample, .. } => Some(sample),
_ => None,
})
.collect();
assert_eq!(
samples.len(),
1,
"only the intact access unit may be delivered, got {} samples",
samples.len()
);
assert_eq!(
samples[0].data.as_ref(),
au2.as_slice(),
"the delivered sample must be the second, whole access unit"
);
}
#[test]
fn cc_gap_emits_continuity_gap() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let payload_a = b"first packet";
let payload_b = b"second different";
let pkt = |cc: u8, payload: &[u8]| -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = 0x00;
p[2] = 0x42; p[3] = 0x10 | (cc & 0x0F); let end = 4 + payload.len().min(TS_PACKET_SIZE - 4);
p[4..end].copy_from_slice(&payload[..end - 4]);
p
};
demux.feed(&pkt(0, payload_a));
demux.feed(&pkt(5, payload_b));
let degraded: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::InputDegraded {
kind, provenance, ..
} => Some((kind, provenance)),
_ => None,
})
.collect();
assert_eq!(
degraded.len(),
1,
"expected exactly one InputDegraded on CC gap"
);
assert_eq!(
degraded[0].0,
InputDegradation::ContinuityGap {
expected: 1,
got: 5
}
);
assert_eq!(degraded[0].1.pid, Some(0x0042));
assert_eq!(
degraded[0].1.packet_index,
Some(mpeg_ts::resync::LOCK_CONFIRMATIONS as u64 + 2)
);
}
#[test]
fn mutation_proof_duplicate_exclusion_is_load_bearing() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let payload = b"identical";
let pkt = |cc: u8| -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = 0x00;
p[2] = 0x55;
p[3] = 0x10 | (cc & 0x0F);
let end = 4 + payload.len().min(TS_PACKET_SIZE - 4);
p[4..end].copy_from_slice(&payload[..end - 4]);
p
};
demux.feed(&pkt(0));
demux.feed(&pkt(0));
let degraded: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::InputDegraded { .. }))
.collect();
assert!(
degraded.is_empty(),
"legal duplicate must not fire InputDegraded; \
would produce ContinuityGap {{ expected: 1, got: 0 }} if exclusion were removed"
);
}
#[test]
fn mutation_proof_changed_payload_same_cc_is_not_a_duplicate() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let pkt = |cc: u8, payload: &[u8]| -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = 0x00;
p[2] = 0x66;
p[3] = 0x10 | (cc & 0x0F);
let end = 4 + payload.len().min(TS_PACKET_SIZE - 4);
p[4..end].copy_from_slice(&payload[..end - 4]);
p
};
demux.feed(&pkt(0, b"first"));
demux.feed(&pkt(0, b"second"));
let degraded: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter_map(|e| match e {
DemuxEvent::InputDegraded { kind, .. } => Some(kind),
_ => None,
})
.collect();
assert_eq!(
degraded.len(),
1,
"same CC + different payload must fire InputDegraded"
);
assert!(matches!(
degraded[0],
InputDegradation::ContinuityGap { .. }
));
}
#[test]
fn mutation_proof_signalled_discontinuity_suppresses_cc_gap() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let payload = b"payload";
demux.feed(&ts_packet_degradation(
false, 0x0077, 0, 0x10, None, payload,
));
let af = [
0x80u8, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, ];
demux.feed(&ts_packet_degradation(
false,
0x0077,
5,
0x30,
Some(&af),
payload, ));
let mut saw_discontinuity = false;
let mut saw_cc_gap = false;
while let Some(event) = demux.poll_event() {
match event {
DemuxEvent::Discontinuity { .. } => saw_discontinuity = true,
DemuxEvent::InputDegraded {
kind: InputDegradation::ContinuityGap { .. },
..
} => saw_cc_gap = true,
_ => {}
}
}
assert!(
saw_discontinuity,
"packet with discontinuity_indicator must emit Discontinuity"
);
assert!(
!saw_cc_gap,
"packet with discontinuity_indicator must NOT emit ContinuityGap — \
the exclusion rule was removed or broken"
);
}
#[test]
fn cc_wraps_at_15_to_0_without_false_gap() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let pkt = |cc: u8| -> [u8; TS_PACKET_SIZE] {
let mut p = [0xFFu8; TS_PACKET_SIZE];
p[0] = 0x47;
p[1] = 0x00;
p[2] = 0x88;
p[3] = 0x10 | (cc & 0x0F);
p[4..11].copy_from_slice(b"payload");
p
};
demux.feed(&pkt(14));
demux.feed(&pkt(15));
demux.feed(&pkt(0));
let degraded: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter(|e| matches!(e, DemuxEvent::InputDegraded { .. }))
.collect();
assert!(
degraded.is_empty(),
"normal CC wrap 15→0 must not emit InputDegraded, got {degraded:?}"
);
}
#[test]
fn post_discontinuity_legal_cc_emits_nothing() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let payload = b"payload";
demux.feed(&ts_packet_degradation(
false, 0x0083, 0, 0x10, None, payload,
));
let af = [0x80u8, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF];
demux.feed(&ts_packet_degradation(
false,
0x0083,
5,
0x30,
Some(&af),
payload,
));
demux.feed(&ts_packet_degradation(
false, 0x0083, 6, 0x10, None, payload,
));
let gaps: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter(|e| {
matches!(
e,
DemuxEvent::InputDegraded {
kind: InputDegradation::ContinuityGap { .. },
..
}
)
})
.collect();
assert!(
gaps.is_empty(),
"post-discontinuity legal CC must not fire ContinuityGap; \
discontinuity_indicator suppresses the event but updates last_cc. \
Got {gaps:?}"
);
}
#[test]
fn pcr_variation_in_adaptation_field_is_still_a_legal_duplicate() {
let mut demux = StreamingTsDemux::new();
let null = null_packet();
for _ in 0..mpeg_ts::resync::LOCK_CONFIRMATIONS + 1 {
demux.feed(&null);
}
let payload = b"identical payload for duplicate test";
let af_pcr_100 = [
0x10u8, 0x00, 0x00, 0x00, 0x00, 0x7E, 0x64, ];
demux.feed(&ts_packet_degradation(
false,
0x0100,
0,
0x30, Some(&af_pcr_100),
payload,
));
let af_pcr_200 = [
0x10u8, 0x00, 0x00, 0x00, 0x00, 0x7E, 0xC8, ];
demux.feed(&ts_packet_degradation(
false,
0x0100,
0, 0x30,
Some(&af_pcr_200),
payload, ));
let gaps: Vec<_> = std::iter::from_fn(|| demux.poll_event())
.filter(|e| {
matches!(
e,
DemuxEvent::InputDegraded {
kind: InputDegradation::ContinuityGap { .. },
..
}
)
})
.collect();
assert!(
gaps.is_empty(),
"PCR-only adaptation-field change must not break duplicate detection; \
duplicate check uses pkt.payload, not raw[4..]. Got {gaps:?}"
);
}
}