zerodds_webdds_bridge/
dynamic_rpc.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! A [`zerodds_dcps::dds_type::DdsType`] for a DDS-RPC request/reply
5//! payload whose structure is only known at run time (synthesized from
6//! an IDL `interface` operation's parameters, DDS-RPC 1.0
7//! §7.5.1.1.4/§7.5.1.1.5) — the same problem
8//! `zerodds_dcps::serialized_type::SerializedSample` solves for a topic
9//! sample, and the same reason it gives for why its own `DdsType::decode`
10//! is a stub: "serialized samples require their topic support", which
11//! `zerodds_rpc::requester::Requester<TIn, TOut>` /
12//! `zerodds_rpc::replier::Replier<TIn, TOut>` have no room for — they
13//! call `TOut::decode(bytes)` as a bare associated function, with no
14//! instance to carry a `SerializedTypeSupport` reference through.
15//!
16//! [`DynamicRpcValue::encode`] needs no such help (`&self` already
17//! carries the `Ast` + type name the value was decoded/built against).
18//! [`DynamicRpcValue::decode`] does, and gets it from
19//! [`with_decode_context`]: a thread-local slot the caller sets
20//! immediately before the one `Requester`/`Replier` call that will
21//! invoke `decode` and clears immediately after.
22//!
23//! **Why this is sound here and would not be in general**: every
24//! `Requester`/`Replier` call that can reach `decode` —
25//! `send_request_blocking` (`crate::requester`'s own doc: "internally
26//! calls `tick()` in a polling loop") and `Replier::tick` ("the caller
27//! orchestrates the tick itself — no dedicated thread") — runs
28//! synchronously on the thread that calls it, never on a background
29//! receive thread. `crate::rpc_dispatch` calls both exactly this way:
30//! one blocking round trip per SOAP request, on the thread handling
31//! that request, with the context set for exactly its duration. A
32//! caller that instead used `send_request_async` (a different thread
33//! polls the `mpsc::Receiver` and calls `tick()` there) would see the
34//! wrong context if it raced this one — `rpc_dispatch` never does.
35
36use std::cell::RefCell;
37use std::sync::Arc;
38
39use zerodds_ast::Ast;
40use zerodds_ast_layout::{Dialect, Encoding, Endianness};
41use zerodds_dcps::dds_type::{DdsType, DecodeError, EncodeError};
42use zerodds_types::dynamic::DynamicData;
43use zerodds_wire_codec::ProbeRequest;
44
45thread_local! {
46    static DECODE_CONTEXT: RefCell<Option<(Arc<Ast>, String, Dialect)>> = const { RefCell::new(None) };
47}
48
49/// Sets the `(Ast, qualified type name, CDR dialect)`
50/// [`DynamicRpcValue::decode`] reads for the duration of `f`, then restores
51/// whatever context (if any) was set before — nested calls on the same
52/// thread compose correctly, though `crate::rpc_dispatch` never nests them.
53/// The dialect is the participant's `wire.codec_dialect`.
54pub fn with_decode_context<R>(
55    ast: &Arc<Ast>,
56    type_name: &str,
57    dialect: Dialect,
58    f: impl FnOnce() -> R,
59) -> R {
60    let previous =
61        DECODE_CONTEXT.with(|c| c.replace(Some((Arc::clone(ast), type_name.to_string(), dialect))));
62    let result = f();
63    DECODE_CONTEXT.with(|c| *c.borrow_mut() = previous);
64    result
65}
66
67/// A DDS-RPC request/reply value: a [`DynamicData`] plus the `Ast`
68/// (and qualified type name) it was built or decoded against, so
69/// [`Self::encode`] and [`crate::dynxml`] can convert it without
70/// needing any other context.
71#[derive(Clone)]
72pub struct DynamicRpcValue {
73    /// The model this value's fields are described against.
74    pub ast: Arc<Ast>,
75    /// The value's qualified type name within `ast`.
76    pub type_name: String,
77    /// The value itself.
78    pub data: DynamicData,
79    /// The CDR dialect it is encoded in; the specification's unless
80    /// [`Self::in_dialect`] names another.
81    pub dialect: Dialect,
82}
83
84impl std::fmt::Debug for DynamicRpcValue {
85    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86        f.debug_struct("DynamicRpcValue")
87            .field("type_name", &self.type_name)
88            .field("data", &self.data)
89            .finish_non_exhaustive()
90    }
91}
92
93impl DynamicRpcValue {
94    /// Constructor.
95    #[must_use]
96    pub fn new(ast: Arc<Ast>, type_name: String, data: DynamicData) -> Self {
97        Self {
98            ast,
99            type_name,
100            data,
101            dialect: Dialect::default(),
102        }
103    }
104
105    /// The same value, encoded in `dialect` — the participant's
106    /// `wire.codec_dialect`.
107    #[must_use]
108    pub fn in_dialect(mut self, dialect: Dialect) -> Self {
109        self.dialect = dialect;
110        self
111    }
112
113    fn probe_request_in(&self, order: Endianness, encoding: Encoding) -> ProbeRequest {
114        ProbeRequest::in_dialect(&self.type_name, encoding, order, self.dialect)
115    }
116
117    /// The value written in `order` and `encoding` at `start`, behind the
118    /// first `start` octets of a sample (`0` for a value on its own), which
119    /// the reflective codec does for either byte order.
120    fn encode_in(
121        &self,
122        order: Endianness,
123        encoding: Encoding,
124        start: usize,
125        out: &mut Vec<u8>,
126    ) -> Result<(), EncodeError> {
127        let bytes = zerodds_wire_codec::encode_continuing(
128            &self.ast,
129            &self.probe_request_in(order, encoding),
130            &self.data,
131            start,
132        )
133        .map_err(|_| EncodeError::Invalid {
134            what: "dynamic RPC payload refused by the wire codec",
135        })?;
136        out.extend(bytes);
137        Ok(())
138    }
139
140    /// A value read in `order` and `encoding` from `start` of `bytes`,
141    /// against the decode context the caller set.
142    fn decode_in(
143        order: Endianness,
144        encoding: Encoding,
145        bytes: &[u8],
146        start: usize,
147    ) -> Result<Self, DecodeError> {
148        let (ast, type_name, dialect) =
149            DECODE_CONTEXT
150                .with(|c| c.borrow().clone())
151                .ok_or(DecodeError::Invalid {
152                    what: "no decode context set — see crate::dynamic_rpc::with_decode_context",
153                })?;
154        let request = ProbeRequest::in_dialect(&type_name, encoding, order, dialect);
155        let data =
156            zerodds_wire_codec::decode_continuing::<DynamicData>(&ast, &request, bytes, start)
157                .map_err(|_| DecodeError::Invalid {
158                    what: "dynamic RPC payload refused by the wire codec",
159                })?;
160        Ok(Self {
161            ast,
162            type_name,
163            data,
164            dialect,
165        })
166    }
167}
168
169impl DdsType for DynamicRpcValue {
170    const TYPE_NAME: &'static str =
171        "zerodds::webdds_bridge::DynamicRpcValue (requires crate::dynamic_rpc's decode context)";
172
173    fn encode(&self, out: &mut Vec<u8>) -> Result<(), EncodeError> {
174        self.encode_in(Endianness::Little, Encoding::PlainCdr2, 0, out)
175    }
176
177    /// The reflective codec writes big-endian as readily as little-endian,
178    /// so this answers instead of taking the trait's refusal.
179    fn encode_be(&self, out: &mut Vec<u8>) -> Result<(), EncodeError> {
180        self.encode_in(Endianness::Big, Encoding::PlainCdr2, 0, out)
181    }
182
183    fn decode(bytes: &[u8]) -> Result<Self, DecodeError> {
184        Self::decode_in(Endianness::Little, Encoding::PlainCdr2, bytes, 0)
185    }
186
187    fn decode_be(bytes: &[u8]) -> Result<Self, DecodeError> {
188        Self::decode_in(Endianness::Big, Encoding::PlainCdr2, bytes, 0)
189    }
190
191    fn encode_xcdr1(&self, out: &mut Vec<u8>) -> Result<(), EncodeError> {
192        self.encode_in(Endianness::Little, Encoding::PlainCdr1, 0, out)
193    }
194
195    fn decode_xcdr1(bytes: &[u8]) -> Result<Self, DecodeError> {
196        Self::decode_in(Endianness::Little, Encoding::PlainCdr1, bytes, 0)
197    }
198
199    /// The value behind the first octets of a DDS-RPC sample, alignment
200    /// counted from the sample (DDS-RPC 1.0 §7.5.1.1.6).
201    fn encode_continuing(&self, prefix: &mut Vec<u8>, xcdr1: bool) -> Result<(), EncodeError> {
202        let encoding = if xcdr1 {
203            Encoding::PlainCdr1
204        } else {
205            Encoding::PlainCdr2
206        };
207        self.encode_in(Endianness::Little, encoding, prefix.len(), prefix)
208    }
209
210    fn decode_continuing(bytes: &[u8], offset: usize, xcdr1: bool) -> Result<Self, DecodeError> {
211        let encoding = if xcdr1 {
212            Encoding::PlainCdr1
213        } else {
214            Encoding::PlainCdr2
215        };
216        Self::decode_in(Endianness::Little, encoding, bytes, offset)
217    }
218
219    fn field_value(&self, path: &str) -> Option<zerodds_dcps::FilterValue> {
220        use zerodds_dcps::FilterValue as Value;
221        let mut value = &self.data;
222        for component in path.split('.') {
223            value = value.member(component)?;
224        }
225        match value {
226            DynamicData::Boolean(v) => Some(Value::Bool(*v)),
227            DynamicData::Integer { value, .. } => i64::try_from(*value).ok().map(Value::Int),
228            DynamicData::Float { value, .. } => Some(Value::Float(*value)),
229            DynamicData::Text(v) | DynamicData::Character { value: v, .. } => {
230                Some(Value::String(v.clone()))
231            }
232            _ => None,
233        }
234    }
235}
236
237#[cfg(test)]
238#[allow(clippy::unwrap_used, clippy::expect_used)]
239mod tests {
240    use super::*;
241    use zerodds_types::dynamic::DynamicType;
242
243    fn model(idl: &str, type_name: &str) -> (Arc<Ast>, DynamicType) {
244        let ast = Arc::new(zerodds_wire_probe::encode::model_from_idl(idl, "t.idl").unwrap());
245        let dt = DynamicType::find(&ast, type_name).unwrap();
246        (ast, dt)
247    }
248
249    #[test]
250    fn encode_needs_no_external_context() {
251        let (ast, _dt) = model(
252            "module M { @final struct Params { long a; long b; }; };",
253            "M::Params",
254        );
255        let data = DynamicData::Aggregate(vec![
256            (
257                "a".to_string(),
258                DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 3),
259            ),
260            (
261                "b".to_string(),
262                DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 4),
263            ),
264        ]);
265        let value = DynamicRpcValue::new(ast, "M::Params".to_string(), data);
266        let mut out = Vec::new();
267        value.encode(&mut out).expect("encode");
268        assert!(!out.is_empty());
269    }
270
271    /// The participant's CDR dialect reaches the octets of a call: in the
272    /// specification's a @key member of a @mutable type carries the
273    /// must-understand flag in its EMHEADER, in Fast DDS's it does not
274    /// (Fast DDS writes `00 00 00 20` for member 0 with LC 2). A value
275    /// decodes back in the dialect it was written in.
276    #[test]
277    fn the_codec_dialect_reaches_the_octets_of_a_call() {
278        let (ast, _dt) = model(
279            "module M { @mutable struct Call { @key long id; long v; }; };",
280            "M::Call",
281        );
282        let data = DynamicData::Aggregate(vec![
283            (
284                "id".to_string(),
285                DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 7),
286            ),
287            (
288                "v".to_string(),
289                DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 9),
290            ),
291        ]);
292        let spec = DynamicRpcValue::new(Arc::clone(&ast), "M::Call".to_string(), data);
293        let fastdds = spec.clone().in_dialect(Dialect::FastDds);
294        let (mut a, mut b) = (Vec::new(), Vec::new());
295        spec.encode(&mut a).expect("encode");
296        fastdds.encode(&mut b).expect("encode");
297        // DHEADER, then the first EMHEADER.
298        assert_eq!(a[4..8], [0x00, 0x00, 0x00, 0xa0], "{a:02x?}");
299        assert_eq!(b[4..8], [0x00, 0x00, 0x00, 0x20], "{b:02x?}");
300        let decoded = with_decode_context(&ast, "M::Call", Dialect::FastDds, || {
301            DynamicRpcValue::decode(&b)
302        })
303        .expect("decode");
304        assert_eq!(decoded.dialect, Dialect::FastDds);
305        assert_eq!(decoded.data.member("id"), fastdds.data.member("id"));
306        assert_eq!(decoded.data.member("v"), fastdds.data.member("v"));
307    }
308
309    #[test]
310    fn decode_without_context_is_a_named_error_not_a_panic() {
311        let err = DynamicRpcValue::decode(&[0, 0, 0, 0]).unwrap_err();
312        assert!(matches!(err, DecodeError::Invalid { .. }));
313    }
314
315    #[test]
316    fn round_trips_through_encode_and_decode_with_context_set() {
317        let (ast, _dt) = model(
318            "module M { @final struct Params { long a; long b; }; };",
319            "M::Params",
320        );
321        let data = DynamicData::Aggregate(vec![
322            (
323                "a".to_string(),
324                DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 7),
325            ),
326            (
327                "b".to_string(),
328                DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 9),
329            ),
330        ]);
331        let value = DynamicRpcValue::new(Arc::clone(&ast), "M::Params".to_string(), data);
332        let mut bytes = Vec::new();
333        value.encode(&mut bytes).expect("encode");
334
335        let decoded = with_decode_context(&ast, "M::Params", Dialect::default(), || {
336            DynamicRpcValue::decode(&bytes)
337        })
338        .expect("decode");
339        assert_eq!(decoded.data.member("a"), value.data.member("a"));
340        assert_eq!(decoded.data.member("b"), value.data.member("b"));
341    }
342
343    #[test]
344    fn context_is_restored_after_the_call_not_leaked() {
345        let (ast, _dt) = model(
346            "module M { @final struct Params { long a; }; };",
347            "M::Params",
348        );
349        let data = DynamicData::Aggregate(vec![(
350            "a".to_string(),
351            DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, 1),
352        )]);
353        let value = DynamicRpcValue::new(Arc::clone(&ast), "M::Params".to_string(), data);
354        let mut bytes = Vec::new();
355        value.encode(&mut bytes).unwrap();
356        let _ = with_decode_context(&ast, "M::Params", Dialect::default(), || {
357            DynamicRpcValue::decode(&bytes)
358        });
359        // Outside the closure, no context is set — the next bare decode
360        // (as the daemon's own request handler will see it between two
361        // unrelated calls) fails loudly instead of reusing a stale one.
362        let err = DynamicRpcValue::decode(&bytes).unwrap_err();
363        assert!(matches!(err, DecodeError::Invalid { .. }));
364    }
365}
zerodds_webdds_bridge/dynamic_rpc.rs

zerodds_webdds_bridge/
dynamic_rpc.rs
