Skip to main content

mpeg_ts/
mux.rs

1//! Section → TS packetiser (the byte-exact inverse of
2//! [`SectionReassembler::feed`](crate::ts::SectionReassembler::feed)).
3//!
4//! Per the PSI carriage rules of ITU-T H.222.0 §2.4.4 (= ISO/IEC 13818-1:2007 §2.4.4)
5//! (`mpeg-ts/docs/tables/h222_0/psi_section.md`): sections are packed into 188-byte packets
6//! with a `pointer_field` where sections begin, concatenated contiguously, and
7//! 0xFF-stuffed at the batch tail.
8
9use alloc::vec::Vec;
10use core::time::Duration;
11
12use crate::pid::{Pid, well_known};
13use crate::ts::{CC_MASK, TS_PACKET_SIZE, TsHeader};
14
15/// Maximum data bytes in a PUSI=1 packet (188 − 4 header − 1 pointer_field). §2.4.4.
16const PUSI_PAYLOAD_CAP: usize = 183;
17/// Maximum data bytes in a continuation packet (188 − 4 header). §2.4.4.
18const PAYLOAD_CAP: usize = 184;
19use crate::ts::STUFFING_BYTE;
20
21/// Packetises PSI/SI sections into 188-byte TS packets.
22///
23/// This is the byte-exact inverse of
24/// [`SectionReassembler::feed`](crate::ts::SectionReassembler::feed): packets
25/// produced here, when fed back through the reassembler, yield the same
26/// sections in order.
27///
28/// ISO/IEC 13818-1:2007 §2.4.4 (`mpeg-ts/docs/tables/h222_0/psi_section.md`).
29pub struct SectionPacketiser {
30    pid: Pid,
31    continuity_counter: u8,
32}
33
34impl SectionPacketiser {
35    /// Start a packetiser for `pid` with continuity_counter = 0.
36    pub fn new(pid: Pid) -> Self {
37        Self {
38            pid,
39            continuity_counter: 0,
40        }
41    }
42
43    /// Start at a specific continuity_counter (0..=15) — for resuming a stream.
44    pub fn with_continuity(pid: Pid, cc: u8) -> Self {
45        Self {
46            pid,
47            continuity_counter: cc & CC_MASK,
48        }
49    }
50
51    /// The PID this packetiser emits packets for.
52    pub fn pid(&self) -> Pid {
53        self.pid
54    }
55
56    /// The continuity_counter for the next emitted packet.
57    pub fn continuity_counter(&self) -> u8 {
58        self.continuity_counter
59    }
60
61    /// Packetise a batch of complete sections into 188-byte TS packets,
62    /// appended to `out` (cleared first).
63    ///
64    /// Returns the number of packets appended.
65    pub fn packetise_into(
66        &mut self,
67        sections: &[&[u8]],
68        out: &mut Vec<[u8; TS_PACKET_SIZE]>,
69    ) -> usize {
70        out.clear();
71
72        if sections.is_empty() {
73            return 0;
74        }
75
76        // Concatenate all sections and record section-start byte offsets.
77        let total_len: usize = sections.iter().map(|s| s.len()).sum();
78        if total_len == 0 {
79            return 0;
80        }
81        let mut data = Vec::with_capacity(total_len);
82        let mut starts = Vec::with_capacity(sections.len());
83        for s in sections {
84            starts.push(data.len());
85            data.extend_from_slice(s);
86        }
87
88        let count_before = out.len();
89        let mut pos = 0usize;
90        // `starts` is ascending and `pos` only advances, so a forward-only
91        // cursor finds the smallest section-start offset >= pos in amortised
92        // O(1) per packet instead of rescanning from the front (audit
93        // r01-O1: O(packets x sections)).
94        let mut next_idx = 0usize;
95
96        while pos < data.len() {
97            while starts.get(next_idx).is_some_and(|&s| s < pos) {
98                next_idx += 1;
99            }
100            let next_start = starts.get(next_idx).copied();
101
102            let pusi: bool;
103            let pointer_field: u8;
104            let cap: usize;
105
106            if let Some(ns) = next_start {
107                let diff = ns.saturating_sub(pos);
108                // W1 (#1074): `diff == PUSI_PAYLOAD_CAP` (183) must NOT take the
109                // PUSI branch — a pointer_field of 183 would claim a section
110                // starts after all 183 payload bytes, i.e. not within this
111                // packet at all (H.222.0 §2.4.4 requires PUSI=1 to mean a
112                // section genuinely starts here). The non-PUSI branch's cap
113                // must in turn be bounded by `diff`, not just the fixed
114                // 184-byte payload size, or a continuation packet at this same
115                // boundary copies one byte past the next section's start —
116                // the reassembler then silently drops it (it only takes what
117                // completes the in-progress section), corrupting the next one.
118                if diff < PUSI_PAYLOAD_CAP {
119                    pusi = true;
120                    pointer_field = diff as u8;
121                    cap = PUSI_PAYLOAD_CAP;
122                } else {
123                    pusi = false;
124                    pointer_field = 0;
125                    cap = PAYLOAD_CAP.min(diff);
126                }
127            } else {
128                pusi = false;
129                pointer_field = 0;
130                cap = PAYLOAD_CAP;
131            }
132
133            let mut pkt = [0u8; TS_PACKET_SIZE];
134
135            let header = TsHeader {
136                tei: false,
137                pusi,
138                pid: self.pid,
139                scrambling: 0,
140                has_adaptation: false,
141                has_payload: true,
142                continuity_counter: self.continuity_counter,
143            };
144            header
145                .serialize_into(&mut pkt[..4])
146                .expect("4-byte header buffer");
147
148            self.continuity_counter = (self.continuity_counter + 1) & CC_MASK;
149
150            let mut write_pos = 4usize;
151
152            if pusi {
153                pkt[write_pos] = pointer_field;
154                write_pos += 1;
155            }
156
157            let remaining = data.len() - pos;
158            let to_copy = remaining.min(cap);
159            pkt[write_pos..write_pos + to_copy].copy_from_slice(&data[pos..pos + to_copy]);
160            pos += to_copy;
161            write_pos += to_copy;
162
163            // 0xFF-stuff remaining payload bytes.
164            for b in &mut pkt[write_pos..] {
165                *b = STUFFING_BYTE;
166            }
167
168            out.push(pkt);
169        }
170
171        out.len() - count_before
172    }
173
174    /// Allocating convenience wrapper over [`packetise_into`](Self::packetise_into).
175    pub fn packetise(&mut self, sections: &[&[u8]]) -> Vec<[u8; TS_PACKET_SIZE]> {
176        let mut out = Vec::new();
177        self.packetise_into(sections, &mut out);
178        out
179    }
180}
181
182// ── Default interval constants ────────────────────────────────────────────────
183
184/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — NIT maximum interval.
185pub const NIT_MAX_INTERVAL: Duration = Duration::from_secs(10);
186/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — BAT maximum interval.
187pub const BAT_MAX_INTERVAL: Duration = Duration::from_secs(10);
188/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — SDT actual maximum interval.
189pub const SDT_ACTUAL_MAX_INTERVAL: Duration = Duration::from_secs(2);
190/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — SDT other maximum interval.
191pub const SDT_OTHER_MAX_INTERVAL: Duration = Duration::from_secs(10);
192/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — EIT p/f actual maximum interval.
193pub const EIT_PF_ACTUAL_MAX_INTERVAL: Duration = Duration::from_secs(2);
194/// dvb-si/docs/tr_101_211.md §4.4.1 — EIT p/f other maximum interval (sat/cable;
195/// terrestrial is 20 s per §4.4.2).
196pub const EIT_PF_OTHER_MAX_INTERVAL: Duration = Duration::from_secs(10);
197/// dvb-si/docs/tr_101_211.md §4.4.1 — EIT schedule first 8 days maximum interval.
198pub const EIT_SCHED_MAX_INTERVAL: Duration = Duration::from_secs(10);
199/// dvb-si/docs/tr_101_211.md §4.4.1 — EIT schedule beyond 8 days maximum interval.
200pub const EIT_SCHED_EXT_MAX_INTERVAL: Duration = Duration::from_secs(30);
201/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — TDT maximum interval.
202pub const TDT_MAX_INTERVAL: Duration = Duration::from_secs(30);
203/// dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2 — TOT maximum interval.
204pub const TOT_MAX_INTERVAL: Duration = Duration::from_secs(30);
205
206/// dvb-si/docs/ts_101_154_av_coding.md §4.1.7 — PAT maximum interval (shall ≤ 100 ms).
207pub const PAT_MAX_INTERVAL: Duration = Duration::from_millis(100);
208/// dvb-si/docs/ts_101_154_av_coding.md §4.1.7 — PMT maximum interval (shall ≤ 100 ms).
209pub const PMT_MAX_INTERVAL: Duration = Duration::from_millis(100);
210
211/// dvb-si/docs/tables/en_300_468/514-repetition-rates-and-random-access-hand-transcribed.md
212/// §5.1.4.1 — minimum inter-section interval floor (≤100 Mbit/s TSs).
213pub const MIN_SECTION_INTERVAL: Duration = Duration::from_millis(25);
214
215// ── SiMux ────────────────────────────────────────────────────────────────────
216
217/// Section-repetition scheduler that builds TS packets on a caller-supplied clock.
218///
219/// Each entry is a PID + table_id + concatenated complete-section bytes + an
220/// emission interval. Call [`poll_into`](Self::poll_into) with
221/// monotonically-increasing `now` values to get 188-byte TS packets for every
222/// entry whose interval has elapsed since its last emission.
223///
224/// The scheduler owns its section bytes — call [`upsert`](Self::upsert) to
225/// update them when SI changes. Entries are keyed by `(pid, table_id)`, not
226/// PID alone: several distinct tables share a PID by spec (TDT/TOT both on
227/// `0x0014`, SDT/BAT both on `0x0011`, EIT present/following vs. schedule
228/// sub-tables on `0x0012`), and `table_id` — the first byte of `sections` —
229/// is exactly the field the spec uses to tell them apart, so `upsert_tot`
230/// cannot silently replace a previously-registered `upsert_tdt` entry.
231///
232/// All entries on the same PID share one [`SectionPacketiser`] (and so one
233/// continuity counter), created lazily the first time that PID emits — this
234/// is what keeps the continuity counter continuous per PID across table
235/// kinds, not just within one kind.
236///
237/// # 25 ms floor
238///
239/// [`MIN_SECTION_INTERVAL`] is the minimum valid interval (EN 300 468 §5.1.4.1).
240/// Supplying an interval below this in [`upsert`](Self::upsert) triggers a
241/// `debug_assert` — in release builds the assertion compiles out; the caller
242/// must ensure compliance.
243pub struct SiMux {
244    entries: Vec<Entry>,
245    /// One packetiser per distinct PID, shared across every table_id on it.
246    packetisers: Vec<SectionPacketiser>,
247}
248
249struct Entry {
250    pid: Pid,
251    table_id: u8,
252    sections: Vec<u8>,
253    interval: Duration,
254    last_emit: Option<Duration>,
255}
256
257/// The dedupe/lookup key for an entry: the first byte of `sections` is
258/// `table_id` for every PSI/SI section (ISO/IEC 13818-1 §2.4.4.4 / EN 300 468
259/// §5.2). An empty `sections` (no section registered yet) has no real
260/// table_id; `0` is used as a sentinel and can only collide with another
261/// empty registration on the same PID, which is a caller no-op either way.
262fn table_id_of(sections: &[u8]) -> u8 {
263    sections.first().copied().unwrap_or(0)
264}
265
266/// Find the packetiser for `pid`, creating one (fresh continuity counter) if
267/// this is the first entry ever registered on that PID. A free function, not
268/// a method, so callers can hold `&mut entries` and this borrow concurrently.
269fn packetiser_for(packetisers: &mut Vec<SectionPacketiser>, pid: Pid) -> &mut SectionPacketiser {
270    if let Some(idx) = packetisers.iter().position(|p| p.pid() == pid) {
271        &mut packetisers[idx]
272    } else {
273        packetisers.push(SectionPacketiser::new(pid));
274        packetisers.last_mut().expect("just pushed")
275    }
276}
277
278impl SiMux {
279    /// Create an empty scheduler (no entries, no pending emissions).
280    pub fn new() -> Self {
281        Self {
282            entries: Vec::new(),
283            packetisers: Vec::new(),
284        }
285    }
286
287    /// Register or replace the sections emitted on `pid` at `interval`.
288    ///
289    /// `sections` is the concatenated complete-section bytes for one emission
290    /// cycle; its first byte (`table_id`) plus `pid` is the dedupe key, so
291    /// re-calling for the same `(pid, table_id)` updates the bytes and
292    /// interval of that one table while preserving the PID's shared
293    /// continuity counter — a *different* table_id on the same `pid` (e.g.
294    /// TOT after TDT, both on `0x0014`) becomes a second, independent entry.
295    ///
296    /// # Panics (debug only)
297    ///
298    /// `debug_assert!(interval >= MIN_SECTION_INTERVAL)` — EN 300 468 §5.1.4.1
299    /// requires a minimum inter-section interval of 25 ms.
300    pub fn upsert(&mut self, pid: Pid, sections: Vec<u8>, interval: Duration) {
301        debug_assert!(
302            interval >= MIN_SECTION_INTERVAL,
303            "interval {interval:?} is below the 25 ms minimum (EN 300 468 §5.1.4.1)"
304        );
305
306        let table_id = table_id_of(&sections);
307        if let Some(entry) = self
308            .entries
309            .iter_mut()
310            .find(|e| e.pid == pid && e.table_id == table_id)
311        {
312            entry.sections = sections;
313            entry.interval = interval;
314        } else {
315            self.entries.push(Entry {
316                pid,
317                table_id,
318                sections,
319                interval,
320                last_emit: None,
321            });
322        }
323    }
324
325    /// PAT at [`PAT_MAX_INTERVAL`] on PID 0x0000.
326    ///
327    /// Interval cites dvb-si/docs/ts_101_154_av_coding.md §4.1.7.
328    pub fn upsert_pat(&mut self, sections: Vec<u8>) {
329        self.upsert(well_known::PAT, sections, PAT_MAX_INTERVAL);
330    }
331
332    /// PMT at [`PMT_MAX_INTERVAL`] on the caller-supplied `pid`.
333    ///
334    /// Interval cites dvb-si/docs/ts_101_154_av_coding.md §4.1.7.
335    pub fn upsert_pmt(&mut self, pid: Pid, sections: Vec<u8>) {
336        self.upsert(pid, sections, PMT_MAX_INTERVAL);
337    }
338
339    /// SDT actual at [`SDT_ACTUAL_MAX_INTERVAL`] on PID 0x0011.
340    ///
341    /// Interval cites dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2.
342    pub fn upsert_sdt_actual(&mut self, sections: Vec<u8>) {
343        self.upsert(well_known::SDT_BAT, sections, SDT_ACTUAL_MAX_INTERVAL);
344    }
345
346    /// NIT at [`NIT_MAX_INTERVAL`] on PID 0x0010.
347    ///
348    /// Interval cites dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2.
349    pub fn upsert_nit(&mut self, sections: Vec<u8>) {
350        self.upsert(well_known::NIT, sections, NIT_MAX_INTERVAL);
351    }
352
353    /// TDT at [`TDT_MAX_INTERVAL`] on PID 0x0014.
354    ///
355    /// Interval cites dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2.
356    pub fn upsert_tdt(&mut self, sections: Vec<u8>) {
357        self.upsert(well_known::TDT_TOT, sections, TDT_MAX_INTERVAL);
358    }
359
360    /// TOT at [`TOT_MAX_INTERVAL`] on PID 0x0014.
361    ///
362    /// Interval cites dvb-si/docs/tr_101_211.md §4.4.1/§4.4.2.
363    pub fn upsert_tot(&mut self, sections: Vec<u8>) {
364        self.upsert(well_known::TDT_TOT, sections, TOT_MAX_INTERVAL);
365    }
366
367    /// Emit every entry due at `now` (i.e. `now - last_emit >= interval`, and
368    /// first call always due), packetising via [`SectionPacketiser`], appended
369    /// to `out` (cleared first).
370    ///
371    /// Updates each emitted entry's `last_emit = now`. Deterministic given the
372    /// fed `now` sequence. Returns the packet count appended.
373    pub fn poll_into(&mut self, now: Duration, out: &mut Vec<[u8; TS_PACKET_SIZE]>) -> usize {
374        out.clear();
375        let before = out.len();
376
377        let mut tmp = Vec::new();
378        let packetisers = &mut self.packetisers;
379        for entry in &mut self.entries {
380            let due = match entry.last_emit {
381                None => true,
382                Some(last) => now.saturating_sub(last) >= entry.interval,
383            };
384            if due {
385                let refs = split_sections(&entry.sections);
386                if !refs.is_empty() {
387                    let packetiser = packetiser_for(packetisers, entry.pid);
388                    packetiser.packetise_into(&refs, &mut tmp);
389                    out.append(&mut tmp);
390                }
391                entry.last_emit = Some(now);
392            }
393        }
394
395        out.len() - before
396    }
397
398    /// Allocating convenience wrapper over [`poll_into`](Self::poll_into).
399    pub fn poll(&mut self, now: Duration) -> Vec<[u8; TS_PACKET_SIZE]> {
400        let mut out = Vec::new();
401        self.poll_into(now, &mut out);
402        out
403    }
404}
405
406impl Default for SiMux {
407    fn default() -> Self {
408        Self::new()
409    }
410}
411
412/// Split concatenated complete-section bytes into individual `&[u8]` slices.
413///
414/// Walks the PSI/SI section headers: byte 0 = table_id, bytes 1-2 =
415/// section_length (12 bits). Each slice is 3 + section_length bytes long.
416/// Trailing bytes that don't form a complete section header are discarded.
417fn split_sections(data: &[u8]) -> Vec<&[u8]> {
418    let mut result = Vec::new();
419    let mut pos = 0;
420    while let Some(end) = crate::ts::section_total_len(&data[pos..]).map(|len| pos + len) {
421        if end > data.len() {
422            break;
423        }
424        result.push(&data[pos..end]);
425        pos = end;
426    }
427    result
428}
429
430#[cfg(test)]
431mod tests {
432    use super::*;
433    use crate::ts::{SectionReassembler, TsPacket};
434    use alloc::vec;
435    use alloc::vec::Vec;
436
437    // ── helpers ──────────────────────────────────────────────────────────────
438
439    /// Build a long-form section with the given table_id and body bytes.
440    /// Returns the full section including its 3-byte header (no CRC — the
441    /// reassembler does not validate CRC).
442    fn build_section(table_id: u8, body_after_length: &[u8]) -> Vec<u8> {
443        let section_length = body_after_length.len() as u16;
444        let mut v = Vec::with_capacity(3 + section_length as usize);
445        v.push(table_id);
446        // SSI=1, PI=0, reserved=11, length upper 4 bits
447        v.push(0xB0 | ((section_length >> 8) as u8 & 0x0F));
448        v.push((section_length & 0xFF) as u8);
449        v.extend_from_slice(body_after_length);
450        v
451    }
452
453    fn concat_sections(sections: &[Vec<u8>]) -> Vec<u8> {
454        let total: usize = sections.iter().map(|s| s.len()).sum();
455        let mut out = Vec::with_capacity(total);
456        for s in sections {
457            out.extend_from_slice(s);
458        }
459        out
460    }
461
462    /// Round-trip `sections` through packetise → reassembler, asserting
463    /// byte-identical output in order and no leftovers.
464    fn assert_round_trip(sections: &[Vec<u8>]) {
465        let mut packetiser = SectionPacketiser::new(Pid::new(0x0100));
466        let refs: Vec<&[u8]> = sections.iter().map(|s| s.as_slice()).collect();
467        let packets = packetiser.packetise(&refs);
468
469        let mut reasm = SectionReassembler::default();
470        for pkt_raw in &packets {
471            let pkt = TsPacket::parse(pkt_raw).expect("parse generated packet");
472            let payload = pkt.payload.expect("payload present");
473            let pusi = pkt.header.pusi;
474            reasm.feed(payload, pusi);
475        }
476
477        let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
478        assert_eq!(
479            got.len(),
480            sections.len(),
481            "section count mismatch: expected {}, got {}",
482            sections.len(),
483            got.len()
484        );
485        for (i, (orig, round)) in sections.iter().zip(got.iter()).enumerate() {
486            assert_eq!(
487                round.as_ref(),
488                orig.as_slice(),
489                "section {i} round-trip mismatch"
490            );
491        }
492        assert!(reasm.is_empty(), "reassembler should be empty after drain");
493    }
494
495    /// 400 sections whose sizes sweep 0..=399 body bytes (so section starts
496    /// land on every offset around the 183/184-byte packet boundaries)
497    /// round-trip byte-identically through packetise -> reassemble (audit
498    /// r01-O1: the section-start lookup is now a forward cursor; this is the
499    /// independent oracle that the rewrite chose the same boundaries).
500    #[test]
501    fn many_sections_with_boundary_straddling_sizes_round_trip() {
502        let sections: Vec<Vec<u8>> = (0..400usize)
503            .map(|n| {
504                let body: Vec<u8> = (0..n).map(|i| (i % 251) as u8).collect();
505                build_section((n % 200) as u8, &body)
506            })
507            .collect();
508        assert_round_trip(&sections);
509    }
510
511    // ── split_sections ──────────────────────────────────────────────────────
512
513    #[test]
514    fn split_single_section() {
515        let s = build_section(0x42, &[0xAA; 5]);
516        let refs = split_sections(&s);
517        assert_eq!(refs.len(), 1);
518        assert_eq!(refs[0], s.as_slice());
519    }
520
521    #[test]
522    fn split_two_sections() {
523        let s1 = build_section(0x42, &[0x01, 0x02]);
524        let s2 = build_section(0x46, &[0x03, 0x04, 0x05]);
525        let both = concat_sections(&[s1.clone(), s2.clone()]);
526        let refs = split_sections(&both);
527        assert_eq!(refs.len(), 2);
528        assert_eq!(refs[0], s1.as_slice());
529        assert_eq!(refs[1], s2.as_slice());
530    }
531
532    #[test]
533    fn split_empty_input() {
534        let refs = split_sections(&[]);
535        assert!(refs.is_empty());
536    }
537
538    #[test]
539    fn split_trailing_garbage_ignored() {
540        let s = build_section(0x42, &[0xAA; 3]);
541        let mut data = s.clone();
542        data.push(0xFF); // trailing byte that doesn't complete a header
543        let refs = split_sections(&data);
544        assert_eq!(refs.len(), 1);
545        assert_eq!(refs[0], s.as_slice());
546    }
547
548    // ── interval constants pinned to spec ────────────────────────────────────
549
550    #[test]
551    fn interval_constants_match_spec() {
552        // TR 101 211 §4.4.1/§4.4.2
553        assert_eq!(NIT_MAX_INTERVAL, Duration::from_secs(10));
554        assert_eq!(BAT_MAX_INTERVAL, Duration::from_secs(10));
555        assert_eq!(SDT_ACTUAL_MAX_INTERVAL, Duration::from_secs(2));
556        assert_eq!(SDT_OTHER_MAX_INTERVAL, Duration::from_secs(10));
557        assert_eq!(EIT_PF_ACTUAL_MAX_INTERVAL, Duration::from_secs(2));
558        assert_eq!(EIT_PF_OTHER_MAX_INTERVAL, Duration::from_secs(10));
559        assert_eq!(EIT_SCHED_MAX_INTERVAL, Duration::from_secs(10));
560        assert_eq!(EIT_SCHED_EXT_MAX_INTERVAL, Duration::from_secs(30));
561        assert_eq!(TDT_MAX_INTERVAL, Duration::from_secs(30));
562        assert_eq!(TOT_MAX_INTERVAL, Duration::from_secs(30));
563
564        // TS 101 154 §4.1.7
565        assert_eq!(PAT_MAX_INTERVAL, Duration::from_millis(100));
566        assert_eq!(PMT_MAX_INTERVAL, Duration::from_millis(100));
567
568        // EN 300 468 §5.1.4.1
569        assert_eq!(MIN_SECTION_INTERVAL, Duration::from_millis(25));
570    }
571
572    // ── SiMux: basic behaviour ───────────────────────────────────────────────
573
574    #[test]
575    fn new_simux_is_empty() {
576        let mut mux = SiMux::new();
577        let pkts = mux.poll(Duration::ZERO);
578        assert!(pkts.is_empty());
579    }
580
581    #[test]
582    fn simux_default_is_empty() {
583        let mut mux = SiMux::default();
584        let pkts = mux.poll(Duration::ZERO);
585        assert!(pkts.is_empty());
586    }
587
588    #[test]
589    fn first_poll_always_emits() {
590        let s = build_section(0x42, &[0xAA; 10]);
591        let mut mux = SiMux::new();
592        mux.upsert(Pid::new(0x0100), s.clone(), Duration::from_millis(100));
593        let pkts = mux.poll(Duration::ZERO);
594        assert!(!pkts.is_empty(), "first poll must emit");
595    }
596
597    #[test]
598    fn first_poll_emits_all_entries() {
599        let s1 = build_section(0x42, &[0x01; 5]);
600        let s2 = build_section(0x46, &[0x02; 5]);
601        let mut mux = SiMux::new();
602        mux.upsert(Pid::new(0x0100), s1.clone(), Duration::from_millis(100));
603        mux.upsert(Pid::new(0x0200), s2.clone(), Duration::from_millis(200));
604        let pkts = mux.poll(Duration::ZERO);
605        // Both entries should emit — we count PID occurrences
606        let pids: Vec<Pid> = pkts
607            .iter()
608            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
609            .collect();
610        assert!(pids.contains(&Pid::new(0x0100)));
611        assert!(pids.contains(&Pid::new(0x0200)));
612    }
613
614    // ── deterministic schedule ───────────────────────────────────────────────
615
616    #[test]
617    fn entry_emits_only_when_due() {
618        let s = build_section(0x42, &[0xAA; 10]);
619        let mut mux = SiMux::new();
620        mux.upsert(Pid::new(0x0100), s.clone(), Duration::from_millis(100));
621
622        // First poll at t=0: emits
623        let pkts0 = mux.poll(Duration::ZERO);
624        assert!(!pkts0.is_empty());
625
626        // t=50: not yet due
627        let pkts50 = mux.poll(Duration::from_millis(50));
628        assert!(pkts50.is_empty(), "should not emit at t=50");
629
630        // t=99: still not due
631        let pkts99 = mux.poll(Duration::from_millis(99));
632        assert!(pkts99.is_empty(), "should not emit at t=99");
633
634        // t=100: due again
635        let pkts100 = mux.poll(Duration::from_millis(100));
636        assert!(!pkts100.is_empty(), "should emit at t=100");
637    }
638
639    #[test]
640    fn two_entries_different_cadence() {
641        let s1 = build_section(0x42, &[0x01; 5]);
642        let s2 = build_section(0x46, &[0x02; 5]);
643        let mut mux = SiMux::new();
644        mux.upsert(
645            Pid::new(0x0100),
646            concat_sections(&[s1]),
647            Duration::from_millis(100),
648        );
649        mux.upsert(
650            Pid::new(0x0200),
651            concat_sections(&[s2]),
652            Duration::from_millis(200),
653        );
654
655        // t=0: both emit
656        let pkts = mux.poll(Duration::ZERO);
657        assert!(!pkts.is_empty());
658
659        // t=100: only 0x0100 emits (due at 100)
660        let pkts100 = mux.poll(Duration::from_millis(100));
661        assert!(!pkts100.is_empty());
662        let pids100: Vec<Pid> = pkts100
663            .iter()
664            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
665            .collect();
666        assert!(
667            pids100.contains(&Pid::new(0x0100)),
668            "PID 0x0100 should emit at t=100"
669        );
670        assert!(
671            !pids100.contains(&Pid::new(0x0200)),
672            "PID 0x0200 should NOT emit at t=100"
673        );
674
675        // t=150: neither emits
676        let pkts150 = mux.poll(Duration::from_millis(150));
677        assert!(pkts150.is_empty(), "neither should emit at t=150");
678
679        // t=200: both emit again
680        let pkts200 = mux.poll(Duration::from_millis(200));
681        let pids200: Vec<Pid> = pkts200
682            .iter()
683            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
684            .collect();
685        assert!(
686            pids200.contains(&Pid::new(0x0100)),
687            "PID 0x0100 should emit at t=200"
688        );
689        assert!(
690            pids200.contains(&Pid::new(0x0200)),
691            "PID 0x0200 should emit at t=200"
692        );
693    }
694
695    #[test]
696    fn entry_emits_again_after_full_interval() {
697        let s = build_section(0x42, &[0xAA; 20]);
698        let mut mux = SiMux::new();
699        mux.upsert(Pid::new(0x0100), s, Duration::from_secs(1));
700
701        // t=0
702        let n0 = mux.poll_into(Duration::ZERO, &mut Vec::new());
703        assert!(n0 > 0);
704
705        // t=999ms: not due
706        let mut buf = Vec::new();
707        let n999 = mux.poll_into(Duration::from_millis(999), &mut buf);
708        assert_eq!(n999, 0);
709
710        // t=1s: due
711        let mut buf2 = Vec::new();
712        let n1000 = mux.poll_into(Duration::from_millis(1000), &mut buf2);
713        assert!(n1000 > 0);
714    }
715
716    // ── upsert replaces existing entry (same pid + table_id only) ───────────
717
718    #[test]
719    fn upsert_updates_existing_entry() {
720        // Same table_id (0x42, SDT actual) both times: this is a genuine
721        // update of the same logical table, so it must replace, not add.
722        let s1 = build_section(0x42, &[0xAA; 5]);
723        let s2 = build_section(0x42, &[0xBB; 10]);
724        let mut mux = SiMux::new();
725
726        mux.upsert(Pid::new(0x0100), s1, Duration::from_millis(100));
727        mux.upsert(Pid::new(0x0100), s2.clone(), Duration::from_millis(200));
728
729        let pkts = mux.poll(Duration::ZERO);
730
731        // Round-trip the output to verify the updated sections
732        let mut reasm = SectionReassembler::default();
733        for raw in &pkts {
734            let pkt = TsPacket::parse(raw).unwrap();
735            if pkt.header.pid == Pid::new(0x0100) {
736                reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
737            }
738        }
739        let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
740        assert_eq!(got.len(), 1);
741        assert_eq!(got[0].as_ref(), s2.as_slice());
742    }
743
744    // ── issue #1000: distinct table_ids sharing a PID must NOT collide ──────
745
746    /// Regression: `upsert_tdt` then `upsert_tot` (both PID 0x0014) used to
747    /// collide because entries were keyed by PID alone — the second call
748    /// silently replaced the first. Both table_ids must now come out, with a
749    /// continuous continuity counter across the shared PID.
750    #[test]
751    fn upsert_tdt_then_tot_both_emitted_with_continuous_cc() {
752        let tdt = build_section(0x70, &[0xAA; 5]); // TDT
753        let tot = build_section(0x73, &[0xBB; 8]); // TOT
754        let mut mux = SiMux::new();
755
756        mux.upsert_tdt(tdt.clone());
757        mux.upsert_tot(tot.clone());
758
759        let pkts = mux.poll(Duration::ZERO);
760        assert!(!pkts.is_empty());
761
762        let mut table_ids_seen = Vec::new();
763        let mut reasm = SectionReassembler::default();
764        let mut prev_cc: Option<u8> = None;
765        for raw in &pkts {
766            let pkt = TsPacket::parse(raw).unwrap();
767            assert_eq!(pkt.header.pid, well_known::TDT_TOT);
768            if let Some(prev) = prev_cc {
769                assert_eq!(
770                    pkt.header.continuity_counter,
771                    (prev + 1) & CC_MASK,
772                    "continuity counter must be continuous across table kinds on one PID"
773                );
774            }
775            prev_cc = Some(pkt.header.continuity_counter);
776            reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
777        }
778        while let Some(section) = reasm.pop_section() {
779            table_ids_seen.push(section.as_ref()[0]);
780        }
781
782        assert!(table_ids_seen.contains(&0x70), "TDT missing from output");
783        assert!(table_ids_seen.contains(&0x73), "TOT missing from output");
784    }
785
786    /// Same collision class for SDT actual (0x42) vs. BAT (0x4A), both on
787    /// PID 0x0011.
788    #[test]
789    fn upsert_sdt_and_bat_both_emitted() {
790        let sdt = build_section(0x42, &[0xCC; 5]);
791        let bat = build_section(0x4A, &[0xDD; 6]);
792        let mut mux = SiMux::new();
793
794        mux.upsert_sdt_actual(sdt.clone());
795        mux.upsert(
796            well_known::SDT_BAT,
797            bat.clone(),
798            Duration::from_millis(1000),
799        );
800
801        let pkts = mux.poll(Duration::ZERO);
802        let mut table_ids_seen = Vec::new();
803        let mut reasm = SectionReassembler::default();
804        for raw in &pkts {
805            let pkt = TsPacket::parse(raw).unwrap();
806            reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
807        }
808        while let Some(section) = reasm.pop_section() {
809            table_ids_seen.push(section.as_ref()[0]);
810        }
811
812        assert!(table_ids_seen.contains(&0x42), "SDT actual missing");
813        assert!(table_ids_seen.contains(&0x4A), "BAT missing");
814    }
815
816    // ── convenience constructors ─────────────────────────────────────────────
817
818    #[test]
819    fn upsert_pat_uses_correct_pid() {
820        let s = build_section(0x00, &[0x01, 0x02]);
821        let mut mux = SiMux::new();
822        mux.upsert_pat(s);
823        let pkts = mux.poll(Duration::ZERO);
824        let pids: Vec<Pid> = pkts
825            .iter()
826            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
827            .collect();
828        assert!(pids.iter().all(|&p| p == well_known::PAT));
829    }
830
831    #[test]
832    fn upsert_sdt_actual_uses_correct_pid() {
833        let s = build_section(0x42, &[0x01]);
834        let mut mux = SiMux::new();
835        mux.upsert_sdt_actual(s);
836        let pkts = mux.poll(Duration::ZERO);
837        let pids: Vec<Pid> = pkts
838            .iter()
839            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
840            .collect();
841        assert!(pids.iter().all(|&p| p == well_known::SDT_BAT));
842    }
843
844    #[test]
845    fn upsert_nit_uses_correct_pid() {
846        let s = build_section(0x40, &[0x01]);
847        let mut mux = SiMux::new();
848        mux.upsert_nit(s);
849        let pkts = mux.poll(Duration::ZERO);
850        let pids: Vec<Pid> = pkts
851            .iter()
852            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
853            .collect();
854        assert!(pids.iter().all(|&p| p == well_known::NIT));
855    }
856
857    #[test]
858    fn upsert_tdt_uses_correct_pid() {
859        let s = build_section(0x70, &[0x01]);
860        let mut mux = SiMux::new();
861        mux.upsert_tdt(s);
862        let pkts = mux.poll(Duration::ZERO);
863        let pids: Vec<Pid> = pkts
864            .iter()
865            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
866            .collect();
867        assert!(pids.iter().all(|&p| p == well_known::TDT_TOT));
868    }
869
870    #[test]
871    fn upsert_tot_uses_correct_pid() {
872        let s = build_section(0x73, &[0x01]);
873        let mut mux = SiMux::new();
874        mux.upsert_tot(s);
875        let pkts = mux.poll(Duration::ZERO);
876        let pids: Vec<Pid> = pkts
877            .iter()
878            .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
879            .collect();
880        assert!(pids.iter().all(|&p| p == well_known::TDT_TOT));
881    }
882
883    // ── round-trip through reassembler ───────────────────────────────────────
884
885    #[test]
886    fn simux_round_trip_single_entry() {
887        let s1 = build_section(0x42, &[0xAA; 20]);
888        let s2 = build_section(0x46, &[0xBB; 15]);
889        let mut mux = SiMux::new();
890        mux.upsert(
891            Pid::new(0x0100),
892            concat_sections(&[s1.clone(), s2.clone()]),
893            Duration::from_millis(100),
894        );
895
896        let pkts = mux.poll(Duration::ZERO);
897
898        let mut reasm = SectionReassembler::default();
899        for raw in &pkts {
900            let pkt = TsPacket::parse(raw).unwrap();
901            if pkt.header.pid == Pid::new(0x0100) {
902                reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
903            }
904        }
905        let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
906        assert_eq!(got.len(), 2, "round-trip must recover both sections");
907        assert_eq!(got[0].as_ref(), s1.as_slice());
908        assert_eq!(got[1].as_ref(), s2.as_slice());
909    }
910
911    #[test]
912    fn simux_round_trip_multi_pid() {
913        let s_a = build_section(0x42, &[0xA0; 10]);
914        let s_b = build_section(0x46, &[0xB0; 10]);
915        let mut mux = SiMux::new();
916        mux.upsert(Pid::new(0x0100), s_a.clone(), Duration::from_millis(100));
917        mux.upsert(Pid::new(0x0200), s_b.clone(), Duration::from_millis(100));
918
919        let pkts = mux.poll(Duration::ZERO);
920
921        let mut reasm_a = SectionReassembler::default();
922        let mut reasm_b = SectionReassembler::default();
923        for raw in &pkts {
924            let pkt = TsPacket::parse(raw).unwrap();
925            match pkt.header.pid.value() {
926                0x0100 => reasm_a.feed(pkt.payload.unwrap(), pkt.header.pusi),
927                0x0200 => reasm_b.feed(pkt.payload.unwrap(), pkt.header.pusi),
928                _ => {}
929            }
930        }
931        let got_a: Vec<_> = core::iter::from_fn(|| reasm_a.pop_section()).collect();
932        let got_b: Vec<_> = core::iter::from_fn(|| reasm_b.pop_section()).collect();
933        assert_eq!(got_a.len(), 1);
934        assert_eq!(got_b.len(), 1);
935        assert_eq!(got_a[0].as_ref(), s_a.as_slice());
936        assert_eq!(got_b[0].as_ref(), s_b.as_slice());
937    }
938
939    // ── CC continuity across polls ───────────────────────────────────────────
940
941    #[test]
942    fn continuity_counter_continuous_across_polls() {
943        // Section large enough to span at least 2 packets.
944        let s = build_section(0x42, &[0xAA; 250]);
945        let mut mux = SiMux::new();
946        mux.upsert(Pid::new(0x0100), s, Duration::from_millis(100));
947
948        let pkts1 = mux.poll(Duration::ZERO);
949        assert!(pkts1.len() >= 2, "need ≥2 packets to test CC continuity");
950
951        let last_cc_pkts1 = TsPacket::parse(&pkts1[pkts1.len() - 1])
952            .unwrap()
953            .header
954            .continuity_counter;
955
956        let pkts2 = mux.poll(Duration::from_millis(100));
957        assert!(!pkts2.is_empty(), "second poll must emit");
958
959        let first_cc_pkts2 = TsPacket::parse(&pkts2[0])
960            .unwrap()
961            .header
962            .continuity_counter;
963
964        assert_eq!(
965            first_cc_pkts2,
966            (last_cc_pkts1 + 1) & 0x0F,
967            "CC must continue across poll cycles"
968        );
969    }
970
971    // ── poll_into clears output ──────────────────────────────────────────────
972
973    #[test]
974    fn poll_into_clears_out_before_appending() {
975        let s = build_section(0x42, &[0xAA; 5]);
976        let mut mux = SiMux::new();
977        mux.upsert(Pid::new(0x0100), s, Duration::from_millis(100));
978
979        let mut out = vec![[0u8; TS_PACKET_SIZE]; 42];
980        let n = mux.poll_into(Duration::ZERO, &mut out);
981        assert_eq!(n, out.len(), "out must contain only new packets");
982    }
983
984    // ── debug_assert on sub-25 ms interval ───────────────────────────────────
985
986    #[test]
987    #[cfg(debug_assertions)]
988    #[should_panic(expected = "25 ms")]
989    fn upsert_rejects_sub_25ms_interval_in_debug() {
990        let s = build_section(0x42, &[0xAA; 5]);
991        let mut mux = SiMux::new();
992        mux.upsert(Pid::new(0x0100), s, Duration::from_millis(10));
993    }
994
995    #[test]
996    fn upsert_accepts_25ms_interval() {
997        let s = build_section(0x42, &[0xAA; 5]);
998        let mut mux = SiMux::new();
999        mux.upsert(Pid::new(0x0100), s, MIN_SECTION_INTERVAL);
1000        let pkts = mux.poll(Duration::ZERO);
1001        assert!(!pkts.is_empty());
1002    }
1003
1004    // ── round-trip property (the mandatory acceptance oracle) ────────────────
1005
1006    #[test]
1007    fn round_trip_single_short_section() {
1008        let s = build_section(0x42, &[0xAA; 10]);
1009        assert_round_trip(&[s]);
1010    }
1011
1012    #[test]
1013    fn round_trip_one_byte_body() {
1014        let s = build_section(0x46, &[0xBB]); // 4 bytes total
1015        assert_round_trip(&[s]);
1016    }
1017
1018    #[test]
1019    fn round_trip_section_exactly_pusi_cap_boundary() {
1020        // A section whose total length is exactly PUSI_PAYLOAD_CAP (183).
1021        let body = vec![0xCC; PUSI_PAYLOAD_CAP - 3];
1022        let s = build_section(0x50, &body);
1023        assert_eq!(s.len(), PUSI_PAYLOAD_CAP);
1024        assert_round_trip(&[s]);
1025    }
1026
1027    #[test]
1028    fn round_trip_section_just_over_pusi_cap() {
1029        // One byte more than fits in a PUSI packet → must span to continuation.
1030        let body = vec![0xDD; PUSI_PAYLOAD_CAP - 3 + 1];
1031        let s = build_section(0x52, &body);
1032        assert_eq!(s.len(), PUSI_PAYLOAD_CAP + 1);
1033        assert_round_trip(&[s]);
1034    }
1035
1036    #[test]
1037    fn round_trip_section_spans_many_packets() {
1038        // A 2000-byte section spans ~11 packets.
1039        let body = vec![0xEE; 2000 - 3];
1040        let s = build_section(0x60, &body);
1041        assert_round_trip(&[s]);
1042    }
1043
1044    #[test]
1045    fn round_trip_section_at_max_size() {
1046        // The maximum section (4096 total, the long-form ceiling), whose final
1047        // continuation packet carries the tail followed by 0xFF stuffing. Since
1048        // #148 the reassembler ignores that trailing stuffing instead of
1049        // counting it toward MAX_SECTION_SIZE, so a full-size section round-trips.
1050        let body = vec![0x11; 4096 - 3];
1051        let s = build_section(0x80, &body);
1052        assert_eq!(s.len(), 4096);
1053        assert_round_trip(&[s]);
1054    }
1055
1056    #[test]
1057    fn round_trip_multiple_short_sections_in_one_batch() {
1058        let s1 = build_section(0x42, &[0x01, 0x02]); // 5 bytes
1059        let s2 = build_section(0x46, &[0x03]); // 4 bytes
1060        let s3 = build_section(0x4A, &[0x04, 0x05, 0x06]); // 6 bytes
1061        assert_round_trip(&[s1, s2, s3]);
1062    }
1063
1064    #[test]
1065    fn round_trip_section_ends_exactly_at_boundary() {
1066        // First section is exactly PUSI_PAYLOAD_CAP bytes — ends at packet
1067        // boundary.  Second section starts fresh in the next packet with
1068        // PUSI=1, pointer_field=0.
1069        let body1 = vec![0xA1; PUSI_PAYLOAD_CAP - 3];
1070        let s1 = build_section(0x50, &body1);
1071        assert_eq!(s1.len(), PUSI_PAYLOAD_CAP);
1072
1073        let s2 = build_section(0x52, &[0xB1, 0xB2]);
1074        assert_round_trip(&[s1, s2]);
1075    }
1076
1077    #[test]
1078    fn continuation_packet_never_overshoots_next_section_start() {
1079        // First section is 366 bytes total: it fills the first (PUSI) packet's
1080        // 183-byte payload exactly, then the *second* packet (a continuation,
1081        // starting at pos=183) is exactly 183 bytes short of the next
1082        // section's start (183 + 183 = 366) — the mid-stream analogue of
1083        // `round_trip_section_ends_exactly_at_boundary` above, which only
1084        // exercises the boundary at pos=0.
1085        let body1 = vec![0xA5; 366 - 3];
1086        let s1 = build_section(0x50, &body1);
1087        assert_eq!(s1.len(), 366);
1088        let s2 = build_section(0x52, &[0xB1, 0xB2, 0xB3]);
1089
1090        let mut packetiser = SectionPacketiser::new(Pid::new(0x0100));
1091        let packets = packetiser.packetise(&[s1.as_slice(), s2.as_slice()]);
1092
1093        for pkt_raw in &packets {
1094            let pkt = TsPacket::parse(pkt_raw).expect("parse generated packet");
1095            let payload = pkt.payload.expect("payload present");
1096            if pkt.header.pusi {
1097                let pointer = payload[0] as usize;
1098                // W1 (#1074): pre-fix this was 183 — a pointer_field pointing
1099                // past the packet's own payload, so no section byte actually
1100                // starts within it.
1101                assert!(
1102                    pointer < PUSI_PAYLOAD_CAP,
1103                    "pointer_field {pointer} leaves no room for a section to \
1104                     start within this packet (H.222.0 §2.4.4)"
1105                );
1106            }
1107        }
1108
1109        assert_round_trip(&[s1, s2]);
1110    }
1111
1112    #[test]
1113    fn round_trip_mix_small_large_sections() {
1114        // Mix of small and spanning sections that stress pointer_field and
1115        // concatenation.
1116        let s1 = build_section(0x10, &[0xAA; 5]);
1117        let body2 = vec![0xBB; 200];
1118        let s2 = build_section(0x20, &body2);
1119        let s3 = build_section(0x30, &[0xCC; 50]);
1120        let body4 = vec![0xDD; 800];
1121        let s4 = build_section(0x40, &body4);
1122        let s5 = build_section(0x50, &[0xEE]); // 1-byte body
1123        assert_round_trip(&[s1, s2, s3, s4, s5]);
1124    }
1125
1126    // ── continuity counter ───────────────────────────────────────────────────
1127
1128    #[test]
1129    fn continuity_counter_increments_per_packet() {
1130        // Use a section large enough to span several packets.
1131        let body = vec![0xAA; 500];
1132        let section = build_section(0x42, &body);
1133        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1134
1135        let packets = p.packetise(&[&section]);
1136        assert!(packets.len() >= 3, "need multiple packets to test CC");
1137
1138        let mut last_cc: Option<u8> = None;
1139        for pkt_raw in &packets {
1140            let pkt = TsPacket::parse(pkt_raw).unwrap();
1141            let cc = pkt.header.continuity_counter;
1142            if let Some(last) = last_cc {
1143                assert_eq!(cc, (last + 1) & 0x0F, "CC must increment per packet");
1144            }
1145            last_cc = Some(cc);
1146        }
1147    }
1148
1149    #[test]
1150    fn continuity_counter_wraps_and_continues_across_calls() {
1151        let mut p = SectionPacketiser::with_continuity(Pid::new(0x0100), 14);
1152        // Section large enough to span at least 3 packets.
1153        let body = vec![0xBB; 500];
1154        let s = build_section(0x42, &body);
1155
1156        // First call: CC 14, 15, 0, …
1157        let pkts1 = p.packetise(&[&s]);
1158        assert!(pkts1.len() >= 3, "section must span ≥3 packets");
1159        let ccs1: Vec<u8> = pkts1
1160            .iter()
1161            .map(|b| TsPacket::parse(b).unwrap().header.continuity_counter)
1162            .collect();
1163        assert_eq!(ccs1[0], 14);
1164        assert_eq!(ccs1[1], 15);
1165        assert_eq!(ccs1[2], 0);
1166
1167        // Second call: CC continues from where first left off.
1168        let pkts2 = p.packetise(&[&s]);
1169        let cc_first_pkt2 = TsPacket::parse(&pkts2[0])
1170            .unwrap()
1171            .header
1172            .continuity_counter;
1173        assert_eq!(cc_first_pkt2, ccs1.last().map(|c| (c + 1) & 0x0F).unwrap());
1174    }
1175
1176    // ── PUSI placement ──────────────────────────────────────────────────────
1177
1178    #[test]
1179    fn pusi_set_when_section_starts() {
1180        let s = build_section(0x42, &[0xAA; 10]);
1181        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1182        let packets = p.packetise(&[&s]);
1183        assert!(!packets.is_empty());
1184        let pkt = TsPacket::parse(&packets[0]).unwrap();
1185        assert!(pkt.header.pusi, "first packet must have PUSI=1");
1186    }
1187
1188    #[test]
1189    fn pusi_not_set_on_mid_section_continuation() {
1190        let body = vec![0xAA; 500];
1191        let s = build_section(0x42, &body);
1192        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1193        let packets = p.packetise(&[&s]);
1194        assert!(packets.len() >= 2);
1195        let pkt1 = TsPacket::parse(&packets[0]).unwrap();
1196        let pkt2 = TsPacket::parse(&packets[1]).unwrap();
1197        assert!(pkt1.header.pusi, "first packet must have PUSI=1");
1198        assert!(
1199            !pkt2.header.pusi,
1200            "second packet is continuation, must have PUSI=0"
1201        );
1202    }
1203
1204    #[test]
1205    fn pointer_field_equals_tail_length_before_new_section() {
1206        // Section1 = 200 bytes.  Section2 = 50 bytes.
1207        // Packet 1: PUSI=1, pointer=0, section1 head.
1208        // Packet 2: PUSI=1, pointer > 0 (tail of section1 before section2).
1209        let body1 = vec![0xA1; 197]; // 200-byte section
1210        let s1 = build_section(0x52, &body1);
1211        assert_eq!(s1.len(), 200);
1212        let s2 = build_section(0x54, &[0xB1; 47]); // 50-byte section
1213        assert_eq!(s2.len(), 50);
1214
1215        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1216        let packets = p.packetise(&[&s1, &s2]);
1217
1218        // Find the packet where PUSI=1 and pointer>0.
1219        let pkt_with_pointer = packets
1220            .iter()
1221            .map(|raw| TsPacket::parse(raw).unwrap())
1222            .find(|pkt| pkt.header.pusi && pkt.payload.is_some_and(|pl| pl.first() != Some(&0)))
1223            .expect("must have a PUSI packet with non-zero pointer");
1224
1225        let payload = pkt_with_pointer.payload.unwrap();
1226        let pointer = payload[0] as usize;
1227        assert!(pointer > 0, "pointer must be non-zero");
1228        // The tail bytes should be from the end of section1.
1229        let tail_start = s1.len() - pointer;
1230        assert_eq!(&payload[1..1 + pointer], &s1[tail_start..]);
1231    }
1232
1233    // ── stuffing ─────────────────────────────────────────────────────────────
1234
1235    #[test]
1236    fn final_packet_unused_tail_is_stuffing() {
1237        let s = build_section(0x42, &[0xAA; 5]); // 8 bytes total
1238        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1239        let packets = p.packetise(&[&s]);
1240
1241        let pkt = TsPacket::parse(&packets[0]).unwrap();
1242        let payload = pkt.payload.unwrap();
1243        assert_eq!(payload[0], 0, "pointer_field should be 0");
1244
1245        let section_end = 1 + s.len(); // after pointer + section
1246        assert!(
1247            section_end < payload.len(),
1248            "must have stuffing after section"
1249        );
1250        for &b in &payload[section_end..] {
1251            assert_eq!(b, STUFFING_BYTE, "all trailing bytes must be 0xFF");
1252        }
1253    }
1254
1255    #[test]
1256    fn reassembler_discards_stuffing() {
1257        let s1 = build_section(0x42, &[0xAA; 10]);
1258        let s2 = build_section(0x46, &[0xBB; 5]);
1259
1260        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1261        let packets = p.packetise(&[&s1, &s2]);
1262
1263        let mut reasm = SectionReassembler::default();
1264        for pkt_raw in &packets {
1265            let pkt = TsPacket::parse(pkt_raw).unwrap();
1266            reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
1267        }
1268
1269        let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
1270        assert_eq!(got.len(), 2);
1271        assert!(
1272            reasm.is_empty(),
1273            "stuffing tail must be discarded, not buffered"
1274        );
1275    }
1276
1277    // ── misc ─────────────────────────────────────────────────────────────────
1278
1279    #[test]
1280    fn empty_batch_produces_no_packets() {
1281        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1282        let packets: Vec<[u8; TS_PACKET_SIZE]> = p.packetise(&[]);
1283        assert!(packets.is_empty());
1284    }
1285
1286    #[test]
1287    fn packetise_into_clears_out_first() {
1288        let s = build_section(0x42, &[0xAA; 5]);
1289        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1290
1291        let mut out = vec![[0u8; TS_PACKET_SIZE]; 99]; // pre-existing junk
1292        let n = p.packetise_into(&[&s], &mut out);
1293        assert_eq!(n, out.len(), "out must contain only the new packets");
1294        // Verify the output is correct (round-trip).
1295        let mut reasm = SectionReassembler::default();
1296        for pkt_raw in &out {
1297            let pkt = TsPacket::parse(pkt_raw).unwrap();
1298            reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
1299        }
1300        let got = reasm.pop_section().unwrap();
1301        assert_eq!(got.as_ref(), s.as_slice());
1302    }
1303
1304    #[test]
1305    fn pid_is_correct() {
1306        let p = SectionPacketiser::new(Pid::new(0x1234));
1307        assert_eq!(p.pid(), Pid::new(0x1234));
1308    }
1309
1310    #[test]
1311    fn with_continuity_masks_to_4_bits() {
1312        let p = SectionPacketiser::with_continuity(Pid::new(0x0100), 0xFE);
1313        assert_eq!(p.continuity_counter(), 0x0E);
1314    }
1315
1316    #[test]
1317    fn has_payload_always_true_no_adaptation() {
1318        let s = build_section(0x42, &[0xAA; 50]);
1319        let mut p = SectionPacketiser::new(Pid::new(0x0100));
1320        let packets = p.packetise(&[&s]);
1321        for pkt_raw in &packets {
1322            let pkt = TsPacket::parse(pkt_raw).unwrap();
1323            assert!(pkt.header.has_payload, "every packet must carry payload");
1324            assert!(!pkt.header.has_adaptation, "no adaptation field is emitted");
1325            assert!(!pkt.header.tei, "TEI must be false");
1326            assert_eq!(pkt.header.scrambling, 0, "scrambling must be 0");
1327        }
1328    }
1329}