use alloc::collections::{BTreeMap, BTreeSet, VecDeque};
use alloc::vec::Vec;
use core::marker::PhantomData;
use broadcast_common::{Serialize, Unpackage};
use mpeg_pes::{PesAssembler, PesPacket};
use mpeg_ts::resync::TsResync;
use mpeg_ts::ts::{SectionReassembler, TS_PACKET_SIZE, TsPacket};
use crate::aac_asc::{AudioSpecificConfig, parse_adts_header};
use crate::ac3::{
AC3_SAMPLES_PER_SYNCFRAME, Ac3SyncframeInfo, 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, ObjectTypeIndication,
SLConfigDescriptor, StreamType as EsdsStreamType,
};
use crate::mpeg_legacy::{Mpeg2SeqHeader, MpegAudioFrameHeader};
use crate::nal::{NalCodec, is_keyframe_nal, nal_unit_type};
use crate::nalu_types::{AvcPps, AvcSps, HevcNalArray, HevcNalUnit};
use crate::pipeline::{CodecConfig, DataCarriage, Sample, SourceTiming, 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 CRC32_LEN: usize = 4;
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 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_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;
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;
const STREAM_TYPE_VISUAL: u8 = 0x04;
const ESDS_ES_ID: u16 = 1;
const ESDS_VIDEO_ES_ID: u16 = 2;
const SL_CONFIG_PREDEFINED_MP4: u8 = 0x02;
const AUDIO_SAMPLE_SIZE_BITS: u16 = 16;
const VIDEO_TIMESCALE: u32 = 90_000;
const AAC_SAMPLES_PER_FRAME: u32 = 1024;
const ADTS_HEADER_SIZE: usize = 7;
const MPEG2_PICTURE_START_CODE: u8 = 0x00;
const MPEG2_PICTURE_CODING_TYPE_I: u8 = 0x01;
const TS_WRAP: u64 = 1 << 33;
const TS_WRAP_HALF: u64 = TS_WRAP / 2;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Codec {
H264,
Hevc,
Mpeg2Video,
MpegAudio(bool),
Aac,
Ac3,
Eac3,
Dts,
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,
_ => Codec::Data(stream_type),
}
}
}
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 {
let mut delta = raw as i128 - prev_raw as i128;
if delta > TS_WRAP_HALF as i128 {
delta -= TS_WRAP as i128; } else if delta < -(TS_WRAP_HALF as i128) {
delta += TS_WRAP as i128; }
prev_unwrapped + delta
}
fn interpolate_ts(anchor_uw: i128, elapsed_samples: u64, sample_rate: u32) -> u64 {
let base = anchor_uw.max(0) as u128;
let offset = (elapsed_samples as u128 * VIDEO_TIMESCALE as u128) / sample_rate.max(1) as u128;
(base + offset) as u64
}
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
}
fn split_mpeg_audio_frames(payload: &[u8]) -> Vec<&[u8]> {
let mut frames = Vec::new();
let mut off = 0usize;
while off + 4 <= payload.len() {
let Ok(hdr) = MpegAudioFrameHeader::parse(&payload[off..]) else {
break;
};
let flen = hdr.frame_length;
if flen < 4 || off + flen > payload.len() {
break;
}
frames.push(&payload[off..off + flen]);
off += flen;
}
frames
}
fn split_adts_frames(payload: &[u8]) -> Vec<&[u8]> {
let mut frames = Vec::new();
let mut off = 0usize;
while off + ADTS_HEADER_SIZE <= payload.len() {
let Ok(hdr) = parse_adts_header(&payload[off..]) else {
break;
};
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]);
off += frame_len;
}
frames
}
fn sfi_to_hz(sfi: u8) -> Option<u32> {
Some(match sfi {
0 => 96000,
1 => 88200,
2 => 64000,
3 => 48000,
4 => 44100,
5 => 32000,
6 => 24000,
7 => 22050,
8 => 16000,
9 => 12000,
10 => 11025,
11 => 8000,
12 => 7350,
_ => return None,
})
}
fn parse_pat(section: &[u8]) -> Result<Vec<u16>> {
if section.first().copied() != Some(TABLE_ID_PAT) {
return Ok(Vec::new());
}
let body = section_body(section, "PAT")?;
let mut pids = 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 {
pids.push(pid);
}
off += PAT_ENTRY_LEN;
}
Ok(pids)
}
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 codec = Codec::from_stream_type(stream_type);
let descriptors = body[desc_start..desc_end].to_vec();
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])
}
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,
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::Data(_) => ConfigProbe::Data,
}
}
#[derive(Clone, Copy)]
enum VideoCodec {
H264,
Hevc,
Mpeg2,
}
enum AudioKind {
Aac,
Ac3,
Eac3,
Dts,
MpegAudio { samples_per_frame: u32 },
}
struct PendingOneBehind {
data: Vec<u8>,
is_sync: bool,
composition_offset: i32,
pts_uw: i128,
dts_uw: i128,
}
enum LiveKind {
Video {
pending: Option<PendingOneBehind>,
last_duration: u32,
codec: VideoCodec,
},
Audio { sample_rate: u32, kind: AudioKind },
Data {
pending: Option<PendingOneBehind>,
last_duration: u32,
},
Section,
}
struct LiveTrack {
track_id: u32,
kind: LiveKind,
}
enum TrackState {
Probing {
probe: ConfigProbe,
backlog: Vec<BufferedAu>,
},
Parked {
config: CodecConfig,
timescale: u32,
kind: LiveKind,
backlog: Vec<BufferedAu>,
},
Live(LiveTrack),
}
#[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,
}
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, dts_uw)
}
}
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,
fallback: (u64, u64),
has_any: bool,
wrap: WrapState,
first_dts_uw: Option<i128>,
track: Option<TrackState>,
}
#[allow(clippy::too_many_arguments)]
fn advance_one_behind(
pending: &mut Option<PendingOneBehind>,
last_duration: &mut u32,
data: Vec<u8>,
is_sync: bool,
composition_offset: i32,
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;
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample {
data: prev.data,
duration,
is_sync: prev.is_sync,
composition_offset: prev.composition_offset,
source_timing: Some(SourceTiming {
pts: prev.pts_uw.max(0) as u64,
dts: prev.dts_uw.max(0) as u64,
}),
},
});
}
*pending = Some(PendingOneBehind {
data,
is_sync,
composition_offset,
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() {
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample {
data: p.data,
duration: last_duration,
is_sync: p.is_sync,
composition_offset: p.composition_offset,
source_timing: Some(SourceTiming {
pts: p.pts_uw.max(0) as u64,
dts: p.dts_uw.max(0) as u64,
}),
},
});
}
}
fn video_sample_bytes(codec: VideoCodec, au_data: &[u8]) -> (Vec<u8>, bool) {
match codec {
VideoCodec::H264 => {
let mut idr = false;
for nal in iter_annexb_nals(au_data) {
if is_keyframe_nal(NalCodec::Avc, nal) {
idr = true;
}
}
(annexb_to_length_prefixed(au_data), idr)
}
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)),
}
}
fn emit_audio_au(
kind: &AudioKind,
sample_rate: u32,
au_data: &[u8],
pts_uw: i128,
dts_uw: i128,
track_id: u32,
events: &mut VecDeque<DemuxEvent>,
) {
let mut elapsed = 0u64;
match kind {
AudioKind::Aac => {
for frame in split_adts_frames(au_data) {
if frame.len() > ADTS_HEADER_SIZE {
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample::from_raw(
frame[ADTS_HEADER_SIZE..].to_vec(),
AAC_SAMPLES_PER_FRAME,
)
.with_source_timing(SourceTiming {
pts: interpolate_ts(pts_uw, elapsed, sample_rate),
dts: interpolate_ts(dts_uw, elapsed, sample_rate),
}),
});
}
elapsed += AAC_SAMPLES_PER_FRAME as u64;
}
}
AudioKind::Ac3 => {
for frame in split_ac3_syncframes(au_data) {
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample::from_raw(frame.to_vec(), AC3_SAMPLES_PER_SYNCFRAME)
.with_source_timing(SourceTiming {
pts: interpolate_ts(pts_uw, elapsed, sample_rate),
dts: interpolate_ts(dts_uw, elapsed, sample_rate),
}),
});
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: Sample::from_raw(split.data, duration).with_source_timing(
SourceTiming {
pts: interpolate_ts(pts_uw, elapsed, sample_rate),
dts: interpolate_ts(dts_uw, elapsed, sample_rate),
},
),
});
elapsed += duration as u64;
}
}
AudioKind::Dts => {
for frame in split_dts_core_frames(au_data) {
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample::from_raw(frame.data.to_vec(), frame.samples)
.with_source_timing(SourceTiming {
pts: interpolate_ts(pts_uw, elapsed, sample_rate),
dts: interpolate_ts(dts_uw, elapsed, sample_rate),
}),
});
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: Sample::from_raw(frame.to_vec(), *samples_per_frame)
.with_source_timing(SourceTiming {
pts: interpolate_ts(pts_uw, elapsed, sample_rate),
dts: interpolate_ts(dts_uw, elapsed, sample_rate),
}),
});
elapsed += *samples_per_frame as u64;
}
}
}
}
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);
let composition_offset = (pts_uw - dts_uw) as i32;
advance_one_behind(
pending,
last_duration,
bytes,
is_sync,
composition_offset,
pts_uw,
dts_uw,
false,
track_id,
events,
);
}
LiveKind::Data {
pending,
last_duration,
} => {
advance_one_behind(
pending,
last_duration,
data.to_vec(),
true,
0,
pts_uw,
dts_uw,
true,
track_id,
events,
);
}
LiveKind::Audio { sample_rate, kind } => {
emit_audio_au(kind, *sample_rate, data, pts_uw, dts_uw, track_id, events);
}
LiveKind::Section => {
events.push_back(DemuxEvent::Sample {
track_id,
sample: Sample::from_raw(data.to_vec(), 0),
});
}
}
}
fn finalize_probe(
codec: Codec,
descriptors: &[u8],
probe: &mut ConfigProbe,
backlog: &[BufferedAu],
) -> Option<(CodecConfig, u32, LiveKind)> {
let latest = backlog
.last()
.expect("finalize_probe is only called after pushing the latest AU");
match probe {
ConfigProbe::Data => {
let Codec::Data(stream_type) = codec else {
unreachable!("ConfigProbe::Data is only created for Codec::Data")
};
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.data) {
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 as u16, i.height 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.data) {
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 = backlog
.iter()
.find_map(|au| Mpeg2SeqHeader::find(&au.data).ok())?;
let esds = EsdsBox::new(ESDescriptor {
es_id: ESDS_VIDEO_ES_ID,
stream_dependence_flag: false,
url_flag: false,
ocr_stream_flag: false,
stream_priority: 0,
depends_on_es_id: None,
url: None,
ocr_es_id: None,
decoder_config: Some(DecoderConfigDescriptor {
object_type_indication: ObjectTypeIndication(OTI_MPEG2_VIDEO_MAIN),
stream_type: EsdsStreamType(STREAM_TYPE_VISUAL),
up_stream: false,
buffer_size_db: 0,
max_bitrate: 0,
avg_bitrate: 0,
decoder_specific_info: None,
}),
sl_config: Some(SLConfigDescriptor {
body: alloc::vec![SL_CONFIG_PREDEFINED_MP4],
}),
});
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 = backlog
.iter()
.find_map(|au| MpegAudioFrameHeader::parse(&au.data).ok())?;
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 {
es_id: ESDS_ES_ID,
stream_dependence_flag: false,
url_flag: false,
ocr_stream_flag: false,
stream_priority: 0,
depends_on_es_id: None,
url: None,
ocr_es_id: None,
decoder_config: Some(DecoderConfigDescriptor {
object_type_indication: ObjectTypeIndication(oti),
stream_type: EsdsStreamType(STREAM_TYPE_AUDIO),
up_stream: false,
buffer_size_db: 0,
max_bitrate: 0,
avg_bitrate: 0,
decoder_specific_info: None,
}),
sl_config: Some(SLConfigDescriptor {
body: alloc::vec![SL_CONFIG_PREDEFINED_MP4],
}),
});
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 },
},
))
}
ConfigProbe::Aac => {
let first_hdr = backlog
.iter()
.find_map(|au| parse_adts_header(&au.data).ok())?;
let asc = AudioSpecificConfig::from_adts_header(&first_hdr);
let sample_rate = sfi_to_hz(first_hdr.sampling_frequency_index)?;
let channel_count = first_hdr.channel_configuration as u16;
let esds = EsdsBox::new(ESDescriptor {
es_id: ESDS_ES_ID,
stream_dependence_flag: false,
url_flag: false,
ocr_stream_flag: false,
stream_priority: 0,
depends_on_es_id: None,
url: None,
ocr_es_id: None,
decoder_config: Some(DecoderConfigDescriptor {
object_type_indication: ObjectTypeIndication(OTI_MPEG4_AUDIO),
stream_type: EsdsStreamType(STREAM_TYPE_AUDIO),
up_stream: false,
buffer_size_db: 0,
max_bitrate: 0,
avg_bitrate: 0,
decoder_specific_info: Some(DecoderSpecificInfo {
data: asc.to_bytes(),
}),
}),
sl_config: Some(SLConfigDescriptor {
body: alloc::vec![SL_CONFIG_PREDEFINED_MP4],
}),
});
Some((
CodecConfig::Aac {
esds,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::Aac,
},
))
}
ConfigProbe::Ac3 => {
let info = backlog
.iter()
.find_map(|au| Ac3SyncframeInfo::from_es(&au.data).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,
},
))
}
ConfigProbe::Eac3 => {
let info = backlog
.iter()
.find_map(|au| Ec3SyncframeInfo::from_es(&au.data).ok())?;
let sample_rate = info.sample_rate;
let channel_count = info.channel_count() as u16;
let config = info.into_dec3();
Some((
CodecConfig::Eac3 {
config,
channel_count,
sample_rate,
sample_size: AUDIO_SAMPLE_SIZE_BITS,
},
sample_rate,
LiveKind::Audio {
sample_rate,
kind: AudioKind::Eac3,
},
))
}
ConfigProbe::Dts => {
let info = backlog
.iter()
.find_map(|au| DtsCoreFrameInfo::from_es(&au.data).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,
},
))
}
}
}
fn advance_track(
stream: &mut StreamState,
data: Vec<u8>,
pts_uw: i128,
dts_uw: i128,
events: &mut VecDeque<DemuxEvent>,
) {
let track = stream
.track
.take()
.expect("StreamState.track is always populated outside this function");
let new_track = match track {
TrackState::Live(mut live) => {
push_live_au(&mut live, &data, pts_uw, dts_uw, events);
TrackState::Live(live)
}
TrackState::Parked {
config,
timescale,
kind,
mut backlog,
} => {
backlog.push(BufferedAu {
data,
pts_uw,
dts_uw,
});
TrackState::Parked {
config,
timescale,
kind,
backlog,
}
}
TrackState::Probing {
mut probe,
mut backlog,
} => {
backlog.push(BufferedAu {
data,
pts_uw,
dts_uw,
});
match finalize_probe(stream.codec, &stream.descriptors, &mut probe, &backlog) {
Some((config, timescale, kind)) => TrackState::Parked {
config,
timescale,
kind,
backlog,
},
None => TrackState::Probing { probe, backlog },
}
}
};
stream.track = Some(new_track);
}
fn on_completed_pes(stream: &mut StreamState, pes_bytes: &[u8], events: &mut VecDeque<DemuxEvent>) {
let Ok(pes) = PesPacket::parse(pes_bytes) else {
return;
};
if pes.payload.is_empty() {
return;
}
let fallback = if stream.has_any {
stream.fallback
} else {
(0, 0)
};
let (pts, dts) = match pes.header.as_ref() {
Some(h) => {
let hp = h.pts.map(|p| p.0);
let hd = h.dts.map(|d| d.0);
let pts = hp.or(hd).unwrap_or(fallback.0);
let dts = hd.unwrap_or(pts);
(pts, dts)
}
None => fallback,
};
stream.fallback = (pts, dts);
stream.has_any = true;
let (pts_uw, dts_uw) = stream.wrap.push(pts, dts);
if stream.first_dts_uw.is_none() {
stream.first_dts_uw = Some(dts_uw);
}
advance_track(stream, pes.payload.to_vec(), pts_uw, dts_uw, events);
}
fn on_completed_section(
stream: &mut StreamState,
section: &[u8],
events: &mut VecDeque<DemuxEvent>,
) {
if section.is_empty() {
return;
}
advance_track(stream, section.to_vec(), 0, 0, events);
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum DemuxEvent {
TrackAdded(Track),
TrackUpdated(Track),
Sample {
track_id: u32,
sample: Sample,
},
Pcr(PcrSample),
Discontinuity {
pid: u16,
},
}
pub struct StreamingTsDemux {
resync: TsResync,
packet_index: u64,
pat_reasm: SectionReassembler,
pmt_reasm: BTreeMap<u16, SectionReassembler>,
es_seen: BTreeSet<u16>,
streams: BTreeMap<u16, StreamState>,
unattributed: BTreeMap<u16, VecDeque<(bool, Vec<u8>)>>,
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>,
}
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(),
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(),
}
}
pub fn feed(&mut self, data: &[u8]) {
let packets = self.resync.feed(data);
for raw in &packets {
self.process_packet(raw);
}
}
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 let Some(Ok(af)) = pkt.adaptation_field() {
if af.discontinuity_indicator {
self.events.push_back(DemuxEvent::Discontinuity {
pid: pkt.header.pid,
});
}
if let Some(pcr) = af.pcr {
self.events.push_back(DemuxEvent::Pcr(PcrSample {
pcr_27mhz: pcr.as_27mhz(),
pid: pkt.header.pid,
packet_index: idx,
discontinuity: af.discontinuity_indicator,
}));
}
}
let pid = pkt.header.pid;
let pusi = pkt.header.pusi;
let Some(payload) = pkt.payload else {
return;
};
if pid == PAT_PID {
self.pat_reasm.feed(payload, pusi);
while let Some(section) = self.pat_reasm.pop_section() {
if let Ok(pmt_pids) = parse_pat(§ion) {
for pmt_pid in pmt_pids {
self.pmt_reasm.entry(pmt_pid).or_default();
}
}
}
return;
}
if let Some(reasm) = self.pmt_reasm.get_mut(&pid) {
reasm.feed(payload, pusi);
let mut newly = Vec::new();
while let Some(section) = reasm.pop_section() {
if let Ok(es_list) = parse_pmt(§ion) {
newly.extend(es_list);
}
}
for (es_pid, codec, descriptors) in newly {
if self.es_seen.insert(es_pid) {
if matches!(codec, Codec::Data(_)) {
self.data_order.push(es_pid);
} else {
self.codec_order.push(es_pid);
}
let mut stream = StreamState {
codec,
descriptors,
carrier: initial_carrier(codec),
fallback: (0, 0),
has_any: false,
wrap: WrapState::default(),
first_dts_uw: None,
track: Some(TrackState::Probing {
probe: initial_probe(codec),
backlog: Vec::new(),
}),
};
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 completed_pes: Option<Vec<u8>> = None;
let mut sections: Vec<Vec<u8>> = Vec::new();
match &mut stream.carrier {
Carrier::Pes(assembler) => {
completed_pes = assembler.feed(buf_pusi, &buf_payload);
}
Carrier::Section(reasm) => {
reasm.feed(&buf_payload, buf_pusi);
while let Some(s) = reasm.pop_section() {
sections.push(s.to_vec());
}
}
}
if let Some(completed) = completed_pes {
on_completed_pes(&mut stream, &completed, &mut self.events);
}
for s in sections {
on_completed_section(&mut stream, &s, &mut self.events);
}
}
}
self.streams.insert(es_pid, stream);
}
}
self.try_promote_ready();
return;
}
if let Some(stream) = self.streams.get_mut(&pid) {
let mut completed_pes: Option<Vec<u8>> = None;
let mut sections: Vec<Vec<u8>> = Vec::new();
match &mut stream.carrier {
Carrier::Pes(assembler) => {
completed_pes = assembler.feed(pusi, payload);
}
Carrier::Section(reasm) => {
reasm.feed(payload, pusi);
while let Some(s) = reasm.pop_section() {
sections.push(s.to_vec());
}
}
}
if let Some(completed) = completed_pes {
on_completed_pes(stream, &completed, &mut self.events);
}
for s in sections {
on_completed_section(stream, &s, &mut self.events);
}
} else if pid != NULL_PACKET_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 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 try_promote_ready(&mut self) {
loop {
let Some(&next_pid) = self
.codec_order
.iter()
.chain(self.data_order.iter())
.find(|p| !self.resolved.contains(p))
else {
return;
};
let Some(stream) = self.streams.get_mut(&next_pid) else {
return;
};
let track = stream
.track
.take()
.expect("StreamState.track is always populated outside this function");
match track {
TrackState::Parked {
config,
timescale,
kind,
backlog,
} => {
let track_id = self.next_track_id;
self.next_track_id += 1;
let anchor = stream.first_dts_uw.unwrap_or(0).max(0) as u64;
let spec = TrackSpec::new(track_id, timescale, config)
.with_source(next_pid, stream.descriptors.clone());
self.events.push_back(DemuxEvent::TrackAdded(Track::new_at(
spec,
Vec::new(),
anchor,
)));
let mut live = LiveTrack { track_id, kind };
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));
self.resolved.insert(next_pid);
}
other @ TrackState::Probing { .. } => {
stream.track = Some(other);
return; }
other @ TrackState::Live(_) => {
stream.track = Some(other);
self.resolved.insert(next_pid);
}
}
}
}
pub fn poll_event(&mut self) -> Option<DemuxEvent> {
self.events.pop_front()
}
pub fn finish(&mut self) {
for stream in self.streams.values_mut() {
let completed = match &mut stream.carrier {
Carrier::Pes(assembler) => assembler.flush(),
Carrier::Section(_) => None,
};
if let Some(completed) = completed {
on_completed_pes(stream, &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.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,
} => {
flush_one_behind(pending, *last_duration, live.track_id, &mut self.events);
}
LiveKind::Audio { .. } => {}
LiveKind::Section => {}
}
}
}
}
}
#[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(track) => {
index_by_id.insert(track.spec.track_id, tracks.len());
tracks.push(track);
}
DemuxEvent::TrackUpdated(track) => {
if let Some(&i) = index_by_id.get(&track.spec.track_id) {
let samples = core::mem::take(&mut tracks[i].samples);
tracks[i] = track;
tracks[i].samples = samples;
}
}
DemuxEvent::Sample { track_id, sample } => {
if let Some(&i) = index_by_id.get(&track_id) {
tracks[i].samples.push(sample);
}
}
DemuxEvent::Pcr(sample) => pcr.push(sample),
DemuxEvent::Discontinuity { .. } => {}
}
}
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::*;
#[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"
);
}
}