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<'a> PayloadDef<'a> 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(®).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<'a> PayloadDef<'a> 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(®).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}