zerodds_rtps/
receiver_state.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! Receiver state (DDSI-RTPS 2.5 §8.3.4 + §8.3.7.4).
4//!
5//! On receipt of an RTPS message the receiver keeps a state
6//! with:
7//!
8//! ```text
9//!   sourceVersion        — ProtocolVersion aus RTPS-Header
10//!   sourceVendorId       — VendorId aus RTPS-Header
11//!   sourceGuidPrefix     — GuidPrefix of the sender
12//!   destGuidPrefix       — GuidPrefix of the receiver itself
13//!   unicastReplyLocators
14//!   multicastReplyLocators
15//!   haveTimestamp        — true if InfoTimestamp/HE.W was seen
16//!   timestamp            — last seen sender wallclock
17//!   messageLength        — if declared by the HE L flag
18//!   messageChecksum      — if declared by the HE C field
19//!   parameters           — if declared by the HE P field
20//!   clockSkewDetected    — heuristic: |timestamp - now| over threshold
21//! ```
22//!
23//! Update triggers:
24//!
25//! - **InfoSource** (§8.3.8.9.4): sets
26//!   `sourceVersion`, `sourceVendorId`, `sourceGuidPrefix` to the values
27//!   given in the InfoSource submessage; `haveTimestamp = false`
28//!   and the reply-locator lists are reset to `LOCATOR_INVALID`.
29//! - **InfoTimestamp** (§8.3.8.5.4): sets `haveTimestamp = true`
30//!   or = false for `I-Flag = 1`, plus `timestamp = …`.
31//! - **HeaderExtension** (§8.3.7.4): combines several effects — the L
32//!   flag updates `messageLength`; the W flag sets
33//!   `haveTimestamp = true` + `timestamp`; the C flag updates
34//!   `messageChecksum`; the P flag updates `parameters`.
35//!
36//! The receiver state is short-lived per RTPS message: before each
37//! `decode_datagram` it is initialized to the default value plus `destGuidPrefix`.
38
39extern crate alloc;
40use alloc::vec::Vec;
41
42use crate::header::RtpsHeader;
43use crate::header_extension::{ChecksumValue, HeTimestamp, HeaderExtension};
44use crate::parameter_list::ParameterList;
45use crate::wire_types::{GuidPrefix, Locator, ProtocolVersion, VendorId};
46
47/// Receiver state per spec table §8.3.4 and update rules
48/// in §8.3.7.4.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct ReceiverState {
51    /// ProtocolVersion from the RTPS header (or overwritten by InfoSource).
52    pub source_version: ProtocolVersion,
53    /// VendorId from the RTPS header (or overwritten by InfoSource).
54    pub source_vendor_id: VendorId,
55    /// GuidPrefix of the sender (RTPS header or InfoSource).
56    pub source_guid_prefix: GuidPrefix,
57    /// GuidPrefix of the receiver (configuration value, fixed).
58    pub dest_guid_prefix: GuidPrefix,
59    /// `true` if the receiver has a sender timestamp.
60    pub have_timestamp: bool,
61    /// Last sender timestamp (valid if `have_timestamp`).
62    pub timestamp: HeTimestamp,
63    /// Set by HE.L — expected remaining length of the RTPS message.
64    pub message_length: Option<u32>,
65    /// Set by HE.C — expected checksum of the RTPS message.
66    pub message_checksum: ChecksumValue,
67    /// Set by HE.P — ParameterList from the HE.
68    pub parameters: Option<ParameterList>,
69    /// Reply locator lists (default `LOCATOR_INVALID` lists, overridable
70    /// by InfoReply).
71    pub unicast_reply_locator_list: Vec<Locator>,
72    /// Reply locator lists (default `LOCATOR_INVALID` lists, overridable
73    /// by InfoReply).
74    pub multicast_reply_locator_list: Vec<Locator>,
75    /// Heuristic flag: `|timestamp - now| > threshold`. Set by
76    /// `note_clock_skew`; the decode module provides only the
77    /// input data.
78    pub clock_skew_detected: bool,
79}
80
81impl ReceiverState {
82    /// Initial state before receiving a message: all fields at
83    /// spec defaults, `dest_guid_prefix` taken from the receiver config.
84    #[must_use]
85    pub fn new(dest_guid_prefix: GuidPrefix) -> Self {
86        Self {
87            source_version: ProtocolVersion::V2_5,
88            source_vendor_id: VendorId([0, 0]),
89            source_guid_prefix: GuidPrefix::from_bytes([0; 12]),
90            dest_guid_prefix,
91            have_timestamp: false,
92            timestamp: HeTimestamp::default(),
93            message_length: None,
94            message_checksum: ChecksumValue::None,
95            parameters: None,
96            unicast_reply_locator_list: Vec::new(),
97            multicast_reply_locator_list: Vec::new(),
98            clock_skew_detected: false,
99        }
100    }
101
102    /// Initializes from an `RtpsHeader` (Spec §8.3.4.1).
103    pub fn init_from_header(&mut self, header: &RtpsHeader) {
104        self.source_version = header.protocol_version;
105        self.source_vendor_id = header.vendor_id;
106        self.source_guid_prefix = header.guid_prefix;
107        // Reset reply locator lists + haveTimestamp:
108        self.unicast_reply_locator_list.clear();
109        self.multicast_reply_locator_list.clear();
110        self.have_timestamp = false;
111    }
112
113    /// Update from an InfoSource submessage (§8.3.8.9.4).
114    ///
115    /// > "An InfoSource Submessage MUST set the receiver's source
116    /// >  GuidPrefix, source ProtocolVersion, source VendorId, and MUST
117    /// >  reset haveTimestamp = false and the reply-locator-lists to
118    /// >  LOCATOR_INVALID."
119    pub fn apply_info_source(
120        &mut self,
121        version: ProtocolVersion,
122        vendor_id: VendorId,
123        guid_prefix: GuidPrefix,
124    ) {
125        self.source_version = version;
126        self.source_vendor_id = vendor_id;
127        self.source_guid_prefix = guid_prefix;
128        self.have_timestamp = false;
129        self.unicast_reply_locator_list.clear();
130        self.multicast_reply_locator_list.clear();
131    }
132
133    /// Update from InfoTimestamp (§8.3.8.5.4). `invalidate = true` (i.e.
134    /// the I flag in the submessage) clears the timestamp.
135    pub fn apply_info_timestamp(&mut self, ts: HeTimestamp, invalidate: bool) {
136        if invalidate {
137            self.have_timestamp = false;
138        } else {
139            self.have_timestamp = true;
140            self.timestamp = ts;
141        }
142    }
143
144    /// Update from InfoReply (§8.3.8.10.4): sets the two reply
145    /// locator lists.
146    pub fn apply_info_reply(&mut self, unicast: Vec<Locator>, multicast: Option<Vec<Locator>>) {
147        self.unicast_reply_locator_list = unicast;
148        if let Some(m) = multicast {
149            self.multicast_reply_locator_list = m;
150        }
151    }
152
153    /// Update from HeaderExtension (§8.3.7.4). Updates `messageLength`,
154    /// `timestamp`, `messageChecksum` and `parameters` depending on the
155    /// set flags.
156    pub fn apply_header_extension(&mut self, he: &HeaderExtension) {
157        if let Some(len) = he.message_length {
158            self.message_length = Some(len);
159        }
160        if let Some(ts) = he.timestamp {
161            self.have_timestamp = true;
162            self.timestamp = ts;
163        }
164        if !matches!(he.checksum, ChecksumValue::None) {
165            self.message_checksum = he.checksum.clone();
166        }
167        if let Some(pl) = &he.parameters {
168            self.parameters = Some(pl.clone());
169        }
170    }
171
172    /// Sets the `clock_skew_detected` flag if the given
173    /// `now` seconds value deviates from the sender timestamp by more than
174    /// `threshold_seconds`. No-op if `!have_timestamp`.
175    pub fn note_clock_skew(&mut self, now_seconds: i32, threshold_seconds: u32) {
176        if !self.have_timestamp {
177            return;
178        }
179        let diff = (now_seconds as i64).saturating_sub(self.timestamp.seconds as i64);
180        if diff.unsigned_abs() > u64::from(threshold_seconds) {
181            self.clock_skew_detected = true;
182        }
183    }
184}
185
186#[cfg(test)]
187mod tests {
188    #![allow(clippy::expect_used, clippy::unwrap_used)]
189    use super::*;
190    use crate::header_extension::ChecksumValue;
191    use alloc::vec;
192
193    fn dummy_prefix(byte: u8) -> GuidPrefix {
194        GuidPrefix::from_bytes([byte; 12])
195    }
196
197    #[test]
198    fn new_state_has_default_fields() {
199        let st = ReceiverState::new(dummy_prefix(7));
200        assert!(!st.have_timestamp);
201        assert_eq!(st.dest_guid_prefix, dummy_prefix(7));
202        assert!(matches!(st.message_checksum, ChecksumValue::None));
203        assert!(st.message_length.is_none());
204        assert!(!st.clock_skew_detected);
205    }
206
207    #[test]
208    fn init_from_header_overrides_source_fields() {
209        let mut st = ReceiverState::new(dummy_prefix(0));
210        let h = RtpsHeader::new(VendorId::ZERODDS, dummy_prefix(0xAB));
211        st.init_from_header(&h);
212        assert_eq!(st.source_vendor_id, VendorId::ZERODDS);
213        assert_eq!(st.source_guid_prefix, dummy_prefix(0xAB));
214    }
215
216    #[test]
217    fn apply_info_source_resets_reply_locators_and_timestamp() {
218        let mut st = ReceiverState::new(dummy_prefix(0));
219        st.have_timestamp = true;
220        st.unicast_reply_locator_list.push(Locator::INVALID);
221        st.apply_info_source(
222            ProtocolVersion { major: 2, minor: 5 },
223            VendorId([0x42, 0x42]),
224            dummy_prefix(0x99),
225        );
226        assert_eq!(st.source_version, ProtocolVersion { major: 2, minor: 5 });
227        assert_eq!(st.source_vendor_id, VendorId([0x42, 0x42]));
228        assert_eq!(st.source_guid_prefix, dummy_prefix(0x99));
229        assert!(!st.have_timestamp);
230        assert!(st.unicast_reply_locator_list.is_empty());
231    }
232
233    #[test]
234    fn apply_info_timestamp_sets_value() {
235        let mut st = ReceiverState::new(dummy_prefix(0));
236        st.apply_info_timestamp(
237            HeTimestamp {
238                seconds: 100,
239                fraction: 200,
240            },
241            false,
242        );
243        assert!(st.have_timestamp);
244        assert_eq!(st.timestamp.seconds, 100);
245        assert_eq!(st.timestamp.fraction, 200);
246    }
247
248    #[test]
249    fn apply_info_timestamp_with_invalidate_clears() {
250        let mut st = ReceiverState::new(dummy_prefix(0));
251        st.have_timestamp = true;
252        st.apply_info_timestamp(HeTimestamp::default(), true);
253        assert!(!st.have_timestamp);
254    }
255
256    #[test]
257    fn apply_info_reply_sets_locators() {
258        let mut st = ReceiverState::new(dummy_prefix(0));
259        let uni = vec![Locator::INVALID];
260        let multi = vec![Locator::INVALID, Locator::INVALID];
261        st.apply_info_reply(uni.clone(), Some(multi.clone()));
262        assert_eq!(st.unicast_reply_locator_list, uni);
263        assert_eq!(st.multicast_reply_locator_list, multi);
264    }
265
266    #[test]
267    fn apply_header_extension_updates_fields() {
268        let mut st = ReceiverState::new(dummy_prefix(0));
269        let he = HeaderExtension {
270            little_endian: true,
271            message_length: Some(99),
272            timestamp: Some(HeTimestamp {
273                seconds: 1,
274                fraction: 2,
275            }),
276            checksum: ChecksumValue::Crc32c(0xCAFE),
277            ..HeaderExtension::default()
278        };
279        st.apply_header_extension(&he);
280        assert_eq!(st.message_length, Some(99));
281        assert!(st.have_timestamp);
282        assert_eq!(st.timestamp.seconds, 1);
283        assert!(matches!(st.message_checksum, ChecksumValue::Crc32c(0xCAFE)));
284    }
285
286    #[test]
287    fn apply_header_extension_with_parameters_sets_pl() {
288        let mut st = ReceiverState::new(dummy_prefix(0));
289        let pl = ParameterList::new();
290        let he = HeaderExtension {
291            little_endian: true,
292            parameters: Some(pl.clone()),
293            ..HeaderExtension::default()
294        };
295        st.apply_header_extension(&he);
296        assert_eq!(st.parameters, Some(pl));
297    }
298
299    #[test]
300    fn note_clock_skew_skipped_without_timestamp() {
301        let mut st = ReceiverState::new(dummy_prefix(0));
302        st.note_clock_skew(1_000_000, 5);
303        assert!(!st.clock_skew_detected);
304    }
305
306    #[test]
307    fn note_clock_skew_within_threshold_does_not_flag() {
308        let mut st = ReceiverState::new(dummy_prefix(0));
309        st.have_timestamp = true;
310        st.timestamp = HeTimestamp {
311            seconds: 100,
312            fraction: 0,
313        };
314        st.note_clock_skew(102, 5); // diff 2s, threshold 5s
315        assert!(!st.clock_skew_detected);
316    }
317
318    #[test]
319    fn note_clock_skew_above_threshold_flags() {
320        let mut st = ReceiverState::new(dummy_prefix(0));
321        st.have_timestamp = true;
322        st.timestamp = HeTimestamp {
323            seconds: 100,
324            fraction: 0,
325        };
326        st.note_clock_skew(200, 5); // diff 100s, threshold 5s
327        assert!(st.clock_skew_detected);
328    }
329}
zerodds_rtps/receiver_state.rs

zerodds_rtps/
receiver_state.rs
