zerodds_rtps/
subscription_data.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3//! SubscriptionBuiltinTopicData (DDSI-RTPS 2.5 §8.5.4.3, §9.6.2.2.4).
4//!
5//! Content of the SEDP subscriptions DATA submessage. Analogous to
6//! [`crate::publication_data::PublicationBuiltinTopicData`] — the same
7//! wire layout, minus writer-specific fields like
8//! lifespan/ownership-strength, plus reader-specific ones like the
9//! time-based filter.
10//!
11//! GUIDs + topic/type
12//! + Durability + Reliability.
13
14extern crate alloc;
15use alloc::string::String;
16use alloc::vec::Vec;
17
18use crate::endpoint_security_info::EndpointSecurityInfo;
19use crate::error::WireError;
20use crate::parameter_list::{Parameter, ParameterList, pid};
21use crate::participant_data::{Duration, ENCAPSULATION_PL_CDR_LE};
22use crate::publication_data::{
23    DurabilityKind, ReliabilityKind, ReliabilityQos, collect_locator_params, decode_cdr_string,
24    decode_duration, decode_i32, decode_liveliness, decode_partition, decode_u32,
25    encode_cdr_string_le, encode_duration_le, encode_liveliness_le, encode_locator_params,
26    encode_partition_le, encode_u32_le, guid_from_param,
27};
28use crate::wire_types::{Guid, Locator};
29
30/// Discovered Subscription / lokaler DataReader — Subset.
31///
32/// Wire-identical to [`PublicationBuiltinTopicData`] in phase 1; a
33/// separate type so the SEDP publications and subscriptions caches stay
34/// clearly distinguishable and extensions (e.g. `expects_inline_qos`,
35/// `time_based_filter`) are not pulled through the publication type.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct SubscriptionBuiltinTopicData {
38    /// Endpoint GUID (= reader GUID).
39    pub key: Guid,
40    /// GUID of the participant the reader belongs to.
41    pub participant_key: Guid,
42    /// Topic name.
43    pub topic_name: String,
44    /// IDL type name.
45    pub type_name: String,
46    /// Durability-QoS.
47    pub durability: DurabilityKind,
48    /// Reliability-QoS.
49    pub reliability: ReliabilityQos,
50    /// Ownership-QoS (Spec §2.2.3.23). Default Shared.
51    pub ownership: zerodds_qos::OwnershipKind,
52    /// Liveliness-QoS (Spec §2.2.3.11).
53    pub liveliness: zerodds_qos::LivelinessQosPolicy,
54    /// Deadline-QoS (Spec §2.2.3.7).
55    pub deadline: zerodds_qos::DeadlineQosPolicy,
56    /// Partition-QoS (Spec §2.2.3.13).
57    pub partition: Vec<String>,
58    /// UserData-QoS (Spec §2.2.3.1) — opaque sequence<octet>.
59    pub user_data: Vec<u8>,
60    /// TopicData-QoS (Spec §2.2.3.3) — opaque sequence<octet>.
61    pub topic_data: Vec<u8>,
62    /// GroupData-QoS (Spec §2.2.3.2) — opaque sequence<octet>.
63    pub group_data: Vec<u8>,
64    /// Type information as opaque bytes (XTypes §7.6.3.2.2).
65    pub type_information: Option<alloc::vec::Vec<u8>>,
66    /// Akzeptierte Data-Representations.
67    pub data_representation: alloc::vec::Vec<i16>,
68    /// Content-filter property (DDSI-RTPS §9.6.3.4 Table 9.14). Only
69    /// set if the reader was created as a `ContentFilteredTopic`.
70    pub content_filter: Option<ContentFilterProperty>,
71    /// Endpoint security info (PID 0x1004, DDS-Security 1.1 §7.4.1.5).
72    /// `None` for legacy peers. WP 4H-c matching checks writer/reader
73    /// pairs for protection compatibility.
74    pub security_info: Option<EndpointSecurityInfo>,
75    /// PID_SERVICE_INSTANCE_NAME (DDS-RPC 1.0 §7.8.2) — logical
76    /// service-instance name of an RPC endpoint.
77    pub service_instance_name: Option<String>,
78    /// PID_RELATED_ENTITY_GUID (DDS-RPC 1.0 §7.8.2) — GUID of the
79    /// counterpart endpoint (e.g. reply reader → reply writer).
80    pub related_entity_guid: Option<Guid>,
81    /// PID_TOPIC_ALIASES (DDS-RPC 1.0 §7.8.2).
82    pub topic_aliases: Option<Vec<String>>,
83    /// PID_ZERODDS_TYPE_ID (vendor PID 0x8002) — reader type identifier
84    /// for XTypes-aware matching (see `PublicationBuiltinTopicData`).
85    pub type_identifier: zerodds_types::TypeIdentifier,
86    /// Endpoint unicast locators (DDSI-RTPS 2.5 §8.5.3.2:
87    /// `DiscoveredReaderData.readerProxy.unicastLocatorList`). Where
88    /// peers send user data to *this* reader endpoint. Empty = the peer
89    /// falls back to the participant `DEFAULT_UNICAST_LOCATOR` from
90    /// SPDP (§8.5.5). OpenDDS stores the real user locator exclusively
91    /// here.
92    pub unicast_locators: Vec<Locator>,
93    /// Endpoint-Multicast-Locators (DDSI-RTPS 2.5 §8.5.3.2).
94    pub multicast_locators: Vec<Locator>,
95}
96
97/// Content-filter property — projects a content-filtered-topic
98/// reference over SEDP.
99///
100/// Spec: OMG DDS 1.4 §9.6.3.4 Table 9.14 / DDSI-RTPS 2.5 §9.6.3.4.
101/// Five strings + one string sequence, all CDR-serialized.
102#[derive(Debug, Clone, PartialEq, Eq, Default)]
103pub struct ContentFilterProperty {
104    /// Name of the content-filtered topic (derivable from the reader
105    /// identifier).
106    pub content_filtered_topic_name: String,
107    /// Name of the underlying topic (same as `topic_name` in the
108    /// SubscriptionBuiltinTopicData).
109    pub related_topic_name: String,
110    /// Filter class, e.g. `"DDSSQL"`. For type constants see
111    /// [`filter_class`].
112    pub filter_class_name: String,
113    /// Filter expression (SQL subset).
114    pub filter_expression: String,
115    /// Positional parameters (`%0`, `%1`, ...).
116    pub expression_parameters: Vec<String>,
117}
118
119/// Standard filter-class names.
120pub mod filter_class {
121    /// OMG SQL filter (DDSSQL) — the only standardized kind.
122    pub const DDSSQL: &str = "DDSSQL";
123}
124
125/// Encodes a ContentFilterProperty as a PL value (five CDR strings +
126/// sequence<string>).
127///
128/// # Errors
129/// `ValueOutOfRange` if a string is too long.
130pub fn encode_content_filter_property_le(
131    cfp: &ContentFilterProperty,
132) -> Result<Vec<u8>, WireError> {
133    let mut out = Vec::new();
134    out.extend_from_slice(&encode_cdr_string_le(&cfp.content_filtered_topic_name)?);
135    out.extend_from_slice(&encode_cdr_string_le(&cfp.related_topic_name)?);
136    out.extend_from_slice(&encode_cdr_string_le(&cfp.filter_class_name)?);
137    out.extend_from_slice(&encode_cdr_string_le(&cfp.filter_expression)?);
138    out.extend_from_slice(&encode_partition_le(&cfp.expression_parameters)?);
139    Ok(out)
140}
141
142/// Decodes a ContentFilterProperty from a PL value. `None` on a wire error.
143pub fn decode_content_filter_property(
144    value: &[u8],
145    little_endian: bool,
146) -> Option<ContentFilterProperty> {
147    let (s1, rest1) = take_cdr_string(value, little_endian)?;
148    let (s2, rest2) = take_cdr_string(rest1, little_endian)?;
149    let (s3, rest3) = take_cdr_string(rest2, little_endian)?;
150    let (s4, rest4) = take_cdr_string(rest3, little_endian)?;
151    let params = decode_partition(rest4, little_endian)?;
152    Some(ContentFilterProperty {
153        content_filtered_topic_name: s1,
154        related_topic_name: s2,
155        filter_class_name: s3,
156        filter_expression: s4,
157        expression_parameters: params,
158    })
159}
160
161/// Helper: reads **one** CDR string from a byte slice and returns the
162/// rest after 4-byte align padding.
163fn take_cdr_string(bytes: &[u8], little_endian: bool) -> Option<(String, &[u8])> {
164    if bytes.len() < 4 {
165        return None;
166    }
167    let mut lb = [0u8; 4];
168    lb.copy_from_slice(&bytes[..4]);
169    let len = if little_endian {
170        u32::from_le_bytes(lb)
171    } else {
172        u32::from_be_bytes(lb)
173    } as usize;
174    let consumed_raw = 4 + len;
175    if consumed_raw > bytes.len() {
176        return None;
177    }
178    let s = decode_cdr_string(&bytes[..consumed_raw], little_endian).ok()?;
179    // 4-byte align to the next element.
180    let padded = (consumed_raw + 3) & !3;
181    let next = padded.min(bytes.len());
182    Some((s, &bytes[next..]))
183}
184
185impl SubscriptionBuiltinTopicData {
186    /// Encodes to PL_CDR_LE bytes (with a 4-byte encapsulation header).
187    ///
188    /// Implemented directly (not delegated via
189    /// [`PublicationBuiltinTopicData`]) so that extension PIDs that are
190    /// writer-only (e.g. `PID_LIFESPAN`, `PID_OWNERSHIP_STRENGTH`) do
191    /// not accidentally end up in the subscription payload.
192    ///
193    /// # Errors
194    /// `ValueOutOfRange` if a string is longer than u32::MAX.
195    pub fn to_pl_cdr_le(&self) -> Result<Vec<u8>, WireError> {
196        let mut params = ParameterList::new();
197
198        params.push(Parameter::new(
199            pid::PARTICIPANT_GUID,
200            self.participant_key.to_bytes().to_vec(),
201        ));
202        params.push(Parameter::new(
203            pid::ENDPOINT_GUID,
204            self.key.to_bytes().to_vec(),
205        ));
206        params.push(Parameter::new(
207            pid::TOPIC_NAME,
208            encode_cdr_string_le(&self.topic_name)?,
209        ));
210        params.push(Parameter::new(
211            pid::TYPE_NAME,
212            encode_cdr_string_le(&self.type_name)?,
213        ));
214        params.push(Parameter::new(
215            pid::DURABILITY,
216            (self.durability as u32).to_le_bytes().to_vec(),
217        ));
218
219        let mut rel = Vec::with_capacity(12);
220        rel.extend_from_slice(&(self.reliability.kind as u32).to_le_bytes());
221        rel.extend_from_slice(&self.reliability.max_blocking_time.to_bytes_le());
222        params.push(Parameter::new(pid::RELIABILITY, rel));
223
224        // OWNERSHIP (reader-side: "ich akzeptiere nur diesen Ownership-Modus").
225        params.push(Parameter::new(
226            pid::OWNERSHIP,
227            encode_u32_le(self.ownership as u32).to_vec(),
228        ));
229
230        // LIVELINESS — requested lease.
231        params.push(Parameter::new(
232            pid::LIVELINESS,
233            encode_liveliness_le(self.liveliness),
234        ));
235
236        // DEADLINE — requested period.
237        params.push(Parameter::new(
238            pid::DEADLINE,
239            encode_duration_le(self.deadline.period).to_vec(),
240        ));
241
242        // PARTITION.
243        if !self.partition.is_empty() {
244            params.push(Parameter::new(
245                pid::PARTITION,
246                encode_partition_le(&self.partition)?,
247            ));
248        }
249
250        // USER_DATA / TOPIC_DATA / GROUP_DATA — opaque sequence<octet>.
251        if !self.user_data.is_empty() {
252            params.push(Parameter::new(
253                pid::USER_DATA,
254                crate::publication_data::encode_octet_seq_le(&self.user_data)?,
255            ));
256        }
257        if !self.topic_data.is_empty() {
258            params.push(Parameter::new(
259                pid::TOPIC_DATA,
260                crate::publication_data::encode_octet_seq_le(&self.topic_data)?,
261            ));
262        }
263        if !self.group_data.is_empty() {
264            params.push(Parameter::new(
265                pid::GROUP_DATA,
266                crate::publication_data::encode_octet_seq_le(&self.group_data)?,
267            ));
268        }
269
270        if let Some(ti) = &self.type_information {
271            params.push(Parameter::new(pid::TYPE_INFORMATION, ti.clone()));
272        }
273
274        if let Some(cfp) = &self.content_filter {
275            params.push(Parameter::new(
276                pid::CONTENT_FILTER_PROPERTY,
277                encode_content_filter_property_le(cfp)?,
278            ));
279        }
280
281        if let Some(info) = self.security_info {
282            params.push(Parameter::new(
283                pid::ENDPOINT_SECURITY_INFO,
284                info.to_bytes(true).to_vec(),
285            ));
286        }
287
288        // ----------------------------------------------------------------
289        // DDS-RPC 1.0 discovery PIDs (§7.8.2) — only if set.
290        // ----------------------------------------------------------------
291        if let Some(name) = &self.service_instance_name {
292            params.push(Parameter::new(
293                pid::SERVICE_INSTANCE_NAME,
294                encode_cdr_string_le(name)?,
295            ));
296        }
297        if let Some(guid) = self.related_entity_guid {
298            params.push(Parameter::new(
299                pid::RELATED_ENTITY_GUID,
300                guid.to_bytes().to_vec(),
301            ));
302        }
303        if let Some(aliases) = &self.topic_aliases {
304            params.push(Parameter::new(
305                pid::TOPIC_ALIASES,
306                encode_partition_le(aliases)?,
307            ));
308        }
309
310        // PID_ZERODDS_TYPE_ID (F-TYPES-3 Wire-up).
311        if self.type_identifier != zerodds_types::TypeIdentifier::None {
312            let mut w = zerodds_cdr::BufferWriter::new(zerodds_cdr::Endianness::Little);
313            self.type_identifier
314                .encode_into(&mut w)
315                .map_err(|_| WireError::ValueOutOfRange {
316                    message: "type_identifier encoding failed",
317                })?;
318            params.push(Parameter::new(pid::ZERODDS_TYPE_ID, w.into_bytes()));
319        }
320
321        if !self.data_representation.is_empty() {
322            let mut dr = Vec::with_capacity(4 + 2 * self.data_representation.len());
323            let len = u32::try_from(self.data_representation.len()).map_err(|_| {
324                WireError::ValueOutOfRange {
325                    message: "data_representation length exceeds u32::MAX",
326                }
327            })?;
328            dr.extend_from_slice(&len.to_le_bytes());
329            for rep in &self.data_representation {
330                dr.extend_from_slice(&rep.to_le_bytes());
331            }
332            params.push(Parameter::new(pid::DATA_REPRESENTATION, dr));
333        }
334
335        // Endpoint locators (§8.5.3.2) — one parameter per locator.
336        encode_locator_params(&mut params, pid::UNICAST_LOCATOR, &self.unicast_locators);
337        encode_locator_params(
338            &mut params,
339            pid::MULTICAST_LOCATOR,
340            &self.multicast_locators,
341        );
342
343        let mut out = Vec::with_capacity(params.parameters.len() * 24 + 16);
344        out.extend_from_slice(&ENCAPSULATION_PL_CDR_LE);
345        out.extend_from_slice(&[0, 0]); // options
346        out.extend_from_slice(&params.to_bytes(true));
347        Ok(out)
348    }
349
350    /// Decoded from PL_CDR_LE bytes (with encapsulation header).
351    ///
352    /// # Errors
353    /// `UnexpectedEof` on too-short bytes,
354    /// `UnsupportedEncapsulation` on an unknown encoding,
355    /// `ValueOutOfRange` if mandatory PIDs are missing.
356    pub fn from_pl_cdr_le(bytes: &[u8]) -> Result<Self, WireError> {
357        if bytes.len() < 4 {
358            return Err(WireError::UnexpectedEof {
359                needed: 4,
360                offset: 0,
361            });
362        }
363        let little_endian = match &bytes[..2] {
364            b if b == ENCAPSULATION_PL_CDR_LE => true,
365            [0x00, 0x02] => false,
366            other => {
367                return Err(WireError::UnsupportedEncapsulation {
368                    kind: [other[0], other[1]],
369                });
370            }
371        };
372        let pl = ParameterList::from_bytes(&bytes[4..], little_endian)?;
373
374        let key = pl
375            .find(pid::ENDPOINT_GUID)
376            .and_then(guid_from_param)
377            .ok_or(WireError::ValueOutOfRange {
378                message: "ENDPOINT_GUID missing or wrong length",
379            })?;
380        let participant_key = pl
381            .find(pid::PARTICIPANT_GUID)
382            .and_then(guid_from_param)
383            .unwrap_or_else(|| Guid::new(key.prefix, crate::wire_types::EntityId::PARTICIPANT));
384        let topic_name = pl
385            .find(pid::TOPIC_NAME)
386            .map(|p| decode_cdr_string(&p.value, little_endian))
387            .transpose()?
388            .ok_or(WireError::ValueOutOfRange {
389                message: "TOPIC_NAME missing",
390            })?;
391        let type_name = pl
392            .find(pid::TYPE_NAME)
393            .map(|p| decode_cdr_string(&p.value, little_endian))
394            .transpose()?
395            .ok_or(WireError::ValueOutOfRange {
396                message: "TYPE_NAME missing",
397            })?;
398
399        let durability = pl
400            .find(pid::DURABILITY)
401            .and_then(|p| {
402                if p.value.len() >= 4 {
403                    let mut b = [0u8; 4];
404                    b.copy_from_slice(&p.value[..4]);
405                    Some(DurabilityKind::from_u32(if little_endian {
406                        u32::from_le_bytes(b)
407                    } else {
408                        u32::from_be_bytes(b)
409                    }))
410                } else {
411                    None
412                }
413            })
414            .unwrap_or_default();
415
416        let reliability = pl
417            .find(pid::RELIABILITY)
418            .and_then(|p| {
419                if p.value.len() >= 12 {
420                    let mut k = [0u8; 4];
421                    k.copy_from_slice(&p.value[..4]);
422                    let kind = ReliabilityKind::from_u32(if little_endian {
423                        u32::from_le_bytes(k)
424                    } else {
425                        u32::from_be_bytes(k)
426                    });
427                    let mut d = [0u8; 8];
428                    d.copy_from_slice(&p.value[4..12]);
429                    let max_blocking_time = if little_endian {
430                        Duration::from_bytes_le(d)
431                    } else {
432                        let mut s = [0u8; 4];
433                        s.copy_from_slice(&d[..4]);
434                        let mut f = [0u8; 4];
435                        f.copy_from_slice(&d[4..]);
436                        Duration {
437                            seconds: i32::from_be_bytes(s),
438                            fraction: u32::from_be_bytes(f),
439                        }
440                    };
441                    Some(ReliabilityQos {
442                        kind,
443                        max_blocking_time,
444                    })
445                } else {
446                    None
447                }
448            })
449            .unwrap_or_default();
450
451        let ownership = pl
452            .find(pid::OWNERSHIP)
453            .and_then(|p| decode_u32(&p.value, little_endian))
454            .map(zerodds_qos::OwnershipKind::from_u32)
455            .unwrap_or_default();
456        // OWNERSHIP_STRENGTH ignored (writer-only).
457        let _ = decode_i32;
458
459        let liveliness = pl
460            .find(pid::LIVELINESS)
461            .and_then(|p| decode_liveliness(&p.value, little_endian))
462            .unwrap_or_default();
463
464        let deadline = pl
465            .find(pid::DEADLINE)
466            .and_then(|p| decode_duration(&p.value, little_endian))
467            .map(|period| zerodds_qos::DeadlineQosPolicy { period })
468            .unwrap_or_default();
469
470        let partition = pl
471            .find(pid::PARTITION)
472            .and_then(|p| decode_partition(&p.value, little_endian))
473            .unwrap_or_default();
474
475        let user_data = pl
476            .find(pid::USER_DATA)
477            .and_then(|p| crate::publication_data::decode_octet_seq(&p.value, little_endian))
478            .unwrap_or_default();
479        let topic_data = pl
480            .find(pid::TOPIC_DATA)
481            .and_then(|p| crate::publication_data::decode_octet_seq(&p.value, little_endian))
482            .unwrap_or_default();
483        let group_data = pl
484            .find(pid::GROUP_DATA)
485            .and_then(|p| crate::publication_data::decode_octet_seq(&p.value, little_endian))
486            .unwrap_or_default();
487
488        let type_information = pl.find(pid::TYPE_INFORMATION).map(|p| p.value.clone());
489        let content_filter = pl
490            .find(pid::CONTENT_FILTER_PROPERTY)
491            .and_then(|p| decode_content_filter_property(&p.value, little_endian));
492
493        let security_info = pl
494            .find(pid::ENDPOINT_SECURITY_INFO)
495            .and_then(|p| EndpointSecurityInfo::from_bytes(&p.value, little_endian).ok());
496
497        let service_instance_name = pl
498            .find(pid::SERVICE_INSTANCE_NAME)
499            .map(|p| decode_cdr_string(&p.value, little_endian))
500            .transpose()
501            .ok()
502            .flatten();
503        let related_entity_guid = pl.find(pid::RELATED_ENTITY_GUID).and_then(guid_from_param);
504        let topic_aliases = pl
505            .find(pid::TOPIC_ALIASES)
506            .and_then(|p| decode_partition(&p.value, little_endian));
507
508        let type_identifier = pl
509            .find(pid::ZERODDS_TYPE_ID)
510            .and_then(|p| {
511                let mut r =
512                    zerodds_cdr::BufferReader::new(&p.value, zerodds_cdr::Endianness::Little);
513                zerodds_types::TypeIdentifier::decode_from(&mut r).ok()
514            })
515            .unwrap_or_default();
516
517        let data_representation = pl
518            .find(pid::DATA_REPRESENTATION)
519            .map(|p| {
520                let v = &p.value;
521                if v.len() < 4 {
522                    return Vec::new();
523                }
524                let mut n_bytes = [0u8; 4];
525                n_bytes.copy_from_slice(&v[..4]);
526                let n = if little_endian {
527                    u32::from_le_bytes(n_bytes)
528                } else {
529                    u32::from_be_bytes(n_bytes)
530                } as usize;
531                let cap = n.min(v.len().saturating_sub(4) / 2);
532                let mut reps = Vec::with_capacity(cap);
533                for i in 0..n {
534                    let off = 4 + i * 2;
535                    if off + 2 > v.len() {
536                        break;
537                    }
538                    let mut b = [0u8; 2];
539                    b.copy_from_slice(&v[off..off + 2]);
540                    reps.push(if little_endian {
541                        i16::from_le_bytes(b)
542                    } else {
543                        i16::from_be_bytes(b)
544                    });
545                }
546                reps
547            })
548            .unwrap_or_default();
549
550        Ok(Self {
551            key,
552            participant_key,
553            topic_name,
554            type_name,
555            durability,
556            reliability,
557            ownership,
558            liveliness,
559            deadline,
560            partition,
561            user_data,
562            topic_data,
563            group_data,
564            type_information,
565            data_representation,
566            content_filter,
567            security_info,
568            service_instance_name,
569            related_entity_guid,
570            topic_aliases,
571            type_identifier,
572            unicast_locators: collect_locator_params(&pl, pid::UNICAST_LOCATOR, little_endian),
573            multicast_locators: collect_locator_params(&pl, pid::MULTICAST_LOCATOR, little_endian),
574        })
575    }
576}
577
578#[cfg(test)]
579#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
580mod tests {
581    use super::*;
582    use crate::participant_data::Duration;
583    use crate::publication_data::ReliabilityKind;
584    use crate::wire_types::{EntityId, GuidPrefix};
585
586    #[test]
587    fn roundtrip_le() {
588        let s = SubscriptionBuiltinTopicData {
589            key: Guid::new(
590                GuidPrefix::from_bytes([2; 12]),
591                EntityId::user_reader_with_key([0xA0, 0xB0, 0xC0]),
592            ),
593            participant_key: Guid::new(GuidPrefix::from_bytes([2; 12]), EntityId::PARTICIPANT),
594            topic_name: "ChatterTopic".into(),
595            type_name: "std_msgs::String".into(),
596            durability: DurabilityKind::TransientLocal,
597            reliability: ReliabilityQos {
598                kind: ReliabilityKind::Reliable,
599                max_blocking_time: Duration::from_secs(5),
600            },
601            ownership: zerodds_qos::OwnershipKind::Shared,
602            liveliness: zerodds_qos::LivelinessQosPolicy::default(),
603            deadline: zerodds_qos::DeadlineQosPolicy::default(),
604            partition: alloc::vec::Vec::new(),
605            user_data: alloc::vec::Vec::new(),
606            topic_data: alloc::vec::Vec::new(),
607            group_data: alloc::vec::Vec::new(),
608            type_information: None,
609            data_representation: alloc::vec::Vec::new(),
610            content_filter: None,
611            security_info: None,
612            service_instance_name: None,
613            related_entity_guid: None,
614            topic_aliases: None,
615            type_identifier: zerodds_types::TypeIdentifier::None,
616            unicast_locators: alloc::vec::Vec::new(),
617            multicast_locators: alloc::vec::Vec::new(),
618        };
619        let bytes = s.to_pl_cdr_le().unwrap();
620        let decoded = SubscriptionBuiltinTopicData::from_pl_cdr_le(&bytes).unwrap();
621        assert_eq!(decoded, s);
622    }
623
624    #[test]
625    fn security_info_roundtrip() {
626        use crate::endpoint_security_info::{EndpointSecurityInfo, attrs, plugin_attrs};
627        let s = SubscriptionBuiltinTopicData {
628            key: Guid::new(
629                GuidPrefix::from_bytes([4; 12]),
630                EntityId::user_reader_with_key([0x10, 0x20, 0x30]),
631            ),
632            participant_key: Guid::new(GuidPrefix::from_bytes([4; 12]), EntityId::PARTICIPANT),
633            topic_name: "ST".into(),
634            type_name: "Foo".into(),
635            durability: DurabilityKind::Volatile,
636            reliability: ReliabilityQos::default(),
637            ownership: zerodds_qos::OwnershipKind::Shared,
638            liveliness: zerodds_qos::LivelinessQosPolicy::default(),
639            deadline: zerodds_qos::DeadlineQosPolicy::default(),
640            partition: alloc::vec::Vec::new(),
641            user_data: alloc::vec::Vec::new(),
642            topic_data: alloc::vec::Vec::new(),
643            group_data: alloc::vec::Vec::new(),
644            type_information: None,
645            data_representation: alloc::vec::Vec::new(),
646            content_filter: None,
647            security_info: Some(EndpointSecurityInfo {
648                endpoint_security_attributes: attrs::IS_VALID | attrs::IS_PAYLOAD_PROTECTED,
649                plugin_endpoint_security_attributes: plugin_attrs::IS_VALID
650                    | plugin_attrs::IS_PAYLOAD_ENCRYPTED,
651            }),
652            service_instance_name: None,
653            related_entity_guid: None,
654            topic_aliases: None,
655            type_identifier: zerodds_types::TypeIdentifier::None,
656            unicast_locators: alloc::vec::Vec::new(),
657            multicast_locators: alloc::vec::Vec::new(),
658        };
659        let bytes = s.to_pl_cdr_le().unwrap();
660        let decoded = SubscriptionBuiltinTopicData::from_pl_cdr_le(&bytes).unwrap();
661        assert_eq!(decoded.security_info, s.security_info);
662    }
663
664    #[test]
665    fn content_filter_property_roundtrip_le() {
666        let cfp = ContentFilterProperty {
667            content_filtered_topic_name: "FilteredShapes".into(),
668            related_topic_name: "Square".into(),
669            filter_class_name: filter_class::DDSSQL.into(),
670            filter_expression: "color = %0 AND x > %1".into(),
671            expression_parameters: alloc::vec!["'RED'".into(), "50".into()],
672        };
673        let bytes = encode_content_filter_property_le(&cfp).unwrap();
674        let decoded = decode_content_filter_property(&bytes, true).unwrap();
675        assert_eq!(decoded, cfp);
676    }
677
678    #[test]
679    fn subscription_with_content_filter_roundtrip_le() {
680        let cfp = ContentFilterProperty {
681            content_filtered_topic_name: "Filt".into(),
682            related_topic_name: "Square".into(),
683            filter_class_name: "DDSSQL".into(),
684            filter_expression: "x > %0".into(),
685            expression_parameters: alloc::vec!["42".into()],
686        };
687        let s = SubscriptionBuiltinTopicData {
688            key: Guid::new(
689                GuidPrefix::from_bytes([3; 12]),
690                EntityId::user_reader_with_key([1, 2, 3]),
691            ),
692            participant_key: Guid::new(GuidPrefix::from_bytes([3; 12]), EntityId::PARTICIPANT),
693            topic_name: "Square".into(),
694            type_name: "ShapeType".into(),
695            durability: DurabilityKind::Volatile,
696            reliability: ReliabilityQos::default(),
697            ownership: zerodds_qos::OwnershipKind::Shared,
698            liveliness: zerodds_qos::LivelinessQosPolicy::default(),
699            deadline: zerodds_qos::DeadlineQosPolicy::default(),
700            partition: alloc::vec::Vec::new(),
701            user_data: alloc::vec::Vec::new(),
702            topic_data: alloc::vec::Vec::new(),
703            group_data: alloc::vec::Vec::new(),
704            type_information: None,
705            data_representation: alloc::vec::Vec::new(),
706            content_filter: Some(cfp.clone()),
707            security_info: None,
708            service_instance_name: None,
709            related_entity_guid: None,
710            topic_aliases: None,
711            type_identifier: zerodds_types::TypeIdentifier::None,
712            unicast_locators: alloc::vec::Vec::new(),
713            multicast_locators: alloc::vec::Vec::new(),
714        };
715        let bytes = s.to_pl_cdr_le().unwrap();
716        let decoded = SubscriptionBuiltinTopicData::from_pl_cdr_le(&bytes).unwrap();
717        assert_eq!(decoded.content_filter, Some(cfp));
718    }
719
720    fn sample_sub() -> SubscriptionBuiltinTopicData {
721        SubscriptionBuiltinTopicData {
722            key: Guid::new(
723                GuidPrefix::from_bytes([5; 12]),
724                EntityId::user_reader_with_key([0xA0, 0xB0, 0xC0]),
725            ),
726            participant_key: Guid::new(GuidPrefix::from_bytes([5; 12]), EntityId::PARTICIPANT),
727            topic_name: "Calc_Reply".into(),
728            type_name: "Calc::Reply".into(),
729            durability: DurabilityKind::Volatile,
730            reliability: ReliabilityQos::default(),
731            ownership: zerodds_qos::OwnershipKind::Shared,
732            liveliness: zerodds_qos::LivelinessQosPolicy::default(),
733            deadline: zerodds_qos::DeadlineQosPolicy::default(),
734            partition: alloc::vec::Vec::new(),
735            user_data: alloc::vec::Vec::new(),
736            topic_data: alloc::vec::Vec::new(),
737            group_data: alloc::vec::Vec::new(),
738            type_information: None,
739            data_representation: alloc::vec::Vec::new(),
740            content_filter: None,
741            security_info: None,
742            service_instance_name: None,
743            related_entity_guid: None,
744            topic_aliases: None,
745            type_identifier: zerodds_types::TypeIdentifier::None,
746            unicast_locators: alloc::vec::Vec::new(),
747            multicast_locators: alloc::vec::Vec::new(),
748        }
749    }
750
751    #[test]
752    fn rpc_discovery_pids_roundtrip_subscription() {
753        let mut s = sample_sub();
754        s.service_instance_name = Some("SrvInst".into());
755        s.related_entity_guid = Some(Guid::new(
756            GuidPrefix::from_bytes([5; 12]),
757            EntityId::user_writer_with_key([1, 2, 3]),
758        ));
759        s.topic_aliases = Some(alloc::vec!["Alias1".into(), "Alias2".into()]);
760        let bytes = s.to_pl_cdr_le().unwrap();
761        let decoded = SubscriptionBuiltinTopicData::from_pl_cdr_le(&bytes).unwrap();
762        assert_eq!(decoded.service_instance_name, s.service_instance_name);
763        assert_eq!(decoded.related_entity_guid, s.related_entity_guid);
764        assert_eq!(decoded.topic_aliases, s.topic_aliases);
765    }
766
767    #[test]
768    fn rpc_pids_optional_legacy_subscription_parses_ok() {
769        let s = sample_sub();
770        let bytes = s.to_pl_cdr_le().unwrap();
771        let decoded = SubscriptionBuiltinTopicData::from_pl_cdr_le(&bytes).unwrap();
772        assert!(decoded.service_instance_name.is_none());
773        assert!(decoded.related_entity_guid.is_none());
774        assert!(decoded.topic_aliases.is_none());
775    }
776}
zerodds_rtps/subscription_data.rs

zerodds_rtps/
subscription_data.rs
