dvb_stream/
t2mi_stream.rs

1//! [`T2miEventStream`] — async [`futures_core::Stream`] of owned T2-MI events.
2//!
3//! Wraps [`dvb_t2mi::pump::T2miPump`] over any [`tokio::io::AsyncRead`] source,
4//! yielding one [`dvb_t2mi::pump::T2miEvent`] per complete, CRC-valid T2-MI packet.
5//! Events own their bytes via `bytes::Bytes` and are `'static`, `Clone`, and
6//! `Send + Sync`.
7//!
8//! # Cancellation
9//!
10//! Dropping the `T2miEventStream` cancels cleanly — no internal tasks are
11//! spawned.
12
13use std::collections::VecDeque;
14use std::pin::Pin;
15use std::task::{Context, Poll};
16
17use dvb_t2mi::pump::{T2miEvent, T2miPump};
18use futures_core::Stream;
19use tokio::io::AsyncRead;
20use tokio::io::ReadBuf;
21
22use crate::resync::{aligned_packets, resync, TS_PACKET_SIZE};
23
24/// Read buffer size: 7 × 188 bytes (matches typical DVB UDP payload size).
25const READ_BUF_SIZE: usize = TS_PACKET_SIZE * 7;
26
27/// Async [`Stream`] of [`T2miEvent`]s from a raw TS byte source.
28///
29/// Feed any [`tokio::io::AsyncRead`] byte source and receive one `T2miEvent`
30/// per complete, CRC-valid T2-MI packet.
31///
32/// The adapter performs 188-byte TS packet alignment using the same resync
33/// logic as [`SectionStream`](crate::SectionStream) (see [`crate::resync`]).
34///
35/// # Cancellation
36///
37/// Drop the stream. No internal tasks are spawned.
38pub struct T2miEventStream<R> {
39    reader: R,
40    pump: T2miPump,
41    queue: VecDeque<T2miEvent>,
42    buf: Vec<u8>,
43    /// Carry-over bytes from the previous read (partial TS packet).
44    filled: usize,
45    /// Whether the reader has reached EOF.
46    eof: bool,
47    /// True once the stream has found an initial 0x47 sync byte.
48    synced: bool,
49}
50
51impl<R: AsyncRead + Unpin> T2miEventStream<R> {
52    /// Create a `T2miEventStream` from a TS-encapsulated source on `pid`.
53    ///
54    /// `pid` is the 13-bit T2-MI PID from the PMT (e.g. `0x0006`).
55    #[must_use]
56    pub fn new(reader: R, pid: u16) -> Self {
57        Self::with_pump(reader, T2miPump::new(pid))
58    }
59
60    /// Create a `T2miEventStream` with an already-constructed [`T2miPump`].
61    #[must_use]
62    pub fn with_pump(reader: R, pump: T2miPump) -> Self {
63        Self {
64            reader,
65            pump,
66            queue: VecDeque::new(),
67            buf: vec![0u8; READ_BUF_SIZE],
68            filled: 0,
69            eof: false,
70            synced: false,
71        }
72    }
73
74    /// Access the underlying pump statistics.
75    #[must_use]
76    pub fn stats(&self) -> dvb_t2mi::pump::Stats {
77        self.pump.stats()
78    }
79
80    /// Feed a completed read into the pump and push events into `queue`.
81    fn feed_buf(&mut self, data: &[u8]) {
82        let start = if self.synced {
83            0
84        } else {
85            match resync(data) {
86                Some(off) => {
87                    self.synced = true;
88                    off
89                }
90                None => return,
91            }
92        };
93
94        for pkt in aligned_packets(&data[start..]) {
95            for event in self.pump.feed_ts(pkt) {
96                self.queue.push_back(event);
97            }
98        }
99
100        let aligned_end = start + (data[start..].len() / TS_PACKET_SIZE) * TS_PACKET_SIZE;
101        let remainder = &data[aligned_end..];
102        if !remainder.is_empty() {
103            self.buf[..remainder.len()].copy_from_slice(remainder);
104            self.filled = remainder.len();
105        } else {
106            self.filled = 0;
107        }
108    }
109}
110
111impl<R: AsyncRead + Unpin> Stream for T2miEventStream<R> {
112    type Item = T2miEvent;
113
114    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
115        let this = self.get_mut();
116
117        loop {
118            if let Some(event) = this.queue.pop_front() {
119                return Poll::Ready(Some(event));
120            }
121
122            if this.eof {
123                return Poll::Ready(None);
124            }
125
126            let buf_len = this.buf.len();
127            let read_from = this.filled;
128            let mut read_buf = ReadBuf::new(&mut this.buf[read_from..buf_len]);
129
130            match Pin::new(&mut this.reader).poll_read(cx, &mut read_buf) {
131                Poll::Pending => return Poll::Pending,
132                Poll::Ready(Err(_)) => {
133                    this.eof = true;
134                    return Poll::Ready(None);
135                }
136                Poll::Ready(Ok(())) => {
137                    let n = read_buf.filled().len();
138                    if n == 0 {
139                        this.eof = true;
140                        return Poll::Ready(None);
141                    }
142                    let total = read_from + n;
143                    let data: Vec<u8> = this.buf[..total].to_vec();
144                    this.feed_buf(&data);
145                }
146            }
147        }
148    }
149}
150
151/// UDP/multicast convenience constructor — enabled by the `udp` feature.
152#[cfg(feature = "udp")]
153impl T2miEventStream<crate::section_stream::UdpReader> {
154    /// Bind a UDP socket to `bind_addr` and join `multicast_addr`.
155    ///
156    /// `pid` is the 13-bit T2-MI PID from the PMT.
157    ///
158    /// # Errors
159    ///
160    /// Returns a [`std::io::Error`] if binding or joining the multicast group
161    /// fails.
162    pub async fn bind_multicast(
163        bind_addr: std::net::SocketAddrV4,
164        multicast_addr: std::net::Ipv4Addr,
165        pid: u16,
166    ) -> std::io::Result<Self> {
167        use tokio::net::UdpSocket;
168        let socket = UdpSocket::bind(bind_addr).await?;
169        socket.join_multicast_v4(multicast_addr, *bind_addr.ip())?;
170        Ok(Self::new(crate::section_stream::UdpReader { socket }, pid))
171    }
172}
dvb_stream/t2mi_stream.rs

dvb_stream/
t2mi_stream.rs
