1use alloc::vec::Vec;
10use core::time::Duration;
11
12use crate::pid::well_known;
13use crate::ts::{CC_MASK, TS_PACKET_SIZE, TsHeader};
14
15const PUSI_PAYLOAD_CAP: usize = 183;
17const PAYLOAD_CAP: usize = 184;
19const STUFFING_BYTE: u8 = 0xFF;
21
22pub struct SectionPacketiser {
31 pid: u16,
32 continuity_counter: u8,
33}
34
35impl SectionPacketiser {
36 pub fn new(pid: u16) -> Self {
38 Self {
39 pid,
40 continuity_counter: 0,
41 }
42 }
43
44 pub fn with_continuity(pid: u16, cc: u8) -> Self {
46 Self {
47 pid,
48 continuity_counter: cc & CC_MASK,
49 }
50 }
51
52 pub fn pid(&self) -> u16 {
54 self.pid
55 }
56
57 pub fn continuity_counter(&self) -> u8 {
59 self.continuity_counter
60 }
61
62 pub fn packetise_into(
67 &mut self,
68 sections: &[&[u8]],
69 out: &mut Vec<[u8; TS_PACKET_SIZE]>,
70 ) -> usize {
71 out.clear();
72
73 if sections.is_empty() {
74 return 0;
75 }
76
77 let total_len: usize = sections.iter().map(|s| s.len()).sum();
79 if total_len == 0 {
80 return 0;
81 }
82 let mut data = Vec::with_capacity(total_len);
83 let mut starts = Vec::with_capacity(sections.len());
84 for s in sections {
85 starts.push(data.len());
86 data.extend_from_slice(s);
87 }
88
89 let count_before = out.len();
90 let mut pos = 0usize;
91 let mut next_idx = 0usize;
96
97 while pos < data.len() {
98 while starts.get(next_idx).is_some_and(|&s| s < pos) {
99 next_idx += 1;
100 }
101 let next_start = starts.get(next_idx).copied();
102
103 let pusi: bool;
104 let pointer_field: u8;
105 let cap: usize;
106
107 if let Some(ns) = next_start {
108 let diff = ns.saturating_sub(pos);
109 if diff < PUSI_PAYLOAD_CAP {
120 pusi = true;
121 pointer_field = diff as u8;
122 cap = PUSI_PAYLOAD_CAP;
123 } else {
124 pusi = false;
125 pointer_field = 0;
126 cap = PAYLOAD_CAP.min(diff);
127 }
128 } else {
129 pusi = false;
130 pointer_field = 0;
131 cap = PAYLOAD_CAP;
132 }
133
134 let mut pkt = [0u8; TS_PACKET_SIZE];
135
136 let header = TsHeader {
137 tei: false,
138 pusi,
139 pid: self.pid,
140 scrambling: 0,
141 has_adaptation: false,
142 has_payload: true,
143 continuity_counter: self.continuity_counter,
144 };
145 header
146 .serialize_into(&mut pkt[..4])
147 .expect("4-byte header buffer");
148
149 self.continuity_counter = (self.continuity_counter + 1) & CC_MASK;
150
151 let mut write_pos = 4usize;
152
153 if pusi {
154 pkt[write_pos] = pointer_field;
155 write_pos += 1;
156 }
157
158 let remaining = data.len() - pos;
159 let to_copy = remaining.min(cap);
160 pkt[write_pos..write_pos + to_copy].copy_from_slice(&data[pos..pos + to_copy]);
161 pos += to_copy;
162 write_pos += to_copy;
163
164 for b in &mut pkt[write_pos..] {
166 *b = STUFFING_BYTE;
167 }
168
169 out.push(pkt);
170 }
171
172 out.len() - count_before
173 }
174
175 pub fn packetise(&mut self, sections: &[&[u8]]) -> Vec<[u8; TS_PACKET_SIZE]> {
177 let mut out = Vec::new();
178 self.packetise_into(sections, &mut out);
179 out
180 }
181}
182
183pub const NIT_MAX_INTERVAL: Duration = Duration::from_secs(10);
187pub const BAT_MAX_INTERVAL: Duration = Duration::from_secs(10);
189pub const SDT_ACTUAL_MAX_INTERVAL: Duration = Duration::from_secs(2);
191pub const SDT_OTHER_MAX_INTERVAL: Duration = Duration::from_secs(10);
193pub const EIT_PF_ACTUAL_MAX_INTERVAL: Duration = Duration::from_secs(2);
195pub const EIT_PF_OTHER_MAX_INTERVAL: Duration = Duration::from_secs(10);
198pub const EIT_SCHED_MAX_INTERVAL: Duration = Duration::from_secs(10);
200pub const EIT_SCHED_EXT_MAX_INTERVAL: Duration = Duration::from_secs(30);
202pub const TDT_MAX_INTERVAL: Duration = Duration::from_secs(30);
204pub const TOT_MAX_INTERVAL: Duration = Duration::from_secs(30);
206
207pub const PAT_MAX_INTERVAL: Duration = Duration::from_millis(100);
209pub const PMT_MAX_INTERVAL: Duration = Duration::from_millis(100);
211
212pub const MIN_SECTION_INTERVAL: Duration = Duration::from_millis(25);
215
216pub struct SiMux {
245 entries: Vec<Entry>,
246 packetisers: Vec<SectionPacketiser>,
248}
249
250struct Entry {
251 pid: u16,
252 table_id: u8,
253 sections: Vec<u8>,
254 interval: Duration,
255 last_emit: Option<Duration>,
256}
257
258fn table_id_of(sections: &[u8]) -> u8 {
264 sections.first().copied().unwrap_or(0)
265}
266
267fn packetiser_for(packetisers: &mut Vec<SectionPacketiser>, pid: u16) -> &mut SectionPacketiser {
271 if let Some(idx) = packetisers.iter().position(|p| p.pid() == pid) {
272 &mut packetisers[idx]
273 } else {
274 packetisers.push(SectionPacketiser::new(pid));
275 packetisers.last_mut().expect("just pushed")
276 }
277}
278
279impl SiMux {
280 pub fn new() -> Self {
282 Self {
283 entries: Vec::new(),
284 packetisers: Vec::new(),
285 }
286 }
287
288 pub fn upsert(&mut self, pid: u16, sections: Vec<u8>, interval: Duration) {
302 debug_assert!(
303 interval >= MIN_SECTION_INTERVAL,
304 "interval {interval:?} is below the 25 ms minimum (EN 300 468 §5.1.4.1)"
305 );
306
307 let table_id = table_id_of(§ions);
308 if let Some(entry) = self
309 .entries
310 .iter_mut()
311 .find(|e| e.pid == pid && e.table_id == table_id)
312 {
313 entry.sections = sections;
314 entry.interval = interval;
315 } else {
316 self.entries.push(Entry {
317 pid,
318 table_id,
319 sections,
320 interval,
321 last_emit: None,
322 });
323 }
324 }
325
326 pub fn upsert_pat(&mut self, sections: Vec<u8>) {
330 self.upsert(well_known::PAT.value(), sections, PAT_MAX_INTERVAL);
331 }
332
333 pub fn upsert_pmt(&mut self, pid: u16, sections: Vec<u8>) {
337 self.upsert(pid, sections, PMT_MAX_INTERVAL);
338 }
339
340 pub fn upsert_sdt_actual(&mut self, sections: Vec<u8>) {
344 self.upsert(
345 well_known::SDT_BAT.value(),
346 sections,
347 SDT_ACTUAL_MAX_INTERVAL,
348 );
349 }
350
351 pub fn upsert_nit(&mut self, sections: Vec<u8>) {
355 self.upsert(well_known::NIT.value(), sections, NIT_MAX_INTERVAL);
356 }
357
358 pub fn upsert_tdt(&mut self, sections: Vec<u8>) {
362 self.upsert(well_known::TDT_TOT.value(), sections, TDT_MAX_INTERVAL);
363 }
364
365 pub fn upsert_tot(&mut self, sections: Vec<u8>) {
369 self.upsert(well_known::TDT_TOT.value(), sections, TOT_MAX_INTERVAL);
370 }
371
372 pub fn poll_into(&mut self, now: Duration, out: &mut Vec<[u8; TS_PACKET_SIZE]>) -> usize {
379 out.clear();
380 let before = out.len();
381
382 let mut tmp = Vec::new();
383 let packetisers = &mut self.packetisers;
384 for entry in &mut self.entries {
385 let due = match entry.last_emit {
386 None => true,
387 Some(last) => now.saturating_sub(last) >= entry.interval,
388 };
389 if due {
390 let refs = split_sections(&entry.sections);
391 if !refs.is_empty() {
392 let packetiser = packetiser_for(packetisers, entry.pid);
393 packetiser.packetise_into(&refs, &mut tmp);
394 out.append(&mut tmp);
395 }
396 entry.last_emit = Some(now);
397 }
398 }
399
400 out.len() - before
401 }
402
403 pub fn poll(&mut self, now: Duration) -> Vec<[u8; TS_PACKET_SIZE]> {
405 let mut out = Vec::new();
406 self.poll_into(now, &mut out);
407 out
408 }
409}
410
411impl Default for SiMux {
412 fn default() -> Self {
413 Self::new()
414 }
415}
416
417fn split_sections(data: &[u8]) -> Vec<&[u8]> {
423 let mut result = Vec::new();
424 let mut pos = 0;
425 while let Some(end) = crate::ts::section_total_len(&data[pos..]).map(|len| pos + len) {
426 if end > data.len() {
427 break;
428 }
429 result.push(&data[pos..end]);
430 pos = end;
431 }
432 result
433}
434
435#[cfg(test)]
436mod tests {
437 use super::*;
438 use crate::ts::{SectionReassembler, TsPacket};
439 use alloc::vec;
440 use alloc::vec::Vec;
441
442 fn build_section(table_id: u8, body_after_length: &[u8]) -> Vec<u8> {
448 let section_length = body_after_length.len() as u16;
449 let mut v = Vec::with_capacity(3 + section_length as usize);
450 v.push(table_id);
451 v.push(0xB0 | ((section_length >> 8) as u8 & 0x0F));
453 v.push((section_length & 0xFF) as u8);
454 v.extend_from_slice(body_after_length);
455 v
456 }
457
458 fn concat_sections(sections: &[Vec<u8>]) -> Vec<u8> {
459 let total: usize = sections.iter().map(|s| s.len()).sum();
460 let mut out = Vec::with_capacity(total);
461 for s in sections {
462 out.extend_from_slice(s);
463 }
464 out
465 }
466
467 fn assert_round_trip(sections: &[Vec<u8>]) {
470 let mut packetiser = SectionPacketiser::new(0x0100);
471 let refs: Vec<&[u8]> = sections.iter().map(|s| s.as_slice()).collect();
472 let packets = packetiser.packetise(&refs);
473
474 let mut reasm = SectionReassembler::default();
475 for pkt_raw in &packets {
476 let pkt = TsPacket::parse(pkt_raw).expect("parse generated packet");
477 let payload = pkt.payload.expect("payload present");
478 let pusi = pkt.header.pusi;
479 reasm.feed(payload, pusi);
480 }
481
482 let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
483 assert_eq!(
484 got.len(),
485 sections.len(),
486 "section count mismatch: expected {}, got {}",
487 sections.len(),
488 got.len()
489 );
490 for (i, (orig, round)) in sections.iter().zip(got.iter()).enumerate() {
491 assert_eq!(
492 round.as_ref(),
493 orig.as_slice(),
494 "section {i} round-trip mismatch"
495 );
496 }
497 assert!(reasm.is_empty(), "reassembler should be empty after drain");
498 }
499
500 #[test]
506 fn many_sections_with_boundary_straddling_sizes_round_trip() {
507 let sections: Vec<Vec<u8>> = (0..400usize)
508 .map(|n| {
509 let body: Vec<u8> = (0..n).map(|i| (i % 251) as u8).collect();
510 build_section((n % 200) as u8, &body)
511 })
512 .collect();
513 assert_round_trip(§ions);
514 }
515
516 #[test]
519 fn split_single_section() {
520 let s = build_section(0x42, &[0xAA; 5]);
521 let refs = split_sections(&s);
522 assert_eq!(refs.len(), 1);
523 assert_eq!(refs[0], s.as_slice());
524 }
525
526 #[test]
527 fn split_two_sections() {
528 let s1 = build_section(0x42, &[0x01, 0x02]);
529 let s2 = build_section(0x46, &[0x03, 0x04, 0x05]);
530 let both = concat_sections(&[s1.clone(), s2.clone()]);
531 let refs = split_sections(&both);
532 assert_eq!(refs.len(), 2);
533 assert_eq!(refs[0], s1.as_slice());
534 assert_eq!(refs[1], s2.as_slice());
535 }
536
537 #[test]
538 fn split_empty_input() {
539 let refs = split_sections(&[]);
540 assert!(refs.is_empty());
541 }
542
543 #[test]
544 fn split_trailing_garbage_ignored() {
545 let s = build_section(0x42, &[0xAA; 3]);
546 let mut data = s.clone();
547 data.push(0xFF); let refs = split_sections(&data);
549 assert_eq!(refs.len(), 1);
550 assert_eq!(refs[0], s.as_slice());
551 }
552
553 #[test]
556 fn interval_constants_match_spec() {
557 assert_eq!(NIT_MAX_INTERVAL, Duration::from_secs(10));
559 assert_eq!(BAT_MAX_INTERVAL, Duration::from_secs(10));
560 assert_eq!(SDT_ACTUAL_MAX_INTERVAL, Duration::from_secs(2));
561 assert_eq!(SDT_OTHER_MAX_INTERVAL, Duration::from_secs(10));
562 assert_eq!(EIT_PF_ACTUAL_MAX_INTERVAL, Duration::from_secs(2));
563 assert_eq!(EIT_PF_OTHER_MAX_INTERVAL, Duration::from_secs(10));
564 assert_eq!(EIT_SCHED_MAX_INTERVAL, Duration::from_secs(10));
565 assert_eq!(EIT_SCHED_EXT_MAX_INTERVAL, Duration::from_secs(30));
566 assert_eq!(TDT_MAX_INTERVAL, Duration::from_secs(30));
567 assert_eq!(TOT_MAX_INTERVAL, Duration::from_secs(30));
568
569 assert_eq!(PAT_MAX_INTERVAL, Duration::from_millis(100));
571 assert_eq!(PMT_MAX_INTERVAL, Duration::from_millis(100));
572
573 assert_eq!(MIN_SECTION_INTERVAL, Duration::from_millis(25));
575 }
576
577 #[test]
580 fn new_simux_is_empty() {
581 let mut mux = SiMux::new();
582 let pkts = mux.poll(Duration::ZERO);
583 assert!(pkts.is_empty());
584 }
585
586 #[test]
587 fn simux_default_is_empty() {
588 let mut mux = SiMux::default();
589 let pkts = mux.poll(Duration::ZERO);
590 assert!(pkts.is_empty());
591 }
592
593 #[test]
594 fn first_poll_always_emits() {
595 let s = build_section(0x42, &[0xAA; 10]);
596 let mut mux = SiMux::new();
597 mux.upsert(0x0100, s.clone(), Duration::from_millis(100));
598 let pkts = mux.poll(Duration::ZERO);
599 assert!(!pkts.is_empty(), "first poll must emit");
600 }
601
602 #[test]
603 fn first_poll_emits_all_entries() {
604 let s1 = build_section(0x42, &[0x01; 5]);
605 let s2 = build_section(0x46, &[0x02; 5]);
606 let mut mux = SiMux::new();
607 mux.upsert(0x0100, s1.clone(), Duration::from_millis(100));
608 mux.upsert(0x0200, s2.clone(), Duration::from_millis(200));
609 let pkts = mux.poll(Duration::ZERO);
610 let pids: Vec<u16> = pkts
612 .iter()
613 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
614 .collect();
615 assert!(pids.contains(&0x0100));
616 assert!(pids.contains(&0x0200));
617 }
618
619 #[test]
622 fn entry_emits_only_when_due() {
623 let s = build_section(0x42, &[0xAA; 10]);
624 let mut mux = SiMux::new();
625 mux.upsert(0x0100, s.clone(), Duration::from_millis(100));
626
627 let pkts0 = mux.poll(Duration::ZERO);
629 assert!(!pkts0.is_empty());
630
631 let pkts50 = mux.poll(Duration::from_millis(50));
633 assert!(pkts50.is_empty(), "should not emit at t=50");
634
635 let pkts99 = mux.poll(Duration::from_millis(99));
637 assert!(pkts99.is_empty(), "should not emit at t=99");
638
639 let pkts100 = mux.poll(Duration::from_millis(100));
641 assert!(!pkts100.is_empty(), "should emit at t=100");
642 }
643
644 #[test]
645 fn two_entries_different_cadence() {
646 let s1 = build_section(0x42, &[0x01; 5]);
647 let s2 = build_section(0x46, &[0x02; 5]);
648 let mut mux = SiMux::new();
649 mux.upsert(0x0100, concat_sections(&[s1]), Duration::from_millis(100));
650 mux.upsert(0x0200, concat_sections(&[s2]), Duration::from_millis(200));
651
652 let pkts = mux.poll(Duration::ZERO);
654 assert!(!pkts.is_empty());
655
656 let pkts100 = mux.poll(Duration::from_millis(100));
658 assert!(!pkts100.is_empty());
659 let pids100: Vec<u16> = pkts100
660 .iter()
661 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
662 .collect();
663 assert!(pids100.contains(&0x0100), "PID 0x0100 should emit at t=100");
664 assert!(
665 !pids100.contains(&0x0200),
666 "PID 0x0200 should NOT emit at t=100"
667 );
668
669 let pkts150 = mux.poll(Duration::from_millis(150));
671 assert!(pkts150.is_empty(), "neither should emit at t=150");
672
673 let pkts200 = mux.poll(Duration::from_millis(200));
675 let pids200: Vec<u16> = pkts200
676 .iter()
677 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
678 .collect();
679 assert!(pids200.contains(&0x0100), "PID 0x0100 should emit at t=200");
680 assert!(pids200.contains(&0x0200), "PID 0x0200 should emit at t=200");
681 }
682
683 #[test]
684 fn entry_emits_again_after_full_interval() {
685 let s = build_section(0x42, &[0xAA; 20]);
686 let mut mux = SiMux::new();
687 mux.upsert(0x0100, s, Duration::from_secs(1));
688
689 let n0 = mux.poll_into(Duration::ZERO, &mut Vec::new());
691 assert!(n0 > 0);
692
693 let mut buf = Vec::new();
695 let n999 = mux.poll_into(Duration::from_millis(999), &mut buf);
696 assert_eq!(n999, 0);
697
698 let mut buf2 = Vec::new();
700 let n1000 = mux.poll_into(Duration::from_millis(1000), &mut buf2);
701 assert!(n1000 > 0);
702 }
703
704 #[test]
707 fn upsert_updates_existing_entry() {
708 let s1 = build_section(0x42, &[0xAA; 5]);
711 let s2 = build_section(0x42, &[0xBB; 10]);
712 let mut mux = SiMux::new();
713
714 mux.upsert(0x0100, s1, Duration::from_millis(100));
715 mux.upsert(0x0100, s2.clone(), Duration::from_millis(200));
716
717 let pkts = mux.poll(Duration::ZERO);
718
719 let mut reasm = SectionReassembler::default();
721 for raw in &pkts {
722 let pkt = TsPacket::parse(raw).unwrap();
723 if pkt.header.pid == 0x0100 {
724 reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
725 }
726 }
727 let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
728 assert_eq!(got.len(), 1);
729 assert_eq!(got[0].as_ref(), s2.as_slice());
730 }
731
732 #[test]
739 fn upsert_tdt_then_tot_both_emitted_with_continuous_cc() {
740 let tdt = build_section(0x70, &[0xAA; 5]); let tot = build_section(0x73, &[0xBB; 8]); let mut mux = SiMux::new();
743
744 mux.upsert_tdt(tdt.clone());
745 mux.upsert_tot(tot.clone());
746
747 let pkts = mux.poll(Duration::ZERO);
748 assert!(!pkts.is_empty());
749
750 let mut table_ids_seen = Vec::new();
751 let mut reasm = SectionReassembler::default();
752 let mut prev_cc: Option<u8> = None;
753 for raw in &pkts {
754 let pkt = TsPacket::parse(raw).unwrap();
755 assert_eq!(pkt.header.pid, well_known::TDT_TOT.value());
756 if let Some(prev) = prev_cc {
757 assert_eq!(
758 pkt.header.continuity_counter,
759 (prev + 1) & CC_MASK,
760 "continuity counter must be continuous across table kinds on one PID"
761 );
762 }
763 prev_cc = Some(pkt.header.continuity_counter);
764 reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
765 }
766 while let Some(section) = reasm.pop_section() {
767 table_ids_seen.push(section.as_ref()[0]);
768 }
769
770 assert!(table_ids_seen.contains(&0x70), "TDT missing from output");
771 assert!(table_ids_seen.contains(&0x73), "TOT missing from output");
772 }
773
774 #[test]
777 fn upsert_sdt_and_bat_both_emitted() {
778 let sdt = build_section(0x42, &[0xCC; 5]);
779 let bat = build_section(0x4A, &[0xDD; 6]);
780 let mut mux = SiMux::new();
781
782 mux.upsert_sdt_actual(sdt.clone());
783 mux.upsert(
784 well_known::SDT_BAT.value(),
785 bat.clone(),
786 Duration::from_millis(1000),
787 );
788
789 let pkts = mux.poll(Duration::ZERO);
790 let mut table_ids_seen = Vec::new();
791 let mut reasm = SectionReassembler::default();
792 for raw in &pkts {
793 let pkt = TsPacket::parse(raw).unwrap();
794 reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
795 }
796 while let Some(section) = reasm.pop_section() {
797 table_ids_seen.push(section.as_ref()[0]);
798 }
799
800 assert!(table_ids_seen.contains(&0x42), "SDT actual missing");
801 assert!(table_ids_seen.contains(&0x4A), "BAT missing");
802 }
803
804 #[test]
807 fn upsert_pat_uses_correct_pid() {
808 let s = build_section(0x00, &[0x01, 0x02]);
809 let mut mux = SiMux::new();
810 mux.upsert_pat(s);
811 let pkts = mux.poll(Duration::ZERO);
812 let pids: Vec<u16> = pkts
813 .iter()
814 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
815 .collect();
816 assert!(pids.iter().all(|&p| p == well_known::PAT.value()));
817 }
818
819 #[test]
820 fn upsert_sdt_actual_uses_correct_pid() {
821 let s = build_section(0x42, &[0x01]);
822 let mut mux = SiMux::new();
823 mux.upsert_sdt_actual(s);
824 let pkts = mux.poll(Duration::ZERO);
825 let pids: Vec<u16> = pkts
826 .iter()
827 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
828 .collect();
829 assert!(pids.iter().all(|&p| p == well_known::SDT_BAT.value()));
830 }
831
832 #[test]
833 fn upsert_nit_uses_correct_pid() {
834 let s = build_section(0x40, &[0x01]);
835 let mut mux = SiMux::new();
836 mux.upsert_nit(s);
837 let pkts = mux.poll(Duration::ZERO);
838 let pids: Vec<u16> = pkts
839 .iter()
840 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
841 .collect();
842 assert!(pids.iter().all(|&p| p == well_known::NIT.value()));
843 }
844
845 #[test]
846 fn upsert_tdt_uses_correct_pid() {
847 let s = build_section(0x70, &[0x01]);
848 let mut mux = SiMux::new();
849 mux.upsert_tdt(s);
850 let pkts = mux.poll(Duration::ZERO);
851 let pids: Vec<u16> = pkts
852 .iter()
853 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
854 .collect();
855 assert!(pids.iter().all(|&p| p == well_known::TDT_TOT.value()));
856 }
857
858 #[test]
859 fn upsert_tot_uses_correct_pid() {
860 let s = build_section(0x73, &[0x01]);
861 let mut mux = SiMux::new();
862 mux.upsert_tot(s);
863 let pkts = mux.poll(Duration::ZERO);
864 let pids: Vec<u16> = pkts
865 .iter()
866 .map(|raw| TsPacket::parse(raw).unwrap().header.pid)
867 .collect();
868 assert!(pids.iter().all(|&p| p == well_known::TDT_TOT.value()));
869 }
870
871 #[test]
874 fn simux_round_trip_single_entry() {
875 let s1 = build_section(0x42, &[0xAA; 20]);
876 let s2 = build_section(0x46, &[0xBB; 15]);
877 let mut mux = SiMux::new();
878 mux.upsert(
879 0x0100,
880 concat_sections(&[s1.clone(), s2.clone()]),
881 Duration::from_millis(100),
882 );
883
884 let pkts = mux.poll(Duration::ZERO);
885
886 let mut reasm = SectionReassembler::default();
887 for raw in &pkts {
888 let pkt = TsPacket::parse(raw).unwrap();
889 if pkt.header.pid == 0x0100 {
890 reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
891 }
892 }
893 let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
894 assert_eq!(got.len(), 2, "round-trip must recover both sections");
895 assert_eq!(got[0].as_ref(), s1.as_slice());
896 assert_eq!(got[1].as_ref(), s2.as_slice());
897 }
898
899 #[test]
900 fn simux_round_trip_multi_pid() {
901 let s_a = build_section(0x42, &[0xA0; 10]);
902 let s_b = build_section(0x46, &[0xB0; 10]);
903 let mut mux = SiMux::new();
904 mux.upsert(0x0100, s_a.clone(), Duration::from_millis(100));
905 mux.upsert(0x0200, s_b.clone(), Duration::from_millis(100));
906
907 let pkts = mux.poll(Duration::ZERO);
908
909 let mut reasm_a = SectionReassembler::default();
910 let mut reasm_b = SectionReassembler::default();
911 for raw in &pkts {
912 let pkt = TsPacket::parse(raw).unwrap();
913 match pkt.header.pid {
914 0x0100 => reasm_a.feed(pkt.payload.unwrap(), pkt.header.pusi),
915 0x0200 => reasm_b.feed(pkt.payload.unwrap(), pkt.header.pusi),
916 _ => {}
917 }
918 }
919 let got_a: Vec<_> = core::iter::from_fn(|| reasm_a.pop_section()).collect();
920 let got_b: Vec<_> = core::iter::from_fn(|| reasm_b.pop_section()).collect();
921 assert_eq!(got_a.len(), 1);
922 assert_eq!(got_b.len(), 1);
923 assert_eq!(got_a[0].as_ref(), s_a.as_slice());
924 assert_eq!(got_b[0].as_ref(), s_b.as_slice());
925 }
926
927 #[test]
930 fn continuity_counter_continuous_across_polls() {
931 let s = build_section(0x42, &[0xAA; 250]);
933 let mut mux = SiMux::new();
934 mux.upsert(0x0100, s, Duration::from_millis(100));
935
936 let pkts1 = mux.poll(Duration::ZERO);
937 assert!(pkts1.len() >= 2, "need ≥2 packets to test CC continuity");
938
939 let last_cc_pkts1 = TsPacket::parse(&pkts1[pkts1.len() - 1])
940 .unwrap()
941 .header
942 .continuity_counter;
943
944 let pkts2 = mux.poll(Duration::from_millis(100));
945 assert!(!pkts2.is_empty(), "second poll must emit");
946
947 let first_cc_pkts2 = TsPacket::parse(&pkts2[0])
948 .unwrap()
949 .header
950 .continuity_counter;
951
952 assert_eq!(
953 first_cc_pkts2,
954 (last_cc_pkts1 + 1) & 0x0F,
955 "CC must continue across poll cycles"
956 );
957 }
958
959 #[test]
962 fn poll_into_clears_out_before_appending() {
963 let s = build_section(0x42, &[0xAA; 5]);
964 let mut mux = SiMux::new();
965 mux.upsert(0x0100, s, Duration::from_millis(100));
966
967 let mut out = vec![[0u8; TS_PACKET_SIZE]; 42];
968 let n = mux.poll_into(Duration::ZERO, &mut out);
969 assert_eq!(n, out.len(), "out must contain only new packets");
970 }
971
972 #[test]
975 #[cfg(debug_assertions)]
976 #[should_panic(expected = "25 ms")]
977 fn upsert_rejects_sub_25ms_interval_in_debug() {
978 let s = build_section(0x42, &[0xAA; 5]);
979 let mut mux = SiMux::new();
980 mux.upsert(0x0100, s, Duration::from_millis(10));
981 }
982
983 #[test]
984 fn upsert_accepts_25ms_interval() {
985 let s = build_section(0x42, &[0xAA; 5]);
986 let mut mux = SiMux::new();
987 mux.upsert(0x0100, s, MIN_SECTION_INTERVAL);
988 let pkts = mux.poll(Duration::ZERO);
989 assert!(!pkts.is_empty());
990 }
991
992 #[test]
995 fn round_trip_single_short_section() {
996 let s = build_section(0x42, &[0xAA; 10]);
997 assert_round_trip(&[s]);
998 }
999
1000 #[test]
1001 fn round_trip_one_byte_body() {
1002 let s = build_section(0x46, &[0xBB]); assert_round_trip(&[s]);
1004 }
1005
1006 #[test]
1007 fn round_trip_section_exactly_pusi_cap_boundary() {
1008 let body = vec![0xCC; PUSI_PAYLOAD_CAP - 3];
1010 let s = build_section(0x50, &body);
1011 assert_eq!(s.len(), PUSI_PAYLOAD_CAP);
1012 assert_round_trip(&[s]);
1013 }
1014
1015 #[test]
1016 fn round_trip_section_just_over_pusi_cap() {
1017 let body = vec![0xDD; PUSI_PAYLOAD_CAP - 3 + 1];
1019 let s = build_section(0x52, &body);
1020 assert_eq!(s.len(), PUSI_PAYLOAD_CAP + 1);
1021 assert_round_trip(&[s]);
1022 }
1023
1024 #[test]
1025 fn round_trip_section_spans_many_packets() {
1026 let body = vec![0xEE; 2000 - 3];
1028 let s = build_section(0x60, &body);
1029 assert_round_trip(&[s]);
1030 }
1031
1032 #[test]
1033 fn round_trip_section_at_max_size() {
1034 let body = vec![0x11; 4096 - 3];
1039 let s = build_section(0x80, &body);
1040 assert_eq!(s.len(), 4096);
1041 assert_round_trip(&[s]);
1042 }
1043
1044 #[test]
1045 fn round_trip_multiple_short_sections_in_one_batch() {
1046 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]);
1050 }
1051
1052 #[test]
1053 fn round_trip_section_ends_exactly_at_boundary() {
1054 let body1 = vec![0xA1; PUSI_PAYLOAD_CAP - 3];
1058 let s1 = build_section(0x50, &body1);
1059 assert_eq!(s1.len(), PUSI_PAYLOAD_CAP);
1060
1061 let s2 = build_section(0x52, &[0xB1, 0xB2]);
1062 assert_round_trip(&[s1, s2]);
1063 }
1064
1065 #[test]
1066 fn continuation_packet_never_overshoots_next_section_start() {
1067 let body1 = vec![0xA5; 366 - 3];
1074 let s1 = build_section(0x50, &body1);
1075 assert_eq!(s1.len(), 366);
1076 let s2 = build_section(0x52, &[0xB1, 0xB2, 0xB3]);
1077
1078 let mut packetiser = SectionPacketiser::new(0x0100);
1079 let packets = packetiser.packetise(&[s1.as_slice(), s2.as_slice()]);
1080
1081 for pkt_raw in &packets {
1082 let pkt = TsPacket::parse(pkt_raw).expect("parse generated packet");
1083 let payload = pkt.payload.expect("payload present");
1084 if pkt.header.pusi {
1085 let pointer = payload[0] as usize;
1086 assert!(
1090 pointer < PUSI_PAYLOAD_CAP,
1091 "pointer_field {pointer} leaves no room for a section to \
1092 start within this packet (H.222.0 §2.4.4)"
1093 );
1094 }
1095 }
1096
1097 assert_round_trip(&[s1, s2]);
1098 }
1099
1100 #[test]
1101 fn round_trip_mix_small_large_sections() {
1102 let s1 = build_section(0x10, &[0xAA; 5]);
1105 let body2 = vec![0xBB; 200];
1106 let s2 = build_section(0x20, &body2);
1107 let s3 = build_section(0x30, &[0xCC; 50]);
1108 let body4 = vec![0xDD; 800];
1109 let s4 = build_section(0x40, &body4);
1110 let s5 = build_section(0x50, &[0xEE]); assert_round_trip(&[s1, s2, s3, s4, s5]);
1112 }
1113
1114 #[test]
1117 fn continuity_counter_increments_per_packet() {
1118 let body = vec![0xAA; 500];
1120 let section = build_section(0x42, &body);
1121 let mut p = SectionPacketiser::new(0x0100);
1122
1123 let packets = p.packetise(&[§ion]);
1124 assert!(packets.len() >= 3, "need multiple packets to test CC");
1125
1126 let mut last_cc: Option<u8> = None;
1127 for pkt_raw in &packets {
1128 let pkt = TsPacket::parse(pkt_raw).unwrap();
1129 let cc = pkt.header.continuity_counter;
1130 if let Some(last) = last_cc {
1131 assert_eq!(cc, (last + 1) & 0x0F, "CC must increment per packet");
1132 }
1133 last_cc = Some(cc);
1134 }
1135 }
1136
1137 #[test]
1138 fn continuity_counter_wraps_and_continues_across_calls() {
1139 let mut p = SectionPacketiser::with_continuity(0x0100, 14);
1140 let body = vec![0xBB; 500];
1142 let s = build_section(0x42, &body);
1143
1144 let pkts1 = p.packetise(&[&s]);
1146 assert!(pkts1.len() >= 3, "section must span ≥3 packets");
1147 let ccs1: Vec<u8> = pkts1
1148 .iter()
1149 .map(|b| TsPacket::parse(b).unwrap().header.continuity_counter)
1150 .collect();
1151 assert_eq!(ccs1[0], 14);
1152 assert_eq!(ccs1[1], 15);
1153 assert_eq!(ccs1[2], 0);
1154
1155 let pkts2 = p.packetise(&[&s]);
1157 let cc_first_pkt2 = TsPacket::parse(&pkts2[0])
1158 .unwrap()
1159 .header
1160 .continuity_counter;
1161 assert_eq!(cc_first_pkt2, ccs1.last().map(|c| (c + 1) & 0x0F).unwrap());
1162 }
1163
1164 #[test]
1167 fn pusi_set_when_section_starts() {
1168 let s = build_section(0x42, &[0xAA; 10]);
1169 let mut p = SectionPacketiser::new(0x0100);
1170 let packets = p.packetise(&[&s]);
1171 assert!(!packets.is_empty());
1172 let pkt = TsPacket::parse(&packets[0]).unwrap();
1173 assert!(pkt.header.pusi, "first packet must have PUSI=1");
1174 }
1175
1176 #[test]
1177 fn pusi_not_set_on_mid_section_continuation() {
1178 let body = vec![0xAA; 500];
1179 let s = build_section(0x42, &body);
1180 let mut p = SectionPacketiser::new(0x0100);
1181 let packets = p.packetise(&[&s]);
1182 assert!(packets.len() >= 2);
1183 let pkt1 = TsPacket::parse(&packets[0]).unwrap();
1184 let pkt2 = TsPacket::parse(&packets[1]).unwrap();
1185 assert!(pkt1.header.pusi, "first packet must have PUSI=1");
1186 assert!(
1187 !pkt2.header.pusi,
1188 "second packet is continuation, must have PUSI=0"
1189 );
1190 }
1191
1192 #[test]
1193 fn pointer_field_equals_tail_length_before_new_section() {
1194 let body1 = vec![0xA1; 197]; let s1 = build_section(0x52, &body1);
1199 assert_eq!(s1.len(), 200);
1200 let s2 = build_section(0x54, &[0xB1; 47]); assert_eq!(s2.len(), 50);
1202
1203 let mut p = SectionPacketiser::new(0x0100);
1204 let packets = p.packetise(&[&s1, &s2]);
1205
1206 let pkt_with_pointer = packets
1208 .iter()
1209 .map(|raw| TsPacket::parse(raw).unwrap())
1210 .find(|pkt| pkt.header.pusi && pkt.payload.is_some_and(|pl| pl.first() != Some(&0)))
1211 .expect("must have a PUSI packet with non-zero pointer");
1212
1213 let payload = pkt_with_pointer.payload.unwrap();
1214 let pointer = payload[0] as usize;
1215 assert!(pointer > 0, "pointer must be non-zero");
1216 let tail_start = s1.len() - pointer;
1218 assert_eq!(&payload[1..1 + pointer], &s1[tail_start..]);
1219 }
1220
1221 #[test]
1224 fn final_packet_unused_tail_is_stuffing() {
1225 let s = build_section(0x42, &[0xAA; 5]); let mut p = SectionPacketiser::new(0x0100);
1227 let packets = p.packetise(&[&s]);
1228
1229 let pkt = TsPacket::parse(&packets[0]).unwrap();
1230 let payload = pkt.payload.unwrap();
1231 assert_eq!(payload[0], 0, "pointer_field should be 0");
1232
1233 let section_end = 1 + s.len(); assert!(
1235 section_end < payload.len(),
1236 "must have stuffing after section"
1237 );
1238 for &b in &payload[section_end..] {
1239 assert_eq!(b, STUFFING_BYTE, "all trailing bytes must be 0xFF");
1240 }
1241 }
1242
1243 #[test]
1244 fn reassembler_discards_stuffing() {
1245 let s1 = build_section(0x42, &[0xAA; 10]);
1246 let s2 = build_section(0x46, &[0xBB; 5]);
1247
1248 let mut p = SectionPacketiser::new(0x0100);
1249 let packets = p.packetise(&[&s1, &s2]);
1250
1251 let mut reasm = SectionReassembler::default();
1252 for pkt_raw in &packets {
1253 let pkt = TsPacket::parse(pkt_raw).unwrap();
1254 reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
1255 }
1256
1257 let got: Vec<_> = core::iter::from_fn(|| reasm.pop_section()).collect();
1258 assert_eq!(got.len(), 2);
1259 assert!(
1260 reasm.is_empty(),
1261 "stuffing tail must be discarded, not buffered"
1262 );
1263 }
1264
1265 #[test]
1268 fn empty_batch_produces_no_packets() {
1269 let mut p = SectionPacketiser::new(0x0100);
1270 let packets: Vec<[u8; TS_PACKET_SIZE]> = p.packetise(&[]);
1271 assert!(packets.is_empty());
1272 }
1273
1274 #[test]
1275 fn packetise_into_clears_out_first() {
1276 let s = build_section(0x42, &[0xAA; 5]);
1277 let mut p = SectionPacketiser::new(0x0100);
1278
1279 let mut out = vec![[0u8; TS_PACKET_SIZE]; 99]; let n = p.packetise_into(&[&s], &mut out);
1281 assert_eq!(n, out.len(), "out must contain only the new packets");
1282 let mut reasm = SectionReassembler::default();
1284 for pkt_raw in &out {
1285 let pkt = TsPacket::parse(pkt_raw).unwrap();
1286 reasm.feed(pkt.payload.unwrap(), pkt.header.pusi);
1287 }
1288 let got = reasm.pop_section().unwrap();
1289 assert_eq!(got.as_ref(), s.as_slice());
1290 }
1291
1292 #[test]
1293 fn pid_is_correct() {
1294 let p = SectionPacketiser::new(0x1234);
1295 assert_eq!(p.pid(), 0x1234);
1296 }
1297
1298 #[test]
1299 fn with_continuity_masks_to_4_bits() {
1300 let p = SectionPacketiser::with_continuity(0x0100, 0xFE);
1301 assert_eq!(p.continuity_counter(), 0x0E);
1302 }
1303
1304 #[test]
1305 fn has_payload_always_true_no_adaptation() {
1306 let s = build_section(0x42, &[0xAA; 50]);
1307 let mut p = SectionPacketiser::new(0x0100);
1308 let packets = p.packetise(&[&s]);
1309 for pkt_raw in &packets {
1310 let pkt = TsPacket::parse(pkt_raw).unwrap();
1311 assert!(pkt.header.has_payload, "every packet must carry payload");
1312 assert!(!pkt.header.has_adaptation, "no adaptation field is emitted");
1313 assert!(!pkt.header.tei, "TEI must be false");
1314 assert_eq!(pkt.header.scrambling, 0, "scrambling must be 0");
1315 }
1316 }
1317}