dvb_ci_runtime/
transport.rs

1//! TPDU transport layer — a sans-IO state machine for the single transport
2//! connection per CI slot (ETSI EN 50221 §A.4).
3//!
4//! Connection lifecycle (Figures 6/7): `Idle → Creating → Active`. The host
5//! sends `Create_T_C`; the module answers `C_T_C_Reply` (→ `Active`) or the
6//! host times out back to `Idle`. In `Active` the host **polls regularly** —
7//! per §A.4, a poll is a `T_Data_Last` with an empty data field — and, whenever
8//! a Status Byte reports Data-Available (DA), sends `T_RCV` to receive the
9//! queued message. Chained module messages arrive as `T_Data_More*` then a
10//! final `T_Data_Last` and are reassembled into one SPDU payload.
11//!
12//! Timing: EN 50221 mandates regular polling and a reply-timeout arc but does
13//! not fix the interval, so [`DEFAULT_POLL_INTERVAL`] / [`DEFAULT_REPLY_TIMEOUT`]
14//! are implementation-chosen defaults. All timing is expressed via the sans-IO
15//! [`Tick`](crate::event::Event::Tick)/timer model so it is deterministic and
16//! testable without a clock.
17
18use std::collections::VecDeque;
19use std::time::Duration;
20
21use broadcast_common::{Parse, Serialize};
22use dvb_ci::tpdu::{CommandTpdu, DataBlock, ResponseTpdu, SbValue, TcObject, create_t_c, tags};
23
24/// Length of a standalone/appended `T_SB` object: `tag · 0x02 · t_c_id · SB`.
25const SB_OBJECT_LEN: usize = 4;
26
27/// Parse a `T_SB` object (`0x80 0x02 t_c_id sb_value`) at the start of `bytes`.
28fn parse_sb(bytes: &[u8]) -> Option<(u8, SbValue)> {
29    if bytes.len() >= SB_OBJECT_LEN && bytes[0] == tags::SB && bytes[1] == 0x02 {
30        Some((bytes[2], SbValue(bytes[3])))
31    } else {
32        None
33    }
34}
35
36/// Conventional host poll interval (implementation-chosen; §A.4 mandates only
37/// "poll regularly").
38pub const DEFAULT_POLL_INTERVAL: Duration = Duration::from_millis(100);
39/// Conventional reply timeout for an expected `R_TPDU` (the §A.4 Figure 6/7
40/// "Timeout" arc; value implementation-chosen).
41pub const DEFAULT_REPLY_TIMEOUT: Duration = Duration::from_millis(1000);
42
43/// Transport connection state (EN 50221 §A.4, Figures 6/7).
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45#[non_exhaustive]
46pub enum TcState {
47    /// No connection; nothing sent.
48    Idle,
49    /// `Create_T_C` sent, awaiting `C_T_C_Reply`.
50    Creating,
51    /// Connection up; polling/exchanging data.
52    Active,
53}
54
55/// What the transport layer wants done after handling an input.
56#[derive(Debug, Default, Clone, PartialEq, Eq)]
57pub struct Out {
58    /// Link-layer TPDU frames to write to the device, in order.
59    pub writes: Vec<Vec<u8>>,
60    /// Fully-reassembled SPDU payloads to pass up to the session layer.
61    pub spdus: Vec<Vec<u8>>,
62    /// Requested delay until the next [`Tick`](crate::event::Event::Tick).
63    pub timer: Option<Duration>,
64    /// A transport error (e.g. reply timeout, unexpected tag).
65    pub error: Option<TransportError>,
66}
67
68/// Transport-layer errors.
69#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
70#[non_exhaustive]
71pub enum TransportError {
72    /// No `C_T_C_Reply` within the reply timeout (§A.4 timeout arc).
73    #[error("transport connection setup timed out")]
74    SetupTimeout,
75    /// A reply arrived for a different `t_c_id` than ours.
76    #[error("unexpected t_c_id {got} (expected {expected})")]
77    WrongTcId {
78        /// The `t_c_id` received.
79        got: u8,
80        /// Our connection's `t_c_id`.
81        expected: u8,
82    },
83    /// The module reported a `T_C_Error`.
84    #[error("module reported T_C_Error")]
85    ModuleError,
86    /// A frame could not be parsed as an `R_TPDU`.
87    #[error("malformed R_TPDU")]
88    Malformed,
89}
90
91/// The single transport connection for a slot.
92#[derive(Debug)]
93pub struct Transport {
94    tcid: u8,
95    state: TcState,
96    reassembly: Vec<u8>,
97    poll_interval: Duration,
98    reply_timeout: Duration,
99    /// Time accumulated since the last poll (drives the poll cadence).
100    since_poll: Duration,
101    /// Time accumulated since a command that expects a reply (drives the
102    /// reply timeout); `None` when not awaiting a reply. While `Some`, a host
103    /// C_TPDU is *in flight* (sent, module not yet answered) — the link is
104    /// half-duplex, so no further data block may be sent until it clears.
105    awaiting: Option<Duration>,
106    /// SPDUs queued to send, one `T_Data_Last` per module turn. EN 50221's link
107    /// is polled half-duplex: the host sends a single data block, then must wait
108    /// for the module's `T_SB` before sending the next. Sending two back-to-back
109    /// makes a real CAM drop the second (issue #337).
110    outbound: VecDeque<Vec<u8>>,
111}
112
113impl Default for Transport {
114    fn default() -> Self {
115        Self::new(1)
116    }
117}
118
119impl Transport {
120    /// New transport for `tcid` with default timings.
121    #[must_use]
122    pub fn new(tcid: u8) -> Self {
123        Self {
124            tcid,
125            state: TcState::Idle,
126            reassembly: Vec::new(),
127            poll_interval: DEFAULT_POLL_INTERVAL,
128            reply_timeout: DEFAULT_REPLY_TIMEOUT,
129            since_poll: Duration::ZERO,
130            awaiting: None,
131            outbound: VecDeque::new(),
132        }
133    }
134
135    /// Override the poll interval / reply timeout.
136    #[must_use]
137    pub fn with_timing(mut self, poll: Duration, reply: Duration) -> Self {
138        self.poll_interval = poll;
139        self.reply_timeout = reply;
140        self
141    }
142
143    /// Current connection state.
144    #[must_use]
145    pub fn state(&self) -> TcState {
146        self.state
147    }
148
149    fn cmd(&self, tag: u8, data: &[u8]) -> Vec<u8> {
150        let c = CommandTpdu {
151            tag,
152            t_c_id: self.tcid,
153            data,
154        };
155        let mut buf = vec![0u8; c.serialized_len()];
156        // serialize_into only fails on a too-small buffer; ours is exact.
157        let n = c.serialize_into(&mut buf).expect("exact buffer");
158        buf.truncate(n);
159        buf
160    }
161
162    fn poll_frame(&self) -> Vec<u8> {
163        // §A.4: poll == T_Data_Last with empty data.
164        self.cmd(tags::DATA_LAST, &[])
165    }
166
167    /// Open the connection: emit `Create_T_C` and arm the reply timeout.
168    pub fn init(&mut self) -> Out {
169        self.state = TcState::Creating;
170        self.awaiting = Some(Duration::ZERO);
171        let obj: TcObject = create_t_c(self.tcid);
172        Out {
173            writes: vec![obj.to_bytes()],
174            timer: Some(self.reply_timeout),
175            ..Out::default()
176        }
177    }
178
179    /// Queue an upper-layer SPDU to send (wrapped in a `T_Data_Last`). The block
180    /// is transmitted now if the link is free, else held until the in-flight
181    /// C_TPDU is answered — one data block per module turn (§A.4 half-duplex).
182    pub fn send_spdu(&mut self, spdu: &[u8]) -> Out {
183        if self.state != TcState::Active {
184            return Out::default();
185        }
186        self.outbound.push_back(spdu.to_vec());
187        self.flush()
188    }
189
190    /// Emit the next queued data block if the link is free (Active and no
191    /// C_TPDU in flight); otherwise nothing (it waits for the module's `T_SB`).
192    fn flush(&mut self) -> Out {
193        if self.state != TcState::Active || self.awaiting.is_some() {
194            return Out::default();
195        }
196        match self.outbound.pop_front() {
197            Some(spdu) => {
198                self.awaiting = Some(Duration::ZERO);
199                self.since_poll = Duration::ZERO;
200                Out {
201                    writes: vec![self.cmd(tags::DATA_LAST, &spdu)],
202                    timer: Some(self.poll_interval),
203                    ..Out::default()
204                }
205            }
206            None => Out::default(),
207        }
208    }
209
210    /// Advance logical time by `elapsed`: poll if due, or time out a pending
211    /// reply.
212    pub fn tick(&mut self, elapsed: Duration) -> Out {
213        match self.state {
214            TcState::Idle => Out::default(),
215            TcState::Creating => {
216                if let Some(w) = self.awaiting.as_mut() {
217                    *w += elapsed;
218                    if *w >= self.reply_timeout {
219                        self.state = TcState::Idle;
220                        self.awaiting = None;
221                        return Out {
222                            error: Some(TransportError::SetupTimeout),
223                            ..Out::default()
224                        };
225                    }
226                }
227                Out {
228                    timer: Some(self.reply_timeout),
229                    ..Out::default()
230                }
231            }
232            TcState::Active => {
233                self.since_poll += elapsed;
234                if self.since_poll >= self.poll_interval {
235                    self.since_poll = Duration::ZERO;
236                    // A queued data block goes out in preference to an empty
237                    // poll, but only when no C_TPDU is in flight.
238                    if self.awaiting.is_none() && !self.outbound.is_empty() {
239                        return self.flush();
240                    }
241                    self.awaiting = Some(Duration::ZERO);
242                    Out {
243                        writes: vec![self.poll_frame()],
244                        timer: Some(self.poll_interval),
245                        ..Out::default()
246                    }
247                } else {
248                    Out {
249                        timer: Some(self.poll_interval - self.since_poll),
250                        ..Out::default()
251                    }
252                }
253            }
254        }
255    }
256
257    /// Handle one link-layer frame read from the device.
258    ///
259    /// A module frame is a leading object (`C_T_C_Reply` / `T_Data_*` / …)
260    /// followed by an appended `T_SB`, or a standalone `T_SB` (the reply to a
261    /// poll with nothing queued). The `T_SB`'s DA bit drives whether the host
262    /// must `T_RCV` next.
263    pub fn on_frame(&mut self, frame: &[u8]) -> Out {
264        self.awaiting = None;
265        match frame.first().copied() {
266            // C_T_C_Reply (+ appended T_SB): connection becomes Active.
267            Some(tags::C_T_C_REPLY) => match TcObject::parse(frame) {
268                Ok(o) if o.t_c_id == self.tcid => {
269                    self.state = TcState::Active;
270                    self.since_poll = Duration::ZERO;
271                    let da = parse_sb(&frame[3..]).is_some_and(|(_, sb)| sb.data_available());
272                    self.after_status(da)
273                }
274                Ok(o) => self.wrong_tcid(o.t_c_id),
275                Err(_) => self.malformed(),
276            },
277            // Standalone T_SB — the reply to a poll.
278            Some(tags::SB) => match parse_sb(frame) {
279                Some((tcid, _)) if tcid != self.tcid => self.wrong_tcid(tcid),
280                Some((_, sb)) => self.after_status(sb.data_available()),
281                None => self.malformed(),
282            },
283            Some(tags::T_C_ERROR) => Out {
284                error: Some(TransportError::ModuleError),
285                ..Out::default()
286            },
287            Some(tags::DATA_LAST | tags::DATA_MORE) => self.on_data(frame),
288            _ => self.malformed(),
289        }
290    }
291
292    fn malformed(&self) -> Out {
293        Out {
294            error: Some(TransportError::Malformed),
295            ..Out::default()
296        }
297    }
298
299    fn wrong_tcid(&self, got: u8) -> Out {
300        Out {
301            error: Some(TransportError::WrongTcId {
302                got,
303                expected: self.tcid,
304            }),
305            ..Out::default()
306        }
307    }
308
309    /// React to a Status Byte: if DA, solicit the queued message with `T_RCV`;
310    /// otherwise resume the idle poll cadence.
311    fn after_status(&mut self, data_available: bool) -> Out {
312        if data_available {
313            self.awaiting = Some(Duration::ZERO);
314            Out {
315                writes: vec![self.cmd(tags::RCV, &[])],
316                ..Out::default()
317            }
318        } else {
319            // Module idle: its `T_SB` freed the link, so send the next queued
320            // data block if any (the #337 fix); otherwise resume polling.
321            if !self.outbound.is_empty() {
322                return self.flush();
323            }
324            self.since_poll = Duration::ZERO;
325            Out {
326                timer: Some(self.poll_interval),
327                ..Out::default()
328            }
329        }
330    }
331
332    fn on_data(&mut self, frame: &[u8]) -> Out {
333        let r = match ResponseTpdu::parse(frame) {
334            Ok(r) => r,
335            Err(_) => {
336                return Out {
337                    error: Some(TransportError::Malformed),
338                    ..Out::default()
339                };
340            }
341        };
342        if r.t_c_id != self.tcid {
343            return Out {
344                error: Some(TransportError::WrongTcId {
345                    got: r.t_c_id,
346                    expected: self.tcid,
347                }),
348                ..Out::default()
349            };
350        }
351        self.reassembly.extend_from_slice(r.data);
352        match r.block {
353            // More chained fragments: each waits for another T_RCV (§A.4 item 10).
354            Some(DataBlock::More) => {
355                self.awaiting = Some(Duration::ZERO);
356                Out {
357                    writes: vec![self.cmd(tags::RCV, &[])],
358                    ..Out::default()
359                }
360            }
361            // Last (or only) fragment: emit the reassembled SPDU, then let the
362            // appended Status Byte decide whether to receive another message.
363            _ => {
364                let mut out = self.after_status(r.sb_value.data_available());
365                if !self.reassembly.is_empty() {
366                    out.spdus.push(core::mem::take(&mut self.reassembly));
367                }
368                out
369            }
370        }
371    }
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377    use dvb_ci::tpdu::SbValue;
378
379    /// Build an R_TPDU frame (module→host) for tests.
380    fn r_tpdu(tag: u8, tcid: u8, data: &[u8], da: bool) -> Vec<u8> {
381        // tag, length_field(=1+data), tcid, data..., SB(0x80), len=2, tcid, sb_value
382        let mut v = vec![tag];
383        v.push((1 + data.len()) as u8);
384        v.push(tcid);
385        v.extend_from_slice(data);
386        v.extend_from_slice(&[tags::SB, 0x02, tcid, SbValue::new(da).0]);
387        v
388    }
389
390    #[test]
391    fn init_sends_create_tc_and_arms_timeout() {
392        let mut t = Transport::new(1);
393        let out = t.init();
394        assert_eq!(out.writes, vec![vec![tags::CREATE_T_C, 0x01, 0x01]]);
395        assert_eq!(t.state(), TcState::Creating);
396        assert_eq!(out.timer, Some(DEFAULT_REPLY_TIMEOUT));
397    }
398
399    #[test]
400    fn setup_times_out_to_idle() {
401        let mut t = Transport::new(1);
402        t.init();
403        let out = t.tick(DEFAULT_REPLY_TIMEOUT);
404        assert_eq!(out.error, Some(TransportError::SetupTimeout));
405        assert_eq!(t.state(), TcState::Idle);
406    }
407
408    #[test]
409    fn reply_activates_then_polls_on_interval() {
410        let mut t = Transport::new(1);
411        t.init();
412        let out = t.on_frame(&[tags::C_T_C_REPLY, 0x01, 0x01]);
413        assert_eq!(t.state(), TcState::Active);
414        assert!(out.error.is_none());
415        // Before the interval: no poll.
416        let early = t.tick(DEFAULT_POLL_INTERVAL / 2);
417        assert!(early.writes.is_empty());
418        // Crossing the interval: an empty T_Data_Last poll.
419        let due = t.tick(DEFAULT_POLL_INTERVAL);
420        assert_eq!(due.writes, vec![vec![tags::DATA_LAST, 0x01, 0x01]]);
421    }
422
423    #[test]
424    fn reassembles_more_then_last_into_one_spdu() {
425        let mut t = Transport::new(1);
426        t.init();
427        t.on_frame(&[tags::C_T_C_REPLY, 0x01, 0x01]);
428        // MORE: partial data, solicits RCV
429        let o1 = t.on_frame(&r_tpdu(tags::DATA_MORE, 1, &[0xAA, 0xBB], false));
430        assert!(o1.spdus.is_empty());
431        assert_eq!(o1.writes, vec![vec![tags::RCV, 0x01, 0x01]]);
432        // LAST: completes the SPDU
433        let o2 = t.on_frame(&r_tpdu(tags::DATA_LAST, 1, &[0xCC], false));
434        assert_eq!(o2.spdus, vec![vec![0xAA, 0xBB, 0xCC]]);
435    }
436
437    #[test]
438    fn data_available_triggers_rcv() {
439        let mut t = Transport::new(1);
440        t.init();
441        t.on_frame(&[tags::C_T_C_REPLY, 0x01, 0x01]);
442        // LAST with DA set → host must RCV the next queued message.
443        let o = t.on_frame(&r_tpdu(tags::DATA_LAST, 1, &[0x01], true));
444        assert_eq!(o.spdus, vec![vec![0x01]]);
445        assert_eq!(o.writes, vec![vec![tags::RCV, 0x01, 0x01]]);
446    }
447
448    #[test]
449    fn two_sends_serialize_one_block_per_module_turn() {
450        // #337: a real CAM drops a second T_Data_Last sent before it answers the
451        // first. Two send_spdu in one turn must emit only ONE write; the second
452        // goes out after the module's T_SB.
453        let mut t = Transport::new(1);
454        t.init();
455        t.on_frame(&[tags::C_T_C_REPLY, 0x01, 0x01]);
456
457        let first = t.send_spdu(&[0x92, 0x07]); // e.g. open_session_response
458        assert_eq!(first.writes.len(), 1);
459        assert_eq!(first.writes[0][0], tags::DATA_LAST);
460
461        // Queued while the first is in flight → no write yet.
462        let second = t.send_spdu(&[0x9F, 0x80, 0x10, 0x00]); // profile_enq
463        assert!(
464            second.writes.is_empty(),
465            "second block must wait for the SB"
466        );
467
468        // Module acknowledges with a standalone T_SB (data_available = 0).
469        let after_sb = t.on_frame(&[tags::SB, 0x02, 0x01, SbValue::new(false).0]);
470        assert_eq!(
471            after_sb.writes.len(),
472            1,
473            "second block flushes after the SB"
474        );
475        assert_eq!(after_sb.writes[0][0], tags::DATA_LAST);
476        // It carries the profile_enq payload.
477        assert!(
478            after_sb.writes[0]
479                .windows(4)
480                .any(|w| w == [0x9F, 0x80, 0x10, 0x00])
481        );
482    }
483
484    #[test]
485    fn wrong_tcid_is_flagged() {
486        let mut t = Transport::new(1);
487        t.init();
488        let o = t.on_frame(&[tags::C_T_C_REPLY, 0x01, 0x09]);
489        assert_eq!(
490            o.error,
491            Some(TransportError::WrongTcId {
492                got: 9,
493                expected: 1
494            })
495        );
496    }
497}
dvb_ci_runtime/transport.rs

dvb_ci_runtime/
transport.rs
