1use alloc::collections::BTreeMap;
16use alloc::sync::Arc;
17use alloc::vec;
18use alloc::vec::Vec;
19use mpeg_ts::pid::Pid;
20
21use crate::descriptors::{AnyDescriptor, DescriptorLoop, DescriptorRegistry};
22use broadcast_common::Parse;
23use mpeg_ts::section::Section;
24
25mod bat;
26mod eit;
27mod nit;
28mod sdt;
29
30pub use bat::*;
31pub use eit::*;
32pub use nit::*;
33pub use sdt::*;
34
35#[derive(Debug)]
37#[non_exhaustive]
38pub struct CompleteTransportStream<'a> {
39 pub transport_stream_id: u16,
41 pub original_network_id: u16,
43 pub descriptors: ParsedDescriptorLoop<'a>,
45}
46
47impl<'a> CompleteTransportStream<'a> {
48 pub(crate) fn collect<I>(entries: I, registry: Option<&'a DescriptorRegistry>) -> Vec<Self>
49 where
50 I: IntoIterator<Item = (u16, u16, DescriptorLoop<'a>)>,
51 {
52 entries
53 .into_iter()
54 .map(
55 |(transport_stream_id, original_network_id, descriptors)| Self {
56 transport_stream_id,
57 original_network_id,
58 descriptors: ParsedDescriptorLoop::parse(descriptors, registry),
59 },
60 )
61 .collect()
62 }
63}
64
65pub const DEFAULT_MAX_PARTIAL_KEYS: usize = 256;
78
79pub type CollectResult<T> = core::result::Result<T, CollectError>;
81
82#[derive(Debug, thiserror::Error, Clone, PartialEq, Eq)]
88#[non_exhaustive]
89pub enum CollectError {
90 #[error("section parse failed: {0}")]
92 Section(#[from] crate::Error),
93
94 #[error(
96 "table_id {table_id:#04x} is a short-form section and cannot be multi-section collected"
97 )]
98 ShortFormSection {
99 table_id: u8,
101 },
102
103 #[error(
105 "section_number {section_number} exceeds last_section_number {last_section_number} for table_id {table_id:#04x}"
106 )]
107 SectionNumberOutOfRange {
108 table_id: u8,
110 section_number: u8,
112 last_section_number: u8,
114 },
115
116 #[error("conflicting bytes for table_id {table_id:#04x} section {section_number}")]
118 ConflictingSection {
119 table_id: u8,
121 section_number: u8,
123 },
124
125 #[error(
127 "EIT schedule table_id {table_id:#04x} is outside advertised range {first_table_id:#04x}..={last_table_id:#04x}"
128 )]
129 EitTableIdOutOfRange {
130 table_id: u8,
132 first_table_id: u8,
134 last_table_id: u8,
136 },
137}
138
139impl From<mpeg_ts::Error> for CollectError {
140 fn from(e: mpeg_ts::Error) -> Self {
141 CollectError::Section(crate::Error::from(e))
142 }
143}
144
145#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
150#[non_exhaustive]
151pub struct SectionSetKey {
152 pub pid: Option<Pid>,
154 pub table_id: u8,
156 pub extension_id: u16,
158 pub current_next_indicator: bool,
160}
161
162#[derive(Debug, Clone, Copy, PartialEq, Eq)]
164#[non_exhaustive]
165pub struct SectionSetMeta {
166 pub key: SectionSetKey,
168 pub version_number: u8,
170 pub last_section_number: u8,
172}
173
174#[derive(Debug)]
175struct PartialSectionSet {
176 meta: SectionSetMeta,
177 slots: Vec<Option<Arc<[u8]>>>,
178 filled: usize,
179 emitted: bool,
180 last_touch: u64,
184}
185
186impl PartialSectionSet {
187 fn new(meta: SectionSetMeta) -> Self {
188 let len = meta.last_section_number as usize + 1;
189 Self {
190 meta,
191 slots: vec![None; len],
192 filled: 0,
193 emitted: false,
194 last_touch: 0,
195 }
196 }
197
198 fn reset(&mut self, meta: SectionSetMeta) {
199 let last_touch = self.last_touch;
200 *self = Self::new(meta);
201 self.last_touch = last_touch;
202 }
203
204 fn insert(&mut self, section_number: u8, bytes: &[u8]) -> CollectResult<bool> {
205 let index = section_number as usize;
206 if let Some(existing) = &self.slots[index] {
207 if existing.as_ref() == bytes {
208 return Ok(false);
209 }
210 return Err(CollectError::ConflictingSection {
211 table_id: self.meta.key.table_id,
212 section_number,
213 });
214 }
215
216 self.slots[index] = Some(Arc::from(bytes));
220 self.filled += 1;
221 self.emitted = false;
222 Ok(true)
223 }
224
225 fn complete(&self) -> bool {
226 self.filled == self.slots.len()
227 }
228
229 fn to_complete(&self) -> Option<CompleteSectionSet> {
230 if !self.complete() || self.emitted {
231 return None;
232 }
233
234 let sections = self
235 .slots
236 .iter()
237 .map(|slot| slot.as_ref().expect("complete set has no holes").clone())
238 .collect();
239 Some(CompleteSectionSet {
240 meta: self.meta,
241 sections,
242 })
243 }
244}
245
246#[derive(Debug)]
272pub struct SectionSetCollector {
273 partial: BTreeMap<SectionSetKey, PartialSectionSet>,
274 touch_order: BTreeMap<u64, SectionSetKey>,
276 tick: u64,
277 max_partial_keys: usize,
278 evicted_for_capacity: u64,
279}
280
281impl Default for SectionSetCollector {
282 fn default() -> Self {
283 Self {
284 partial: BTreeMap::new(),
285 touch_order: BTreeMap::new(),
286 tick: 0,
287 max_partial_keys: DEFAULT_MAX_PARTIAL_KEYS,
288 evicted_for_capacity: 0,
289 }
290 }
291}
292
293impl SectionSetCollector {
294 #[must_use]
297 pub fn new() -> Self {
298 Self::default()
299 }
300
301 #[must_use]
305 pub fn with_max_partial_keys(mut self, max_partial_keys: usize) -> Self {
306 self.max_partial_keys = max_partial_keys;
307 self
308 }
309
310 #[must_use]
315 pub fn evicted_for_capacity(&self) -> u64 {
316 self.evicted_for_capacity
317 }
318
319 fn touch_or_evict(&mut self, key: SectionSetKey) -> bool {
324 self.tick = self.tick.wrapping_add(1).max(1);
325 if !self.partial.contains_key(&key) && self.partial.len() >= self.max_partial_keys {
326 let oldest = self.touch_order.keys().next().copied();
327 match oldest {
328 Some(oldest_tick) => {
329 if let Some(oldest_key) = self.touch_order.remove(&oldest_tick) {
330 self.partial.remove(&oldest_key);
331 self.evicted_for_capacity += 1;
332 }
333 }
334 None => return false,
335 }
336 }
337 true
338 }
339
340 pub fn push_section(
350 &mut self,
351 bytes: impl AsRef<[u8]>,
352 ) -> CollectResult<Option<CompleteSectionSet>> {
353 self.push_section_with_pid(None, bytes)
354 }
355
356 pub fn push_section_with_pid(
361 &mut self,
362 pid: Option<Pid>,
363 bytes: impl AsRef<[u8]>,
364 ) -> CollectResult<Option<CompleteSectionSet>> {
365 let raw = bytes.as_ref();
366 let section = Section::parse(raw)?;
367 if !section.section_syntax_indicator {
368 return Err(CollectError::ShortFormSection {
369 table_id: section.table_id,
370 });
371 }
372 if section.section_number > section.last_section_number {
373 return Err(CollectError::SectionNumberOutOfRange {
374 table_id: section.table_id,
375 section_number: section.section_number,
376 last_section_number: section.last_section_number,
377 });
378 }
379 section.validate_crc(raw)?;
380
381 let key = SectionSetKey {
382 pid,
383 table_id: section.table_id,
384 extension_id: section.extension_id,
385 current_next_indicator: section.current_next_indicator,
386 };
387 let meta = SectionSetMeta {
388 key,
389 version_number: section.version_number,
390 last_section_number: section.last_section_number,
391 };
392 if !self.touch_or_evict(key) {
393 return Ok(None);
394 }
395 let now = self.tick;
396
397 let partial = self
398 .partial
399 .entry(key)
400 .or_insert_with(|| PartialSectionSet::new(meta));
401
402 if partial.meta.version_number != meta.version_number
403 || partial.meta.last_section_number != meta.last_section_number
404 {
405 partial.reset(meta);
406 }
407
408 self.touch_order.remove(&partial.last_touch);
412 partial.last_touch = now;
413 self.touch_order.insert(now, key);
414
415 partial.insert(section.section_number, raw)?;
416 let complete = partial.to_complete();
417 if complete.is_some() {
418 partial.emitted = true;
419 }
420 Ok(complete)
421 }
422
423 pub fn clear(&mut self) {
425 self.partial.clear();
426 self.touch_order.clear();
427 }
428
429 #[must_use]
431 pub fn len(&self) -> usize {
432 self.partial.len()
433 }
434
435 #[must_use]
437 pub fn is_empty(&self) -> bool {
438 self.partial.is_empty()
439 }
440}
441
442#[derive(Debug, Clone)]
445pub struct CompleteSectionSet {
446 meta: SectionSetMeta,
447 sections: Vec<Arc<[u8]>>,
448}
449
450#[derive(Debug)]
458pub struct CompleteTable<T> {
459 meta: SectionSetMeta,
460 sections: Vec<T>,
461}
462
463impl<T> CompleteTable<T> {
464 #[must_use]
466 pub const fn meta(&self) -> SectionSetMeta {
467 self.meta
468 }
469
470 #[must_use]
472 pub fn sections(&self) -> &[T] {
473 &self.sections
474 }
475
476 #[must_use]
478 pub fn into_sections(self) -> Vec<T> {
479 self.sections
480 }
481}
482
483impl CompleteSectionSet {
484 #[must_use]
486 pub const fn meta(&self) -> SectionSetMeta {
487 self.meta
488 }
489
490 #[must_use]
492 pub fn section_bytes(&self) -> impl ExactSizeIterator<Item = &[u8]> {
493 self.sections.iter().map(AsRef::as_ref)
494 }
495
496 pub fn parse_sections<'a, T>(&'a self) -> crate::Result<Vec<T>>
501 where
502 T: Parse<'a, Error = crate::Error>,
503 {
504 self.section_bytes().map(T::parse).collect()
505 }
506
507 pub fn table<'a, T>(&'a self) -> crate::Result<CompleteTable<T>>
512 where
513 T: Parse<'a, Error = crate::Error>,
514 {
515 Ok(CompleteTable {
516 meta: self.meta,
517 sections: self.parse_sections()?,
518 })
519 }
520
521 pub fn nit(&self) -> crate::Result<CompleteNit<'_>> {
523 CompleteNit::parse(self, None)
524 }
525
526 pub fn nit_with_registry<'a>(
528 &'a self,
529 registry: &'a DescriptorRegistry,
530 ) -> crate::Result<CompleteNit<'a>> {
531 CompleteNit::parse(self, Some(registry))
532 }
533
534 pub fn bat(&self) -> crate::Result<CompleteBat<'_>> {
536 CompleteBat::parse(self, None)
537 }
538
539 pub fn bat_with_registry<'a>(
541 &'a self,
542 registry: &'a DescriptorRegistry,
543 ) -> crate::Result<CompleteBat<'a>> {
544 CompleteBat::parse(self, Some(registry))
545 }
546
547 pub fn sdt(&self) -> crate::Result<CompleteSdt<'_>> {
549 CompleteSdt::parse(self, None)
550 }
551
552 pub fn sdt_with_registry<'a>(
554 &'a self,
555 registry: &'a DescriptorRegistry,
556 ) -> crate::Result<CompleteSdt<'a>> {
557 CompleteSdt::parse(self, Some(registry))
558 }
559
560 pub fn eit(&self) -> crate::Result<CompleteEit<'_>> {
562 CompleteEit::parse(self, None)
563 }
564
565 pub fn eit_with_registry<'a>(
567 &'a self,
568 registry: &'a DescriptorRegistry,
569 ) -> crate::Result<CompleteEit<'a>> {
570 CompleteEit::parse(self, Some(registry))
571 }
572}
573
574#[derive(Debug)]
584pub struct ParsedDescriptorLoop<'a> {
585 raw: Vec<DescriptorLoop<'a>>,
586 descriptors: Vec<crate::Result<AnyDescriptor<'a>>>,
587}
588
589impl<'a> ParsedDescriptorLoop<'a> {
590 pub(crate) fn parse(raw: DescriptorLoop<'a>, registry: Option<&'a DescriptorRegistry>) -> Self {
591 Self::parse_all(core::iter::once(raw), registry)
592 }
593
594 pub(crate) fn parse_all(
597 raws: impl IntoIterator<Item = DescriptorLoop<'a>>,
598 registry: Option<&'a DescriptorRegistry>,
599 ) -> Self {
600 let raw: Vec<DescriptorLoop<'a>> = raws.into_iter().collect();
601 let mut descriptors = Vec::new();
602 for loop_ in &raw {
603 match registry {
604 Some(registry) => descriptors.extend(registry.parse_loop(loop_.raw())),
605 None => descriptors.extend(loop_.iter()),
606 }
607 }
608 Self { raw, descriptors }
609 }
610
611 #[must_use]
617 pub fn raw(&self) -> &[DescriptorLoop<'a>] {
618 &self.raw
619 }
620
621 pub fn descriptors(&self) -> &[crate::Result<AnyDescriptor<'a>>] {
624 &self.descriptors
625 }
626}
627
628#[cfg(test)]
629mod tests {
630 use super::*;
631
632 const TEST_TABLE_ID: u8 = 0x42;
633
634 fn min_section(extension_id: u16) -> Vec<u8> {
635 let section_length: u16 = 9; let mut buf = vec![0u8; 12];
637 buf[0] = TEST_TABLE_ID;
638 buf[1] = 0xB0 | ((section_length >> 8) as u8 & 0x0F);
639 buf[2] = (section_length & 0xFF) as u8;
640 buf[3..5].copy_from_slice(&extension_id.to_be_bytes());
641 buf[5] = 0xC1;
642 buf[6] = 0;
643 buf[7] = 0;
644 let crc = broadcast_common::crc32_mpeg2::compute(&buf[..8]);
645 buf[8..12].copy_from_slice(&crc.to_be_bytes());
646 buf
647 }
648
649 #[test]
650 fn collect_single_section_is_complete() {
651 let mut c = SectionSetCollector::new();
652 let sec = min_section(0);
653 let result = c.push_section(&sec).unwrap();
654 assert!(result.is_some());
655 assert_eq!(c.len(), 1);
656 }
657
658 #[test]
665 fn new_key_beyond_cap_evicts_least_recently_touched_completed_entry() {
666 let mut c = SectionSetCollector::new().with_max_partial_keys(3);
667
668 for eid in 0..3u16 {
670 let sec = min_section(eid);
671 let result = c.push_section(&sec).unwrap();
672 assert!(
673 result.is_some(),
674 "single-section set for eid {eid} completes"
675 );
676 }
677 assert_eq!(c.len(), 3);
678 assert_eq!(c.evicted_for_capacity(), 0);
679
680 let sec4 = min_section(3);
683 let result = c.push_section(&sec4).unwrap();
684 assert!(
685 result.is_some(),
686 "new key beyond cap must evict, not be skipped"
687 );
688 assert_eq!(c.len(), 3, "cap stays honored, not just grown");
689 assert_eq!(c.evicted_for_capacity(), 1);
690
691 let sec0_again = min_section(0);
694 let result = c.push_section(&sec0_again).unwrap();
695 assert!(result.is_some(), "evicted key re-admits like a fresh one");
696
697 c.clear();
699 assert!(c.is_empty());
700 let result = c.push_section(&sec4).unwrap();
701 assert!(result.is_some());
702 assert_eq!(c.len(), 1);
703 }
704
705 #[test]
706 fn partial_keys_cap_does_not_skip_existing_key() {
707 let mut c = SectionSetCollector::new().with_max_partial_keys(1);
708
709 let sec0 = {
711 let mut buf = min_section(0xAB);
712 buf[7] = 1;
714 let crc = broadcast_common::crc32_mpeg2::compute(&buf[..8]);
716 buf[8..12].copy_from_slice(&crc.to_be_bytes());
717 buf
718 };
719 let result = c.push_section(&sec0).unwrap();
720 assert!(result.is_none(), "incomplete section set yields None");
721
722 let mut sec1 = min_section(0xAB);
725 sec1[6] = 1; sec1[7] = 1; let crc = broadcast_common::crc32_mpeg2::compute(&sec1[..8]);
728 sec1[8..12].copy_from_slice(&crc.to_be_bytes());
729
730 let result = c.push_section(&sec1).unwrap();
731 assert!(
732 result.is_some(),
733 "existing key must NOT be skipped when cap full"
734 );
735 assert_eq!(c.len(), 1);
736 }
737
738 fn long_section(table_id: u8, ext: u16, payload: &[u8]) -> Vec<u8> {
739 let section_length = (5 + payload.len() + 4) as u16;
740 let mut buf = vec![0u8; 3 + section_length as usize];
741 buf[0] = table_id;
742 buf[1] = 0xB0 | ((section_length >> 8) as u8 & 0x0F);
743 buf[2] = (section_length & 0xFF) as u8;
744 buf[3..5].copy_from_slice(&ext.to_be_bytes());
745 buf[5] = 0xC1;
746 buf[8..8 + payload.len()].copy_from_slice(payload);
747 let crc_off = buf.len() - 4;
748 let crc = broadcast_common::crc32_mpeg2::compute(&buf[..crc_off]);
749 buf[crc_off..].copy_from_slice(&crc.to_be_bytes());
750 buf
751 }
752
753 fn complete_set(section: &[u8]) -> CompleteSectionSet {
754 SectionSetCollector::new()
755 .push_section(section)
756 .unwrap()
757 .unwrap()
758 }
759
760 const LCN_DESCRIPTOR: [u8; 6] = [0x83, 4, 0x00, 0x01, 0xFC, 0x05];
761
762 fn only_descriptor<'a, 'b>(loop_: &'b ParsedDescriptorLoop<'a>) -> &'b AnyDescriptor<'a> {
763 assert_eq!(loop_.descriptors().len(), 1);
764 loop_.descriptors()[0].as_ref().unwrap()
765 }
766
767 fn lcn_registry() -> DescriptorRegistry {
768 let mut registry = DescriptorRegistry::new();
769 registry.with_logical_channel();
770 registry
771 }
772
773 #[test]
774 fn bat_with_registry_types_the_opted_in_private_descriptor() {
775 let mut payload = vec![0xF0, LCN_DESCRIPTOR.len() as u8];
776 payload.extend_from_slice(&LCN_DESCRIPTOR);
777 payload.extend_from_slice(&[0xF0, 0x00]);
778 let set = complete_set(&long_section(0x4A, 0x1234, &payload));
779 let registry = lcn_registry();
780 let with = set.bat_with_registry(®istry).unwrap();
781 assert_eq!(with.bouquet_id, 0x1234);
782 assert!(matches!(
783 only_descriptor(&with.bouquet_descriptors),
784 AnyDescriptor::LogicalChannel(_)
785 ));
786 let plain = set.bat().unwrap();
787 assert!(matches!(
788 only_descriptor(&plain.bouquet_descriptors),
789 AnyDescriptor::Unknown { tag: 0x83, .. }
790 ));
791 }
792
793 #[test]
794 fn sdt_with_registry_types_the_opted_in_private_descriptor() {
795 let mut payload = vec![
796 0,
797 1,
798 0xFF,
799 0x00,
800 0x07,
801 0xFC,
802 0x80,
803 LCN_DESCRIPTOR.len() as u8,
804 ];
805 payload.extend_from_slice(&LCN_DESCRIPTOR);
806 let set = complete_set(&long_section(0x42, 7, &payload));
807 let registry = lcn_registry();
808 let with = set.sdt_with_registry(®istry).unwrap();
809 assert_eq!(with.original_network_id, 1);
810 assert_eq!(with.services.len(), 1);
811 assert_eq!(with.services[0].service_id, 7);
812 assert!(matches!(
813 only_descriptor(&with.services[0].descriptors),
814 AnyDescriptor::LogicalChannel(_)
815 ));
816 let plain = set.sdt().unwrap();
817 assert!(matches!(
818 only_descriptor(&plain.services[0].descriptors),
819 AnyDescriptor::Unknown { tag: 0x83, .. }
820 ));
821 }
822
823 #[test]
824 fn eit_with_registry_types_the_opted_in_private_descriptor() {
825 let mut payload = vec![
826 0,
827 1,
828 0,
829 2,
830 0,
831 0x4E,
832 0x00,
833 0x09,
834 0xE4,
835 0x09,
836 0x12,
837 0x34,
838 0x56,
839 0x00,
840 0x01,
841 0x00,
842 0x80,
843 LCN_DESCRIPTOR.len() as u8,
844 ];
845 payload.extend_from_slice(&LCN_DESCRIPTOR);
846 let set = complete_set(&long_section(0x4E, 100, &payload));
847 let registry = lcn_registry();
848 let with = set.eit_with_registry(®istry).unwrap();
849 assert_eq!(with.service_id, 100);
850 assert_eq!(with.events.len(), 1);
851 assert_eq!(with.events[0].event_id, 9);
852 assert!(matches!(
853 only_descriptor(&with.events[0].descriptors),
854 AnyDescriptor::LogicalChannel(_)
855 ));
856 let plain = set.eit().unwrap();
857 assert!(matches!(
858 only_descriptor(&plain.events[0].descriptors),
859 AnyDescriptor::Unknown { tag: 0x83, .. }
860 ));
861 }
862
863 #[test]
864 fn into_sections_returns_every_parsed_section() {
865 let set = complete_set(&long_section(0x4A, 9, &[0xF0, 0, 0xF0, 0]));
866 let table = set.table::<crate::tables::bat::BatSection>().unwrap();
867 assert_eq!(table.sections().len(), 1);
868 let sections = table.into_sections();
869 assert_eq!(sections.len(), 1);
870 assert_eq!(sections[0].bouquet_id, 9);
871 }
872}