1use alloc::vec::Vec;
10use core::time::Duration;
11
12use crate::pid::{Pid, well_known};
13use crate::ts::{CC_MASK, TS_PACKET_SIZE, TsHeader};
14
15const PUSI_PAYLOAD_CAP: usize = 183;
17const PAYLOAD_CAP: usize = 184;
19use crate::ts::STUFFING_BYTE;
20
21pub struct SectionPacketiser {
30 pid: Pid,
31 continuity_counter: u8,
32}
33
34impl SectionPacketiser {
35 pub fn new(pid: Pid) -> Self {
37 Self {
38 pid,
39 continuity_counter: 0,
40 }
41 }
42
43 pub fn with_continuity(pid: Pid, cc: u8) -> Self {
45 Self {
46 pid,
47 continuity_counter: cc & CC_MASK,
48 }
49 }
50
51 pub fn pid(&self) -> Pid {
53 self.pid
54 }
55
56 pub fn continuity_counter(&self) -> u8 {
58 self.continuity_counter
59 }
60
61 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 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 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 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 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 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
182pub const NIT_MAX_INTERVAL: Duration = Duration::from_secs(10);
186pub const BAT_MAX_INTERVAL: Duration = Duration::from_secs(10);
188pub const SDT_ACTUAL_MAX_INTERVAL: Duration = Duration::from_secs(2);
190pub const SDT_OTHER_MAX_INTERVAL: Duration = Duration::from_secs(10);
192pub const EIT_PF_ACTUAL_MAX_INTERVAL: Duration = Duration::from_secs(2);
194pub const EIT_PF_OTHER_MAX_INTERVAL: Duration = Duration::from_secs(10);
197pub const EIT_SCHED_MAX_INTERVAL: Duration = Duration::from_secs(10);
199pub const EIT_SCHED_EXT_MAX_INTERVAL: Duration = Duration::from_secs(30);
201pub const TDT_MAX_INTERVAL: Duration = Duration::from_secs(30);
203pub const TOT_MAX_INTERVAL: Duration = Duration::from_secs(30);
205
206pub const PAT_MAX_INTERVAL: Duration = Duration::from_millis(100);
208pub const PMT_MAX_INTERVAL: Duration = Duration::from_millis(100);
210
211pub const MIN_SECTION_INTERVAL: Duration = Duration::from_millis(25);
214
215pub struct SiMux {
244 entries: Vec<Entry>,
245 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
257fn table_id_of(sections: &[u8]) -> u8 {
263 sections.first().copied().unwrap_or(0)
264}
265
266fn 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 pub fn new() -> Self {
281 Self {
282 entries: Vec::new(),
283 packetisers: Vec::new(),
284 }
285 }
286
287 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(§ions);
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 pub fn upsert_pat(&mut self, sections: Vec<u8>) {
329 self.upsert(well_known::PAT, sections, PAT_MAX_INTERVAL);
330 }
331
332 pub fn upsert_pmt(&mut self, pid: Pid, sections: Vec<u8>) {
336 self.upsert(pid, sections, PMT_MAX_INTERVAL);
337 }
338
339 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 pub fn upsert_nit(&mut self, sections: Vec<u8>) {
350 self.upsert(well_known::NIT, sections, NIT_MAX_INTERVAL);
351 }
352
353 pub fn upsert_tdt(&mut self, sections: Vec<u8>) {
357 self.upsert(well_known::TDT_TOT, sections, TDT_MAX_INTERVAL);
358 }
359
360 pub fn upsert_tot(&mut self, sections: Vec<u8>) {
364 self.upsert(well_known::TDT_TOT, sections, TOT_MAX_INTERVAL);
365 }
366
367 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 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
412fn 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 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 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 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 #[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(§ions);
509 }
510
511 #[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); let refs = split_sections(&data);
544 assert_eq!(refs.len(), 1);
545 assert_eq!(refs[0], s.as_slice());
546 }
547
548 #[test]
551 fn interval_constants_match_spec() {
552 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 assert_eq!(PAT_MAX_INTERVAL, Duration::from_millis(100));
566 assert_eq!(PMT_MAX_INTERVAL, Duration::from_millis(100));
567
568 assert_eq!(MIN_SECTION_INTERVAL, Duration::from_millis(25));
570 }
571
572 #[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 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 #[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 let pkts0 = mux.poll(Duration::ZERO);
624 assert!(!pkts0.is_empty());
625
626 let pkts50 = mux.poll(Duration::from_millis(50));
628 assert!(pkts50.is_empty(), "should not emit at t=50");
629
630 let pkts99 = mux.poll(Duration::from_millis(99));
632 assert!(pkts99.is_empty(), "should not emit at t=99");
633
634 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 let pkts = mux.poll(Duration::ZERO);
657 assert!(!pkts.is_empty());
658
659 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 let pkts150 = mux.poll(Duration::from_millis(150));
677 assert!(pkts150.is_empty(), "neither should emit at t=150");
678
679 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 let n0 = mux.poll_into(Duration::ZERO, &mut Vec::new());
703 assert!(n0 > 0);
704
705 let mut buf = Vec::new();
707 let n999 = mux.poll_into(Duration::from_millis(999), &mut buf);
708 assert_eq!(n999, 0);
709
710 let mut buf2 = Vec::new();
712 let n1000 = mux.poll_into(Duration::from_millis(1000), &mut buf2);
713 assert!(n1000 > 0);
714 }
715
716 #[test]
719 fn upsert_updates_existing_entry() {
720 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 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 #[test]
751 fn upsert_tdt_then_tot_both_emitted_with_continuous_cc() {
752 let tdt = build_section(0x70, &[0xAA; 5]); let tot = build_section(0x73, &[0xBB; 8]); 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 #[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 #[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 #[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 #[test]
942 fn continuity_counter_continuous_across_polls() {
943 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 #[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 #[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 #[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]); assert_round_trip(&[s]);
1016 }
1017
1018 #[test]
1019 fn round_trip_section_exactly_pusi_cap_boundary() {
1020 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 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 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 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]); let s2 = build_section(0x46, &[0x03]); let s3 = build_section(0x4A, &[0x04, 0x05, 0x06]); assert_round_trip(&[s1, s2, s3]);
1062 }
1063
1064 #[test]
1065 fn round_trip_section_ends_exactly_at_boundary() {
1066 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 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 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 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]); assert_round_trip(&[s1, s2, s3, s4, s5]);
1124 }
1125
1126 #[test]
1129 fn continuity_counter_increments_per_packet() {
1130 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(&[§ion]);
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 let body = vec![0xBB; 500];
1154 let s = build_section(0x42, &body);
1155
1156 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 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 #[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 let body1 = vec![0xA1; 197]; let s1 = build_section(0x52, &body1);
1211 assert_eq!(s1.len(), 200);
1212 let s2 = build_section(0x54, &[0xB1; 47]); assert_eq!(s2.len(), 50);
1214
1215 let mut p = SectionPacketiser::new(Pid::new(0x0100));
1216 let packets = p.packetise(&[&s1, &s2]);
1217
1218 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 let tail_start = s1.len() - pointer;
1230 assert_eq!(&payload[1..1 + pointer], &s1[tail_start..]);
1231 }
1232
1233 #[test]
1236 fn final_packet_unused_tail_is_stuffing() {
1237 let s = build_section(0x42, &[0xAA; 5]); 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(); 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 #[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]; let n = p.packetise_into(&[&s], &mut out);
1293 assert_eq!(n, out.len(), "out must contain only the new packets");
1294 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}