Skip to main content

dvb_t2mi/
pump.rs

1//! [`T2miPump`] — owning-[`Bytes`] feed-and-iterate T2-MI pump.
2//!
3//! Feed raw bytes (TS-encapsulated or bare T2-MI stream) in; get back an
4//! iterator of [`T2miEvent`]s — one per **CRC-valid** complete T2-MI packet.
5//! Lazy zero-copy: events own their [`bytes::Bytes`] slice and expose typed
6//! views ([`T2miEvent::header`], [`T2miEvent::payload`]) that borrow from it
7//! on demand.
8//!
9//! ```no_run
10//! use dvb_t2mi::pump::T2miPump;
11//! use dvb_t2mi::payload::AnyPayload;
12//!
13//! let mut pump = T2miPump::new(0x0006); // T2-MI PID from the PMT
14//! let ts_packet = [0u8; 188]; // a real TS packet from your source
15//! for event in pump.feed_ts(&ts_packet) {
16//!     if let Ok(AnyPayload::Bbframe(bb)) = event.payload() {
17//!         println!("BBFrame plp_id={}", bb.plp_id);
18//!     }
19//! }
20//! ```
21//!
22//! # CRC policy
23//!
24//! Every complete packet is validated against its 4-byte CRC-32 trailer
25//! (ETSI TS 102 773 Annex A / [`crate::crc::validate_crc`]) before being
26//! emitted.  Packets that fail CRC are silently dropped and counted in
27//! [`Stats::crc_failures`].  The caller never sees a corrupted packet.
28//!
29//! # TS header parsing
30//!
31//! [`T2miPump::feed_ts`] extracts the MPEG-TS payload in-place — sync byte
32//! 0x47, PID, PUSI flag, and adaptation-field skip per ISO/IEC 13818-1
33//! §2.4.3.2 — and passes it to [`crate::ts::PacketReassembler`].  No
34//! `dvb-si` dependency is introduced; the TS header reader is a private
35//! helper below.
36
37use alloc::vec::Vec;
38use bytes::Bytes;
39
40use crate::crc;
41use crate::packet::Header;
42use crate::payload::AnyPayload;
43use crate::ts::PacketReassembler;
44
45// ── TS header constants (ISO/IEC 13818-1 §2.4.3.2) ──────────────────────────
46
47/// TS sync byte.
48const TS_SYNC_BYTE: u8 = 0x47;
49/// Expected size of one MPEG-TS packet.
50const TS_PACKET_SIZE: usize = 188;
51/// Byte 1 bit 6 = PUSI (Payload Unit Start Indicator).
52const PUSI_MASK: u8 = 0x40;
53/// Byte 1 bits 4..=0 = PID upper 5 bits.
54const PID_MASK_HI: u8 = 0x1F;
55/// Byte 3 bit 5 = adaptation_field_control bit 1 (adaptation field present).
56const ADAPTATION_FLAG: u8 = 0x20;
57/// Byte 3 bit 4 = adaptation_field_control bit 0 (payload present).
58const PAYLOAD_FLAG: u8 = 0x10;
59
60/// Minimal result of TS header parsing needed by the pump.
61struct TsInfo {
62    pid: u16,
63    pusi: bool,
64    /// Byte offset within the 188-byte packet where the payload starts.
65    payload_start: usize,
66}
67
68/// Parse the 4-byte MPEG-TS header and skip any adaptation field.
69///
70/// Returns `None` when:
71/// - `buf` is shorter than [`TS_PACKET_SIZE`],
72/// - the sync byte is not `0x47`,
73/// - the payload-present flag is clear, or
74/// - the adaptation field length overflows the packet.
75///
76/// Citation: ISO/IEC 13818-1:2019 §2.4.3.2 (transport_packet header) and
77/// §2.4.3.5 (adaptation_field length).
78fn parse_ts_header(buf: &[u8]) -> Option<TsInfo> {
79    if buf.len() < TS_PACKET_SIZE || buf[0] != TS_SYNC_BYTE {
80        return None;
81    }
82    let b1 = buf[1];
83    let b3 = buf[3];
84
85    let pusi = (b1 & PUSI_MASK) != 0;
86    let pid = (((b1 & PID_MASK_HI) as u16) << 8) | (buf[2] as u16);
87    let has_adaptation = (b3 & ADAPTATION_FLAG) != 0;
88    let has_payload = (b3 & PAYLOAD_FLAG) != 0;
89
90    if !has_payload {
91        return None;
92    }
93
94    let mut cursor: usize = 4;
95    if has_adaptation {
96        let af_len = buf[cursor] as usize;
97        cursor += 1 + af_len;
98        if cursor > TS_PACKET_SIZE {
99            return None;
100        }
101    }
102
103    Some(TsInfo {
104        pid,
105        pusi,
106        payload_start: cursor,
107    })
108}
109
110// ── T2miEvent ─────────────────────────────────────────────────────────────────
111
112/// One complete, CRC-valid T2-MI packet. Owns its bytes — `'static`, cheap clone.
113///
114/// Only constructed after CRC-32 validation (ETSI TS 102 773 Annex A).
115/// [`T2miEvent::header`] and [`T2miEvent::payload`] are lazy: they borrow from
116/// the owned [`Bytes`] on demand.
117#[derive(Debug, Clone)]
118pub struct T2miEvent {
119    bytes: Bytes,
120}
121
122impl T2miEvent {
123    /// The full packet bytes (header + payload + CRC trailer).
124    #[must_use]
125    pub fn bytes(&self) -> &Bytes {
126        &self.bytes
127    }
128
129    /// The raw `packet_type` byte (byte 0 of the T2-MI header per §5.1).
130    ///
131    /// Never panics — events are only built for CRC-valid packets which are at
132    /// least `6` (header) + `4` (CRC) = 10 bytes.
133    #[must_use]
134    pub fn packet_type(&self) -> u8 {
135        self.bytes[0]
136    }
137
138    /// Parse the 6-byte T2-MI packet header (lazy, borrows this event's bytes).
139    ///
140    /// # Errors
141    ///
142    /// Propagates [`crate::Error`] from [`broadcast_common::Parse::parse`] on [`Header`].
143    pub fn header(&self) -> crate::Result<Header> {
144        use broadcast_common::Parse;
145        Header::parse(&self.bytes)
146    }
147
148    /// Extract the `packet_type` byte and payload slice from this event's
149    /// bytes — shared logic for [`payload`](Self::payload) and
150    /// [`payload_with`](Self::payload_with).
151    ///
152    /// Uses [`Header::raw_payload_bytes`] so that genuinely-private
153    /// `packet_type` values (not in [`PacketType`](crate::packet::PacketType))
154    /// are not rejected.  The packet is already CRC-validated by the pump.
155    fn payload_parts(&self) -> crate::Result<(u8, &[u8])> {
156        let payload_bytes = Header::raw_payload_bytes(&self.bytes)?;
157        let packet_type = self.bytes[0];
158        Ok((packet_type, payload_bytes))
159    }
160
161    /// Parse the payload by dispatching on `packet_type`.
162    ///
163    /// Extracts the payload slice via [`Header::raw_payload_bytes`] (no
164    /// `packet_type` enum conversion), then calls
165    /// [`AnyPayload::dispatch`].  Unrecognised packet types produce
166    /// [`AnyPayload::Unknown`] with the raw payload bytes.
167    ///
168    /// # Errors
169    ///
170    /// Returns [`crate::Error`] from extracting the payload slice or from the
171    /// typed payload parser.
172    pub fn payload(&self) -> crate::Result<AnyPayload<'_>> {
173        let (packet_type, payload_bytes) = self.payload_parts()?;
174        Ok(match AnyPayload::dispatch(packet_type, payload_bytes) {
175            Some(result) => result?,
176            None => AnyPayload::Unknown {
177                packet_type,
178                body: payload_bytes,
179            },
180        })
181    }
182
183    /// Parse the payload by dispatching on `packet_type`, preferring the
184    /// registry's custom parsers over the built-in dispatch.
185    ///
186    /// Like [`payload`](Self::payload), but calls
187    /// [`AnyPayload::dispatch_with`] so that runtime-registered custom
188    /// packet types are resolved to [`AnyPayload::Other`].  Unrecognised
189    /// packet types produce [`AnyPayload::Unknown`] with the raw payload
190    /// bytes, exactly as [`payload`](Self::payload) does.
191    ///
192    /// # Errors
193    ///
194    /// Returns [`crate::Error`] from extracting the payload slice or from the
195    /// typed payload parser (built-in or custom).
196    pub fn payload_with(
197        &self,
198        registry: &crate::payload::PayloadRegistry,
199    ) -> crate::Result<AnyPayload<'_>> {
200        let (packet_type, payload_bytes) = self.payload_parts()?;
201        Ok(
202            match AnyPayload::dispatch_with(registry, packet_type, payload_bytes) {
203                Some(result) => result?,
204                None => AnyPayload::Unknown {
205                    packet_type,
206                    body: payload_bytes,
207                },
208            },
209        )
210    }
211}
212
213// ── Stats ─────────────────────────────────────────────────────────────────────
214
215/// Accumulated pump statistics (monotonically growing across all `feed` calls).
216///
217/// New counter fields may be added in a future release; construction is via
218/// [`Default`] only.
219#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
220#[non_exhaustive]
221pub struct Stats {
222    /// TS packets fed via [`T2miPump::feed_ts`].
223    pub ts_packets: u64,
224    /// Complete T2-MI packets produced by the reassembler (pre-CRC check).
225    pub t2mi_packets: u64,
226    /// Packets dropped due to CRC-32 mismatch (ETSI TS 102 773 Annex A).
227    pub crc_failures: u64,
228    /// Malformed inputs: bad TS sync byte, truncated TS packet, overflowed
229    /// adaptation field, or `feed_ts` called on a raw-mode pump.
230    pub malformed_packets: u64,
231}
232
233// ── T2miPump ──────────────────────────────────────────────────────────────────
234
235/// Feed-and-iterate T2-MI pump.
236///
237/// Supports two operating modes:
238///
239/// - **TS-encapsulated** (most common): construct with [`T2miPump::new`],
240///   passing the 13-bit PID carrying T2-MI (from the PMT).  Feed 188-byte
241///   MPEG-TS packets with [`T2miPump::feed_ts`].  The pump filters by PID,
242///   strips the TS header per ISO/IEC 13818-1 §2.4.3.2, and forwards the
243///   payload to the internal [`PacketReassembler`] (ETSI TS 102 773 §6.1.1).
244///
245/// - **Raw** (un-encapsulated): construct with [`T2miPump::raw`].  Feed
246///   arbitrary byte slices with [`T2miPump::feed_raw`].  The pump buffers bytes
247///   and emits events once a full packet (determined by the header's
248///   `payload_len_bits`) is available.
249///
250/// # PID note
251///
252/// PIDs are 13-bit values (0x0000–0x1FFF per ISO/IEC 13818-1 §2.4.3.2).
253/// This type uses `u16` directly; no newtype is introduced.  Values above
254/// 0x1FFF are accepted without error — the PID filter simply never matches.
255pub struct T2miPump {
256    mode: PumpMode,
257    reasm: PacketReassembler,
258    stats: Stats,
259    scratch: Vec<T2miEvent>,
260    /// Raw-mode sync flag: true once the first raw feed has initialised the
261    /// reassembler via a PUSI=true, pointer=0 signal.
262    raw_started: bool,
263}
264
265enum PumpMode {
266    /// TS-encapsulated: filter packets to this PID.
267    Ts { pid: u16 },
268    /// Un-encapsulated raw byte stream.
269    Raw,
270}
271
272impl T2miPump {
273    /// Create a TS-encapsulated pump that filters to `pid`.
274    ///
275    /// `pid` is the 13-bit T2-MI PID from the PMT (e.g. 0x0006 for data
276    /// piping).
277    ///
278    /// # PID range
279    ///
280    /// Valid MPEG-TS PIDs are 13-bit (0x0000–0x1FFF); this parameter is `u16`.
281    /// No newtype is introduced to keep the API lightweight.
282    #[must_use]
283    pub fn new(pid: u16) -> Self {
284        Self {
285            mode: PumpMode::Ts { pid },
286            reasm: PacketReassembler::new(),
287            stats: Stats::default(),
288            scratch: Vec::new(),
289            raw_started: false,
290        }
291    }
292
293    /// Create an un-encapsulated raw-stream pump.
294    ///
295    /// Use [`T2miPump::feed_raw`] to supply bytes.  The pump buffers internally
296    /// and emits events by packet boundary, not by call boundary — a packet
297    /// split across two `feed_raw` calls produces exactly one event.
298    #[must_use]
299    pub fn raw() -> Self {
300        Self {
301            mode: PumpMode::Raw,
302            reasm: PacketReassembler::new(),
303            stats: Stats::default(),
304            scratch: Vec::new(),
305            raw_started: false,
306        }
307    }
308
309    /// Accumulated statistics.
310    #[must_use]
311    pub fn stats(&self) -> Stats {
312        self.stats
313    }
314
315    /// Feed one 188-byte MPEG-TS packet. Infallible: malformed packets are
316    /// counted in [`Stats::malformed_packets`] and discarded.
317    ///
318    /// Packets on the wrong PID are silently ignored (only [`Stats::ts_packets`]
319    /// is incremented).
320    ///
321    /// Returns a draining iterator over any T2-MI events completed by this feed.
322    pub fn feed_ts(&mut self, packet: &[u8]) -> impl Iterator<Item = T2miEvent> + '_ {
323        self.scratch.clear();
324
325        match self.mode {
326            PumpMode::Raw => {
327                // feed_ts on a raw-mode pump is a caller error.
328                self.stats.malformed_packets += 1;
329            }
330            PumpMode::Ts { pid: filter_pid } => {
331                self.stats.ts_packets += 1;
332                match parse_ts_header(packet) {
333                    None => {
334                        self.stats.malformed_packets += 1;
335                    }
336                    Some(info) => {
337                        if info.pid == filter_pid {
338                            let payload = &packet[info.payload_start..TS_PACKET_SIZE];
339                            self.reasm.feed(payload, info.pusi);
340                            Self::drain_reasm_into(
341                                &mut self.reasm,
342                                &mut self.stats,
343                                &mut self.scratch,
344                            );
345                        }
346                        // Wrong PID: ignored cheaply — no stats beyond ts_packets.
347                    }
348                }
349            }
350        }
351
352        self.scratch.drain(..)
353    }
354
355    /// Feed raw T2-MI bytes (un-encapsulated mode).
356    ///
357    /// The slice may contain a partial packet; bytes are buffered internally.
358    /// A packet split across two `feed_raw` calls produces exactly one event.
359    ///
360    /// Returns a draining iterator over any T2-MI events completed by this feed.
361    pub fn feed_raw(&mut self, data: &[u8]) -> impl Iterator<Item = T2miEvent> + '_ {
362        self.scratch.clear();
363
364        match self.mode {
365            PumpMode::Ts { .. } => {
366                // feed_raw on a TS-mode pump is a caller error.
367                self.stats.malformed_packets += 1;
368            }
369            PumpMode::Raw => {
370                if !self.raw_started {
371                    // First call: initialise the reassembler with PUSI=true and
372                    // pointer_field=0.  PacketReassembler::feed interprets the
373                    // first byte of the payload as the pointer_field when PUSI is
374                    // set (ETSI TS 102 773 §6.1.1).  We prepend a 0x00 byte so the
375                    // reassembler sees pointer=0 and treats the rest as the start
376                    // of a new T2-MI packet.
377                    let mut buf = Vec::with_capacity(1 + data.len());
378                    buf.push(0x00); // pointer_field = 0
379                    buf.extend_from_slice(data);
380                    self.reasm.feed(&buf, true);
381                    self.raw_started = true;
382                } else {
383                    // Continuation: feed without PUSI — bytes extend the
384                    // current T2-MI packet in progress.
385                    self.reasm.feed(data, false);
386                }
387                Self::drain_reasm_into(&mut self.reasm, &mut self.stats, &mut self.scratch);
388            }
389        }
390
391        self.scratch.drain(..)
392    }
393
394    /// Drain all pending packets from the reassembler, CRC-validate each one,
395    /// and push valid packets to `scratch`.
396    fn drain_reasm_into(
397        reasm: &mut PacketReassembler,
398        stats: &mut Stats,
399        scratch: &mut Vec<T2miEvent>,
400    ) {
401        for raw in reasm.drain_packets() {
402            stats.t2mi_packets += 1;
403            match crc::validate_crc(&raw) {
404                Ok(()) => scratch.push(T2miEvent { bytes: raw }),
405                Err(_) => stats.crc_failures += 1,
406            }
407        }
408    }
409}
410
411// ── Tests ─────────────────────────────────────────────────────────────────────
412
413#[cfg(test)]
414mod tests {
415    use super::*;
416    use broadcast_common::crc32_mpeg2;
417
418    // ── Test helpers ─────────────────────────────────────────────────────────
419
420    /// Build a syntactically valid T2-MI packet (header + payload + CRC-32).
421    ///
422    /// `packet_type` is the raw byte (Table 1 of TS 102 773).
423    /// `payload` is the post-header, pre-CRC data.
424    /// Returns the full byte vector including the 4-byte CRC trailer.
425    fn make_t2mi_packet(packet_type: u8, payload: &[u8]) -> Vec<u8> {
426        let payload_len_bits = (payload.len() * 8) as u16;
427        let mut pkt = Vec::with_capacity(6 + payload.len() + 4);
428        pkt.push(packet_type);
429        pkt.push(0x01); // packet_count
430        pkt.push(0x00); // superframe_idx=0, rfu=0, t2mi_stream_id=0
431        pkt.push(0x00); // rfu byte = 0
432        pkt.extend_from_slice(&payload_len_bits.to_be_bytes());
433        pkt.extend_from_slice(payload);
434        let crc = crc32_mpeg2::compute(&pkt);
435        pkt.extend_from_slice(&crc.to_be_bytes());
436        pkt
437    }
438
439    /// Wrap a T2-MI payload slice in a single 188-byte MPEG-TS packet.
440    ///
441    /// Sets PUSI=true and pointer_field=0 so the reassembler treats
442    /// the T2-MI data as starting at byte 0 of the payload.
443    /// The T2-MI bytes must fit in 183 bytes (188 − 4 header − 1 pointer).
444    fn ts_packet(pid: u16, t2mi_data: &[u8], pusi: bool, pointer_field: u8) -> [u8; 188] {
445        let mut pkt = [0xFFu8; 188];
446        pkt[0] = TS_SYNC_BYTE;
447        pkt[1] = if pusi { PUSI_MASK } else { 0 };
448        pkt[1] |= ((pid >> 8) as u8) & PID_MASK_HI;
449        pkt[2] = (pid & 0xFF) as u8;
450        pkt[3] = PAYLOAD_FLAG; // payload present, no adaptation field
451        if pusi {
452            pkt[4] = pointer_field;
453            let start = 5 + pointer_field as usize;
454            assert!(
455                start + t2mi_data.len() <= 188,
456                "T2-MI data too large for one TS packet"
457            );
458            pkt[start..start + t2mi_data.len()].copy_from_slice(t2mi_data);
459        } else {
460            let start = 4;
461            assert!(
462                start + t2mi_data.len() <= 188,
463                "T2-MI data too large for one TS packet"
464            );
465            pkt[start..start + t2mi_data.len()].copy_from_slice(t2mi_data);
466        }
467        pkt
468    }
469
470    // ── (a) valid T2-MI packet in TS → one event, typed payload ──────────────
471
472    #[test]
473    fn ts_packet_emits_one_event_with_typed_payload() {
474        // Build a valid BBFrame T2-MI packet.
475        // BbframePayload minimum: frame_idx(1) + plp_id(1) + flags(1) = 3 bytes.
476        let bbframe_payload = [0x01u8, 0x02, 0x00];
477        let t2mi = make_t2mi_packet(0x00, &bbframe_payload);
478
479        let pkt = ts_packet(0x0006, &t2mi, true, 0);
480        let mut pump = T2miPump::new(0x0006);
481        let events: Vec<_> = pump.feed_ts(&pkt).collect();
482
483        assert_eq!(events.len(), 1, "expected exactly one event");
484        assert_eq!(events[0].packet_type(), 0x00);
485
486        let payload = events[0].payload().expect("payload parse should succeed");
487        assert!(
488            matches!(payload, AnyPayload::Bbframe(_)),
489            "expected Bbframe, got {payload:?}"
490        );
491
492        let stats = pump.stats();
493        assert_eq!(stats.ts_packets, 1);
494        assert_eq!(stats.t2mi_packets, 1);
495        assert_eq!(stats.crc_failures, 0);
496        assert_eq!(stats.malformed_packets, 0);
497    }
498
499    // ── (b) corrupted CRC → zero events, crc_failures=1 ─────────────────────
500
501    #[test]
502    fn corrupted_crc_drops_packet_and_counts() {
503        let payload = [0x00u8, 0x00, 0x00]; // minimal BBFrame payload
504        let mut t2mi = make_t2mi_packet(0x00, &payload);
505        // Corrupt the last CRC byte.
506        *t2mi.last_mut().unwrap() ^= 0xFF;
507
508        let pkt = ts_packet(0x0006, &t2mi, true, 0);
509        let mut pump = T2miPump::new(0x0006);
510        let events: Vec<_> = pump.feed_ts(&pkt).collect();
511
512        assert_eq!(events.len(), 0, "corrupted packet must not emit");
513        let stats = pump.stats();
514        assert_eq!(stats.crc_failures, 1);
515        assert_eq!(stats.t2mi_packets, 1); // reassembler produced it, CRC gate dropped it
516    }
517
518    // ── (c) feed_raw with packet split across two calls → one event ──────────
519
520    #[test]
521    fn feed_raw_split_across_two_calls_emits_one_event() {
522        // Use a timestamp payload (11 bytes, all zeros), packet_type=0x20.
523        let ts_payload = [0x00u8; 11];
524        let t2mi = make_t2mi_packet(0x20, &ts_payload);
525
526        // Split at an arbitrary boundary (e.g. after the header).
527        let split = 6;
528        let first = &t2mi[..split];
529        let second = &t2mi[split..];
530
531        let mut pump = T2miPump::raw();
532
533        let ev1: Vec<_> = pump.feed_raw(first).collect();
534        assert_eq!(ev1.len(), 0, "no complete packet yet after first chunk");
535
536        let ev2: Vec<_> = pump.feed_raw(second).collect();
537        assert_eq!(
538            ev2.len(),
539            1,
540            "one event after second chunk completes the packet"
541        );
542
543        let stats = pump.stats();
544        assert_eq!(stats.t2mi_packets, 1);
545        assert_eq!(stats.crc_failures, 0);
546    }
547
548    // ── (d) garbage TS packet → malformed counted, no panic ──────────────────
549
550    #[test]
551    fn garbage_ts_packet_counted_no_panic() {
552        let mut pump = T2miPump::new(0x0006);
553        let garbage = [0x00u8; 188]; // bad sync byte
554        let events: Vec<_> = pump.feed_ts(&garbage).collect();
555        assert_eq!(events.len(), 0);
556        assert_eq!(pump.stats().malformed_packets, 1);
557        assert_eq!(pump.stats().ts_packets, 1);
558    }
559
560    // ── (e) wrong-PID TS packet → ignored cheaply ────────────────────────────
561
562    #[test]
563    fn wrong_pid_ts_packet_ignored() {
564        let payload = [0x00u8, 0x00, 0x00];
565        let t2mi = make_t2mi_packet(0x00, &payload);
566        let pkt = ts_packet(0x0100, &t2mi, true, 0); // PID 0x0100, pump listens on 0x0006
567
568        let mut pump = T2miPump::new(0x0006);
569        let events: Vec<_> = pump.feed_ts(&pkt).collect();
570
571        assert_eq!(events.len(), 0, "wrong-PID packet must not emit");
572        // ts_packets incremented, but nothing else moves.
573        let stats = pump.stats();
574        assert_eq!(stats.ts_packets, 1);
575        assert_eq!(stats.t2mi_packets, 0);
576        assert_eq!(stats.crc_failures, 0);
577        assert_eq!(stats.malformed_packets, 0);
578    }
579
580    // ── additional: header() lazy parse ──────────────────────────────────────
581
582    #[test]
583    fn event_header_lazy_parse_matches_packet_type() {
584        let payload = [0x00u8; 11]; // Timestamp payload
585        let t2mi = make_t2mi_packet(0x20, &payload);
586        let pkt = ts_packet(0x0010, &t2mi, true, 0);
587
588        let mut pump = T2miPump::new(0x0010);
589        let events: Vec<_> = pump.feed_ts(&pkt).collect();
590        assert_eq!(events.len(), 1);
591
592        let hdr = events[0].header().expect("header parse should succeed");
593        assert_eq!(hdr.packet_type as u8, 0x20);
594        assert_eq!(hdr.packet_count, 0x01);
595    }
596
597    // ── additional: stats() method ───────────────────────────────────────────
598
599    #[test]
600    fn stats_accumulate_across_feeds() {
601        let payload = [0x00u8, 0x00, 0x00];
602        let t2mi = make_t2mi_packet(0x00, &payload);
603        let pkt = ts_packet(0x0006, &t2mi, true, 0);
604
605        let mut pump = T2miPump::new(0x0006);
606        pump.feed_ts(&pkt).for_each(drop);
607        pump.feed_ts(&pkt).for_each(drop);
608
609        let stats = pump.stats();
610        assert_eq!(stats.ts_packets, 2);
611        // The reassembler resets on PUSI so we get 2 complete packets.
612        assert_eq!(stats.t2mi_packets, 2);
613    }
614
615    // ── payload_with registry seam ───────────────────────────────────────────
616
617    #[test]
618    fn payload_with_dispatches_custom_registered_type() {
619        use crate::payload::registry::PayloadRegistry;
620        use crate::traits::PayloadDef;
621        use broadcast_common::Parse;
622
623        #[derive(Debug)]
624        #[cfg_attr(feature = "serde", derive(serde::Serialize))]
625        struct TestPrivatePayload {
626            val: u8,
627        }
628
629        impl<'a> Parse<'a> for TestPrivatePayload {
630            type Error = crate::Error;
631            fn parse(bytes: &'a [u8]) -> crate::Result<Self> {
632                if bytes.is_empty() {
633                    return Err(crate::Error::BufferTooShort {
634                        need: 1,
635                        have: 0,
636                        what: "TestPrivatePayload",
637                    });
638                }
639                Ok(Self { val: bytes[0] })
640            }
641        }
642
643        impl PayloadDef<'_> for TestPrivatePayload {
644            const PACKET_TYPE: u8 = 0x00;
645            const NAME: &'static str = "TEST_PRIVATE";
646        }
647
648        let mut reg = PayloadRegistry::new();
649        reg.register::<TestPrivatePayload>();
650
651        let private_payload = [0x42u8, 0x02, 0x00];
652        let t2mi = make_t2mi_packet(0x00, &private_payload);
653        let pkt = ts_packet(0x0006, &t2mi, true, 0);
654
655        let mut pump = T2miPump::new(0x0006);
656        let events: Vec<_> = pump.feed_ts(&pkt).collect();
657        assert_eq!(events.len(), 1, "expected one event");
658
659        let result = events[0].payload_with(&reg).expect("payload_with parse");
660        match result {
661            AnyPayload::Other {
662                packet_type,
663                ref value,
664            } => {
665                assert_eq!(packet_type, 0x00);
666                let downcast = value.downcast_ref::<TestPrivatePayload>().unwrap();
667                assert_eq!(downcast.val, 0x42);
668            }
669            other => panic!("expected Other, got {other:?}"),
670        }
671
672        let built_in = events[0].payload().expect("payload parse");
673        assert!(
674            matches!(built_in, AnyPayload::Bbframe(_)),
675            "expected Bbframe via built-in dispatch, got {built_in:?}"
676        );
677    }
678
679    // ── payload_with with genuinely-private packet type (not in PacketType) ──
680
681    #[test]
682    fn payload_with_dispatches_genuinely_private_packet_type() {
683        use crate::payload::registry::PayloadRegistry;
684        use crate::traits::PayloadDef;
685        use broadcast_common::Parse;
686
687        #[derive(Debug)]
688        #[cfg_attr(feature = "serde", derive(serde::Serialize))]
689        struct PrivatePayload {
690            val: u8,
691        }
692
693        impl<'a> Parse<'a> for PrivatePayload {
694            type Error = crate::Error;
695            fn parse(bytes: &'a [u8]) -> crate::Result<Self> {
696                if bytes.is_empty() {
697                    return Err(crate::Error::BufferTooShort {
698                        need: 1,
699                        have: 0,
700                        what: "PrivatePayload",
701                    });
702                }
703                Ok(Self { val: bytes[0] })
704            }
705        }
706
707        impl PayloadDef<'_> for PrivatePayload {
708            const PACKET_TYPE: u8 = 0x42;
709            const NAME: &'static str = "PRIVATE_0X42";
710        }
711
712        let mut reg = PayloadRegistry::new();
713        reg.register::<PrivatePayload>();
714
715        let private_body = [0xABu8];
716        let t2mi = make_t2mi_packet(0x42, &private_body);
717        let pkt = ts_packet(0x0006, &t2mi, true, 0);
718
719        let mut pump = T2miPump::new(0x0006);
720        let events: Vec<_> = pump.feed_ts(&pkt).collect();
721        assert_eq!(events.len(), 1, "expected one event");
722
723        let result = events[0].payload_with(&reg).expect("payload_with parse");
724        match result {
725            AnyPayload::Other {
726                packet_type,
727                ref value,
728            } => {
729                assert_eq!(packet_type, 0x42);
730                let downcast = value.downcast_ref::<PrivatePayload>().unwrap();
731                assert_eq!(downcast.val, 0xAB);
732            }
733            other => panic!("expected Other, got {other:?}"),
734        }
735
736        let no_reg = events[0].payload().expect("payload without registry");
737        match no_reg {
738            AnyPayload::Unknown {
739                packet_type,
740                body: _,
741            } => {
742                assert_eq!(packet_type, 0x42);
743            }
744            other => panic!("expected Unknown, got {other:?}"),
745        }
746    }
747}