zerodds_rtps/
participant_message_data.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! `ParticipantMessageData` Wire-Encoding (DDSI-RTPS 2.5 §9.6.3.1).
4//!
5//! Payload structure of the writer liveliness protocol (WLP, §8.4.13).
6//! Published periodically by the `BUILTIN_PARTICIPANT_MESSAGE_WRITER`
7//! as a DATA submessage on the `DCPSParticipantMessage` topic. Readers
8//! use the reception as an implicit `assert_liveliness()` and thereby
9//! drive the lease tracking per peer participant.
10//!
11//! # Wire-Layout (§9.6.3.1)
12//!
13//! ```text
14//! struct ParticipantMessageData {
15//!     GUID_t   participantGuidPrefix; // 12 byte (GuidPrefix only!)
16//!     octet    kind[4];               // 4 byte u32 (BE/LE per CDR)
17//!     sequence<octet> data;           // 4 byte length + N byte
18//! };
19//! ```
20//!
21//! Spec pitfall: despite the field name `participantGuidPrefix`, in
22//! practice Cyclone DDS and Fast-DDS send a full 16-byte GUID (prefix +
23//! ENTITYID_PARTICIPANT). We therefore write 16 bytes and parse
24//! tolerantly: 16 bytes → full GUID, 12 bytes → prefix-only.
25//!
26//! The `data` sequence is semantically a `vec<octet>` with a leading
27//! 32-bit length. Spec §9.6.3.1 defines it as an opaque token; ZeroDDS
28//! uses it for MANUAL_BY_TOPIC to transport the topic token (topic hash
29//! fingerprint).
30//!
31//! # CDR encoding
32//!
33//! Encoded as XCDR1 plain (encapsulation 0x0000 BE / 0x0001 LE) or
34//! XCDR2 plain (0x0006 BE / 0x0007 LE). Cyclone sends the topic by
35//! default as XCDR1 plain. We accept all four encapsulation kinds and
36//! write LE by default.
37//!
38//! # DoS caps
39//!
40//! `data.len()` is capped at [`MAX_DATA_LEN`] = 4096 bytes. The encoder
41//! does not truncate — the caller must cap before the call — the
42//! decoder rejects over-long data with [`WireError::ValueOutOfRange`].
43
44extern crate alloc;
45use alloc::vec::Vec;
46
47use crate::error::WireError;
48use crate::wire_types::GuidPrefix;
49
50/// CDR-Encapsulation-Header: XCDR1 Plain Big-Endian (`0x0000`).
51pub const ENCAPSULATION_CDR_BE: [u8; 2] = [0x00, 0x00];
52/// CDR-Encapsulation-Header: XCDR1 Plain Little-Endian (`0x0001`).
53pub const ENCAPSULATION_CDR_LE: [u8; 2] = [0x00, 0x01];
54/// CDR-Encapsulation-Header: XCDR2 Plain Big-Endian (`0x0006`).
55pub const ENCAPSULATION_CDR2_BE: [u8; 2] = [0x00, 0x06];
56/// CDR-Encapsulation-Header: XCDR2 Plain Little-Endian (`0x0007`).
57pub const ENCAPSULATION_CDR2_LE: [u8; 2] = [0x00, 0x07];
58
59/// DoS cap for the `data` sequence (topic token / vendor opaque).
60/// Deliberately chosen small — WLP heartbeats should be lightweight, a
61/// peer that sends more is either buggy or malicious.
62pub const MAX_DATA_LEN: usize = 4096;
63
64/// Spec-defined `kind` code: AUTOMATIC_LIVELINESS_UPDATE (§9.6.3.1).
65/// Sent by the builtin WLP writer when LIVELINESS=AUTOMATIC.
66pub const PARTICIPANT_MESSAGE_DATA_KIND_AUTOMATIC_LIVELINESS_UPDATE: u32 = 0x0000_0000;
67
68/// Spec-defined `kind` code: MANUAL_BY_PARTICIPANT_LIVELINESS_UPDATE
69/// (§9.6.3.1). Triggered by `assert_liveliness()` on the
70/// `DomainParticipant`.
71pub const PARTICIPANT_MESSAGE_DATA_KIND_MANUAL_BY_PARTICIPANT_LIVELINESS_UPDATE: u32 = 0x0000_0001;
72
73/// Vendor-specific kind range: top bit set
74/// (`0x80000000..=0xFFFFFFFF`). Spec §9.6.3.1 reserves this for
75/// vendor-own kinds (e.g. ZeroDDS MANUAL_BY_TOPIC with a topic token
76/// in the `data` sequence).
77pub const PARTICIPANT_MESSAGE_DATA_KIND_VENDOR_BASE: u32 = 0x8000_0000;
78
79/// ZeroDDS vendor kind: MANUAL_BY_TOPIC. Triggered by
80/// `DataWriter::assert_liveliness()`; the `data` sequence carries a
81/// topic token (typically a 4-byte hash). Cyclone peers ignore the
82/// vendor kind (spec §9.6.3.1: "If kind has its MSB set,
83/// implementations not understanding the kind shall ignore the
84/// message").
85pub const PARTICIPANT_MESSAGE_DATA_KIND_ZERODDS_MANUAL_BY_TOPIC: u32 = 0x8000_0001;
86
87/// `ParticipantMessageData` (DDSI-RTPS 2.5 §9.6.3.1) — payload of the
88/// `DCPSParticipantMessage` topic.
89#[derive(Debug, Clone, PartialEq, Eq)]
90pub struct ParticipantMessageData {
91    /// 16-byte GUID of the sender (prefix + EntityId::PARTICIPANT).
92    /// Even though the spec says "prefix", Cyclone and Fast-DDS write
93    /// the full GUID here; we follow.
94    pub participant_guid: [u8; 16],
95    /// Liveliness kind (see `PARTICIPANT_MESSAGE_DATA_KIND_*`).
96    pub kind: u32,
97    /// Opaque token. On MANUAL_BY_TOPIC a topic hash, otherwise empty.
98    pub data: Vec<u8>,
99}
100
101impl ParticipantMessageData {
102    /// Constructor for an AUTOMATIC heartbeat (data empty).
103    #[must_use]
104    pub fn automatic(prefix: GuidPrefix) -> Self {
105        Self {
106            participant_guid: full_guid_bytes(prefix),
107            kind: PARTICIPANT_MESSAGE_DATA_KIND_AUTOMATIC_LIVELINESS_UPDATE,
108            data: Vec::new(),
109        }
110    }
111
112    /// Constructor for a MANUAL_BY_PARTICIPANT heartbeat (data empty).
113    #[must_use]
114    pub fn manual_by_participant(prefix: GuidPrefix) -> Self {
115        Self {
116            participant_guid: full_guid_bytes(prefix),
117            kind: PARTICIPANT_MESSAGE_DATA_KIND_MANUAL_BY_PARTICIPANT_LIVELINESS_UPDATE,
118            data: Vec::new(),
119        }
120    }
121
122    /// Constructor for ZeroDDS MANUAL_BY_TOPIC (data = topic token).
123    #[must_use]
124    pub fn manual_by_topic(prefix: GuidPrefix, topic_token: Vec<u8>) -> Self {
125        Self {
126            participant_guid: full_guid_bytes(prefix),
127            kind: PARTICIPANT_MESSAGE_DATA_KIND_ZERODDS_MANUAL_BY_TOPIC,
128            data: topic_token,
129        }
130    }
131
132    /// Encodes to CDR bytes (with a 4-byte encapsulation header).
133    /// `little_endian = true` → `ENCAPSULATION_CDR_LE`, otherwise BE.
134    ///
135    /// # Errors
136    /// `WireError::ValueOutOfRange` if `data.len() > MAX_DATA_LEN`.
137    pub fn to_cdr(&self, little_endian: bool) -> Result<Vec<u8>, WireError> {
138        if self.data.len() > MAX_DATA_LEN {
139            return Err(WireError::ValueOutOfRange {
140                message: "ParticipantMessageData.data exceeds MAX_DATA_LEN",
141            });
142        }
143        let data_len_u32 =
144            u32::try_from(self.data.len()).map_err(|_| WireError::ValueOutOfRange {
145                message: "ParticipantMessageData.data length exceeds u32",
146            })?;
147        let mut out = Vec::with_capacity(4 + 16 + 4 + 4 + self.data.len());
148        // Encapsulation-Header
149        if little_endian {
150            out.extend_from_slice(&ENCAPSULATION_CDR_LE);
151        } else {
152            out.extend_from_slice(&ENCAPSULATION_CDR_BE);
153        }
154        out.extend_from_slice(&[0, 0]); // options
155        // Body Start (CDR-Offset 0)
156        // GUID — 16 bytes raw, no endian swap (bytes are opaque).
157        out.extend_from_slice(&self.participant_guid);
158        // kind — 4 byte u32 BE/LE
159        let kind_bytes = if little_endian {
160            self.kind.to_le_bytes()
161        } else {
162            self.kind.to_be_bytes()
163        };
164        out.extend_from_slice(&kind_bytes);
165        // data: u32 length + N byte
166        let len_bytes = if little_endian {
167            data_len_u32.to_le_bytes()
168        } else {
169            data_len_u32.to_be_bytes()
170        };
171        out.extend_from_slice(&len_bytes);
172        out.extend_from_slice(&self.data);
173        Ok(out)
174    }
175
176    /// Decodes from CDR bytes (with an encapsulation header).
177    ///
178    /// Accepts `0x0000`/`0x0001` (XCDR1 plain) and `0x0006`/`0x0007`
179    /// (XCDR2 plain). Other encapsulation kinds → error.
180    ///
181    /// Tolerant of the 12-byte prefix-only encoding (pads with 0 to 16
182    /// bytes).
183    ///
184    /// # Errors
185    /// - `UnsupportedEncapsulation` on non-CDR encapsulation
186    /// - `UnexpectedEof` if the bytes are too short for header / body
187    /// - `ValueOutOfRange` if `data.len > MAX_DATA_LEN`
188    pub fn from_cdr(bytes: &[u8]) -> Result<Self, WireError> {
189        if bytes.len() < 4 {
190            return Err(WireError::UnexpectedEof {
191                needed: 4,
192                offset: 0,
193            });
194        }
195        let little_endian = match (bytes[0], bytes[1]) {
196            (0x00, 0x00) | (0x00, 0x06) => false,
197            (0x00, 0x01) | (0x00, 0x07) => true,
198            (a, b) => {
199                return Err(WireError::UnsupportedEncapsulation { kind: [a, b] });
200            }
201        };
202        // Body starts at offset 4 (after options).
203        let body = &bytes[4..];
204        // We accept a 16-byte GUID (spec de-facto) OR 12-byte
205        // prefix-only (an alternative reading of the spec wording
206        // "GUID_t guidPrefix" — strict 12-byte encoders exist).
207        // Heuristic: 16 bytes is the default, 12 bytes is only valid if
208        // the following fields (kind + data-len) parse correctly with 12
209        // bytes.
210        let (guid_bytes, after_guid_offset) = parse_guid(body)?;
211        if body.len() < after_guid_offset + 4 {
212            return Err(WireError::UnexpectedEof {
213                needed: after_guid_offset + 4,
214                offset: 4,
215            });
216        }
217        let kind_slice = &body[after_guid_offset..after_guid_offset + 4];
218        let mut kind_arr = [0u8; 4];
219        kind_arr.copy_from_slice(kind_slice);
220        let kind = if little_endian {
221            u32::from_le_bytes(kind_arr)
222        } else {
223            u32::from_be_bytes(kind_arr)
224        };
225        let len_offset = after_guid_offset + 4;
226        if body.len() < len_offset + 4 {
227            return Err(WireError::UnexpectedEof {
228                needed: len_offset + 4,
229                offset: 4,
230            });
231        }
232        let mut len_arr = [0u8; 4];
233        len_arr.copy_from_slice(&body[len_offset..len_offset + 4]);
234        let data_len = if little_endian {
235            u32::from_le_bytes(len_arr)
236        } else {
237            u32::from_be_bytes(len_arr)
238        } as usize;
239        if data_len > MAX_DATA_LEN {
240            return Err(WireError::ValueOutOfRange {
241                message: "ParticipantMessageData.data exceeds MAX_DATA_LEN",
242            });
243        }
244        let data_offset = len_offset + 4;
245        if body.len() < data_offset + data_len {
246            return Err(WireError::UnexpectedEof {
247                needed: data_offset + data_len,
248                offset: 4,
249            });
250        }
251        let data = body[data_offset..data_offset + data_len].to_vec();
252        Ok(Self {
253            participant_guid: guid_bytes,
254            kind,
255            data,
256        })
257    }
258
259    /// Returns the GuidPrefix (first 12 bytes).
260    #[must_use]
261    pub fn prefix(&self) -> GuidPrefix {
262        let mut p = [0u8; 12];
263        p.copy_from_slice(&self.participant_guid[..12]);
264        GuidPrefix::from_bytes(p)
265    }
266
267    /// `true` if the `kind` value is vendor-specific (MSB set).
268    #[must_use]
269    pub fn is_vendor_kind(&self) -> bool {
270        self.kind >= PARTICIPANT_MESSAGE_DATA_KIND_VENDOR_BASE
271    }
272}
273
274fn full_guid_bytes(prefix: GuidPrefix) -> [u8; 16] {
275    let mut g = [0u8; 16];
276    g[..12].copy_from_slice(&prefix.to_bytes());
277    // EntityId::PARTICIPANT = [0, 0, 1, 0xC1]
278    g[12] = 0;
279    g[13] = 0;
280    g[14] = 1;
281    g[15] = 0xC1;
282    g
283}
284
285/// Parses the GUID body. Tries 16 bytes first (Cyclone/Fast-DDS
286/// default), falls back to 12 bytes (strict spec reading).
287fn parse_guid(body: &[u8]) -> Result<([u8; 16], usize), WireError> {
288    // 16-byte variant: needs at least 16 + 4 (kind) + 4 (data-len).
289    if body.len() >= 24 {
290        let mut g = [0u8; 16];
291        g.copy_from_slice(&body[..16]);
292        return Ok((g, 16));
293    }
294    // 12-byte variant: needs at least 12 + 4 + 4.
295    if body.len() >= 20 {
296        let mut g = [0u8; 16];
297        g[..12].copy_from_slice(&body[..12]);
298        // EntityId-Default: PARTICIPANT
299        g[14] = 1;
300        g[15] = 0xC1;
301        return Ok((g, 12));
302    }
303    Err(WireError::UnexpectedEof {
304        needed: 24,
305        offset: 4,
306    })
307}
308
309#[cfg(test)]
310mod tests {
311    #![allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
312    use super::*;
313    use alloc::vec;
314
315    fn sample_prefix() -> GuidPrefix {
316        GuidPrefix::from_bytes([0xA, 0xB, 0xC, 0xD, 1, 2, 3, 4, 5, 6, 7, 8])
317    }
318
319    #[test]
320    fn participant_message_data_automatic_default_data_empty() {
321        let m = ParticipantMessageData::automatic(sample_prefix());
322        assert_eq!(
323            m.kind,
324            PARTICIPANT_MESSAGE_DATA_KIND_AUTOMATIC_LIVELINESS_UPDATE
325        );
326        assert!(m.data.is_empty());
327        assert_eq!(m.prefix(), sample_prefix());
328    }
329
330    #[test]
331    fn participant_message_data_kind_constants_match_spec() {
332        // §9.6.3.1: AUTOMATIC = 0x00000000, MANUAL_BY_PARTICIPANT = 0x00000001.
333        // Vendor range: MSB set, i.e. >= 0x80000000.
334        assert_eq!(
335            PARTICIPANT_MESSAGE_DATA_KIND_AUTOMATIC_LIVELINESS_UPDATE,
336            0x0000_0000
337        );
338        assert_eq!(
339            PARTICIPANT_MESSAGE_DATA_KIND_MANUAL_BY_PARTICIPANT_LIVELINESS_UPDATE,
340            0x0000_0001
341        );
342        assert_eq!(PARTICIPANT_MESSAGE_DATA_KIND_VENDOR_BASE, 0x8000_0000);
343        // The ZeroDDS vendor kind must be in the vendor range (MSB set).
344        // `assert_eq!` instead of `assert!` because clippy
345        // `assertions_on_constants` would otherwise complain; here we
346        // compare against a constant expected value.
347        assert_eq!(
348            PARTICIPANT_MESSAGE_DATA_KIND_ZERODDS_MANUAL_BY_TOPIC
349                & PARTICIPANT_MESSAGE_DATA_KIND_VENDOR_BASE,
350            PARTICIPANT_MESSAGE_DATA_KIND_VENDOR_BASE
351        );
352    }
353
354    #[test]
355    fn participant_message_data_roundtrip_le() {
356        let m = ParticipantMessageData::manual_by_participant(sample_prefix());
357        let bytes = m.to_cdr(true).unwrap();
358        // Encapsulation-Header LE
359        assert_eq!(&bytes[..4], &[0x00, 0x01, 0x00, 0x00]);
360        let decoded = ParticipantMessageData::from_cdr(&bytes).unwrap();
361        assert_eq!(decoded, m);
362    }
363
364    #[test]
365    fn participant_message_data_roundtrip_be() {
366        let m = ParticipantMessageData::automatic(sample_prefix());
367        let bytes = m.to_cdr(false).unwrap();
368        assert_eq!(&bytes[..4], &[0x00, 0x00, 0x00, 0x00]);
369        let decoded = ParticipantMessageData::from_cdr(&bytes).unwrap();
370        assert_eq!(decoded, m);
371    }
372
373    #[test]
374    fn participant_message_data_roundtrip_with_topic_token() {
375        let m =
376            ParticipantMessageData::manual_by_topic(sample_prefix(), vec![0xDE, 0xAD, 0xBE, 0xEF]);
377        assert!(m.is_vendor_kind());
378        let bytes = m.to_cdr(true).unwrap();
379        let decoded = ParticipantMessageData::from_cdr(&bytes).unwrap();
380        assert_eq!(decoded.data, vec![0xDE, 0xAD, 0xBE, 0xEF]);
381        assert_eq!(
382            decoded.kind,
383            PARTICIPANT_MESSAGE_DATA_KIND_ZERODDS_MANUAL_BY_TOPIC
384        );
385    }
386
387    #[test]
388    fn participant_message_data_accepts_xcdr2_le_encapsulation() {
389        // The ZeroDDS default for user topics is XCDR2-LE (0x0007). If a
390        // peer sends the WLP topic with XCDR2, we must be able to decode
391        // it (Cyclone does this with the spec-2.5 default rep).
392        let m = ParticipantMessageData::automatic(sample_prefix());
393        let mut bytes = m.to_cdr(true).unwrap();
394        bytes[0] = 0x00;
395        bytes[1] = 0x07; // XCDR2 LE
396        let decoded = ParticipantMessageData::from_cdr(&bytes).unwrap();
397        assert_eq!(decoded, m);
398    }
399
400    #[test]
401    fn participant_message_data_accepts_xcdr2_be_encapsulation() {
402        let m = ParticipantMessageData::automatic(sample_prefix());
403        let mut bytes = m.to_cdr(false).unwrap();
404        bytes[0] = 0x00;
405        bytes[1] = 0x06; // XCDR2 BE
406        let decoded = ParticipantMessageData::from_cdr(&bytes).unwrap();
407        assert_eq!(decoded, m);
408    }
409
410    #[test]
411    fn participant_message_data_rejects_unknown_encapsulation() {
412        let mut bytes = vec![0x99, 0x99, 0, 0];
413        bytes.extend_from_slice(&[0u8; 24]);
414        let res = ParticipantMessageData::from_cdr(&bytes);
415        assert!(matches!(
416            res,
417            Err(WireError::UnsupportedEncapsulation { kind: [0x99, 0x99] })
418        ));
419    }
420
421    #[test]
422    fn participant_message_data_rejects_overlong_data() {
423        // Build manually: encapsulation + 16 byte guid + 4 byte kind +
424        // 4 byte length=MAX+1.
425        let mut bytes = vec![0x00, 0x01, 0x00, 0x00];
426        bytes.extend_from_slice(&[0u8; 16]);
427        bytes.extend_from_slice(&0u32.to_le_bytes()); // kind
428        let too_big = (MAX_DATA_LEN as u32) + 1;
429        bytes.extend_from_slice(&too_big.to_le_bytes());
430        // data missing — that's fine, the cap check fires first.
431        let res = ParticipantMessageData::from_cdr(&bytes);
432        assert!(matches!(res, Err(WireError::ValueOutOfRange { .. })));
433    }
434
435    #[test]
436    fn participant_message_data_encoder_caps_data_length() {
437        let mut m = ParticipantMessageData::automatic(sample_prefix());
438        m.data = vec![0u8; MAX_DATA_LEN + 1];
439        let res = m.to_cdr(true);
440        assert!(matches!(res, Err(WireError::ValueOutOfRange { .. })));
441    }
442
443    #[test]
444    fn participant_message_data_too_short_encapsulation() {
445        let bytes = [0x00];
446        let res = ParticipantMessageData::from_cdr(&bytes);
447        assert!(matches!(res, Err(WireError::UnexpectedEof { .. })));
448    }
449
450    #[test]
451    fn participant_message_data_too_short_body() {
452        // Encapsulation valid, body only 8 byte (less than 12-byte prefix variant).
453        let bytes = vec![0x00, 0x01, 0x00, 0x00, 0, 0, 0, 0, 0, 0, 0, 0];
454        let res = ParticipantMessageData::from_cdr(&bytes);
455        assert!(matches!(res, Err(WireError::UnexpectedEof { .. })));
456    }
457
458    #[test]
459    fn participant_message_data_truncated_data_section() {
460        // length=8 but only 4 byte follow.
461        let mut bytes = vec![0x00, 0x01, 0x00, 0x00];
462        bytes.extend_from_slice(&[0u8; 16]);
463        bytes.extend_from_slice(&0u32.to_le_bytes());
464        bytes.extend_from_slice(&8u32.to_le_bytes());
465        bytes.extend_from_slice(&[1, 2, 3, 4]);
466        let res = ParticipantMessageData::from_cdr(&bytes);
467        assert!(matches!(res, Err(WireError::UnexpectedEof { .. })));
468    }
469
470    #[test]
471    fn participant_message_data_le_be_bytes_differ_for_kind() {
472        // Sanity: BE and LE encoding differ via kind.
473        let mut m = ParticipantMessageData::automatic(sample_prefix());
474        m.kind = 0x0102_0304;
475        let le = m.to_cdr(true).unwrap();
476        let be = m.to_cdr(false).unwrap();
477        assert_ne!(le, be);
478        // Both must yield the same value again.
479        assert_eq!(ParticipantMessageData::from_cdr(&le).unwrap(), m);
480        assert_eq!(ParticipantMessageData::from_cdr(&be).unwrap(), m);
481    }
482
483    #[test]
484    fn participant_message_data_accepts_12_byte_prefix_only_encoding() {
485        // A legacy/strict encoder writes only 12 bytes (prefix). We
486        // must be able to decode it + fill in the EntityId default
487        // (PARTICIPANT = 0xC1).
488        let mut bytes = vec![0x00, 0x01, 0x00, 0x00];
489        let prefix = sample_prefix().to_bytes();
490        bytes.extend_from_slice(&prefix); // 12 byte
491        bytes.extend_from_slice(&0u32.to_le_bytes()); // kind
492        bytes.extend_from_slice(&0u32.to_le_bytes()); // data len = 0
493        let decoded = ParticipantMessageData::from_cdr(&bytes).unwrap();
494        assert_eq!(decoded.prefix(), sample_prefix());
495        // EntityId-Default-Auffuellung
496        assert_eq!(&decoded.participant_guid[12..], &[0, 0, 1, 0xC1]);
497    }
498}
zerodds_rtps/participant_message_data.rs

zerodds_rtps/
participant_message_data.rs
