use alloc::vec::Vec;
use mpeg_ts::pid::Pid;
use broadcast_common::{Package, crc32_mpeg2};
use mpeg_pes::{Pts as PesPts, StreamId};
use mpeg_ts::mux::SectionPacketiser;
use mpeg_ts::ts::{Pcr, TS_PACKET_SIZE, TsHeader};
use crate::aac_asc::AudioSpecificConfig;
use crate::annexb::{iter_length_prefixed_nals, length_prefixed_to_annexb};
use crate::error::{Error, Result};
use crate::media::{Media, TimelineOrigin, Track, relative_decode_times};
use crate::mp4esds::EsdsBox;
use crate::nal::{NalCodec, nal_unit_type};
use crate::pipeline::{CodecConfig, DataCarriage, Sample};
const PAT_PID: Pid = Pid::new(0x0000);
const PMT_PID: Pid = Pid::new(0x1000);
const ES_PID_BASE: Pid = Pid::new(0x0100);
const PROGRAM_NUMBER: u16 = 1;
const TABLE_ID_PAT: u8 = 0x00;
const TABLE_ID_PMT: u8 = 0x02;
const CRC32_LEN: usize = 4;
const SECTION_SYNTAX_FLAGS_HI: u8 = 0xB0;
const SECTION_LENGTH_HI_MASK: u8 = 0x0F;
const MAX_SECTION_LENGTH: usize = 0x3FD;
const VERSION_CURRENT_NEXT: u8 = 0xC1;
const PID_RESERVED_HI: u8 = 0xE0;
const INFO_RESERVED_HI: u8 = 0xF0;
const STREAM_TYPE_MPEG2_VIDEO: u8 = 0x02;
const STREAM_TYPE_MPEG1_AUDIO: u8 = 0x03;
const STREAM_TYPE_MPEG2_AUDIO: u8 = 0x04;
const STREAM_TYPE_AAC_ADTS: u8 = 0x0F;
const STREAM_TYPE_AVC: u8 = 0x1B;
const STREAM_TYPE_HEVC: u8 = 0x24;
const STREAM_TYPE_MPEGH: u8 = 0x2D;
const STREAM_TYPE_AC3: u8 = 0x81;
const STREAM_TYPE_DTS: u8 = 0x82;
const STREAM_TYPE_EAC3: u8 = 0x87;
const OTI_MPEG1_AUDIO: u8 = 0x6B;
const OTI_MPEG2_AUDIO: u8 = 0x69;
const MAX_ES_INFO_LENGTH: usize = 0x0FFF;
const DESCRIPTOR_TAG_CA: u8 = 0x09;
const DESCRIPTOR_TAG_EXTENSION: u8 = 0x3F;
const MPEGH_REFERENCE_CHANNEL_LAYOUT_MAX: u8 = 0x3F;
const MPEGH_REFERENCE_CHANNEL_LAYOUT_FALLBACK: u8 = 0;
fn mpegh_3daudio_descriptor(
profile_level_indication: u8,
reference_channel_layout: u8,
) -> Result<Vec<u8>> {
use broadcast_common::Serialize as _;
use dvb_si::descriptors::mpeg2_extension::{Mpeg2ExtensionBody, Mpeg2ExtensionDescriptor};
use dvb_si::descriptors::mpegh_3d_audio::MpegH3dAudio;
let reference_channel_layout = if reference_channel_layout > MPEGH_REFERENCE_CHANNEL_LAYOUT_MAX
{
MPEGH_REFERENCE_CHANNEL_LAYOUT_FALLBACK
} else {
reference_channel_layout
};
let body = MpegH3dAudio::new(
profile_level_indication,
false,
None,
reference_channel_layout,
);
Mpeg2ExtensionDescriptor::new(Mpeg2ExtensionBody::MpegH3dAudio(body))
.try_to_bytes()
.map_err(|_| Error::InvalidInput("MPEG-H_3dAudio_descriptor does not encode"))
}
const STREAM_ID_VIDEO_BASE: u8 = 0xE0;
const STREAM_ID_VIDEO_LIMIT: u8 = 0xF0;
const STREAM_ID_AUDIO_BASE: u8 = 0xC0;
const STREAM_ID_AUDIO_LIMIT: u8 = 0xE0;
const MAX_VIDEO_STREAMS: u8 = STREAM_ID_VIDEO_LIMIT - STREAM_ID_VIDEO_BASE;
const MAX_AUDIO_STREAMS: u8 = STREAM_ID_AUDIO_LIMIT - STREAM_ID_AUDIO_BASE;
const STREAM_ID_PRIVATE_1: u8 = 0xBD;
const PES_START_CODE: [u8; 3] = [0x00, 0x00, 0x01];
const HEADER_FIXED: usize = 3;
const MIN_LEN: usize = 6;
const PES_OPTIONAL_MARKER: u8 = 0x80;
const PTS_DTS_FLAGS_PTS_ONLY: u8 = 0x80;
const PTS_DTS_FLAGS_BOTH: u8 = 0xC0;
const TS_PREFIX_PTS_WITH_DTS: u8 = 0b0011;
const TS_PREFIX_DTS: u8 = 0b0001;
const H264_NAL_TYPE_MASK: u8 = 0x1F;
const H264_NAL_AUD: u8 = 9;
const H264_NAL_SPS: u8 = 7;
const H264_NAL_AUD_BYTE: u8 = 0x09;
const H264_AUD_PRIMARY_PIC_TYPE: u8 = 0xF0;
const HEVC_NAL_VPS: u8 = 32;
const HEVC_NAL_SPS: u8 = 33;
const HEVC_NAL_PPS: u8 = 34;
const HEVC_NAL_AUD: u8 = 35;
const HEVC_AUD_FIRST_BYTE: u8 = 0x46;
const HEVC_AUD_SECOND_BYTE: u8 = 0x01;
const HEVC_AUD_PIC_TYPE: u8 = 0x50;
const AF_CTRL_ADAPTATION: u8 = 0x20;
const AF_CTRL_PAYLOAD: u8 = 0x10;
const AF_PCR_FLAG: u8 = 0x10;
const PCR_FIELD_LEN: usize = 6;
const STUFFING_BYTE: u8 = 0xFF;
const TS_CLOCK_HZ: u64 = 90_000;
const TS_TIMESTAMP_MOD: u64 = broadcast_common::clock33::WRAP_33BIT;
const PCR_LEAD_TICKS: u64 = 9_000;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EsKind {
Avc,
Hevc,
Mpeg2Video,
Aac,
MpegAudio {
is_mpeg2: bool,
},
Ac3,
Eac3,
Dts,
MpegH,
Data {
stream_type: u8,
carriage: DataCarriage,
},
}
impl EsKind {
fn stream_type(self) -> u8 {
match self {
EsKind::Avc => STREAM_TYPE_AVC,
EsKind::Hevc => STREAM_TYPE_HEVC,
EsKind::Mpeg2Video => STREAM_TYPE_MPEG2_VIDEO,
EsKind::Aac => STREAM_TYPE_AAC_ADTS,
EsKind::MpegAudio { is_mpeg2: true } => STREAM_TYPE_MPEG2_AUDIO,
EsKind::MpegAudio { is_mpeg2: false } => STREAM_TYPE_MPEG1_AUDIO,
EsKind::Ac3 => STREAM_TYPE_AC3,
EsKind::Eac3 => STREAM_TYPE_EAC3,
EsKind::Dts => STREAM_TYPE_DTS,
EsKind::MpegH => STREAM_TYPE_MPEGH,
EsKind::Data { stream_type, .. } => stream_type,
}
}
fn is_video(self) -> bool {
matches!(self, EsKind::Avc | EsKind::Hevc | EsKind::Mpeg2Video)
}
fn is_continuous_audio(self) -> bool {
matches!(
self,
EsKind::Aac
| EsKind::MpegAudio { .. }
| EsKind::Ac3
| EsKind::Eac3
| EsKind::Dts
| EsKind::MpegH
)
}
fn is_section_carried(self) -> bool {
matches!(
self,
EsKind::Data {
carriage: DataCarriage::Sections,
..
}
)
}
fn from_config(config: &CodecConfig) -> Option<Self> {
match config {
CodecConfig::Avc { .. } => Some(EsKind::Avc),
CodecConfig::Hevc { .. } => Some(EsKind::Hevc),
CodecConfig::Mpeg2Video { .. } => Some(EsKind::Mpeg2Video),
CodecConfig::Aac { .. } => Some(EsKind::Aac),
CodecConfig::MpegAudio { esds, .. } => Some(EsKind::MpegAudio {
is_mpeg2: mpeg_audio_is_mpeg2(esds),
}),
CodecConfig::Ac3 { .. } => Some(EsKind::Ac3),
CodecConfig::Eac3 { .. } => Some(EsKind::Eac3),
CodecConfig::Dts { .. } => Some(EsKind::Dts),
CodecConfig::MpegH { .. } => Some(EsKind::MpegH),
CodecConfig::Data {
stream_type,
carriage,
..
} => Some(EsKind::Data {
stream_type: *stream_type,
carriage: *carriage,
}),
_ => None,
}
}
}
fn mpeg_audio_is_mpeg2(esds: &EsdsBox) -> bool {
let oti = esds
.es_descriptor
.decoder_config
.as_ref()
.map(|dc| dc.object_type_indication.0)
.unwrap_or(OTI_MPEG1_AUDIO);
oti == OTI_MPEG2_AUDIO
}
pub(crate) struct EsPlan {
pid: Pid,
stream_id: StreamId,
kind: EsKind,
asc: Option<AudioSpecificConfig>,
avc_sps_pps: Vec<Vec<u8>>,
hevc_vps_sps_pps: Vec<Vec<u8>>,
descriptors: Vec<u8>,
}
const TS_PAYLOAD_CAPACITY: usize = TS_PACKET_SIZE - 4;
const PACKETS_PER_SAMPLE_SLACK: usize = 2;
struct TaggedPacket {
sort_key: u64,
packet: [u8; TS_PACKET_SIZE],
}
#[derive(Debug, Clone)]
pub struct TsMux {
psi_interval_ms: Option<u32>,
}
pub const DEFAULT_PSI_INTERVAL_MS: u32 = 100;
const MS_PER_SECOND: u64 = 1000;
impl Default for TsMux {
fn default() -> Self {
Self::new()
}
}
impl TsMux {
pub fn new() -> Self {
Self {
psi_interval_ms: Some(DEFAULT_PSI_INTERVAL_MS),
}
}
#[must_use]
pub fn with_psi_interval_ms(mut self, interval_ms: Option<u32>) -> Self {
self.psi_interval_ms = interval_ms.filter(|&ms| ms > 0);
self
}
pub fn psi_interval_ms(&self) -> Option<u32> {
self.psi_interval_ms
}
}
impl Package for TsMux {
type Media = Media;
type Output = Vec<u8>;
type Error = Error;
fn package(&mut self, media: &Media) -> Result<Vec<u8>> {
if media.tracks.is_empty() {
return Err(Error::InvalidInput("cannot package a Media with no tracks"));
}
let samples: Vec<&[Sample]> = media.tracks.iter().map(|t| t.samples.as_slice()).collect();
let interval_ticks = self
.psi_interval_ms
.map(|ms| u64::from(ms) * TS_CLOCK_HZ / MS_PER_SECOND);
mux_tracks(&media.tracks, &samples, interval_ticks)
}
}
#[derive(Debug, Clone)]
pub struct StreamingTsMux {
psi_interval_ms: Option<u32>,
origin: Option<TimelineOrigin>,
continuity: TsContinuity,
psi: PsiState,
}
impl Default for StreamingTsMux {
fn default() -> Self {
Self::new()
}
}
impl StreamingTsMux {
pub fn new() -> Self {
Self::with_psi_interval_ms(Some(DEFAULT_PSI_INTERVAL_MS))
}
pub fn with_psi_interval_ms(interval_ms: Option<u32>) -> Self {
let interval_ms = interval_ms.filter(|&ms| ms > 0);
Self {
psi_interval_ms: interval_ms,
origin: None,
continuity: TsContinuity::new(),
psi: PsiState::new(interval_ms.map(|ms| u64::from(ms) * TS_CLOCK_HZ / MS_PER_SECOND)),
}
}
pub fn psi_interval_ms(&self) -> Option<u32> {
self.psi_interval_ms
}
}
impl Package for StreamingTsMux {
type Media = Media;
type Output = Vec<u8>;
type Error = Error;
fn package(&mut self, media: &Media) -> Result<Vec<u8>> {
if media.tracks.is_empty() {
return Err(Error::InvalidInput("cannot package a Media with no tracks"));
}
let has_dts = media
.tracks
.iter()
.any(|t| t.samples.iter().any(|s| s.dts.is_some()));
if !has_dts && self.origin.is_none() {
return Ok(Vec::new());
}
let origin = *self
.origin
.get_or_insert_with(|| TimelineOrigin::of(&media.tracks));
let times: Vec<Vec<i64>> = media
.tracks
.iter()
.map(|t| relative_decode_times(t, &origin))
.collect();
let times: Vec<&[i64]> = times.iter().map(Vec::as_slice).collect();
let samples: Vec<&[Sample]> = media.tracks.iter().map(|t| t.samples.as_slice()).collect();
mux_tracks_timed_with_cc(
&media.tracks,
&samples,
×,
&mut self.continuity,
&mut self.psi,
)
}
}
pub(crate) fn plan_elementary_streams(tracks: &[Track]) -> Result<(Vec<EsPlan>, Vec<usize>)> {
let mut plans: Vec<EsPlan> = Vec::new();
let mut planned_idx: Vec<usize> = Vec::new();
let mut next_pid = ES_PID_BASE.value();
let (mut n_video, mut n_audio) = (0u8, 0u8);
for (idx, track) in tracks.iter().enumerate() {
let Some(kind) = EsKind::from_config(&track.spec.config) else {
continue; };
let stream_id = if kind.is_video() {
if n_video >= MAX_VIDEO_STREAMS {
return Err(Error::TooManyElementaryStreams {
family: "video",
max: MAX_VIDEO_STREAMS,
});
}
let id = StreamId(STREAM_ID_VIDEO_BASE + n_video);
n_video += 1;
id
} else {
match kind {
EsKind::Data {
carriage: DataCarriage::Pes,
..
} => StreamId(STREAM_ID_PRIVATE_1),
EsKind::Data {
carriage: DataCarriage::Sections,
..
} => StreamId(0),
_ => {
if n_audio >= MAX_AUDIO_STREAMS {
return Err(Error::TooManyElementaryStreams {
family: "audio",
max: MAX_AUDIO_STREAMS,
});
}
let id = StreamId(STREAM_ID_AUDIO_BASE + n_audio);
n_audio += 1;
id
}
}
};
let asc = match &track.spec.config {
CodecConfig::Aac { esds, .. } => Some(esds.audio_specific_config()?),
_ => None,
};
let avc_sps_pps = match &track.spec.config {
CodecConfig::Avc { config, .. } => {
let r = &config.config;
let mut sets = Vec::new();
for sps in &r.sps {
sets.push(sps.0.clone());
}
for pps in &r.pps {
sets.push(pps.0.clone());
}
sets
}
_ => Vec::new(),
};
let hevc_vps_sps_pps = match &track.spec.config {
CodecConfig::Hevc { config, .. } => hevc_parameter_sets(&config.config),
_ => Vec::new(),
};
let synthesized: Vec<Vec<u8>> = match &track.spec.config {
CodecConfig::MpegH { config, .. } => {
alloc::vec![mpegh_3daudio_descriptor(
config.mpegh3da_profile_level_indication,
config.reference_channel_layout,
)?]
}
_ => Vec::new(),
};
let descriptors = merge_es_info_descriptors(&track.spec.es_info_descriptors, &synthesized)?;
plans.push(EsPlan {
pid: Pid::try_new(next_pid).map_err(|_| {
Error::InvalidInput("too many tracks: elementary PID exceeds 0x1FFF")
})?,
stream_id,
kind,
asc,
avc_sps_pps,
hevc_vps_sps_pps,
descriptors,
});
planned_idx.push(idx);
next_pid += 1;
}
if plans.is_empty() {
return Err(Error::InvalidInput(
"no track carries a TS-representable codec (AVC/HEVC/MPEG-2-video/AAC/\
MPEG-audio/AC-3/E-AC-3/DTS/MPEG-H/Data)",
));
}
Ok((plans, planned_idx))
}
fn merge_es_info_descriptors(inherited: &[u8], synthesized: &[Vec<u8>]) -> Result<Vec<u8>> {
let synthesized_keys: Vec<(u8, Option<u8>)> = synthesized
.iter()
.filter_map(|d| {
let tag = *d.first()?;
Some(descriptor_dedup_key(tag, d.get(2..).unwrap_or(&[])))
})
.collect();
let mut out = Vec::with_capacity(inherited.len() + total_param_len(synthesized));
let mut off = 0usize;
while off + 2 <= inherited.len() {
let tag = inherited[off];
let len = inherited[off + 1] as usize;
let end = (off + 2 + len).min(inherited.len());
let body = &inherited[(off + 2).min(end)..end];
let key = descriptor_dedup_key(tag, body);
if tag != DESCRIPTOR_TAG_CA && !synthesized_keys.contains(&key) {
out.extend_from_slice(&inherited[off..end]);
}
off = end;
}
for d in synthesized {
out.extend_from_slice(d);
}
if out.len() > MAX_ES_INFO_LENGTH {
return Err(Error::BufferCapExceeded {
what: "PMT ES_info descriptor loop",
cap: MAX_ES_INFO_LENGTH,
});
}
Ok(out)
}
fn descriptor_dedup_key(tag: u8, body: &[u8]) -> (u8, Option<u8>) {
if tag == DESCRIPTOR_TAG_EXTENSION {
(tag, body.first().copied())
} else {
(tag, None)
}
}
fn hevc_parameter_sets(
record: &crate::hevc_config::HEVCDecoderConfigurationRecord,
) -> Vec<Vec<u8>> {
let mut vps = Vec::new();
let mut sps = Vec::new();
let mut pps = Vec::new();
for array in &record.arrays {
let bucket = match array.nal_unit_type {
HEVC_NAL_VPS => &mut vps,
HEVC_NAL_SPS => &mut sps,
HEVC_NAL_PPS => &mut pps,
_ => continue,
};
for nalu in &array.nalus {
bucket.push(nalu.0.clone());
}
}
vps.into_iter().chain(sps).chain(pps).collect()
}
pub(crate) fn mux_tracks(
tracks: &[Track],
samples: &[&[Sample]],
psi_interval_ticks: Option<u64>,
) -> Result<Vec<u8>> {
let origin = TimelineOrigin::of(tracks);
let times: Vec<Vec<i64>> = tracks
.iter()
.map(|t| relative_decode_times(t, &origin))
.collect();
let times: Vec<&[i64]> = times.iter().map(Vec::as_slice).collect();
let mut cc = TsContinuity::default();
let mut psi = PsiState::new(psi_interval_ticks);
mux_tracks_timed_with_cc(tracks, samples, ×, &mut cc, &mut psi)
}
pub(crate) fn mux_tracks_at_continuing(
tracks: &[Track],
samples: &[&[Sample]],
base_dts_ticks: &[u64],
cc: &mut TsContinuity,
) -> Result<Vec<u8>> {
debug_assert_eq!(tracks.len(), base_dts_ticks.len());
let times: Vec<Vec<i64>> = samples
.iter()
.zip(base_dts_ticks)
.map(|(ss, &base)| {
let mut t = i64::try_from(base).unwrap_or(i64::MAX);
ss.iter()
.map(|s| {
let now = t;
t = t.saturating_add(i64::from(s.duration.unwrap_or(0)));
now
})
.collect()
})
.collect();
let times: Vec<&[i64]> = times.iter().map(Vec::as_slice).collect();
mux_tracks_timed_with_cc(tracks, samples, ×, cc, &mut PsiState::new(None))
}
#[derive(Debug, Clone)]
struct PsiState {
interval_ticks: Option<u64>,
started: bool,
due: Option<u64>,
}
impl PsiState {
fn new(interval_ticks: Option<u64>) -> Self {
Self {
interval_ticks,
started: false,
due: None,
}
}
}
#[derive(Debug, Default, Clone)]
pub struct TsContinuity {
counters: Vec<Option<u8>>,
}
impl TsContinuity {
pub fn new() -> Self {
Self::default()
}
fn next_for(&self, pid: Pid) -> u8 {
self.counters
.get(usize::from(pid.value()))
.copied()
.flatten()
.unwrap_or(0)
}
fn advance(&mut self, pid: Pid, cc: u8) {
self.set_next_for(pid, (cc + 1) & 0x0F);
}
fn set_next_for(&mut self, pid: Pid, cc: u8) {
let idx = usize::from(pid.value());
if self.counters.len() <= idx {
self.counters.resize(idx + 1, None);
}
self.counters[idx] = Some(cc & 0x0F);
}
}
fn mux_tracks_timed_with_cc(
tracks: &[Track],
samples: &[&[Sample]],
dts_ticks: &[&[i64]],
continuity: &mut TsContinuity,
psi: &mut PsiState,
) -> Result<Vec<u8>> {
let psi_interval_ticks = psi.interval_ticks;
debug_assert_eq!(tracks.len(), samples.len());
debug_assert_eq!(tracks.len(), dts_ticks.len());
let (plans, planned_idx) = plan_elementary_streams(tracks)?;
let pcr_pid = plans
.iter()
.find(|p| p.kind.is_video())
.or_else(|| plans.iter().find(|p| p.kind.is_continuous_audio()))
.or_else(|| plans.iter().find(|p| !p.kind.is_section_carried()))
.map(|p| p.pid)
.unwrap_or(plans[0].pid);
let mut out: Vec<u8> = Vec::new();
let pat = build_pat_section(PMT_PID)?;
let pmt = build_pmt_section(pcr_pid, &plans)?;
if !psi.started {
let pat_pkts = packetise_section(PAT_PID, &pat, continuity.next_for(PAT_PID));
let pat_last_cc = pat_pkts.last().map(|p| p[3] & 0x0F).unwrap_or(0);
for pkt in pat_pkts {
out.extend_from_slice(&pkt);
}
continuity.advance(PAT_PID, pat_last_cc);
let pmt_pkts = packetise_section(PMT_PID, &pmt, continuity.next_for(PMT_PID));
let pmt_last_cc = pmt_pkts.last().map(|p| p[3] & 0x0F).unwrap_or(0);
for pkt in pmt_pkts {
out.extend_from_slice(&pkt);
}
continuity.advance(PMT_PID, pmt_last_cc);
}
let packet_hint: usize = planned_idx
.iter()
.map(|&i| {
samples[i]
.iter()
.map(|s| s.data.len() / TS_PAYLOAD_CAPACITY + PACKETS_PER_SAMPLE_SLACK)
.sum::<usize>()
})
.sum();
let mut tagged: Vec<TaggedPacket> = Vec::with_capacity(packet_hint);
let mut pcr_stamps: Vec<u64> = Vec::new();
for (plan, &track_idx) in plans.iter().zip(&planned_idx) {
let track = &tracks[track_idx];
let ts_scale = track.spec.timescale.max(1) as u64;
let mut last_key: u64 = 0;
let mut cc: u8 = continuity.next_for(plan.pid);
let mut section_packetiser = SectionPacketiser::with_continuity(plan.pid, cc);
for (sample, &dts_local) in samples[track_idx].iter().zip(dts_ticks[track_idx]) {
let dts_ticks_local = dts_local.max(0) as u64;
let dts90 = rescale(dts_ticks_local, ts_scale) + PCR_LEAD_TICKS;
let sort_key = rescale_for_ordering(dts_ticks_local, ts_scale).max(last_key);
last_key = sort_key;
if plan.kind.is_section_carried() {
for pkt in section_packetiser.packetise(&[&sample.data[..]]) {
tagged.push(TaggedPacket {
sort_key,
packet: pkt,
});
}
} else {
let pts_local = dts_local.max(0) + sample.composition_offset() as i64;
let pts90 = rescale_signed(pts_local, ts_scale) + PCR_LEAD_TICKS;
let es_payload = build_es_payload(plan, sample)?;
let carry_pcr = plan.pid == pcr_pid;
packetise_pes(
plan,
&es_payload,
pts90,
dts90,
carry_pcr,
&mut cc,
sort_key,
&mut tagged,
)?;
if carry_pcr {
pcr_stamps.push(sort_key);
}
}
}
let next_cc = if plan.kind.is_section_carried() {
section_packetiser.continuity_counter()
} else {
cc
};
continuity.set_next_for(plan.pid, next_cc);
}
emit_pcr_only_packets(pcr_pid, &mut pcr_stamps, &mut tagged);
tagged.sort_by_key(|t| t.sort_key);
let mut pat_cc = continuity.next_for(PAT_PID);
let mut pmt_cc = continuity.next_for(PMT_PID);
let mut psi_due = match (psi_interval_ticks, tagged.first()) {
(Some(_), Some(_)) if psi.due.is_some() => psi.due,
(Some(iv), Some(first)) => Some(first.sort_key.saturating_add(iv)),
_ => None,
};
let mut prev_key: Option<u64> = None;
let psi_repeats = match (psi_interval_ticks, tagged.first(), tagged.last()) {
(Some(iv), Some(first), Some(last)) => {
usize::try_from((last.sort_key - first.sort_key) / iv).unwrap_or(0) + 1
}
_ => 0,
};
let psi_repeat_bytes = (pat.len() + 1).div_ceil(TS_PAYLOAD_CAPACITY) * TS_PACKET_SIZE
+ (pmt.len() + 1).div_ceil(TS_PAYLOAD_CAPACITY) * TS_PACKET_SIZE;
out.reserve_exact(tagged.len() * TS_PACKET_SIZE + psi_repeats * psi_repeat_bytes);
for t in &tagged {
if let (Some(due), Some(iv)) = (psi_due, psi_interval_ticks)
&& t.sort_key >= due
&& prev_key != Some(t.sort_key)
{
for (pid, section, next_cc) in
[(PAT_PID, &pat, &mut pat_cc), (PMT_PID, &pmt, &mut pmt_cc)]
{
for pkt in packetise_section(pid, section, *next_cc) {
*next_cc = ((pkt[3] & 0x0F) + 1) & 0x0F;
out.extend_from_slice(&pkt);
}
}
psi_due = Some(t.sort_key.saturating_add(iv));
}
prev_key = Some(t.sort_key);
out.extend_from_slice(&t.packet);
}
continuity.set_next_for(PAT_PID, pat_cc);
continuity.set_next_for(PMT_PID, pmt_cc);
psi.started = true;
psi.due = psi_due.or(psi.due);
debug_assert_eq!(out.len() % TS_PACKET_SIZE, 0);
Ok(out)
}
const PCR_ONLY_AF_LEN: usize = 1 + PCR_FIELD_LEN;
const MAX_PCR_INTERVAL_TICKS: u64 = PCR_INTERVAL_MS * TS_CLOCK_HZ / 1000;
const PCR_INTERVAL_MS: u64 = 40;
fn emit_pcr_only_packets(pcr_pid: Pid, pcr_stamps: &mut [u64], tagged: &mut Vec<TaggedPacket>) {
if pcr_stamps.is_empty() {
return;
}
pcr_stamps.sort_unstable();
let mut insertions: Vec<u64> = Vec::new();
for w in pcr_stamps.windows(2) {
let (a, b) = (w[0], w[1]);
if b - a <= MAX_PCR_INTERVAL_TICKS {
continue;
}
let mut t = a + MAX_PCR_INTERVAL_TICKS;
while t < b {
insertions.push(t);
t += MAX_PCR_INTERVAL_TICKS;
}
}
if insertions.is_empty() {
return;
}
let mut cc_at: Vec<(u64, u8)> = tagged
.iter()
.filter(|tp| pid_of_packet(&tp.packet) == pcr_pid.value())
.map(|tp| (tp.sort_key, tp.packet[3] & 0x0F))
.collect();
cc_at.sort_by_key(|&(k, _)| k);
for t in insertions {
let cc = cc_at
.iter()
.rfind(|&&(k, _)| k <= t)
.map(|&(_, cc)| cc)
.unwrap_or(0);
let mut pkt = [STUFFING_BYTE; TS_PACKET_SIZE];
write_pcr_only_packet(&mut pkt, pcr_pid, cc, pcr_for(t + PCR_LEAD_TICKS));
tagged.push(TaggedPacket {
sort_key: t,
packet: pkt,
});
}
}
fn pid_of_packet(pkt: &[u8; TS_PACKET_SIZE]) -> u16 {
(((pkt[1] & 0x1F) as u16) << 8) | pkt[2] as u16
}
fn write_pcr_only_packet(pkt: &mut [u8; TS_PACKET_SIZE], pid: Pid, cc: u8, pcr: Pcr) {
let hdr = TsHeader {
tei: false,
pusi: false,
pid,
scrambling: 0,
has_adaptation: true,
has_payload: false,
continuity_counter: cc,
};
hdr.serialize_into(&mut pkt[..4]).expect("4-byte TS header");
pkt[4] = PCR_ONLY_AF_LEN as u8;
pkt[5] = AF_PCR_FLAG;
pkt[6..6 + PCR_FIELD_LEN].copy_from_slice(&pcr.to_field_bytes());
}
fn rescale(ticks: u64, timescale: u64) -> u64 {
let scaled = (ticks * TS_CLOCK_HZ + timescale / 2) / timescale;
scaled % TS_TIMESTAMP_MOD
}
fn rescale_signed(ticks: i64, timescale: u64) -> u64 {
if ticks <= 0 {
return 0;
}
rescale(ticks as u64, timescale)
}
fn rescale_for_ordering(ticks: u64, timescale: u64) -> u64 {
let scaled = (ticks as u128 * TS_CLOCK_HZ as u128 + timescale as u128 / 2) / timescale as u128;
scaled.min(u64::MAX as u128) as u64
}
fn build_es_payload(plan: &EsPlan, sample: &Sample) -> Result<Vec<u8>> {
match plan.kind {
EsKind::Avc => {
let au = build_annexb_au(&sample.data, sample.flags.is_sync, &plan.avc_sps_pps)?;
ensure_avc_aud(au)
}
EsKind::Hevc => {
let au =
build_hevc_annexb_au(&sample.data, sample.flags.is_sync, &plan.hevc_vps_sps_pps)?;
ensure_hevc_aud(au)
}
EsKind::Aac => {
let asc = plan
.asc
.as_ref()
.ok_or(Error::InvalidInput("AAC ES has no AudioSpecificConfig"))?;
let frame_len_usize = sample.data.len() + 7; let frame_len: u16 = frame_len_usize
.try_into()
.map_err(|_| Error::InvalidValue {
field: "frame_len",
value: frame_len_usize as u64,
reason: "exceeds ADTS 13-bit frame_length maximum (8191)",
})?;
let header = asc.to_adts_header(frame_len)?;
let mut out = Vec::with_capacity(header.len() + sample.data.len());
out.extend_from_slice(&header);
out.extend_from_slice(&sample.data);
Ok(out)
}
EsKind::Mpeg2Video
| EsKind::MpegAudio { .. }
| EsKind::Ac3
| EsKind::Eac3
| EsKind::Dts
| EsKind::MpegH
| EsKind::Data { .. } => Ok(sample.data.to_vec()),
}
}
fn build_annexb_au(length_prefixed: &[u8], is_sync: bool, sps_pps: &[Vec<u8>]) -> Result<Vec<u8>> {
let nals = iter_length_prefixed_nals(length_prefixed)?;
let needs_params = is_sync
&& !sps_pps.is_empty()
&& !nals
.iter()
.any(|n| !n.is_empty() && (n[0] & H264_NAL_TYPE_MASK) == H264_NAL_SPS);
if !needs_params {
return length_prefixed_to_annexb(length_prefixed);
}
let mut out = Vec::with_capacity(length_prefixed.len() + total_param_len(sps_pps));
let mut inserted = false;
for nal in &nals {
let nal_type = nal.first().map(|b| b & H264_NAL_TYPE_MASK);
if !inserted && nal_type != Some(H264_NAL_AUD) {
append_param_sets(&mut out, sps_pps);
inserted = true;
}
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(nal);
}
if !inserted {
append_param_sets(&mut out, sps_pps);
}
Ok(out)
}
fn build_hevc_annexb_au(
length_prefixed: &[u8],
is_sync: bool,
vps_sps_pps: &[Vec<u8>],
) -> Result<Vec<u8>> {
let nals = iter_length_prefixed_nals(length_prefixed)?;
let needs_params = is_sync
&& !vps_sps_pps.is_empty()
&& !nals
.iter()
.any(|n| nal_unit_type(NalCodec::Hevc, n) == Some(HEVC_NAL_SPS));
if !needs_params {
return length_prefixed_to_annexb(length_prefixed);
}
let mut out = Vec::with_capacity(length_prefixed.len() + total_param_len(vps_sps_pps));
let mut inserted = false;
for nal in &nals {
let nal_type = nal_unit_type(NalCodec::Hevc, nal);
if !inserted && nal_type != Some(HEVC_NAL_AUD) {
append_param_sets(&mut out, vps_sps_pps);
inserted = true;
}
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(nal);
}
if !inserted {
append_param_sets(&mut out, vps_sps_pps);
}
Ok(out)
}
fn total_param_len(sps_pps: &[Vec<u8>]) -> usize {
sps_pps.iter().map(|p| 4 + p.len()).sum()
}
fn append_param_sets(out: &mut Vec<u8>, sps_pps: &[Vec<u8>]) {
for p in sps_pps {
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(p);
}
}
const ANNEXB_START_CODE: [u8; 4] = [0, 0, 0, 1];
fn ensure_avc_aud(mut au: Vec<u8>) -> Result<Vec<u8>> {
if first_avc_nal_type(&au) == Some(H264_NAL_AUD) {
return Ok(au);
}
let mut out = Vec::with_capacity(ANNEXB_START_CODE.len() + 2 + au.len());
out.extend_from_slice(&ANNEXB_START_CODE);
out.extend_from_slice(&[H264_NAL_AUD_BYTE, H264_AUD_PRIMARY_PIC_TYPE]);
out.append(&mut au);
Ok(out)
}
fn ensure_hevc_aud(mut au: Vec<u8>) -> Result<Vec<u8>> {
if first_hevc_nal_type(&au) == Some(HEVC_NAL_AUD) {
return Ok(au);
}
let mut out = Vec::with_capacity(ANNEXB_START_CODE.len() + 3 + au.len());
out.extend_from_slice(&ANNEXB_START_CODE);
out.extend_from_slice(&[HEVC_AUD_FIRST_BYTE, HEVC_AUD_SECOND_BYTE, HEVC_AUD_PIC_TYPE]);
out.append(&mut au);
Ok(out)
}
fn first_avc_nal_type(annexb: &[u8]) -> Option<u8> {
annexb_first_nal(annexb).and_then(|n| n.first().map(|b| b & H264_NAL_TYPE_MASK))
}
fn first_hevc_nal_type(annexb: &[u8]) -> Option<u8> {
annexb_first_nal(annexb).and_then(|n| nal_unit_type(NalCodec::Hevc, n))
}
fn annexb_first_nal(annexb: &[u8]) -> Option<&[u8]> {
let skip = if annexb.starts_with(&ANNEXB_START_CODE) {
ANNEXB_START_CODE.len()
} else if annexb.starts_with(&[0, 0, 1]) {
3
} else {
0
};
let rest = annexb.get(skip..)?;
let end = rest
.windows(3)
.position(|w| w == [0, 0, 1])
.unwrap_or(rest.len());
let end = if rest.get(end.wrapping_sub(1)) == Some(&0) && end > 0 {
end - 1
} else {
end
};
rest.get(..end)
}
fn build_pat_section(pmt_pid: Pid) -> Result<Vec<u8>> {
let mut body = Vec::new();
body.extend_from_slice(&1u16.to_be_bytes()); body.push(VERSION_CURRENT_NEXT);
body.push(0); body.push(0); body.extend_from_slice(&PROGRAM_NUMBER.to_be_bytes());
body.push(PID_RESERVED_HI | ((pmt_pid.value() >> 8) as u8 & !PID_RESERVED_HI));
body.push((pmt_pid.value() & 0xFF) as u8);
finish_section(TABLE_ID_PAT, body)
}
fn build_pmt_section(pcr_pid: Pid, plans: &[EsPlan]) -> Result<Vec<u8>> {
let mut body = Vec::new();
body.extend_from_slice(&PROGRAM_NUMBER.to_be_bytes());
body.push(VERSION_CURRENT_NEXT);
body.push(0); body.push(0); body.push(PID_RESERVED_HI | ((pcr_pid.value() >> 8) as u8 & !PID_RESERVED_HI));
body.push((pcr_pid.value() & 0xFF) as u8);
body.push(INFO_RESERVED_HI);
body.push(0);
for p in plans {
body.push(p.kind.stream_type());
body.push(PID_RESERVED_HI | ((p.pid.value() >> 8) as u8 & !PID_RESERVED_HI));
body.push((p.pid.value() & 0xFF) as u8);
let es_info_length = p.descriptors.len().min(MAX_ES_INFO_LENGTH);
body.push(INFO_RESERVED_HI | ((es_info_length >> 8) as u8 & !INFO_RESERVED_HI));
body.push((es_info_length & 0xFF) as u8);
body.extend_from_slice(&p.descriptors[..es_info_length]);
}
finish_section(TABLE_ID_PMT, body)
}
fn finish_section(table_id: u8, body: Vec<u8>) -> Result<Vec<u8>> {
let section_length = body.len() + CRC32_LEN;
if section_length > MAX_SECTION_LENGTH {
return Err(Error::BufferCapExceeded {
what: "PSI section_length",
cap: MAX_SECTION_LENGTH,
});
}
let mut section = Vec::with_capacity(3 + section_length);
section.push(table_id);
section.push(SECTION_SYNTAX_FLAGS_HI | ((section_length >> 8) as u8 & SECTION_LENGTH_HI_MASK));
section.push((section_length & 0xFF) as u8);
section.extend_from_slice(&body);
let crc = crc32_mpeg2::compute(§ion);
section.extend_from_slice(&crc.to_be_bytes());
Ok(section)
}
fn packetise_section(pid: Pid, section: &[u8], start_cc: u8) -> Vec<[u8; TS_PACKET_SIZE]> {
let mut packets = Vec::new();
let mut cc: u8 = start_cc & 0x0F;
let mut pos = 0usize;
let mut first = true;
while pos < section.len() || first {
let mut pkt = [STUFFING_BYTE; TS_PACKET_SIZE];
let hdr = TsHeader {
tei: false,
pusi: first,
pid,
scrambling: 0,
has_adaptation: false,
has_payload: true,
continuity_counter: cc,
};
hdr.serialize_into(&mut pkt[..4]).expect("4-byte TS header");
cc = (cc + 1) & 0x0F;
let mut w = 4usize;
let cap = if first {
pkt[w] = 0; w += 1;
TS_PACKET_SIZE - w
} else {
TS_PACKET_SIZE - w
};
let take = (section.len() - pos).min(cap);
pkt[w..w + take].copy_from_slice(§ion[pos..pos + take]);
pos += take;
packets.push(pkt);
first = false;
}
packets
}
#[allow(clippy::too_many_arguments)]
fn packetise_pes(
plan: &EsPlan,
es_payload: &[u8],
pts90: u64,
dts90: u64,
carry_pcr: bool,
cc: &mut u8,
sort_key: u64,
tagged: &mut Vec<TaggedPacket>,
) -> Result<()> {
let pes = build_pes_bytes(plan, es_payload, pts90, dts90)?;
let mut pos = 0usize;
let mut first = true;
while pos < pes.len() {
let mut pkt = [STUFFING_BYTE; TS_PACKET_SIZE];
let remaining = pes.len() - pos;
let want_pcr = first && carry_pcr;
let pcr_af_content = if want_pcr { 1 + PCR_FIELD_LEN } else { 0 };
let forced_header = 4 + if want_pcr { 1 + pcr_af_content } else { 0 };
let cap = TS_PACKET_SIZE - forced_header;
let is_last = remaining <= cap;
let to_copy = remaining.min(cap);
let stuff = cap - to_copy;
if want_pcr {
let af_len = pcr_af_content + stuff;
write_af_packet(
&mut pkt,
plan.pid,
first,
*cc,
af_len,
true,
Some(pcr_for(dts90)),
&pes[pos..pos + to_copy],
);
pos += to_copy;
} else if is_last && stuff > 0 {
let af_len = TS_PACKET_SIZE - 4 - 1 - to_copy;
write_af_packet(
&mut pkt,
plan.pid,
first,
*cc,
af_len,
false,
None,
&pes[pos..pos + to_copy],
);
pos += to_copy;
} else {
let hdr = TsHeader {
tei: false,
pusi: first,
pid: plan.pid,
scrambling: 0,
has_adaptation: false,
has_payload: true,
continuity_counter: *cc,
};
hdr.serialize_into(&mut pkt[..4]).expect("4-byte TS header");
pkt[4..4 + to_copy].copy_from_slice(&pes[pos..pos + to_copy]);
pos += to_copy;
}
*cc = (*cc + 1) & 0x0F;
tagged.push(TaggedPacket {
sort_key,
packet: pkt,
});
first = false;
}
Ok(())
}
fn pcr_for(dts90: u64) -> Pcr {
let base = dts90.saturating_sub(PCR_LEAD_TICKS);
Pcr::from_27mhz(broadcast_common::clock33::pcr_from_parts(base, 0))
}
#[allow(clippy::too_many_arguments)]
fn write_af_packet(
pkt: &mut [u8; TS_PACKET_SIZE],
pid: Pid,
pusi: bool,
cc: u8,
af_len: usize,
has_pcr: bool,
pcr: Option<Pcr>,
payload: &[u8],
) {
let hdr = TsHeader {
tei: false,
pusi,
pid,
scrambling: 0,
has_adaptation: true,
has_payload: true,
continuity_counter: cc,
};
hdr.serialize_into(&mut pkt[..4]).expect("4-byte TS header");
debug_assert_eq!(pkt[3] & (AF_CTRL_ADAPTATION | AF_CTRL_PAYLOAD), 0x30);
pkt[4] = af_len as u8;
if af_len >= 1 {
pkt[5] = if has_pcr { AF_PCR_FLAG } else { 0 };
if has_pcr && let Some(p) = pcr {
pkt[6..6 + PCR_FIELD_LEN].copy_from_slice(&p.to_field_bytes());
}
}
let payload_start = 5 + af_len;
pkt[payload_start..payload_start + payload.len()].copy_from_slice(payload);
}
fn build_pes_bytes(plan: &EsPlan, es_payload: &[u8], pts90: u64, dts90: u64) -> Result<Vec<u8>> {
let include_dts = dts90 != pts90;
let opt_content = if include_dts { 10 } else { 5 };
let after_len = HEADER_FIXED + opt_content + es_payload.len();
let pes_packet_length = if plan.kind.is_video() {
0u16
} else {
u16::try_from(after_len).map_err(|_| Error::BufferCapExceeded {
what: "PES_packet_length",
cap: u16::MAX as usize,
})?
};
let mut out = Vec::with_capacity(MIN_LEN + HEADER_FIXED + opt_content + es_payload.len());
out.extend_from_slice(&PES_START_CODE);
out.push(plan.stream_id.0);
out.extend_from_slice(&pes_packet_length.to_be_bytes());
out.push(PES_OPTIONAL_MARKER); out.push(if include_dts {
PTS_DTS_FLAGS_BOTH
} else {
PTS_DTS_FLAGS_PTS_ONLY
});
out.push(opt_content as u8); if include_dts {
out.extend_from_slice(&encode_timestamp(pts90, TS_PREFIX_PTS_WITH_DTS));
out.extend_from_slice(&encode_timestamp(dts90, TS_PREFIX_DTS));
} else {
out.extend_from_slice(&PesPts(pts90).to_field_bytes());
}
out.extend_from_slice(es_payload);
Ok(out)
}
fn encode_timestamp(ts: u64, prefix: u8) -> [u8; 5] {
broadcast_common::clock33::pack_marked(ts, prefix)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn es_kind_stream_types_mirror_demux() {
assert_eq!(EsKind::Avc.stream_type(), 0x1B);
assert_eq!(EsKind::Hevc.stream_type(), 0x24);
assert_eq!(EsKind::Mpeg2Video.stream_type(), 0x02);
assert_eq!(EsKind::Aac.stream_type(), 0x0F);
assert_eq!(
EsKind::MpegAudio { is_mpeg2: false }.stream_type(),
0x03,
"MPEG-1 audio"
);
assert_eq!(
EsKind::MpegAudio { is_mpeg2: true }.stream_type(),
0x04,
"MPEG-2 audio"
);
assert_eq!(EsKind::Ac3.stream_type(), 0x81);
assert_eq!(EsKind::Eac3.stream_type(), 0x87);
assert_eq!(EsKind::Dts.stream_type(), 0x82);
assert_eq!(EsKind::MpegH.stream_type(), 0x2D);
assert!(EsKind::Avc.is_video());
assert!(EsKind::Hevc.is_video());
assert!(EsKind::Mpeg2Video.is_video());
assert!(!EsKind::Aac.is_video());
assert!(!EsKind::MpegAudio { is_mpeg2: false }.is_video());
assert_eq!(
EsKind::Data {
stream_type: 0x06,
carriage: DataCarriage::Pes,
}
.stream_type(),
0x06
);
assert_eq!(
EsKind::Data {
stream_type: 0x86,
carriage: DataCarriage::Sections,
}
.stream_type(),
0x86
);
assert!(
EsKind::Data {
stream_type: 0x86,
carriage: DataCarriage::Sections,
}
.is_section_carried()
);
assert!(
!EsKind::Data {
stream_type: 0x06,
carriage: DataCarriage::Pes,
}
.is_section_carried()
);
}
#[test]
fn pat_section_crc_is_valid() {
let pat = build_pat_section(PMT_PID).unwrap();
assert_eq!(crc32_mpeg2::compute(&pat), 0);
assert_eq!(pat[0], TABLE_ID_PAT);
}
#[test]
fn pmt_section_crc_is_valid() {
let plans = alloc::vec![EsPlan {
pid: ES_PID_BASE,
stream_id: StreamId(STREAM_ID_VIDEO_BASE),
kind: EsKind::Avc,
asc: None,
avc_sps_pps: Vec::new(),
hevc_vps_sps_pps: Vec::new(),
descriptors: Vec::new(),
}];
let pmt = build_pmt_section(ES_PID_BASE, &plans).unwrap();
assert_eq!(crc32_mpeg2::compute(&pmt), 0);
assert_eq!(pmt[0], TABLE_ID_PMT);
}
#[test]
fn pmt_section_carries_es_info_descriptors() {
let descriptors = alloc::vec![0x59, 0x02, 0xAA, 0xBB]; let plans = alloc::vec![EsPlan {
pid: ES_PID_BASE,
stream_id: StreamId(0),
kind: EsKind::Data {
stream_type: 0x06,
carriage: DataCarriage::Pes,
},
asc: None,
avc_sps_pps: Vec::new(),
hevc_vps_sps_pps: Vec::new(),
descriptors: descriptors.clone(),
}];
let pmt = build_pmt_section(ES_PID_BASE, &plans).unwrap();
assert_eq!(crc32_mpeg2::compute(&pmt), 0);
let es_loop_start = 8 + 4;
assert_eq!(pmt[es_loop_start], 0x06, "stream_type");
let es_info_length =
(((pmt[es_loop_start + 3] & 0x0F) as usize) << 8) | pmt[es_loop_start + 4] as usize;
assert_eq!(es_info_length, descriptors.len());
let desc_start = es_loop_start + 5;
assert_eq!(
&pmt[desc_start..desc_start + es_info_length],
&descriptors[..]
);
}
#[test]
fn section_packets_are_whole_and_pusi() {
let pat = build_pat_section(PMT_PID).unwrap();
let pkts = packetise_section(PAT_PID, &pat, 0);
assert_eq!(pkts.len(), 1);
assert_eq!(pkts[0][0], 0x47);
assert_ne!(pkts[0][1] & 0x40, 0, "PUSI must be set on the first packet");
}
const MPEGH_EXT_TAG_FOR_TEST: u8 = 0x08;
#[test]
fn mpegh_3daudio_descriptor_matches_real_fixture_bytes() {
let bytes = mpegh_3daudio_descriptor(0x10, 0x01).expect("encodes");
assert_eq!(bytes, alloc::vec![0x3F, 0x04, 0x08, 0x10, 0x7F, 0xC1]);
}
#[test]
fn mpegh_3daudio_descriptor_fields_each_change_the_bytes() {
let bytes = mpegh_3daudio_descriptor(0x0B, 0x06).expect("encodes");
assert_eq!(bytes, alloc::vec![0x3F, 0x04, 0x08, 0x0B, 0x7F, 0xC6]);
assert_ne!(bytes, mpegh_3daudio_descriptor(0x10, 0x06).unwrap());
assert_ne!(bytes, mpegh_3daudio_descriptor(0x0B, 0x01).unwrap());
}
#[test]
fn mpegh_3daudio_descriptor_layout_wider_than_six_bits_falls_back_to_zero() {
assert_eq!(
mpegh_3daudio_descriptor(0x10, 0x3F).unwrap(),
alloc::vec![0x3F, 0x04, 0x08, 0x10, 0x7F, 0xFF]
);
for wide in [0x40u8, 0x81, 0xFF] {
assert_eq!(
mpegh_3daudio_descriptor(0x10, wide).unwrap(),
alloc::vec![0x3F, 0x04, 0x08, 0x10, 0x7F, 0xC0],
"layout {wide:#04x}"
);
}
}
#[test]
fn merge_es_info_descriptors_denies_ca_and_preserves_order() {
let lang = alloc::vec![
DESCRIPTOR_TAG_ISO_639_LANGUAGE_FOR_TEST,
0x04,
b'e',
b'n',
b'g',
0x00
];
let ca = alloc::vec![DESCRIPTOR_TAG_CA, 0x04, 0x00, 0x01, 0x00, 0x82];
let private = alloc::vec![0x88u8, 0x02, 0xAA, 0xBB];
let mut inherited = lang.clone();
inherited.extend_from_slice(&ca);
inherited.extend_from_slice(&private);
let merged = merge_es_info_descriptors(&inherited, &[]).expect("no overflow");
let mut expected = lang;
expected.extend_from_slice(&private);
assert_eq!(
merged, expected,
"CA_descriptor must be dropped; siblings survive in source order"
);
}
const DESCRIPTOR_TAG_ISO_639_LANGUAGE_FOR_TEST: u8 = 0x0A;
#[test]
fn merge_es_info_descriptors_dedups_synthesized_tag_favouring_synthesized_bytes() {
let before = alloc::vec![0x88u8, 0x01, 0x01];
let stale_mpegh = alloc::vec![
DESCRIPTOR_TAG_EXTENSION,
0x02, MPEGH_EXT_TAG_FOR_TEST,
0xFF, ];
let after = alloc::vec![0x89u8, 0x01, 0x02];
let mut inherited = before.clone();
inherited.extend_from_slice(&stale_mpegh);
inherited.extend_from_slice(&after);
let synthesized = alloc::vec![mpegh_3daudio_descriptor(0x0B, 0x01).unwrap()];
let merged = merge_es_info_descriptors(&inherited, &synthesized).expect("no overflow");
let mut expected = before;
expected.extend_from_slice(&after);
expected.extend_from_slice(&mpegh_3daudio_descriptor(0x0B, 0x01).unwrap());
assert_eq!(
merged, expected,
"inherited duplicate of a synthesized tag must be dropped, keeping only the \
synthesized copy, appended after the surviving unrelated siblings"
);
assert_eq!(
merged
.iter()
.enumerate()
.filter(|&(i, &b)| b == DESCRIPTOR_TAG_EXTENSION && i + 1 < merged.len())
.count(),
1,
"extension_descriptor tag (0x3F) must appear exactly once in the merged loop"
);
}
#[test]
fn merge_es_info_descriptors_keeps_distinct_extension_descriptor_tag_siblings() {
const OTHER_EXTENSION_TAG_FOR_TEST: u8 = 0x15;
let other_extension = alloc::vec![
DESCRIPTOR_TAG_EXTENSION,
0x02, OTHER_EXTENSION_TAG_FOR_TEST,
0xAB, ];
let inherited = other_extension.clone();
let synthesized = alloc::vec![mpegh_3daudio_descriptor(0x0B, 0x01).unwrap()];
let merged = merge_es_info_descriptors(&inherited, &synthesized).expect("no overflow");
let mut expected = other_extension;
expected.extend_from_slice(&mpegh_3daudio_descriptor(0x0B, 0x01).unwrap());
assert_eq!(
merged, expected,
"two extension_descriptor entries with distinct extension_descriptor_tag values \
are not duplicates and must both survive"
);
}
#[test]
fn merge_es_info_descriptors_overflow_returns_typed_error_not_truncated() {
const BODY_LEN: u8 = 0xFD; const N: usize = 17;
let mut inherited = Vec::new();
for _ in 0..N {
inherited.push(0x80u8); inherited.push(BODY_LEN);
inherited.extend(core::iter::repeat_n(0xAAu8, BODY_LEN as usize));
}
assert!(
inherited.len() > MAX_ES_INFO_LENGTH,
"test setup must actually overflow"
);
let result = merge_es_info_descriptors(&inherited, &[]);
match result {
Err(Error::BufferCapExceeded { what, cap }) => {
assert_eq!(what, "PMT ES_info descriptor loop");
assert_eq!(cap, MAX_ES_INFO_LENGTH);
}
other => panic!(
"expected Err(Error::BufferCapExceeded {{ .. }}) for an over-cap ES_info loop, \
got {other:?} — a truncated loop would be a malformed PMT"
),
}
}
#[test]
fn pmt_section_length_past_1021_errors() {
let plans = alloc::vec![EsPlan {
pid: ES_PID_BASE,
stream_id: StreamId(0),
kind: EsKind::Data {
stream_type: 0x06,
carriage: DataCarriage::Pes,
},
asc: None,
avc_sps_pps: Vec::new(),
hevc_vps_sps_pps: Vec::new(),
descriptors: alloc::vec![0u8; 1004],
}];
let err = build_pmt_section(ES_PID_BASE, &plans).unwrap_err();
assert!(
matches!(
err,
Error::BufferCapExceeded {
what: "PSI section_length",
cap: MAX_SECTION_LENGTH,
}
),
"expected BufferCapExceeded for PSI section_length, got {err:?}"
);
}
#[test]
fn pmt_section_length_at_1021_succeeds() {
let plans = alloc::vec![EsPlan {
pid: ES_PID_BASE,
stream_id: StreamId(0),
kind: EsKind::Data {
stream_type: 0x06,
carriage: DataCarriage::Pes,
},
asc: None,
avc_sps_pps: Vec::new(),
hevc_vps_sps_pps: Vec::new(),
descriptors: alloc::vec![0u8; 1003],
}];
let pmt = build_pmt_section(ES_PID_BASE, &plans).unwrap();
let section_length = (((pmt[1] & SECTION_LENGTH_HI_MASK) as usize) << 8) | pmt[2] as usize;
assert_eq!(section_length, MAX_SECTION_LENGTH);
assert_eq!(crc32_mpeg2::compute(&pmt), 0);
}
#[test]
fn audio_pes_packet_length_past_65535_errors() {
let plan = EsPlan {
pid: ES_PID_BASE,
stream_id: StreamId(STREAM_ID_AUDIO_BASE),
kind: EsKind::Ac3,
asc: None,
avc_sps_pps: Vec::new(),
hevc_vps_sps_pps: Vec::new(),
descriptors: Vec::new(),
};
let fits = alloc::vec![0u8; u16::MAX as usize - HEADER_FIXED - 5];
let ok = build_pes_bytes(&plan, &fits, 0, 0).expect("the maximum legal PES must build");
assert_eq!(
u16::from_be_bytes([ok[4], ok[5]]),
u16::MAX,
"PES_packet_length = 65535 at the boundary"
);
let over = alloc::vec![0u8; fits.len() + 1];
let err = build_pes_bytes(&plan, &over, 0, 0).unwrap_err();
assert!(
matches!(
err,
Error::BufferCapExceeded {
what: "PES_packet_length",
cap: 65535,
}
),
"one byte past the cap must be an error, got {err:?}"
);
}
#[test]
fn video_pes_packet_length_above_65535_is_unbounded() {
let plan = EsPlan {
pid: ES_PID_BASE,
stream_id: StreamId(STREAM_ID_VIDEO_BASE),
kind: EsKind::Avc,
asc: None,
avc_sps_pps: Vec::new(),
hevc_vps_sps_pps: Vec::new(),
descriptors: Vec::new(),
};
let big = alloc::vec![0u8; u16::MAX as usize + 1];
let pes = build_pes_bytes(&plan, &big, 0, 0).expect("video may be unbounded");
assert_eq!(
u16::from_be_bytes([pes[4], pes[5]]),
0,
"video PES_packet_length is the unbounded 0 form"
);
}
}