zerodds_rtps/
inline_qos.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! Inline-QoS helpers for DDS-RPC and SEDP (DDS-RPC 1.0 §7.8.2).
4//!
5//! Inline QoS is a [`crate::parameter_list::ParameterList`] in the body
6//! of a DATA/DATA_FRAG submessage (Q flag, Spec §9.4.5.3). Phase 2 already uses
7//! it for `PID_KEY_HASH`, `PID_STATUS_INFO` etc. — this module
8//! adds helpers for the RPC-specific inline-QoS PIDs:
9//!
10//! * `PID_RELATED_SAMPLE_IDENTITY = 0x0083` (DDS-RPC 1.0 §7.8.2): is
11//!   set by the reply writer in the inline QoS of every reply DATA; the value
12//!   is the `request_id` (`SampleIdentity` = 16 byte writer_guid +
13//!   8 byte sequence_number) of the correlated request. Encoding:
14//!   XCDR2-Final, alignment cap=4, i.e. 24 bytes without padding.
15//!
16//! We encapsulate encode/decode here so that `zerodds-rpc` can work without a
17//! hard dependency on RTPS internals like the `Parameter` layout.
18
19use crate::error::WireError;
20use crate::parameter_list::{
21    MUST_UNDERSTAND_BIT, Parameter, ParameterList, VENDOR_SPECIFIC_BIT, pid,
22};
23
24/// Spec constant: 16 byte writer GUID + 8 byte sequence number.
25pub const SAMPLE_IDENTITY_WIRE_SIZE: usize = 24;
26
27/// Wire representation of a `SampleIdentity` (DDS-RPC 1.0 §7.5.1.1.1).
28///
29/// Identical to the layout in `zerodds-rpc::common_types::SampleIdentity`, but
30/// modeled here as a pure byte helper — `zerodds-rtps` must not depend on
31/// `zerodds-rpc` (crate dependency direction).
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
33pub struct SampleIdentityBytes {
34    /// 16 byte writer-GUID.
35    pub writer_guid: [u8; 16],
36    /// 64-bit sequence number.
37    pub sequence_number: u64,
38}
39
40impl SampleIdentityBytes {
41    /// Constructor.
42    #[must_use]
43    pub const fn new(writer_guid: [u8; 16], sequence_number: u64) -> Self {
44        Self {
45            writer_guid,
46            sequence_number,
47        }
48    }
49
50    /// XCDR2-Final encoder with the given endianness.
51    /// Layout: 16 byte GUID + 8 byte u64 (alignment cap=4 ⇒ no padding).
52    #[must_use]
53    pub fn to_bytes(&self, little_endian: bool) -> [u8; SAMPLE_IDENTITY_WIRE_SIZE] {
54        let mut out = [0u8; SAMPLE_IDENTITY_WIRE_SIZE];
55        out[..16].copy_from_slice(&self.writer_guid);
56        let sn = if little_endian {
57            self.sequence_number.to_le_bytes()
58        } else {
59            self.sequence_number.to_be_bytes()
60        };
61        out[16..].copy_from_slice(&sn);
62        out
63    }
64
65    /// XCDR2-Final decoder.
66    ///
67    /// # Errors
68    /// `WireError::UnexpectedEof` if the buffer is shorter than 24 bytes.
69    pub fn from_bytes(bytes: &[u8], little_endian: bool) -> Result<Self, WireError> {
70        if bytes.len() < SAMPLE_IDENTITY_WIRE_SIZE {
71            return Err(WireError::UnexpectedEof {
72                needed: SAMPLE_IDENTITY_WIRE_SIZE,
73                offset: 0,
74            });
75        }
76        let mut writer_guid = [0u8; 16];
77        writer_guid.copy_from_slice(&bytes[..16]);
78        let mut sn = [0u8; 8];
79        sn.copy_from_slice(&bytes[16..24]);
80        let sequence_number = if little_endian {
81            u64::from_le_bytes(sn)
82        } else {
83            u64::from_be_bytes(sn)
84        };
85        Ok(Self {
86            writer_guid,
87            sequence_number,
88        })
89    }
90}
91
92/// Builds a `Parameter` with `PID_RELATED_SAMPLE_IDENTITY` (0x0083) and
93/// the 24-byte XCDR2 encoding of the `SampleIdentity` as the value.
94#[must_use]
95pub fn related_sample_identity_param(id: SampleIdentityBytes, little_endian: bool) -> Parameter {
96    Parameter::new(
97        pid::RELATED_SAMPLE_IDENTITY,
98        id.to_bytes(little_endian).to_vec(),
99    )
100}
101
102/// Builds an inline-QoS `ParameterList` that carries only the
103/// `PID_RELATED_SAMPLE_IDENTITY` — good enough for the reply DATA
104/// of a simple RPC operation. Callers can add further parameters
105/// (`PID_KEY_HASH` etc.) via `push`.
106#[must_use]
107pub fn reply_inline_qos(id: SampleIdentityBytes, little_endian: bool) -> ParameterList {
108    let mut pl = ParameterList::new();
109    pl.push(related_sample_identity_param(id, little_endian));
110    pl
111}
112
113/// Spec §9.6.3.9 — `PID_STATUS_INFO` bits.
114pub mod status_info {
115    /// Bit 0: the sample was marked NOT_ALIVE_DISPOSED via `dispose`.
116    pub const DISPOSED: u32 = 0x0000_0001;
117    /// Bit 1: the sample was marked NOT_ALIVE_NO_WRITERS via
118    /// `unregister_instance`.
119    pub const UNREGISTERED: u32 = 0x0000_0002;
120    /// Bit 2: the sample was filtered by the writer via a content filter.
121    pub const FILTERED: u32 = 0x0000_0004;
122}
123
124/// Builds a `Parameter` with `PID_STATUS_INFO` (0x0071). The value is a
125/// 4-byte status word (bits per [`status_info`]). The spec requires
126/// **big-endian** encoding independent of the RTPS header endianness
127/// (DDSI-RTPS 2.5 §9.6.3.9).
128#[must_use]
129pub fn status_info_param(bits: u32) -> Parameter {
130    Parameter::new(pid::STATUS_INFO, bits.to_be_bytes().to_vec())
131}
132
133/// Reads `PID_STATUS_INFO` from an inline-QoS list. Returns the
134/// 4-byte status word, or `None` if the PID is missing / the value
135/// is not 4 bytes.
136#[must_use]
137pub fn find_status_info(pl: &ParameterList) -> Option<u32> {
138    let p = pl.find(pid::STATUS_INFO)?;
139    if p.value.len() != 4 {
140        return None;
141    }
142    let mut b = [0u8; 4];
143    b.copy_from_slice(&p.value);
144    Some(u32::from_be_bytes(b))
145}
146
147/// Reads `PID_KEY_HASH` from an inline-QoS list (Spec §9.6.4.8 +
148/// XTypes 1.3 §7.6.8). Returns the 16-byte identity of the instance,
149/// or `None` if the PID is missing / the value has an invalid
150/// length.
151#[must_use]
152pub fn find_key_hash(pl: &ParameterList) -> Option<[u8; 16]> {
153    let p = pl.find(pid::KEY_HASH)?;
154    if p.value.len() != 16 {
155        return None;
156    }
157    let mut b = [0u8; 16];
158    b.copy_from_slice(&p.value);
159    Some(b)
160}
161
162/// Builds an inline-QoS `ParameterList` for a lifecycle marker —
163/// `PID_KEY_HASH` (16 byte) + `PID_STATUS_INFO` (4 byte). Sent by the
164/// writer on `dispose`/`unregister_instance`.
165#[must_use]
166pub fn lifecycle_inline_qos(key_hash: [u8; 16], status_bits: u32) -> ParameterList {
167    let mut pl = ParameterList::new();
168    pl.push(Parameter::new(pid::KEY_HASH, key_hash.to_vec()));
169    pl.push(status_info_param(status_bits));
170    pl
171}
172
173/// Spec §8.7.9 — `PID_ORIGINAL_WRITER_INFO` as inline QoS. 24-byte
174/// value: 16 byte original GUID + 8 byte SequenceNumber (signed i64).
175/// Set by the persistence service when it forwards a historical sample
176/// on behalf of another writer.
177#[must_use]
178pub fn original_writer_info_param(
179    original_guid: [u8; 16],
180    original_sn: i64,
181    little_endian: bool,
182) -> Parameter {
183    let mut value = alloc::vec::Vec::with_capacity(24);
184    value.extend_from_slice(&original_guid);
185    let sn_bytes = if little_endian {
186        original_sn.to_le_bytes()
187    } else {
188        original_sn.to_be_bytes()
189    };
190    value.extend_from_slice(&sn_bytes);
191    Parameter::new(pid::ORIGINAL_WRITER_INFO, value)
192}
193
194/// Reads `PID_ORIGINAL_WRITER_INFO` from an inline-QoS list.
195///
196/// # Errors
197/// `WireError::UnexpectedEof` if the value slice is below 24 bytes.
198pub fn find_original_writer_info(
199    pl: &ParameterList,
200    little_endian: bool,
201) -> Result<Option<([u8; 16], i64)>, WireError> {
202    let target = pid::ORIGINAL_WRITER_INFO;
203    for p in &pl.parameters {
204        let masked = p.id & !(MUST_UNDERSTAND_BIT | VENDOR_SPECIFIC_BIT);
205        if masked == target {
206            if p.value.len() < 24 {
207                return Err(WireError::UnexpectedEof {
208                    needed: 24,
209                    offset: 0,
210                });
211            }
212            let mut g = [0u8; 16];
213            g.copy_from_slice(&p.value[..16]);
214            let mut s = [0u8; 8];
215            s.copy_from_slice(&p.value[16..24]);
216            let sn = if little_endian {
217                i64::from_le_bytes(s)
218            } else {
219                i64::from_be_bytes(s)
220            };
221            return Ok(Some((g, sn)));
222        }
223    }
224    Ok(None)
225}
226
227/// Spec §8.7.7 / §9.6.2.2.5 — `PID_DIRECTED_WRITE` as inline QoS.
228/// Marks a sample as a point-to-point send to a single
229/// target reader (16 byte GUID). Other readers that receive the sample
230/// (e.g. via multicast) MUST discard it.
231#[must_use]
232pub fn directed_write_param(target_reader_guid: [u8; 16]) -> Parameter {
233    Parameter::new(pid::DIRECTED_WRITE, target_reader_guid.to_vec())
234}
235
236/// Reads `PID_DIRECTED_WRITE` from an inline-QoS list.
237///
238/// Returns `Ok(None)` if the PID is not set (the sample is not
239/// directed). Otherwise returns the 16-byte target GUID.
240///
241/// # Errors
242/// `WireError::UnexpectedEof` if the value slice is below 16 bytes.
243pub fn find_directed_write(pl: &ParameterList) -> Result<Option<[u8; 16]>, WireError> {
244    let target = pid::DIRECTED_WRITE;
245    for p in &pl.parameters {
246        let masked = p.id & !(MUST_UNDERSTAND_BIT | VENDOR_SPECIFIC_BIT);
247        if masked == target {
248            if p.value.len() < 16 {
249                return Err(WireError::UnexpectedEof {
250                    needed: 16,
251                    offset: 0,
252                });
253            }
254            let mut g = [0u8; 16];
255            g.copy_from_slice(&p.value[..16]);
256            return Ok(Some(g));
257        }
258    }
259    Ok(None)
260}
261
262/// Spec §8.7.7 — receiver-side filter: returns `true` if the sample
263/// is addressed to the reader with `own_reader_guid` (or no
264/// directed write is set). `false` if a PID_DIRECTED_WRITE
265/// is present and the value does NOT match own_guid — the
266/// caller MUST then discard the sample.
267///
268/// # Errors
269/// Wire decoding error from `find_directed_write`.
270pub fn directed_write_matches_reader(
271    pl: &ParameterList,
272    own_reader_guid: [u8; 16],
273) -> Result<bool, WireError> {
274    Ok(match find_directed_write(pl)? {
275        None => true,
276        Some(target) => target == own_reader_guid,
277    })
278}
279
280/// Reads `PID_RELATED_SAMPLE_IDENTITY` from an inline-QoS list.
281///
282/// Masks the must-understand bit + vendor bit. Returns `Ok(None)` if
283/// the PID is not present (inline QoS without RPC correlation —
284/// e.g. a user-sample DATA that has nothing to do with RPC).
285///
286/// # Errors
287/// `WireError::UnexpectedEof` if the value slice is below 24 bytes.
288pub fn find_related_sample_identity(
289    pl: &ParameterList,
290    little_endian: bool,
291) -> Result<Option<SampleIdentityBytes>, WireError> {
292    let target = pid::RELATED_SAMPLE_IDENTITY;
293    for p in &pl.parameters {
294        let masked = p.id & !(MUST_UNDERSTAND_BIT | VENDOR_SPECIFIC_BIT);
295        if masked == target {
296            return Ok(Some(SampleIdentityBytes::from_bytes(
297                &p.value,
298                little_endian,
299            )?));
300        }
301    }
302    Ok(None)
303}
304
305#[cfg(test)]
306#[allow(clippy::expect_used, clippy::unwrap_used)]
307mod tests {
308    use super::*;
309    use alloc::sync::Arc;
310
311    fn sample_id() -> SampleIdentityBytes {
312        SampleIdentityBytes::new([0xAB; 16], 0x0102_0304_0506_0708)
313    }
314
315    #[test]
316    fn sample_identity_layout_is_24_bytes() {
317        let id = sample_id();
318        let bytes_le = id.to_bytes(true);
319        let bytes_be = id.to_bytes(false);
320        assert_eq!(bytes_le.len(), SAMPLE_IDENTITY_WIRE_SIZE);
321        assert_eq!(bytes_be.len(), SAMPLE_IDENTITY_WIRE_SIZE);
322        // First 16 bytes = GUID (endianness-independent).
323        assert_eq!(&bytes_le[..16], &[0xAB; 16]);
324        assert_eq!(&bytes_be[..16], &[0xAB; 16]);
325        // Last 8 bytes = sequence number, different per endianness.
326        assert_eq!(&bytes_le[16..], &0x0102_0304_0506_0708u64.to_le_bytes());
327        assert_eq!(&bytes_be[16..], &0x0102_0304_0506_0708u64.to_be_bytes());
328    }
329
330    #[test]
331    fn sample_identity_roundtrip_le() {
332        let id = sample_id();
333        let bytes = id.to_bytes(true);
334        assert_eq!(SampleIdentityBytes::from_bytes(&bytes, true).unwrap(), id);
335    }
336
337    #[test]
338    fn sample_identity_roundtrip_be() {
339        let id = sample_id();
340        let bytes = id.to_bytes(false);
341        assert_eq!(SampleIdentityBytes::from_bytes(&bytes, false).unwrap(), id);
342    }
343
344    #[test]
345    fn sample_identity_too_short_is_eof_error() {
346        let res = SampleIdentityBytes::from_bytes(&[0u8; 10], true);
347        assert!(matches!(res, Err(WireError::UnexpectedEof { .. })));
348    }
349
350    #[test]
351    fn related_sample_identity_param_uses_pid_0x0083() {
352        let p = related_sample_identity_param(sample_id(), true);
353        assert_eq!(p.id, 0x0083);
354        assert_eq!(p.id, pid::RELATED_SAMPLE_IDENTITY);
355        assert_eq!(p.value.len(), SAMPLE_IDENTITY_WIRE_SIZE);
356    }
357
358    #[test]
359    fn reply_inline_qos_contains_only_pid_0x0083() {
360        let pl = reply_inline_qos(sample_id(), true);
361        assert_eq!(pl.parameters.len(), 1);
362        assert_eq!(pl.parameters[0].id, pid::RELATED_SAMPLE_IDENTITY);
363    }
364
365    #[test]
366    fn find_related_sample_identity_finds_pid() {
367        let id = sample_id();
368        let pl = reply_inline_qos(id, true);
369        let got = find_related_sample_identity(&pl, true).unwrap();
370        assert_eq!(got, Some(id));
371    }
372
373    #[test]
374    fn find_related_sample_identity_missing_returns_none() {
375        let pl = ParameterList::new();
376        let got = find_related_sample_identity(&pl, true).unwrap();
377        assert_eq!(got, None);
378    }
379
380    #[test]
381    fn find_related_sample_identity_with_must_understand_bit_works() {
382        // Cyclone likes to set inline-QoS PIDs with the MU bit. Masking must
383        // take effect.
384        let id = sample_id();
385        let mut pl = ParameterList::new();
386        pl.push(Parameter::new(
387            MUST_UNDERSTAND_BIT | pid::RELATED_SAMPLE_IDENTITY,
388            id.to_bytes(true).to_vec(),
389        ));
390        let got = find_related_sample_identity(&pl, true).unwrap();
391        assert_eq!(got, Some(id));
392    }
393
394    #[test]
395    fn find_related_sample_identity_truncated_value_is_error() {
396        let mut pl = ParameterList::new();
397        pl.push(Parameter::new(
398            pid::RELATED_SAMPLE_IDENTITY,
399            alloc::vec![0u8; 8],
400        ));
401        let res = find_related_sample_identity(&pl, true);
402        assert!(matches!(res, Err(WireError::UnexpectedEof { .. })));
403    }
404
405    #[test]
406    fn roundtrip_in_data_submessage_inline_qos() {
407        // E2E: DataSubmessage with Q flag + PID 0x0083 in inline QoS,
408        // then roundtrip through write_body / read_body_with_flags.
409        use crate::submessages::{DATA_FLAG_INLINE_QOS, DataSubmessage};
410        use crate::wire_types::{EntityId, SequenceNumber};
411
412        let id = sample_id();
413        let inline = reply_inline_qos(id, true);
414        let payload: Arc<[u8]> = Arc::from(alloc::vec![1u8, 2, 3, 4].as_slice());
415        let data = DataSubmessage {
416            extra_flags: 0,
417            reader_id: EntityId::user_reader_with_key([1, 2, 3]),
418            writer_id: EntityId::user_writer_with_key([4, 5, 6]),
419            writer_sn: SequenceNumber(42),
420            inline_qos: Some(inline),
421            key_flag: false,
422            non_standard_flag: false,
423            serialized_payload: payload.clone(),
424        };
425        let (body, flags) = data.write_body(true);
426        assert_ne!(flags & DATA_FLAG_INLINE_QOS, 0);
427
428        let decoded = DataSubmessage::read_body_with_flags(&body, true, flags).unwrap();
429        assert!(decoded.inline_qos.is_some());
430        let pl = decoded.inline_qos.unwrap();
431        let back = find_related_sample_identity(&pl, true).unwrap();
432        assert_eq!(back, Some(id));
433        // Payload byte-identical.
434        assert_eq!(&decoded.serialized_payload[..], &[1, 2, 3, 4]);
435    }
436
437    // ---- Spec §8.7.7 Directed Write ----
438
439    #[test]
440    fn directed_write_param_carries_target_guid() {
441        let target = [0xCA; 16];
442        let p = directed_write_param(target);
443        assert_eq!(p.id, pid::DIRECTED_WRITE);
444        assert_eq!(p.value, target.to_vec());
445    }
446
447    #[test]
448    fn find_directed_write_returns_target_when_present() {
449        let mut pl = ParameterList::new();
450        let target = [0xAB; 16];
451        pl.push(directed_write_param(target));
452        assert_eq!(find_directed_write(&pl).unwrap(), Some(target));
453    }
454
455    #[test]
456    fn find_directed_write_returns_none_when_absent() {
457        let pl = ParameterList::new();
458        assert_eq!(find_directed_write(&pl).unwrap(), None);
459    }
460
461    #[test]
462    fn find_directed_write_rejects_truncated_value() {
463        let mut pl = ParameterList::new();
464        pl.push(Parameter::new(pid::DIRECTED_WRITE, alloc::vec![0; 8]));
465        let r = find_directed_write(&pl);
466        assert!(matches!(r, Err(WireError::UnexpectedEof { .. })));
467    }
468
469    #[test]
470    fn directed_write_matches_reader_returns_true_for_matching_target() {
471        let mut pl = ParameterList::new();
472        let me = [0xAB; 16];
473        pl.push(directed_write_param(me));
474        assert!(directed_write_matches_reader(&pl, me).unwrap());
475    }
476
477    #[test]
478    fn directed_write_matches_reader_returns_false_for_other_target() {
479        let mut pl = ParameterList::new();
480        pl.push(directed_write_param([0xAB; 16]));
481        let other = [0xCD; 16];
482        assert!(!directed_write_matches_reader(&pl, other).unwrap());
483    }
484
485    #[test]
486    fn directed_write_matches_reader_returns_true_when_no_directed_write() {
487        let pl = ParameterList::new();
488        // No directed write → every reader matches.
489        assert!(directed_write_matches_reader(&pl, [0; 16]).unwrap());
490    }
491
492    // ---- Spec §8.7.9 OriginalWriterInfo ----
493
494    #[test]
495    fn original_writer_info_param_24_byte_layout_le() {
496        let p = original_writer_info_param([0xAB; 16], 0x0102_0304_0506_0708, true);
497        assert_eq!(p.id, pid::ORIGINAL_WRITER_INFO);
498        assert_eq!(p.value.len(), 24);
499        assert_eq!(&p.value[..16], &[0xAB; 16]);
500        assert_eq!(&p.value[16..], &0x0102_0304_0506_0708i64.to_le_bytes());
501    }
502
503    #[test]
504    fn original_writer_info_roundtrip_le() {
505        let mut pl = ParameterList::new();
506        pl.push(original_writer_info_param([0xCD; 16], 42, true));
507        let back = find_original_writer_info(&pl, true).unwrap();
508        assert_eq!(back, Some(([0xCD; 16], 42)));
509    }
510
511    #[test]
512    fn original_writer_info_roundtrip_be() {
513        let mut pl = ParameterList::new();
514        pl.push(original_writer_info_param([0xCD; 16], 42, false));
515        let back = find_original_writer_info(&pl, false).unwrap();
516        assert_eq!(back, Some(([0xCD; 16], 42)));
517    }
518
519    #[test]
520    fn find_original_writer_info_returns_none_when_absent() {
521        let pl = ParameterList::new();
522        assert_eq!(find_original_writer_info(&pl, true).unwrap(), None);
523    }
524
525    #[test]
526    fn find_original_writer_info_rejects_truncated() {
527        let mut pl = ParameterList::new();
528        pl.push(Parameter::new(
529            pid::ORIGINAL_WRITER_INFO,
530            alloc::vec![0; 16],
531        ));
532        let r = find_original_writer_info(&pl, true);
533        assert!(matches!(r, Err(WireError::UnexpectedEof { .. })));
534    }
535
536    #[test]
537    fn vendor_trace_context_pid_constant() {
538        assert_eq!(pid::VENDOR_TRACE_CONTEXT, 0x0D00);
539    }
540
541    #[test]
542    fn vendor_trace_context_param_roundtrip() {
543        let mut pl = ParameterList::new();
544        let payload = alloc::vec![1, 2, 3, 4, 5, 6, 7, 8];
545        pl.push(Parameter::new(pid::VENDOR_TRACE_CONTEXT, payload.clone()));
546        let bytes = pl.to_bytes(true);
547        let pl2 = ParameterList::from_bytes(&bytes, true).expect("decode");
548        let p = pl2.find(pid::VENDOR_TRACE_CONTEXT).expect("present");
549        assert_eq!(p.value, payload);
550    }
551}
zerodds_rtps/inline_qos.rs

zerodds_rtps/
inline_qos.rs
