diff --git a/Cargo.lock b/Cargo.lock index 069a18ca4..ab50fa53c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -273,6 +273,26 @@ version = "3.19.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" +[[package]] +name = "bytemuck" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" +dependencies = [ + "bytemuck_derive", +] + +[[package]] +name = "bytemuck_derive" +version = "1.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9abbd1bc6865053c427f7198e6af43bfdedc55ab791faed4fbd361d789575ff" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + [[package]] name = "byteorder" version = "1.5.0" @@ -2748,6 +2768,7 @@ dependencies = [ name = "ros-z-cdr" version = "0.1.0" dependencies = [ + "bytemuck", "byteorder", "serde", "thiserror 1.0.69", @@ -2812,6 +2833,7 @@ name = "ros-z-msgs" version = "0.1.0" dependencies = [ "anyhow", + "bytemuck", "byteorder", "glob", "once_cell", diff --git a/crates/ros-z-cdr/Cargo.toml b/crates/ros-z-cdr/Cargo.toml index c6ba0545e..0ddd49663 100644 --- a/crates/ros-z-cdr/Cargo.toml +++ b/crates/ros-z-cdr/Cargo.toml @@ -7,6 +7,7 @@ description = "CDR serialization for ros-z" [dependencies] serde = { workspace = true } byteorder = { workspace = true } +bytemuck = { version = "1", features = ["derive", "extern_crate_alloc"] } zenoh-buffers = { workspace = true } thiserror = "1.0" diff --git a/crates/ros-z-cdr/src/lib.rs b/crates/ros-z-cdr/src/lib.rs index 3544acf58..a9f8ee44e 100644 --- a/crates/ros-z-cdr/src/lib.rs +++ b/crates/ros-z-cdr/src/lib.rs @@ -17,8 +17,10 @@ pub mod buffer; pub mod deserializer; pub mod error; +pub mod plain; pub mod primitives; pub mod serializer; +pub mod traits; pub mod zbuf_writer; use std::cell::RefCell; @@ -36,8 +38,11 @@ pub use buffer::CdrBuffer; pub use byteorder::{BigEndian, LittleEndian}; pub use deserializer::{CdrDeserializer, from_bytes, from_bytes_with}; pub use error::{Error, Result}; +#[cfg(target_endian = "little")] +pub use plain::CdrPlain; pub use primitives::{CdrReader, CdrWriter}; pub use serializer::{CdrSerializer, to_buffer, to_vec, to_vec_reuse}; +pub use traits::{CdrDeserialize, CdrSerialize, CdrSerializedSize, cdr_to_vec}; pub use zbuf_writer::ZBufWriter; /// Native endian type alias for the current platform. diff --git a/crates/ros-z-cdr/src/plain.rs b/crates/ros-z-cdr/src/plain.rs new file mode 100644 index 000000000..66c83813a --- /dev/null +++ b/crates/ros-z-cdr/src/plain.rs @@ -0,0 +1,65 @@ +//! `CdrPlain` marker trait for types whose CDR wire layout equals their memory layout. +//! +//! When a type is `CdrPlain`, sequences of that type can be serialized and deserialized +//! with a single bulk memcpy instead of element-by-element encoding. +//! +//! # Safety invariants +//! +//! A type `T: CdrPlain` must satisfy **all** of: +//! 1. No padding bytes — `bytemuck::Pod` guarantees this at compile time. +//! 2. CDR wire layout == in-memory layout on little-endian hosts. For all ROS primitive +//! numeric types this holds: CDR encodes them in native byte order (LE) without +//! reordering fields or adding framing. +//! 3. Every possible bit pattern is a valid `T` — again `bytemuck::Pod`. +//! +//! This trait is only defined on little-endian targets because CDR uses little-endian +//! encoding for all primitive types. On a big-endian host the wire bytes would need +//! byte-swapping per element, making the bulk-copy path incorrect. + +/// Marker trait for types whose CDR serialized form is identical to their in-memory +/// representation on little-endian hosts. +/// +/// # Safety +/// Implementors must guarantee that: +/// - The type has no padding bytes. +/// - The CDR wire layout of the type matches its in-memory layout (true for all ROS +/// numeric primitives on LE hosts). +/// - Every possible bit pattern is a valid value (i.e. the type is `bytemuck::Pod`). +/// +/// The `bytemuck::Pod` bound is enforced at the usage sites (`write_pod_slice`, +/// `read_pod_slice`) rather than here so that blanket impls for `[T; N]` can be +/// expressed — `bytemuck::Pod` is only impl'd for fixed array sizes up to 64. +/// +/// This trait should only be implemented by codegen for generated message types, or +/// manually for well-known primitive types defined in this crate. +#[cfg(target_endian = "little")] +pub unsafe trait CdrPlain: Copy + 'static {} + +// ── Primitive impls ────────────────────────────────────────────────────────── +// bool is excluded: bytemuck::Pod is not impl'd for bool (only 0/1 are valid). +// char is excluded: Rust char is 4-byte Unicode; CDR wchar is 2 bytes. + +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for f32 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for f64 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for i8 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for u8 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for i16 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for u16 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for i32 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for u32 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for i64 {} +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for u64 {} + +// Fixed arrays of plain types are themselves plain. +#[cfg(target_endian = "little")] +unsafe impl CdrPlain for [T; N] {} diff --git a/crates/ros-z-cdr/src/primitives.rs b/crates/ros-z-cdr/src/primitives.rs index 28aedf243..0843fe0de 100644 --- a/crates/ros-z-cdr/src/primitives.rs +++ b/crates/ros-z-cdr/src/primitives.rs @@ -5,6 +5,8 @@ //! by the serde-based serializer/deserializer and can also be used directly //! for schema-driven (dynamic) message handling. +#[cfg(target_endian = "little")] +use bytemuck; use byteorder::{ByteOrder, ReadBytesExt}; use std::marker::PhantomData; @@ -159,6 +161,20 @@ impl<'a, BO: ByteOrder, B: CdrBuffer> CdrWriter<'a, BO, B> { pub fn write_sequence_length(&mut self, len: usize) { self.write_u32(len as u32); } + + /// Bulk-write a slice of plain (POD) values as raw bytes. + /// + /// The caller must write the sequence length prefix separately before calling this. + /// Alignment is handled internally based on `T`'s alignment requirement. + /// + /// Only available on little-endian hosts where CDR wire layout == memory layout. + #[cfg(target_endian = "little")] + #[inline] + pub fn write_pod_slice(&mut self, slice: &[T]) { + debug_assert!(!slice.is_empty()); + self.align(std::mem::align_of::()); + self.buffer.extend_from_slice(bytemuck::cast_slice(slice)); + } } /// Low-level CDR reader with alignment handling. @@ -356,6 +372,31 @@ impl<'a, BO: ByteOrder> CdrReader<'a, BO> { let len = self.read_u32()? as usize; self.read_bytes(len) } + + /// Bulk-read `count` plain (POD) values as a zero-copy borrowed slice. + /// + /// The caller must have already read the sequence length prefix. + /// Alignment is handled internally based on `T`'s alignment requirement. + /// + /// Only available on little-endian hosts where CDR wire layout == memory layout. + #[cfg(target_endian = "little")] + #[inline] + pub fn read_pod_slice( + &mut self, + count: usize, + ) -> Result> { + if count == 0 { + return Ok(vec![]); + } + self.align(std::mem::align_of::())?; + let byte_count = count + .checked_mul(std::mem::size_of::()) + .ok_or(Error::UnexpectedEof)?; + let bytes = self.read_bytes(byte_count)?; + // `pod_collect_to_vec` handles misaligned input buffers safely (copies into + // a freshly aligned allocation). `cast_slice` would panic on misaligned network data. + Ok(bytemuck::pod_collect_to_vec(bytes)) + } } #[cfg(test)] diff --git a/crates/ros-z-cdr/src/traits.rs b/crates/ros-z-cdr/src/traits.rs new file mode 100644 index 000000000..3b13423be --- /dev/null +++ b/crates/ros-z-cdr/src/traits.rs @@ -0,0 +1,428 @@ +//! `CdrSerialize`, `CdrDeserialize`, and `CdrSerializedSize` traits. +//! +//! These bypass serde for CDR serialization, enabling the bulk POD fast path +//! for sequences of plain types while keeping serde derives for non-CDR uses. + +use byteorder::{ByteOrder, LittleEndian}; + +use crate::buffer::CdrBuffer; +use crate::error::Result; +use crate::primitives::{CdrReader, CdrWriter}; +use crate::zbuf_writer::ZBufWriter; + +// ── Core traits ────────────────────────────────────────────────────────────── + +pub trait CdrSerialize { + fn cdr_serialize(&self, writer: &mut CdrWriter<'_, BO, B>); +} + +pub trait CdrDeserialize: Sized { + fn cdr_deserialize<'de, BO: ByteOrder>(reader: &mut CdrReader<'de, BO>) -> Result; +} + +/// Size in bytes of the CDR-serialized form, given the current stream alignment position. +pub trait CdrSerializedSize { + fn cdr_serialized_size(&self, current_alignment: usize) -> usize; +} + +// ── Entry points ───────────────────────────────────────────────────────────── + +/// Serialize `value` into a new `Vec` using CDR little-endian encoding. +pub fn cdr_to_vec(value: &T, capacity_hint: usize) -> Vec { + let mut buffer = Vec::with_capacity(capacity_hint); + let mut writer = CdrWriter::::new(&mut buffer); + value.cdr_serialize(&mut writer); + buffer +} + +/// Serialize `value` into a `ZBufWriter` (for zero-copy Zenoh transport). +pub fn cdr_to_zbuf_writer(value: &T, writer: &mut ZBufWriter) { + let mut cdr_writer = CdrWriter::::new(writer); + value.cdr_serialize(&mut cdr_writer); +} + +// ── Primitive CdrSerialize impls ───────────────────────────────────────────── + +impl CdrSerialize for bool { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_bool(*self); + } +} + +impl CdrSerialize for i8 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_i8(*self); + } +} + +impl CdrSerialize for u8 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_u8(*self); + } +} + +impl CdrSerialize for i16 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_i16(*self); + } +} + +impl CdrSerialize for u16 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_u16(*self); + } +} + +impl CdrSerialize for i32 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_i32(*self); + } +} + +impl CdrSerialize for u32 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_u32(*self); + } +} + +impl CdrSerialize for i64 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_i64(*self); + } +} + +impl CdrSerialize for u64 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_u64(*self); + } +} + +impl CdrSerialize for f32 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_f32(*self); + } +} + +impl CdrSerialize for f64 { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_f64(*self); + } +} + +impl CdrSerialize for String { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_string(self); + } +} + +impl CdrSerialize for str { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_string(self); + } +} + +// Generic Vec — element-by-element with length prefix. +// Note: Vec uses this path too (sequence of u8 with u32 length prefix). +// ZBuf is handled separately in the ZBuf CdrSerialize impl in ros-z. +impl CdrSerialize for Vec { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + w.write_sequence_length(self.len()); + for item in self { + item.cdr_serialize(w); + } + } +} + +// Fixed arrays — no length prefix, element-by-element. +impl CdrSerialize for [T; N] { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + for item in self { + item.cdr_serialize(w); + } + } +} + +// ── Primitive CdrDeserialize impls ─────────────────────────────────────────── + +impl CdrDeserialize for bool { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_bool() + } +} + +impl CdrDeserialize for i8 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_i8() + } +} + +impl CdrDeserialize for u8 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_u8() + } +} + +impl CdrDeserialize for i16 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_i16() + } +} + +impl CdrDeserialize for u16 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_u16() + } +} + +impl CdrDeserialize for i32 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_i32() + } +} + +impl CdrDeserialize for u32 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_u32() + } +} + +impl CdrDeserialize for i64 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_i64() + } +} + +impl CdrDeserialize for u64 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_u64() + } +} + +impl CdrDeserialize for f32 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_f32() + } +} + +impl CdrDeserialize for f64 { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_f64() + } +} + +impl CdrDeserialize for String { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + r.read_string() + } +} + +// Generic Vec — element-by-element with length prefix. +impl CdrDeserialize for Vec { + #[inline] + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + let count = r.read_sequence_length()?; + let mut out = Vec::with_capacity(count); + for _ in 0..count { + out.push(T::cdr_deserialize(r)?); + } + Ok(out) + } +} + +// Fixed arrays — no length prefix, element-by-element. +impl CdrDeserialize for [T; N] { + fn cdr_deserialize<'de, BO: ByteOrder>(r: &mut CdrReader<'de, BO>) -> Result { + // Can't use array::try_from_fn on stable yet, so build via Vec. + let mut v: Vec = Vec::with_capacity(N); + for _ in 0..N { + v.push(T::cdr_deserialize(r)?); + } + // SAFETY: we just pushed exactly N elements. + let arr: [T; N] = v.try_into().unwrap_or_else(|_| unreachable!()); + Ok(arr) + } +} + +// ── Primitive CdrSerializedSize impls ──────────────────────────────────────── + +/// Alignment helper: bytes needed to align `pos` to `align`. +#[inline] +fn padding(pos: usize, align: usize) -> usize { + let modulo = pos % align; + if modulo == 0 { 0 } else { align - modulo } +} + +impl CdrSerializedSize for bool { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + 1 + } +} + +impl CdrSerializedSize for i8 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + 1 + } +} + +impl CdrSerializedSize for u8 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + 1 + } +} + +impl CdrSerializedSize for i16 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 2) + 2 + } +} + +impl CdrSerializedSize for u16 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 2) + 2 + } +} + +impl CdrSerializedSize for i32 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 4) + 4 + } +} + +impl CdrSerializedSize for u32 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 4) + 4 + } +} + +impl CdrSerializedSize for i64 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 8) + 8 + } +} + +impl CdrSerializedSize for u64 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 8) + 8 + } +} + +impl CdrSerializedSize for f32 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 4) + 4 + } +} + +impl CdrSerializedSize for f64 { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + padding(pos, 8) + 8 + } +} + +impl CdrSerializedSize for String { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + // u32 length prefix (4-byte aligned) + string bytes + null terminator + let after_len = pos + padding(pos, 4) + 4; + after_len + self.len() + 1 + } +} + +impl CdrSerializedSize for Vec { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + let mut p = pos + padding(pos, 4) + 4; // sequence length u32 + for item in self { + p = item.cdr_serialized_size(p); + } + p + } +} + +impl CdrSerializedSize for [T; N] { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + let mut p = pos; + for item in self { + p = item.cdr_serialized_size(p); + } + p + } +} + +#[cfg(test)] +mod tests { + use super::*; + use byteorder::LittleEndian; + + fn roundtrip(value: &T) -> T { + let buf = cdr_to_vec(value, 64); + let mut reader = CdrReader::::new(&buf); + T::cdr_deserialize(&mut reader).expect("deserialize failed") + } + + #[test] + #[allow(clippy::approx_constant, clippy::bool_assert_comparison)] + fn test_primitives_roundtrip() { + assert_eq!(roundtrip(&42i32), 42i32); + assert_eq!(roundtrip(&3.14f64), 3.14f64); + assert_eq!(roundtrip(&true), true); + assert_eq!(roundtrip(&"hello".to_string()), "hello".to_string()); + } + + #[test] + fn test_vec_roundtrip() { + let v: Vec = vec![1, 2, 3, 4, 5]; + assert_eq!(roundtrip(&v), v); + } + + #[test] + fn test_vec_u8_roundtrip() { + // Vec uses the generic Vec path (u32 count + elements) + let v: Vec = vec![10, 20, 30]; + assert_eq!(roundtrip(&v), v); + } + + #[test] + fn test_string_roundtrip() { + let s = "hello, world".to_string(); + assert_eq!(roundtrip(&s), s); + } +} diff --git a/crates/ros-z-codegen/src/generator/rust.rs b/crates/ros-z-codegen/src/generator/rust.rs index 5990a9854..939855aba 100644 --- a/crates/ros-z-codegen/src/generator/rust.rs +++ b/crates/ros-z-codegen/src/generator/rust.rs @@ -30,6 +30,474 @@ impl GenerationContext { } } +// ── Plainness detection ─────────────────────────────────────────────────────── + +/// Returns true if the field type has a CDR wire layout identical to its +/// in-memory layout (i.e., it qualifies for `CdrPlain`). +/// +/// A type is plain iff: +/// - It is a fixed-size numeric primitive (not bool, string, char/wchar). +/// - OR it is a nested struct that is itself plain (looked up in `plain_types`). +/// - AND it has no unbounded/bounded sequence dimension. +/// +/// Fixed arrays of plain types are themselves plain. +pub fn is_field_plain(field_type: &FieldType, plain_types: &HashSet) -> bool { + // Sequences (Vec) are never plain — they carry a variable-length prefix. + if matches!( + field_type.array, + ArrayType::Unbounded | ArrayType::Bounded(_) + ) { + return false; + } + // base type check + is_base_type_plain( + &field_type.base_type, + field_type.package.as_deref(), + plain_types, + ) +} + +fn is_base_type_plain( + base_type: &str, + package: Option<&str>, + plain_types: &HashSet, +) -> bool { + match base_type { + // bool: CDR bool is u8(0|1), bytemuck::Pod not impl'd for bool + "bool" | "string" | "wstring" | "char" | "wchar" => false, + "byte" | "uint8" | "int8" | "uint16" | "int16" | "uint32" | "int32" | "uint64" + | "int64" | "float32" | "float64" => true, + custom => { + // Look up in the set of already-confirmed-plain struct types. + let key = match package { + Some(pkg) => format!("{}::{}", pkg, custom), + None => custom.to_string(), + }; + plain_types.contains(&key) + } + } +} + +/// Returns the alignment (in bytes) of a primitive base type, or None for +/// non-primitive / custom types. +fn primitive_align(base_type: &str) -> Option { + match base_type { + "byte" | "uint8" | "int8" => Some(1), + "uint16" | "int16" => Some(2), + "uint32" | "int32" | "float32" => Some(4), + "uint64" | "int64" | "float64" => Some(8), + _ => None, + } +} + +/// Compute the set of plain struct types for a slice of resolved messages. +/// +/// Returns a `HashSet` where each entry is `"package::TypeName"`. +/// The computation is bottom-up: a struct is plain iff all its fields are plain +/// AND the struct has no inter-field or trailing padding in C/Rust repr(C). +/// +/// Padding detection: a struct has no padding iff all fields share the same +/// alignment (or more precisely, every field's alignment divides the struct's +/// natural alignment uniformly). We use the conservative rule: all primitive +/// fields must have the same alignment. Nested plain structs are assumed to +/// already satisfy this invariant. +pub fn compute_plain_types(messages: &[ResolvedMessage]) -> HashSet { + // Iterate until stable (handles mutually-dependent plain structs, though rare). + let mut plain: HashSet = HashSet::new(); + loop { + let before = plain.len(); + 'msg: for msg in messages { + let key = format!("{}::{}", msg.parsed.package, msg.parsed.name); + if plain.contains(&key) { + continue; + } + // All fields must individually be plain. + if !msg + .parsed + .fields + .iter() + .all(|f| is_field_plain(&f.field_type, &plain)) + { + continue; + } + // Additionally, all direct primitive fields must share the same alignment + // to guarantee no inter-field or trailing padding. + let mut max_align: Option = None; + for field in &msg.parsed.fields { + if let Some(align) = primitive_align(&field.field_type.base_type) { + match max_align { + None => max_align = Some(align), + Some(existing) if existing != align => continue 'msg, // mixed alignment → padding + _ => {} + } + } + // Custom (nested plain) types: their alignment is inherited from their + // own max primitive. We don't re-check here — they are already validated. + } + plain.insert(key); + } + if plain.len() == before { + break; // stable + } + } + plain +} + +// ── CDR trait codegen ───────────────────────────────────────────────────────── + +/// Generate `CdrSerialize`, `CdrDeserialize`, `CdrSerializedSize` impls, +/// and (when the struct is plain) the `CdrPlain` + `bytemuck::Pod/Zeroable` derives. +fn generate_cdr_impls( + msg: &ResolvedMessage, + plain_types: &HashSet, + ctx: &GenerationContext, +) -> Result { + let name = format_ident!("{}", msg.parsed.name); + let fields = &msg.parsed.fields; + let pkg = &msg.parsed.package; + let is_plain = plain_types.contains(&format!("{}::{}", pkg, &msg.parsed.name)); + + // ── CdrSerialize ────────────────────────────────────────────────────────── + let ser_fields: Vec = fields + .iter() + .map(|f| generate_cdr_serialize_field(f, pkg, plain_types, ctx)) + .collect::>>()?; + + let ser_impl = quote! { + impl ::ros_z_cdr::CdrSerialize for #name { + fn cdr_serialize( + &self, + __w: &mut ::ros_z_cdr::CdrWriter<'_, BO, B>, + ) + where + BO: ::byteorder::ByteOrder, + B: ::ros_z_cdr::CdrBuffer, + { + #(#ser_fields)* + } + } + }; + + // ── CdrDeserialize ──────────────────────────────────────────────────────── + let de_fields: Vec = fields + .iter() + .map(|f| generate_cdr_deserialize_field(f, pkg, plain_types, ctx)) + .collect::>>()?; + + let field_idents: Vec = fields.iter().map(|f| escape_field_name(&f.name)).collect(); + + let de_impl = quote! { + impl ::ros_z_cdr::CdrDeserialize for #name { + fn cdr_deserialize<'__de, BO>( + __r: &mut ::ros_z_cdr::CdrReader<'__de, BO>, + ) -> ::ros_z_cdr::Result + where + BO: ::byteorder::ByteOrder, + { + #(#de_fields)* + Ok(Self { #(#field_idents),* }) + } + } + }; + + // ── CdrSerializedSize ───────────────────────────────────────────────────── + let size_fields: Vec = fields + .iter() + .map(|f| generate_cdr_size_field(f, pkg, plain_types, ctx)) + .collect::>>()?; + + let size_impl = quote! { + impl ::ros_z_cdr::CdrSerializedSize for #name { + fn cdr_serialized_size(&self, __pos: usize) -> usize { + let mut __p = __pos; + #(#size_fields)* + __p + } + } + }; + + // ── CdrPlain (only when struct is plain) ────────────────────────────────── + let plain_impl = if is_plain { + quote! { + #[cfg(target_endian = "little")] + unsafe impl ::ros_z_cdr::CdrPlain for #name {} + } + } else { + quote! {} + }; + + Ok(quote! { + #ser_impl + #de_impl + #size_impl + #plain_impl + }) +} + +/// Generate a single field's CdrSerialize statement. +fn generate_cdr_serialize_field( + field: &Field, + source_pkg: &str, + plain_types: &HashSet, + _ctx: &GenerationContext, +) -> Result { + let fname = escape_field_name(&field.name); + let ft = &field.field_type; + + // ZBuf fields: byte sequences stored as ros_z::ZBuf (zero-copy Zenoh type). + // CdrSerialize is implemented for ros_z::ZBuf in the ros-z crate. + if is_zbuf_field(field) { + return Ok(quote! { + ::ros_z_cdr::CdrSerialize::cdr_serialize(&self.#fname, __w); + }); + } + + match &ft.array { + ArrayType::Single => Ok(quote! { + ::ros_z_cdr::CdrSerialize::cdr_serialize(&self.#fname, __w); + }), + ArrayType::Fixed(_) => { + let elem_plain = is_base_type_plain( + &ft.base_type, + ft.package.as_deref().or(Some(source_pkg)), + plain_types, + ); + if elem_plain && !matches!(ft.base_type.as_str(), "bool" | "string" | "wstring") { + Ok(quote! { + #[cfg(target_endian = "little")] + __w.write_pod_slice(&self.#fname); + #[cfg(not(target_endian = "little"))] + for __item in &self.#fname { + ::ros_z_cdr::CdrSerialize::cdr_serialize(__item, __w); + } + }) + } else { + Ok(quote! { + for __item in &self.#fname { + ::ros_z_cdr::CdrSerialize::cdr_serialize(__item, __w); + } + }) + } + } + ArrayType::Unbounded | ArrayType::Bounded(_) => { + // Check if element type is plain → bulk path + let elem_plain = is_base_type_plain( + &ft.base_type, + ft.package.as_deref().or(Some(source_pkg)), + plain_types, + ); + if elem_plain && !matches!(ft.base_type.as_str(), "bool" | "string" | "wstring") { + Ok(quote! { + __w.write_sequence_length(self.#fname.len()); + #[cfg(target_endian = "little")] + if !self.#fname.is_empty() { + __w.write_pod_slice(&self.#fname); + } + #[cfg(not(target_endian = "little"))] + for __item in &self.#fname { + ::ros_z_cdr::CdrSerialize::cdr_serialize(__item, __w); + } + }) + } else { + Ok(quote! { + __w.write_sequence_length(self.#fname.len()); + for __item in &self.#fname { + ::ros_z_cdr::CdrSerialize::cdr_serialize(__item, __w); + } + }) + } + } + } +} + +/// Generate a single field's CdrDeserialize statement (binds a local variable). +fn generate_cdr_deserialize_field( + field: &Field, + source_pkg: &str, + plain_types: &HashSet, + ctx: &GenerationContext, +) -> Result { + let fname = escape_field_name(&field.name); + let ft = &field.field_type; + let rust_elem_ty = generate_field_type_tokens_with_context(ft, source_pkg, ctx)?; + + // ZBuf: CdrDeserialize is implemented for ros_z::ZBuf in the ros-z crate. + if is_zbuf_field(field) { + return Ok(quote! { + let #fname: #rust_elem_ty = ::ros_z_cdr::CdrDeserialize::cdr_deserialize(__r)?; + }); + } + + match &ft.array { + ArrayType::Single => Ok(quote! { + let #fname = ::ros_z_cdr::CdrDeserialize::cdr_deserialize(__r)?; + }), + ArrayType::Fixed(n) => { + let n_lit = proc_macro2::Literal::usize_unsuffixed(*n); + let elem_plain = is_base_type_plain( + &ft.base_type, + ft.package.as_deref().or(Some(source_pkg)), + plain_types, + ); + let base_ty = generate_base_type_tokens_with_context(ft, source_pkg, ctx)?; + if elem_plain && !matches!(ft.base_type.as_str(), "bool" | "string" | "wstring") { + Ok(quote! { + let #fname: #rust_elem_ty = { + #[cfg(target_endian = "little")] + { + let __slice = __r.read_pod_slice::<#base_ty>(#n_lit)?; + ::std::convert::TryInto::try_into(__slice) + .map_err(|_| ::ros_z_cdr::Error::UnexpectedEof)? + } + #[cfg(not(target_endian = "little"))] + { + let mut __arr = [Default::default(); #n_lit]; + for __slot in __arr.iter_mut() { + *__slot = ::ros_z_cdr::CdrDeserialize::cdr_deserialize(__r)?; + } + __arr + } + }; + }) + } else { + Ok(quote! { + let #fname: #rust_elem_ty = { + let mut __arr = [Default::default(); #n_lit]; + for __slot in __arr.iter_mut() { + *__slot = ::ros_z_cdr::CdrDeserialize::cdr_deserialize(__r)?; + } + __arr + }; + }) + } + } + ArrayType::Unbounded | ArrayType::Bounded(_) => { + let elem_plain = is_base_type_plain( + &ft.base_type, + ft.package.as_deref().or(Some(source_pkg)), + plain_types, + ); + let base_ty = generate_base_type_tokens_with_context(ft, source_pkg, ctx)?; + if elem_plain && !matches!(ft.base_type.as_str(), "bool" | "string" | "wstring") { + Ok(quote! { + let #fname: Vec<#base_ty> = { + let __count = __r.read_sequence_length()?; + #[cfg(target_endian = "little")] + { + if __count > 0 { + __r.read_pod_slice::<#base_ty>(__count)? + } else { + vec![] + } + } + #[cfg(not(target_endian = "little"))] + { + let mut __v = Vec::with_capacity(__count); + for _ in 0..__count { + __v.push(::ros_z_cdr::CdrDeserialize::cdr_deserialize(__r)?); + } + __v + } + }; + }) + } else { + Ok(quote! { + let #fname: Vec<#base_ty> = { + let __count = __r.read_sequence_length()?; + let mut __v = Vec::with_capacity(__count); + for _ in 0..__count { + __v.push(::ros_z_cdr::CdrDeserialize::cdr_deserialize(__r)?); + } + __v + }; + }) + } + } + } +} + +/// Generate a single field's CdrSerializedSize statement (updates `__p`). +fn generate_cdr_size_field( + field: &Field, + source_pkg: &str, + plain_types: &HashSet, + _ctx: &GenerationContext, +) -> Result { + let fname = escape_field_name(&field.name); + let ft = &field.field_type; + + // ZBuf: u32 length prefix (4-byte aligned) + byte contents. + if is_zbuf_field(field) { + return Ok(quote! { + { + use ::zenoh_buffers::buffer::Buffer; + __p += (__p % 4 != 0) as usize * (4 - __p % 4) + 4; + __p += self.#fname.len(); + } + }); + } + + match &ft.array { + ArrayType::Single => Ok(quote! { + __p = ::ros_z_cdr::CdrSerializedSize::cdr_serialized_size(&self.#fname, __p); + }), + ArrayType::Fixed(_) => { + let elem_plain = is_base_type_plain( + &ft.base_type, + ft.package.as_deref().or(Some(source_pkg)), + plain_types, + ); + if elem_plain && !matches!(ft.base_type.as_str(), "bool" | "string" | "wstring") { + // O(1) size for plain fixed arrays: align + N * sizeof(T) + Ok(quote! { + if !self.#fname.is_empty() { + let __elem_align = ::std::mem::align_of_val(&self.#fname[0]); + __p += (__p % __elem_align != 0) as usize + * (__elem_align - __p % __elem_align); + __p += self.#fname.len() + * ::std::mem::size_of_val(&self.#fname[0]); + } + }) + } else { + Ok(quote! { + for __item in &self.#fname { + __p = ::ros_z_cdr::CdrSerializedSize::cdr_serialized_size(__item, __p); + } + }) + } + } + ArrayType::Unbounded | ArrayType::Bounded(_) => { + let elem_plain = is_base_type_plain( + &ft.base_type, + ft.package.as_deref().or(Some(source_pkg)), + plain_types, + ); + if elem_plain && !matches!(ft.base_type.as_str(), "bool" | "string" | "wstring") { + // O(1) size for plain sequences: align + count * sizeof(T) + Ok(quote! { + // u32 sequence length prefix + __p += (__p % 4 != 0) as usize * (4 - __p % 4) + 4; + if !self.#fname.is_empty() { + let __elem_align = ::std::mem::align_of_val(&self.#fname[0]); + __p += (__p % __elem_align != 0) as usize + * (__elem_align - __p % __elem_align); + __p += self.#fname.len() + * ::std::mem::size_of_val(&self.#fname[0]); + } + }) + } else { + Ok(quote! { + // u32 sequence length prefix + __p += (__p % 4 != 0) as usize * (4 - __p % 4) + 4; + for __item in &self.#fname { + __p = ::ros_z_cdr::CdrSerializedSize::cdr_serialized_size(__item, __p); + } + }) + } + } + } +} + /// Generate Rust module for a package containing messages pub fn generate_package_module(package: &str, messages: &[ResolvedMessage]) -> Result { let package_ident = format_ident!("{}", package); @@ -54,29 +522,41 @@ pub fn generate_message_impl(msg: &ResolvedMessage) -> Result { pub fn generate_message_impl_with_context( msg: &ResolvedMessage, ctx: &GenerationContext, +) -> Result { + generate_message_impl_with_cdr(msg, ctx, &HashSet::new()) +} + +/// Generate Rust implementation with CDR trait impls and plainness information. +pub fn generate_message_impl_with_cdr( + msg: &ResolvedMessage, + ctx: &GenerationContext, + plain_types: &HashSet, ) -> Result { let name = format_ident!("{}", msg.parsed.name); + let msg_is_plain = + plain_types.contains(&format!("{}::{}", msg.parsed.package, msg.parsed.name)); let struct_def = generate_struct_with_context( &msg.parsed.package, &msg.parsed.name, &msg.parsed.fields, &msg.parsed.constants, ctx, + msg_is_plain, )?; let type_info = generate_message_type_info(&msg.parsed.package, &msg.parsed.name, &msg.type_hash); - // No longer need custom serde - ros_z::ZBuf implements Serialize/Deserialize - - // Generate size estimation implementation let size_estimation_impl = generate_size_estimation_impl(&name, &msg.parsed.fields, &msg.parsed.package, ctx)?; + let cdr_impls = generate_cdr_impls(msg, plain_types, ctx)?; + Ok(quote! { #struct_def #type_info #size_estimation_impl + #cdr_impls }) } @@ -94,6 +574,7 @@ fn generate_struct( fields, constants, &GenerationContext::default(), + false, ) } @@ -104,6 +585,7 @@ fn generate_struct_with_context( fields: &[Field], constants: &[crate::types::Constant], ctx: &GenerationContext, + is_plain: bool, ) -> Result { let name_ident = format_ident!("{}", name); let field_defs: Vec = fields @@ -132,12 +614,22 @@ fn generate_struct_with_context( // Python bridge module path for derive macros let py_module_path = format!("ros_z_msgs_py.types.{}", package); + let bytemuck_derives = if is_plain { + quote! { + #[cfg_attr(target_endian = "little", repr(C))] + #[cfg_attr(target_endian = "little", derive(Copy, ::bytemuck::Pod, ::bytemuck::Zeroable))] + } + } else { + quote! {} + }; + if has_large_array { // Large array messages need smart-default for arrays >32 elements Ok(quote! { #[derive(Debug, Clone, ::smart_default::SmartDefault, ::serde::Serialize, ::serde::Deserialize)] #[cfg_attr(feature = "python_registry", derive(::ros_z_derive::FromPyMessage, ::ros_z_derive::IntoPyMessage))] #[cfg_attr(feature = "python_registry", ros_msg(module = #py_module_path))] + #bytemuck_derives pub struct #name_ident { #(#field_defs),* } @@ -152,6 +644,7 @@ fn generate_struct_with_context( #[derive(Debug, Clone, Default, ::serde::Serialize, ::serde::Deserialize)] #[cfg_attr(feature = "python_registry", derive(::ros_z_derive::FromPyMessage, ::ros_z_derive::IntoPyMessage))] #[cfg_attr(feature = "python_registry", ros_msg(module = #py_module_path))] + #bytemuck_derives pub struct #name_ident { #(#field_defs),* } diff --git a/crates/ros-z-codegen/src/lib.rs b/crates/ros-z-codegen/src/lib.rs index 9eb9f56de..cedbcfb3c 100644 --- a/crates/ros-z-codegen/src/lib.rs +++ b/crates/ros-z-codegen/src/lib.rs @@ -285,6 +285,11 @@ impl MessageGenerator { self.config.local_packages.clone(), ); + // Compute plain types across all messages once (bottom-up over full type graph) + let all_msgs_vec: Vec = + packages.values().flatten().map(|m| (*m).clone()).collect(); + let plain_types = generator::rust::compute_plain_types(&all_msgs_vec); + for package_name in all_package_names { let package_ident = quote::format_ident!("{}", &package_name); @@ -294,7 +299,11 @@ impl MessageGenerator { .map(|msgs| { msgs.iter() .map(|msg| { - generator::rust::generate_message_impl_with_context(msg, &gen_ctx) + generator::rust::generate_message_impl_with_cdr( + msg, + &gen_ctx, + &plain_types, + ) }) .collect::>>() }) @@ -360,7 +369,7 @@ impl MessageGenerator { // Wrap in ros module for namespacing let wrapped_tokens = quote! { - #[allow(clippy::approx_constant)] + #[allow(clippy::approx_constant, clippy::manual_is_multiple_of, clippy::let_and_return)] pub mod ros { #all_tokens } diff --git a/crates/ros-z-console/src/core/dynamic_subscriber.rs b/crates/ros-z-console/src/core/dynamic_subscriber.rs index 1105fc184..4d5a55e94 100644 --- a/crates/ros-z-console/src/core/dynamic_subscriber.rs +++ b/crates/ros-z-console/src/core/dynamic_subscriber.rs @@ -7,7 +7,7 @@ use std::{sync::Arc, time::Duration}; use flume::Receiver; use ros_z::{ - dynamic::{DynamicCdrSerdes, DynamicMessage, MessageSchema}, + dynamic::{DynamicMessage, DynamicSerdeCdrSerdes, MessageSchema}, node::ZNode, pubsub::ZSub, }; @@ -25,7 +25,7 @@ pub struct DynamicTopicSubscriber { /// Channel for receiving messages asynchronously message_rx: Receiver, /// Subscriber handle (kept alive to maintain subscription) - _subscriber: Arc>, + _subscriber: Arc>, } impl DynamicTopicSubscriber { diff --git a/crates/ros-z-msgs/Cargo.toml b/crates/ros-z-msgs/Cargo.toml index 4d923825e..d24921fab 100644 --- a/crates/ros-z-msgs/Cargo.toml +++ b/crates/ros-z-msgs/Cargo.toml @@ -17,12 +17,13 @@ zenoh-buffers = { workspace = true } prost = { workspace = true, optional = true } pyo3 = { workspace = true, optional = true } ros-z-derive = { path = "../ros-z-derive", optional = true } -ros-z-cdr = { workspace = true, optional = true } +ros-z-cdr = { workspace = true } +byteorder = { workspace = true } +bytemuck = { version = "1", features = ["derive"] } parking_lot = { workspace = true, optional = true } once_cell = { workspace = true, optional = true } [dev-dependencies] -ros-z-cdr = { workspace = true } byteorder = { workspace = true } [build-dependencies] @@ -76,7 +77,6 @@ rolling = ["ros-z/rolling"] kilted = ["ros-z/kilted"] python_registry = [ "pyo3", - "ros-z-cdr", "parking_lot", "once_cell", "ros-z-derive", diff --git a/crates/ros-z-msgs/tests/size_estimation_performance.rs b/crates/ros-z-msgs/tests/size_estimation_performance.rs index d91fbf085..f21645f57 100644 --- a/crates/ros-z-msgs/tests/size_estimation_performance.rs +++ b/crates/ros-z-msgs/tests/size_estimation_performance.rs @@ -171,7 +171,7 @@ fn test_estimated_size_matches_actual() { #[test] fn test_capacity_hint_api() { - use ros_z::msg::{CdrSerdes, ZSerializer}; + use ros_z::msg::{SerdeCdrSerdes, ZSerializer}; let cloud = PointCloud2 { header: Header { @@ -190,7 +190,7 @@ fn test_capacity_hint_api() { // Test the low-level API with explicit hint let hint = cloud.estimated_serialized_size(); - let zbuf = CdrSerdes::::serialize_to_zbuf_with_hint(&cloud, hint); + let zbuf = SerdeCdrSerdes::::serialize_to_zbuf_with_hint(&cloud, hint); assert!(zbuf.len() > 50_000); println!("Serialized with explicit hint: {} bytes", zbuf.len()); diff --git a/crates/ros-z-tests/tests/pubsub_interop.rs b/crates/ros-z-tests/tests/pubsub_interop.rs index 674ae1cc0..2528a89f2 100644 --- a/crates/ros-z-tests/tests/pubsub_interop.rs +++ b/crates/ros-z-tests/tests/pubsub_interop.rs @@ -37,7 +37,7 @@ use std::{ use common::*; use ros_z::{ Builder, WithTypeInfo, - msg::{CdrSerdes, ZDeserializer, ZMessage, ZSerializer}, + msg::{ZDeserializer, ZMessage, ZSerializer}, }; use ros_z_msgs::{ geometry_msgs::{PoseStamped, Twist, TwistStamped}, @@ -110,7 +110,7 @@ const CASES: &[InteropCase] = &[ fn ros_z_pub_to_ros2_sub(case: &InteropCase) where T: ZMessage + WithTypeInfo + Default + 'static, - CdrSerdes: for<'a> ZSerializer = &'a T>, + T::Serdes: for<'a> ZSerializer = &'a T>, { if !check_ros2_available() { eprintln!("Skipping {}: ros2 CLI not available", case.type_name); @@ -192,7 +192,7 @@ where fn ros2_pub_to_ros_z_sub(case: &InteropCase) where T: ZMessage + WithTypeInfo + 'static, - CdrSerdes: for<'a> ZDeserializer = &'a [u8]>, + T::Serdes: for<'a> ZDeserializer = &'a [u8]>, { if !check_ros2_available() { eprintln!("Skipping {}: ros2 CLI not available", case.type_name); diff --git a/crates/ros-z-tests/tests/type_description_interop.rs b/crates/ros-z-tests/tests/type_description_interop.rs index 13cd1b2ae..68814c892 100644 --- a/crates/ros-z-tests/tests/type_description_interop.rs +++ b/crates/ros-z-tests/tests/type_description_interop.rs @@ -23,7 +23,9 @@ use std::{ use common::*; use ros_z::{ Builder, - dynamic::{DynamicCdrSerdes, DynamicMessage, MessageSchema, type_description_msg_to_schema}, + dynamic::{ + DynamicMessage, DynamicSerdeCdrSerdes, MessageSchema, type_description_msg_to_schema, + }, }; use ros_z_msgs::type_description_interfaces::{ self, GetTypeDescriptionRequest, srv::GetTypeDescription, @@ -421,7 +423,7 @@ fn test_dynamic_subscriber_from_type_description() { println!("Step 3: Creating dynamic subscriber..."); let zsub = node .create_sub_impl::("chatter", None) - .with_serdes::() + .with_serdes::() .with_dyn_schema(schema.clone()) .build() .expect("Failed to create dynamic subscriber"); diff --git a/crates/ros-z/examples/protobuf_demo/src/types.rs b/crates/ros-z/examples/protobuf_demo/src/types.rs index ba93c652d..253eb72c3 100644 --- a/crates/ros-z/examples/protobuf_demo/src/types.rs +++ b/crates/ros-z/examples/protobuf_demo/src/types.rs @@ -26,6 +26,10 @@ impl MessageTypeInfo for SensorData { impl WithTypeInfo for SensorData {} +impl ZMessage for SensorData { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + // SensorData uses serde/CDR for backward compatibility with the original pub/sub demo // ========== CalculateRequest Trait Implementations ========== diff --git a/crates/ros-z/examples/z_custom_message.rs b/crates/ros-z/examples/z_custom_message.rs index 52c6b4d24..9f3905bb7 100644 --- a/crates/ros-z/examples/z_custom_message.rs +++ b/crates/ros-z/examples/z_custom_message.rs @@ -29,6 +29,10 @@ impl MessageTypeInfo for RobotStatus { impl ros_z::WithTypeInfo for RobotStatus {} +impl ros_z::msg::ZMessage for RobotStatus { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + // Custom service request #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct NavigateToRequest { @@ -49,6 +53,10 @@ impl MessageTypeInfo for NavigateToRequest { impl ros_z::WithTypeInfo for NavigateToRequest {} +impl ros_z::msg::ZMessage for NavigateToRequest { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + // Custom service response #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct NavigateToResponse { @@ -69,6 +77,10 @@ impl MessageTypeInfo for NavigateToResponse { impl ros_z::WithTypeInfo for NavigateToResponse {} +impl ros_z::msg::ZMessage for NavigateToResponse { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + // Service type definition pub struct NavigateTo; diff --git a/crates/ros-z/src/action/macros.rs b/crates/ros-z/src/action/macros.rs index 711c5e446..2c19f4b07 100644 --- a/crates/ros-z/src/action/macros.rs +++ b/crates/ros-z/src/action/macros.rs @@ -66,5 +66,28 @@ macro_rules! define_action { $action_name } } + + // Provide ZMessage impls via the serde (SerdeCdrSerdes) path for types that + // do not implement the CDR traits. Types that do implement CdrSerialize + + // CdrDeserialize + CdrSerializedSize get ZMessage automatically from the + // blanket impl in ros_z::msg and should NOT use define_action!. + impl $crate::msg::ZMessage for $goal_type + where + $goal_type: Send + Sync + 'static, + { + type Serdes = $crate::msg::SerdeCdrSerdes<$goal_type>; + } + impl $crate::msg::ZMessage for $result_type + where + $result_type: Send + Sync + 'static, + { + type Serdes = $crate::msg::SerdeCdrSerdes<$result_type>; + } + impl $crate::msg::ZMessage for $feedback_type + where + $feedback_type: Send + Sync + 'static, + { + type Serdes = $crate::msg::SerdeCdrSerdes<$feedback_type>; + } }; } diff --git a/crates/ros-z/src/action/messages.rs b/crates/ros-z/src/action/messages.rs index 4a03252d6..787b1d817 100644 --- a/crates/ros-z/src/action/messages.rs +++ b/crates/ros-z/src/action/messages.rs @@ -1,5 +1,6 @@ use std::marker::PhantomData; +use ros_z_cdr::{CdrBuffer, CdrDeserialize, CdrReader, CdrSerialize, CdrSerializedSize, CdrWriter}; use serde::{Deserialize, Serialize}; use super::{GoalId, GoalInfo, GoalStatus, ZAction}; @@ -309,3 +310,269 @@ impl crate::ServiceTypeInfo for CancelService { A::cancel_goal_type_info() } } + +// ── CDR serialization impls ─────────────────────────────────────────────────── +// Concrete (non-generic) action message types now implement CdrSerialize + +// CdrDeserialize + CdrSerializedSize directly, so the blanket +// `impl ZMessage for T` covers them automatically. +// +// Generic types (GoalRequest, etc.) still use the serde path via explicit +// ZMessage impls below until ZAction's associated type bounds are updated. + +impl CdrSerialize for GoalStatusInfo { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.goal_info.cdr_serialize(w); + self.status.cdr_serialize(w); + } +} +impl CdrDeserialize for GoalStatusInfo { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GoalStatusInfo { + goal_info: GoalInfo::cdr_deserialize(r)?, + status: GoalStatus::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for GoalStatusInfo { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.goal_info.cdr_serialized_size(pos); + self.status.cdr_serialized_size(p) + } +} + +impl CdrSerialize for CancelGoalRequest { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.goal_info.cdr_serialize(w); + } +} +impl CdrDeserialize for CancelGoalRequest { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(CancelGoalRequest { + goal_info: GoalInfo::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for CancelGoalRequest { + fn cdr_serialized_size(&self, pos: usize) -> usize { + self.goal_info.cdr_serialized_size(pos) + } +} + +impl CdrSerialize for CancelGoalResponse { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.return_code.cdr_serialize(w); + self.goals_canceling.cdr_serialize(w); + } +} +impl CdrDeserialize for CancelGoalResponse { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(CancelGoalResponse { + return_code: i8::cdr_deserialize(r)?, + goals_canceling: Vec::::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for CancelGoalResponse { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.return_code.cdr_serialized_size(pos); + self.goals_canceling.cdr_serialized_size(p) + } +} + +impl CdrSerialize for GoalResponse { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.accepted.cdr_serialize(w); + self.stamp_sec.cdr_serialize(w); + self.stamp_nanosec.cdr_serialize(w); + } +} +impl CdrDeserialize for GoalResponse { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GoalResponse { + accepted: bool::cdr_deserialize(r)?, + stamp_sec: i32::cdr_deserialize(r)?, + stamp_nanosec: u32::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for GoalResponse { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.accepted.cdr_serialized_size(pos); + let p = self.stamp_sec.cdr_serialized_size(p); + self.stamp_nanosec.cdr_serialized_size(p) + } +} + +impl CdrSerialize for ResultRequest { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.goal_id.cdr_serialize(w); + } +} +impl CdrDeserialize for ResultRequest { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(ResultRequest { + goal_id: GoalId::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for ResultRequest { + fn cdr_serialized_size(&self, pos: usize) -> usize { + self.goal_id.cdr_serialized_size(pos) + } +} + +impl CdrSerialize for StatusMessage { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.status_list.cdr_serialize(w); + } +} +impl CdrDeserialize for StatusMessage { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(StatusMessage { + status_list: Vec::::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for StatusMessage { + fn cdr_serialized_size(&self, pos: usize) -> usize { + self.status_list.cdr_serialized_size(pos) + } +} + +impl CdrSerialize for SendGoalResponse { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.accepted.cdr_serialize(w); + self.stamp_sec.cdr_serialize(w); + self.stamp_nanosec.cdr_serialize(w); + } +} +impl CdrDeserialize for SendGoalResponse { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(SendGoalResponse { + accepted: bool::cdr_deserialize(r)?, + stamp_sec: i32::cdr_deserialize(r)?, + stamp_nanosec: u32::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for SendGoalResponse { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.accepted.cdr_serialized_size(pos); + let p = self.stamp_sec.cdr_serialized_size(p); + self.stamp_nanosec.cdr_serialized_size(p) + } +} + +impl CdrSerialize for GetResultRequest { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.goal_id.cdr_serialize(w); + } +} +impl CdrDeserialize for GetResultRequest { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GetResultRequest { + goal_id: GoalId::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for GetResultRequest { + fn cdr_serialized_size(&self, pos: usize) -> usize { + self.goal_id.cdr_serialized_size(pos) + } +} + +impl CdrSerialize for CancelGoalServiceRequest { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.goal_info.cdr_serialize(w); + } +} +impl CdrDeserialize for CancelGoalServiceRequest { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(CancelGoalServiceRequest { + goal_info: GoalInfo::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for CancelGoalServiceRequest { + fn cdr_serialized_size(&self, pos: usize) -> usize { + self.goal_info.cdr_serialized_size(pos) + } +} + +impl CdrSerialize for CancelGoalServiceResponse { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.return_code.cdr_serialize(w); + self.goals_canceling.cdr_serialize(w); + } +} +impl CdrDeserialize for CancelGoalServiceResponse { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(CancelGoalServiceResponse { + return_code: i8::cdr_deserialize(r)?, + goals_canceling: Vec::::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for CancelGoalServiceResponse { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.return_code.cdr_serialized_size(pos); + self.goals_canceling.cdr_serialized_size(p) + } +} + +// ── Generic types: still use serde path until ZAction gains CDR bounds ──────── + +impl crate::msg::ZMessage for GoalRequest +where + A::Goal: Send + Sync + serde::Serialize + for<'de> serde::Deserialize<'de> + 'static, +{ + type Serdes = crate::msg::SerdeCdrSerdes>; +} + +impl crate::msg::ZMessage for ResultResponse +where + A::Result: Send + Sync + serde::Serialize + for<'de> serde::Deserialize<'de> + 'static, +{ + type Serdes = crate::msg::SerdeCdrSerdes>; +} + +impl crate::msg::ZMessage for FeedbackMessage +where + A::Feedback: Send + Sync + serde::Serialize + for<'de> serde::Deserialize<'de> + 'static, +{ + type Serdes = crate::msg::SerdeCdrSerdes>; +} + +impl crate::msg::ZMessage for SendGoalRequest +where + A::Goal: Send + Sync + serde::Serialize + for<'de> serde::Deserialize<'de> + 'static, +{ + type Serdes = crate::msg::SerdeCdrSerdes>; +} + +impl crate::msg::ZMessage for GetResultResponse +where + A::Result: Send + Sync + serde::Serialize + for<'de> serde::Deserialize<'de> + 'static, +{ + type Serdes = crate::msg::SerdeCdrSerdes>; +} diff --git a/crates/ros-z/src/action/mod.rs b/crates/ros-z/src/action/mod.rs index 822602741..b200a6fa1 100644 --- a/crates/ros-z/src/action/mod.rs +++ b/crates/ros-z/src/action/mod.rs @@ -1,4 +1,5 @@ use crate::msg::ZMessage; +use ros_z_cdr::{CdrBuffer, CdrDeserialize, CdrReader, CdrSerialize, CdrSerializedSize, CdrWriter}; use serde::{Deserialize, Serialize}; use std::time::SystemTime; @@ -369,3 +370,111 @@ pub fn transition_goal_state(current: GoalStatus, event: GoalEvent) -> GoalStatu _ => GoalStatus::Unknown, } } + +// ── CDR serialization impls ─────────────────────────────────────────────────── +// These allow GoalId, GoalStatus, Time, and GoalInfo to satisfy the +// CdrSerialize + CdrDeserialize + CdrSerializedSize bounds, which in turn +// lets the action message types use the NativeCdrSerdes blanket ZMessage impl. + +impl CdrSerialize for GoalId { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.0.cdr_serialize(w); + } +} + +impl CdrDeserialize for GoalId { + #[inline] + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GoalId(<[u8; 16]>::cdr_deserialize(r)?)) + } +} + +impl CdrSerializedSize for GoalId { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + self.0.cdr_serialized_size(pos) + } +} + +impl CdrSerialize for GoalStatus { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + (*self as i8).cdr_serialize(w); + } +} + +impl CdrDeserialize for GoalStatus { + #[inline] + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + let v = i8::cdr_deserialize(r)?; + GoalStatus::try_from(v).map_err(ros_z_cdr::error::Error::Custom) + } +} + +impl CdrSerializedSize for GoalStatus { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + pos + 1 + } +} + +impl CdrSerialize for Time { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.sec.cdr_serialize(w); + self.nanosec.cdr_serialize(w); + } +} + +impl CdrDeserialize for Time { + #[inline] + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(Time { + sec: i32::cdr_deserialize(r)?, + nanosec: u32::cdr_deserialize(r)?, + }) + } +} + +impl CdrSerializedSize for Time { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.sec.cdr_serialized_size(pos); + self.nanosec.cdr_serialized_size(p) + } +} + +impl CdrSerialize for GoalInfo { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.goal_id.cdr_serialize(w); + self.stamp.cdr_serialize(w); + } +} + +impl CdrDeserialize for GoalInfo { + #[inline] + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GoalInfo { + goal_id: GoalId::cdr_deserialize(r)?, + stamp: Time::cdr_deserialize(r)?, + }) + } +} + +impl CdrSerializedSize for GoalInfo { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.goal_id.cdr_serialized_size(pos); + self.stamp.cdr_serialized_size(p) + } +} diff --git a/crates/ros-z/src/dynamic/mod.rs b/crates/ros-z/src/dynamic/mod.rs index ecd8fe2fc..2e028604b 100644 --- a/crates/ros-z/src/dynamic/mod.rs +++ b/crates/ros-z/src/dynamic/mod.rs @@ -73,7 +73,7 @@ pub use error::DynamicError; pub use message::{DynamicMessage, DynamicMessageBuilder}; pub use registry::{SchemaRegistry, get_schema, has_schema, register_schema}; pub use schema::{FieldSchema, FieldType, MessageSchema, MessageSchemaBuilder}; -pub use serdes::DynamicCdrSerdes; +pub use serdes::DynamicSerdeCdrSerdes; pub use serialization::SerializationFormat; pub use type_description::{MessageSchemaTypeDescription, type_description_msg_to_schema}; pub use type_description_client::TypeDescriptionClient; @@ -92,18 +92,18 @@ use crate::pubsub::{ZPub, ZPubBuilder, ZSub, ZSubBuilder}; // Implement ZMessage for DynamicMessage impl ZMessage for DynamicMessage { - type Serdes = DynamicCdrSerdes; + type Serdes = DynamicSerdeCdrSerdes; } // Type aliases for convenience /// Type alias for a dynamic message publisher. -pub type DynPub = ZPub; +pub type DynPub = ZPub; /// Type alias for a dynamic message subscriber. -pub type DynSub = ZSub; +pub type DynSub = ZSub; /// Type alias for a dynamic message publisher builder. -pub type DynPubBuilder = ZPubBuilder; +pub type DynPubBuilder = ZPubBuilder; /// Type alias for a dynamic message subscriber builder. -pub type DynSubBuilder = ZSubBuilder; +pub type DynSubBuilder = ZSubBuilder; diff --git a/crates/ros-z/src/dynamic/serdes.rs b/crates/ros-z/src/dynamic/serdes.rs index 3405fd1cd..a11789555 100644 --- a/crates/ros-z/src/dynamic/serdes.rs +++ b/crates/ros-z/src/dynamic/serdes.rs @@ -1,6 +1,6 @@ //! Serialization/deserialization implementations for dynamic messages. //! -//! This module provides `DynamicCdrSerdes` which implements the `ZSerializer` +//! This module provides `DynamicSerdeCdrSerdes` which implements the `ZSerializer` //! and `ZDeserializer` traits, allowing `DynamicMessage` to be used with //! the standard `ZPub`/`ZSub` infrastructure. @@ -22,25 +22,25 @@ use super::schema::MessageSchema; /// # Example /// /// ```ignore -/// use ros_z::dynamic::{DynamicMessage, DynamicCdrSerdes, MessageSchema}; +/// use ros_z::dynamic::{DynamicMessage, DynamicSerdeCdrSerdes, MessageSchema}; /// use ros_z::pubsub::{ZPub, ZSub}; /// /// // Publisher - schema is embedded in DynamicMessage -/// let publisher: ZPub = node +/// let publisher: ZPub = node /// .create_pub("/topic") -/// .with_serdes::() +/// .with_serdes::() /// .build()?; /// /// // Subscriber - schema must be provided via with_dyn_schema() -/// let subscriber: ZSub = node +/// let subscriber: ZSub = node /// .create_sub("/topic") -/// .with_serdes::() +/// .with_serdes::() /// .with_dyn_schema(schema) /// .build()?; /// ``` -pub struct DynamicCdrSerdes; +pub struct DynamicSerdeCdrSerdes; -impl ZSerializer for DynamicCdrSerdes { +impl ZSerializer for DynamicSerdeCdrSerdes { type Input<'a> = &'a DynamicMessage; fn serialize_to_zbuf(input: &DynamicMessage) -> ZBuf { @@ -97,7 +97,7 @@ impl ZSerializer for DynamicCdrSerdes { } } -impl ZDeserializer for DynamicCdrSerdes { +impl ZDeserializer for DynamicSerdeCdrSerdes { type Input<'a> = (&'a [u8], &'a Arc); type Output = DynamicMessage; type Error = DynamicError; @@ -131,7 +131,7 @@ mod tests { msg.set("y", 2.0f64).unwrap(); msg.set("z", 3.0f64).unwrap(); - let zbuf = DynamicCdrSerdes::serialize_to_zbuf(&msg); + let zbuf = DynamicSerdeCdrSerdes::serialize_to_zbuf(&msg); assert!(zbuf.len() > 0); } @@ -144,10 +144,10 @@ mod tests { msg.set("z", 3.5f64).unwrap(); // Serialize - let bytes = DynamicCdrSerdes::serialize(&msg); + let bytes = DynamicSerdeCdrSerdes::serialize(&msg); // Deserialize - let deserialized = DynamicCdrSerdes::deserialize((&bytes, &schema)).unwrap(); + let deserialized = DynamicSerdeCdrSerdes::deserialize((&bytes, &schema)).unwrap(); assert_eq!(deserialized.get::("x").unwrap(), 1.5); assert_eq!(deserialized.get::("y").unwrap(), 2.5); @@ -163,10 +163,10 @@ mod tests { msg.set("z", 3.0f64).unwrap(); let mut buffer = Vec::new(); - DynamicCdrSerdes::serialize_to_buf(&msg, &mut buffer); + DynamicSerdeCdrSerdes::serialize_to_buf(&msg, &mut buffer); // Should match serialize() output - let direct = DynamicCdrSerdes::serialize(&msg); + let direct = DynamicSerdeCdrSerdes::serialize(&msg); assert_eq!(buffer, direct); } } diff --git a/crates/ros-z/src/dynamic/tests/pubsub_tests.rs b/crates/ros-z/src/dynamic/tests/pubsub_tests.rs index fb7ccf3be..f4e330718 100644 --- a/crates/ros-z/src/dynamic/tests/pubsub_tests.rs +++ b/crates/ros-z/src/dynamic/tests/pubsub_tests.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use crate::dynamic::message::DynamicMessage; use crate::dynamic::schema::{FieldType, MessageSchema}; -use crate::dynamic::serdes::DynamicCdrSerdes; +use crate::dynamic::serdes::DynamicSerdeCdrSerdes; use crate::msg::{ZDeserializer, ZSerializer}; fn create_test_schema() -> Arc { @@ -54,7 +54,7 @@ fn test_complex_schema_for_pubsub() { assert_eq!(twist.fields.len(), 2); } -// Tests for unified pub/sub using DynamicCdrSerdes +// Tests for unified pub/sub using DynamicSerdeCdrSerdes #[test] fn test_dynamic_cdr_serdes_roundtrip() { @@ -64,12 +64,12 @@ fn test_dynamic_cdr_serdes_roundtrip() { msg.set("y", 2.5f64).unwrap(); msg.set("z", 3.5f64).unwrap(); - // Serialize using DynamicCdrSerdes (ZSerializer trait) - let bytes = DynamicCdrSerdes::serialize(&msg); + // Serialize using DynamicSerdeCdrSerdes (ZSerializer trait) + let bytes = DynamicSerdeCdrSerdes::serialize(&msg); assert!(!bytes.is_empty()); - // Deserialize using DynamicCdrSerdes (ZDeserializer trait) - let deserialized = DynamicCdrSerdes::deserialize((&bytes, &schema)).unwrap(); + // Deserialize using DynamicSerdeCdrSerdes (ZDeserializer trait) + let deserialized = DynamicSerdeCdrSerdes::deserialize((&bytes, &schema)).unwrap(); assert_eq!(deserialized.get::("x").unwrap(), 1.5); assert_eq!(deserialized.get::("y").unwrap(), 2.5); @@ -85,12 +85,12 @@ fn test_dynamic_cdr_serdes_zbuf() { msg.set("data", "Hello, unified pubsub!").unwrap(); // Serialize to ZBuf - let zbuf = DynamicCdrSerdes::serialize_to_zbuf(&msg); + let zbuf = DynamicSerdeCdrSerdes::serialize_to_zbuf(&msg); assert!(zbuf.len() > 0); // Convert to bytes and deserialize let bytes: Vec = zbuf.contiguous().to_vec(); - let deserialized = DynamicCdrSerdes::deserialize((&bytes, &schema)).unwrap(); + let deserialized = DynamicSerdeCdrSerdes::deserialize((&bytes, &schema)).unwrap(); assert_eq!( deserialized.get::("data").unwrap(), @@ -108,14 +108,14 @@ fn test_dynamic_cdr_serdes_to_buf() { // Serialize to existing buffer let mut buffer = Vec::new(); - DynamicCdrSerdes::serialize_to_buf(&msg, &mut buffer); + DynamicSerdeCdrSerdes::serialize_to_buf(&msg, &mut buffer); // Should match serialize() output - let direct = DynamicCdrSerdes::serialize(&msg); + let direct = DynamicSerdeCdrSerdes::serialize(&msg); assert_eq!(buffer, direct); // Verify deserialize works - let deserialized = DynamicCdrSerdes::deserialize((&buffer, &schema)).unwrap(); + let deserialized = DynamicSerdeCdrSerdes::deserialize((&buffer, &schema)).unwrap(); assert_eq!(deserialized.get::("x").unwrap(), 10.0); } @@ -188,7 +188,7 @@ fn test_zpub_builder_with_dyn_schema() { #[test] fn test_zpub_builder_with_serdes_preserves_schema() { - use crate::dynamic::{DynamicCdrSerdes, DynamicMessage}; + use crate::dynamic::{DynamicMessage, DynamicSerdeCdrSerdes}; use crate::pubsub::ZPubBuilder; use std::marker::PhantomData; @@ -210,7 +210,7 @@ fn test_zpub_builder_with_serdes_preserves_schema() { }; // Convert serdes type - schema should be preserved - let builder: ZPubBuilder = builder.with_serdes(); + let builder: ZPubBuilder = builder.with_serdes(); assert!(builder.dyn_schema.is_some()); assert_eq!( builder.dyn_schema.as_ref().unwrap().type_name, diff --git a/crates/ros-z/src/dynamic/type_description_service.rs b/crates/ros-z/src/dynamic/type_description_service.rs index 24ea5bebd..0107ab89d 100644 --- a/crates/ros-z/src/dynamic/type_description_service.rs +++ b/crates/ros-z/src/dynamic/type_description_service.rs @@ -38,6 +38,7 @@ use std::collections::HashMap; use std::sync::{Arc, RwLock}; +use ros_z_cdr::{CdrBuffer, CdrDeserialize, CdrReader, CdrSerialize, CdrSerializedSize, CdrWriter}; use serde::{Deserialize, Serialize}; use tracing::{debug, info, trace, warn}; use zenoh::query::Query; @@ -112,7 +113,183 @@ pub struct GetTypeDescriptionRequest { pub include_type_sources: bool, } -// GetTypeDescriptionRequest implements ZMessage via blanket impl +// ── CDR impls for wire types ────────────────────────────────────────────────── + +impl CdrSerialize for WireFieldType { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.type_id.cdr_serialize(w); + self.capacity.cdr_serialize(w); + self.string_capacity.cdr_serialize(w); + self.nested_type_name.cdr_serialize(w); + } +} +impl CdrDeserialize for WireFieldType { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(WireFieldType { + type_id: u8::cdr_deserialize(r)?, + capacity: u64::cdr_deserialize(r)?, + string_capacity: u64::cdr_deserialize(r)?, + nested_type_name: String::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for WireFieldType { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.type_id.cdr_serialized_size(pos); + let p = self.capacity.cdr_serialized_size(p); + let p = self.string_capacity.cdr_serialized_size(p); + self.nested_type_name.cdr_serialized_size(p) + } +} + +impl CdrSerialize for WireField { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.name.cdr_serialize(w); + self.field_type.cdr_serialize(w); + self.default_value.cdr_serialize(w); + } +} +impl CdrDeserialize for WireField { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(WireField { + name: String::cdr_deserialize(r)?, + field_type: WireFieldType::cdr_deserialize(r)?, + default_value: String::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for WireField { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.name.cdr_serialized_size(pos); + let p = self.field_type.cdr_serialized_size(p); + self.default_value.cdr_serialized_size(p) + } +} + +impl CdrSerialize for WireIndividualTypeDescription { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.type_name.cdr_serialize(w); + self.fields.cdr_serialize(w); + } +} +impl CdrDeserialize for WireIndividualTypeDescription { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(WireIndividualTypeDescription { + type_name: String::cdr_deserialize(r)?, + fields: Vec::::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for WireIndividualTypeDescription { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.type_name.cdr_serialized_size(pos); + self.fields.cdr_serialized_size(p) + } +} + +impl CdrSerialize for WireTypeDescription { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.type_description.cdr_serialize(w); + self.referenced_type_descriptions.cdr_serialize(w); + } +} +impl CdrDeserialize for WireTypeDescription { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(WireTypeDescription { + type_description: WireIndividualTypeDescription::cdr_deserialize(r)?, + referenced_type_descriptions: Vec::::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for WireTypeDescription { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.type_description.cdr_serialized_size(pos); + self.referenced_type_descriptions.cdr_serialized_size(p) + } +} + +impl CdrSerialize for WireTypeSource { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.type_name.cdr_serialize(w); + self.encoding.cdr_serialize(w); + self.raw_file_contents.cdr_serialize(w); + } +} +impl CdrDeserialize for WireTypeSource { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(WireTypeSource { + type_name: String::cdr_deserialize(r)?, + encoding: String::cdr_deserialize(r)?, + raw_file_contents: String::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for WireTypeSource { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.type_name.cdr_serialized_size(pos); + let p = self.encoding.cdr_serialized_size(p); + self.raw_file_contents.cdr_serialized_size(p) + } +} + +impl CdrSerialize for WireKeyValue { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.key.cdr_serialize(w); + self.value.cdr_serialize(w); + } +} +impl CdrDeserialize for WireKeyValue { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(WireKeyValue { + key: String::cdr_deserialize(r)?, + value: String::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for WireKeyValue { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.key.cdr_serialized_size(pos); + self.value.cdr_serialized_size(p) + } +} + +impl CdrSerialize for GetTypeDescriptionRequest { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.type_name.cdr_serialize(w); + self.type_hash.cdr_serialize(w); + self.include_type_sources.cdr_serialize(w); + } +} +impl CdrDeserialize for GetTypeDescriptionRequest { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GetTypeDescriptionRequest { + type_name: String::cdr_deserialize(r)?, + type_hash: String::cdr_deserialize(r)?, + include_type_sources: bool::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for GetTypeDescriptionRequest { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.type_name.cdr_serialized_size(pos); + let p = self.type_hash.cdr_serialized_size(p); + self.include_type_sources.cdr_serialized_size(p) + } +} /// GetTypeDescription service response. #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -124,7 +301,37 @@ pub struct GetTypeDescriptionResponse { pub extra_information: Vec, } -// GetTypeDescriptionResponse implements ZMessage via blanket impl +impl CdrSerialize for GetTypeDescriptionResponse { + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + self.successful.cdr_serialize(w); + self.failure_reason.cdr_serialize(w); + self.type_description.cdr_serialize(w); + self.type_sources.cdr_serialize(w); + self.extra_information.cdr_serialize(w); + } +} +impl CdrDeserialize for GetTypeDescriptionResponse { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(GetTypeDescriptionResponse { + successful: bool::cdr_deserialize(r)?, + failure_reason: String::cdr_deserialize(r)?, + type_description: WireTypeDescription::cdr_deserialize(r)?, + type_sources: Vec::::cdr_deserialize(r)?, + extra_information: Vec::::cdr_deserialize(r)?, + }) + } +} +impl CdrSerializedSize for GetTypeDescriptionResponse { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.successful.cdr_serialized_size(pos); + let p = self.failure_reason.cdr_serialized_size(p); + let p = self.type_description.cdr_serialized_size(p); + let p = self.type_sources.cdr_serialized_size(p); + self.extra_information.cdr_serialized_size(p) + } +} /// Marker type for GetTypeDescription service. pub struct GetTypeDescription; @@ -384,11 +591,11 @@ impl TypeDescriptionService { /// This is called from the Zenoh callback when a query is received. /// It deserializes the request, looks up the schema, and sends the response. fn handle_query(schemas: &Arc>>, query: Query) { - use crate::msg::{CdrSerdes, ZDeserializer, ZSerializer}; + use crate::msg::{SerdeCdrSerdes, ZDeserializer, ZSerializer}; // Deserialize the request let request: GetTypeDescriptionRequest = match query.payload() { - Some(payload) => match CdrSerdes::deserialize(payload.to_bytes().as_ref()) { + Some(payload) => match SerdeCdrSerdes::deserialize(payload.to_bytes().as_ref()) { Ok(req) => req, Err(e) => { warn!("[TDS] Failed to deserialize request: {}", e); @@ -420,7 +627,7 @@ impl TypeDescriptionService { ); // Serialize and send the response - let bytes = CdrSerdes::serialize(&response); + let bytes = SerdeCdrSerdes::serialize(&response); use zenoh::Wait; if let Err(e) = query.reply(query.key_expr().clone(), bytes).wait() { warn!("[TDS] Failed to send response: {}", e); diff --git a/crates/ros-z/src/msg.rs b/crates/ros-z/src/msg.rs index d23551e84..d898f0c11 100644 --- a/crates/ros-z/src/msg.rs +++ b/crates/ros-z/src/msg.rs @@ -1,7 +1,10 @@ use byteorder::LittleEndian; #[cfg(feature = "protobuf")] use prost::Message as ProstMessage; -use ros_z_cdr::{CdrBuffer, CdrSerializer, ZBufWriter}; +use ros_z_cdr::{ + CdrBuffer, CdrDeserialize, CdrSerialize, CdrSerializedSize, CdrSerializer, CdrWriter, + ZBufWriter, +}; use serde::{Deserialize, Serialize}; use std::marker::PhantomData; use zenoh_buffers::ZBuf; @@ -45,7 +48,7 @@ pub trait ZSerializer { /// # Example /// /// ```rust,no_run - /// use ros_z::msg::{ZSerializer, CdrSerdes}; + /// use ros_z::msg::{ZSerializer, SerdeCdrSerdes}; /// use serde::Serialize; /// /// #[derive(Serialize)] @@ -53,7 +56,7 @@ pub trait ZSerializer { /// /// let msg = LargeMsg { data: vec![0; 1_000_000] }; /// let hint = 4 + 4 + 1_000_000; // header + length + data - /// let zbuf = CdrSerdes::::serialize_to_zbuf_with_hint(&msg, hint); + /// let zbuf = SerdeCdrSerdes::::serialize_to_zbuf_with_hint(&msg, hint); /// ``` fn serialize_to_zbuf_with_hint(input: Self::Input<'_>, capacity_hint: usize) -> ZBuf; @@ -81,7 +84,7 @@ pub trait ZSerializer { /// # Example /// /// ```rust,no_run - /// use ros_z::msg::{ZSerializer, CdrSerdes}; + /// use ros_z::msg::{ZSerializer, SerdeCdrSerdes}; /// use ros_z::shm::ShmProviderBuilder; /// use serde::Serialize; /// @@ -92,7 +95,7 @@ pub trait ZSerializer { /// let msg = MyMsg { value: 42 }; /// let provider = ShmProviderBuilder::new(1024 * 1024).build()?; /// - /// let (zbuf, size) = CdrSerdes::::serialize_to_shm(&msg, 128, &provider)?; + /// let (zbuf, size) = SerdeCdrSerdes::::serialize_to_shm(&msg, 128, &provider)?; /// println!("Serialized {} bytes to SHM", size); /// # Ok(()) /// # } @@ -198,22 +201,29 @@ pub trait ZMessage: Send + Sync + Sized + 'static { } } -// Blanket implementation for serde-compatible types using CDR +// Blanket implementation for types with dedicated CDR traits (fast path). +// All generated message types satisfy these bounds; internal ros-z types that +// only have serde get explicit ZMessage impls below using SerdeCdrSerdes instead. impl ZMessage for T where - T: Send + Sync + Serialize + for<'a> Deserialize<'a> + 'static, + T: Send + + Sync + + ros_z_cdr::CdrSerialize + + ros_z_cdr::CdrDeserialize + + ros_z_cdr::CdrSerializedSize + + 'static, { - type Serdes = CdrSerdes; + type Serdes = NativeCdrSerdes; } -// CDR +// ── Serde-based CDR serialization (existing path, kept for non-generated types) ─────────── -pub struct CdrSerdes(PhantomData); +pub struct SerdeCdrSerdes(PhantomData); /// CDR encapsulation header for little-endian encoding -const CDR_HEADER_LE: [u8; 4] = [0x00, 0x01, 0x00, 0x00]; +pub const CDR_HEADER_LE: [u8; 4] = [0x00, 0x01, 0x00, 0x00]; -impl ZSerializer for CdrSerdes +impl ZSerializer for SerdeCdrSerdes where T: Serialize, { @@ -223,22 +233,14 @@ where T: 'a; fn serialize_to_zbuf(input: &T) -> ZBuf { - // Use fixed 256-byte capacity for backward compatibility Self::serialize_to_zbuf_with_hint(input, 256) } fn serialize_to_zbuf_with_hint(input: &T, capacity_hint: usize) -> ZBuf { - // Create ZBufWriter with provided capacity hint for optimal allocation let mut writer = ZBufWriter::with_capacity(capacity_hint); - - // Write CDR header writer.extend_from_slice(&CDR_HEADER_LE); - - // Serialize payload directly to ZBufWriter let mut serializer = CdrSerializer::::new(&mut writer); input.serialize(&mut serializer).unwrap(); - - // Convert to ZBuf (transfers ownership, no copy) writer.into_zbuf() } @@ -247,24 +249,14 @@ where estimated_size: usize, provider: &zenoh::shm::ShmProvider, ) -> zenoh::Result<(ZBuf, usize)> { - // Create SHM writer with estimated size let mut writer = crate::shm::ShmWriter::new(provider, estimated_size)?; - - // Write CDR header writer.extend_from_slice(&CDR_HEADER_LE); - - // Serialize payload directly to SHM buffer let mut serializer = CdrSerializer::::new(&mut writer); input .serialize(&mut serializer) .map_err(|e| zenoh::Error::from(format!("CDR serialization failed: {}", e)))?; - - // Get actual serialized size let actual_size = writer.position(); - - // Convert to ZBuf (SHM-backed, zero-copy!) let zbuf = writer.into_zbuf()?; - Ok((zbuf, actual_size)) } @@ -276,22 +268,13 @@ where fn serialize_to_buf(input: &T, buffer: &mut Vec) { buffer.clear(); - - // STEP 1: Write CDR header FIRST (4 bytes) - // This avoids the O(n) memmove that would be needed if we prepended it later buffer.extend_from_slice(&CDR_HEADER_LE); - - // STEP 2: Serialize payload using CdrSerializer - // Zero buffer swaps - the serializer works directly on our buffer! let mut fast_ser = CdrSerializer::::new(buffer); input.serialize(&mut fast_ser).unwrap(); - // Buffer is automatically updated through the mutable reference - - // Done! Header is at position 0, payload follows immediately after } } -impl ZDeserializer for CdrSerdes +impl ZDeserializer for SerdeCdrSerdes where for<'a> T: Deserialize<'a>, { @@ -303,8 +286,7 @@ where if input.len() < 4 { return Err(CdrError("CDR data too short for header".into())); } - let header = &input[0..4]; - let representation_identifier = &header[0..2]; + let representation_identifier = &input[0..2]; if representation_identifier != [0x00, 0x01] { return Err(CdrError(format!( "Expected CDR_LE encapsulation ({:?}), found {:?}", @@ -319,6 +301,90 @@ where } } +// ── Fast CdrSerialize-based CDR serialization (new path for generated types) ──────────── + +/// CDR serialization using the `CdrSerialize`/`CdrDeserialize` traits directly. +/// +/// Generated message types implement these traits and use `NativeCdrSerdes` as their +/// `ZMessage::Serdes` type. This enables the POD bulk fast path for sequences of +/// plain types (e.g., `Vec`, `Vec`). +pub struct NativeCdrSerdes(PhantomData); + +impl ZSerializer for NativeCdrSerdes +where + T: CdrSerialize + CdrSerializedSize, +{ + type Input<'a> + = &'a T + where + T: 'a; + + fn serialize_to_zbuf(input: &T) -> ZBuf { + let capacity_hint = input.cdr_serialized_size(0) + 4; + Self::serialize_to_zbuf_with_hint(input, capacity_hint) + } + + fn serialize_to_zbuf_with_hint(input: &T, capacity_hint: usize) -> ZBuf { + let mut writer = ZBufWriter::with_capacity(capacity_hint); + writer.extend_from_slice(&CDR_HEADER_LE); + ros_z_cdr::traits::cdr_to_zbuf_writer(input, &mut writer); + writer.into_zbuf() + } + + fn serialize_to_shm( + input: &T, + estimated_size: usize, + provider: &zenoh::shm::ShmProvider, + ) -> zenoh::Result<(ZBuf, usize)> { + let mut writer = crate::shm::ShmWriter::new(provider, estimated_size)?; + writer.extend_from_slice(&CDR_HEADER_LE); + let mut cdr_writer = CdrWriter::::new(&mut writer); + input.cdr_serialize(&mut cdr_writer); + let actual_size = writer.position(); + let zbuf = writer.into_zbuf()?; + Ok((zbuf, actual_size)) + } + + fn serialize(input: &T) -> Vec { + let mut buffer = Vec::new(); + Self::serialize_to_buf(input, &mut buffer); + buffer + } + + fn serialize_to_buf(input: &T, buffer: &mut Vec) { + buffer.clear(); + buffer.extend_from_slice(&CDR_HEADER_LE); + let mut cdr_writer = CdrWriter::::new(buffer); + input.cdr_serialize(&mut cdr_writer); + } +} + +impl ZDeserializer for NativeCdrSerdes +where + T: CdrDeserialize, +{ + type Input<'b> = &'b [u8]; + type Output = T; + type Error = CdrError; + + fn deserialize(input: Self::Input<'_>) -> Result { + if input.len() < 4 { + return Err(CdrError("CDR data too short for header".into())); + } + let representation_identifier = &input[0..2]; + if representation_identifier != [0x00, 0x01] { + return Err(CdrError(format!( + "Expected CDR_LE encapsulation ({:?}), found {:?}", + [0x00, 0x01], + representation_identifier + ))); + } + let payload = &input[4..]; + let mut reader = ros_z_cdr::CdrReader::::new(payload); + T::cdr_deserialize(&mut reader).map_err(|e| CdrError(e.to_string())) + } +} + // Protobuf #[cfg(feature = "protobuf")] @@ -409,6 +475,10 @@ mod tests { text: String, } + impl ZMessage for SimpleMessage { + type Serdes = SerdeCdrSerdes; + } + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] struct LargeMessage { data: Vec, @@ -423,14 +493,14 @@ mod tests { text: "Hello, ZBuf!".to_string(), }; - let zbuf = CdrSerdes::::serialize_to_zbuf(&msg); + let zbuf = SerdeCdrSerdes::::serialize_to_zbuf(&msg); let bytes = zbuf.contiguous(); // Verify CDR header assert_eq!(&bytes[0..4], &CDR_HEADER_LE); // Verify roundtrip - let deserialized = CdrSerdes::::deserialize(&bytes).unwrap(); + let deserialized = SerdeCdrSerdes::::deserialize(&bytes).unwrap(); assert_eq!(deserialized, msg); } @@ -442,8 +512,8 @@ mod tests { }; // Both methods should produce identical bytes - let zbuf = CdrSerdes::::serialize_to_zbuf(&msg); - let vec = CdrSerdes::::serialize(&msg); + let zbuf = SerdeCdrSerdes::::serialize_to_zbuf(&msg); + let vec = SerdeCdrSerdes::::serialize(&msg); let zbuf_bytes = zbuf.contiguous(); assert_eq!(&*zbuf_bytes, &vec[..]); @@ -463,19 +533,19 @@ mod tests { let mut buffer = Vec::with_capacity(1024); // First serialization - let zbuf1 = CdrSerdes::::serialize_to_zbuf_reuse(&msg1, &mut buffer); + let zbuf1 = SerdeCdrSerdes::::serialize_to_zbuf_reuse(&msg1, &mut buffer); let bytes1 = zbuf1.contiguous(); // Buffer should be empty after take assert!(buffer.is_empty()); // Second serialization (buffer will be reallocated) - let zbuf2 = CdrSerdes::::serialize_to_zbuf_reuse(&msg2, &mut buffer); + let zbuf2 = SerdeCdrSerdes::::serialize_to_zbuf_reuse(&msg2, &mut buffer); let bytes2 = zbuf2.contiguous(); // Verify roundtrips - let decoded1 = CdrSerdes::::deserialize(&bytes1).unwrap(); - let decoded2 = CdrSerdes::::deserialize(&bytes2).unwrap(); + let decoded1 = SerdeCdrSerdes::::deserialize(&bytes1).unwrap(); + let decoded2 = SerdeCdrSerdes::::deserialize(&bytes2).unwrap(); assert_eq!(decoded1, msg1); assert_eq!(decoded2, msg2); @@ -506,9 +576,9 @@ mod tests { }; // Serialize using both methods - let vec1 = CdrSerdes::::serialize(&msg); + let vec1 = SerdeCdrSerdes::::serialize(&msg); let mut vec2 = Vec::new(); - CdrSerdes::::serialize_to_buf(&msg, &mut vec2); + SerdeCdrSerdes::::serialize_to_buf(&msg, &mut vec2); // Results should be identical assert_eq!(vec1, vec2); @@ -524,13 +594,13 @@ mod tests { }; let mut buffer = Vec::with_capacity(1024); - CdrSerdes::::serialize_to_buf(&msg, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(&msg, &mut buffer); let capacity_after_first = buffer.capacity(); assert_eq!(capacity_after_first, 1024); // Serialize again - should reuse capacity - CdrSerdes::::serialize_to_buf(&msg, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(&msg, &mut buffer); assert_eq!(buffer.capacity(), capacity_after_first); } @@ -550,12 +620,12 @@ mod tests { let mut buffer = Vec::new(); // Serialize large message - CdrSerdes::::serialize_to_buf(&msg1, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(&msg1, &mut buffer); let len1 = buffer.len(); assert!(len1 > 100); // Serialize small message - should clear buffer first - CdrSerdes::::serialize_to_buf(&msg2, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(&msg2, &mut buffer); let len2 = buffer.len(); assert!(len2 < len1); @@ -582,11 +652,11 @@ mod tests { // Serialize using serialize_to_buf let mut buffer = Vec::new(); - CdrSerdes::::serialize_to_buf(&original, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(&original, &mut buffer); // Deserialize let deserialized = - CdrSerdes::::deserialize(&buffer).expect("Failed to deserialize"); + SerdeCdrSerdes::::deserialize(&buffer).expect("Failed to deserialize"); // Should match original assert_eq!(deserialized, original); @@ -602,7 +672,7 @@ mod tests { let mut buffer = Vec::new(); assert_eq!(buffer.capacity(), 0); - CdrSerdes::::serialize_to_buf(&msg, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(&msg, &mut buffer); assert!(!buffer.is_empty()); assert!(buffer.capacity() > 0); @@ -630,12 +700,12 @@ mod tests { let mut all_serialized = Vec::new(); for msg in &messages { - CdrSerdes::::serialize_to_buf(msg, &mut buffer); + SerdeCdrSerdes::::serialize_to_buf(msg, &mut buffer); all_serialized.push(buffer.clone()); // Verify each serialization is correct - let deserialized = - CdrSerdes::::deserialize(&buffer).expect("Failed to deserialize"); + let deserialized = SerdeCdrSerdes::::deserialize(&buffer) + .expect("Failed to deserialize"); assert_eq!(&deserialized, msg); } @@ -716,3 +786,290 @@ mod tests { assert_eq!(&*bytes, &vec[..]); } } + +/// Tests for `NativeCdrSerdes` — the `CdrSerialize`-based CDR fast path. +/// +/// These tests verify: +/// 1. Byte-identical wire output between `SerdeCdrSerdes` (serde path) and `NativeCdrSerdes` (CDR trait path). +/// 2. Roundtrip correctness for `NativeCdrSerdes`. +/// 3. POD bulk path produces the same bytes for plain sequences as the element loop. +#[cfg(test)] +mod fast_cdr_tests { + use super::*; + use ros_z_cdr::{ + CdrBuffer, CdrDeserialize, CdrReader, CdrSerialize, CdrSerializedSize, CdrWriter, + }; + + // ── Test types ──────────────────────────────────────────────────────────── + + /// A struct with a string field — NOT plain (element-by-element path). + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] + struct Header { + seq: u32, + frame_id: String, + } + + impl CdrSerialize for Header { + fn cdr_serialize( + &self, + w: &mut CdrWriter<'_, BO, B>, + ) { + self.seq.cdr_serialize(w); + self.frame_id.cdr_serialize(w); + } + } + + impl CdrDeserialize for Header { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(Self { + seq: u32::cdr_deserialize(r)?, + frame_id: String::cdr_deserialize(r)?, + }) + } + } + + impl CdrSerializedSize for Header { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.seq.cdr_serialized_size(pos); + self.frame_id.cdr_serialized_size(p) + } + } + + /// A plain struct — all fields are f64, no strings/sequences. + /// On LE hosts this satisfies `CdrPlain` (verified in ros-z-cdr tests). + #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)] + struct Point3d { + x: f64, + y: f64, + z: f64, + } + + impl CdrSerialize for Point3d { + fn cdr_serialize( + &self, + w: &mut CdrWriter<'_, BO, B>, + ) { + self.x.cdr_serialize(w); + self.y.cdr_serialize(w); + self.z.cdr_serialize(w); + } + } + + impl CdrDeserialize for Point3d { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + Ok(Self { + x: f64::cdr_deserialize(r)?, + y: f64::cdr_deserialize(r)?, + z: f64::cdr_deserialize(r)?, + }) + } + } + + impl CdrSerializedSize for Point3d { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.x.cdr_serialized_size(pos); + let p = self.y.cdr_serialized_size(p); + self.z.cdr_serialized_size(p) + } + } + + /// A message with a Vec — this is the key fast-path case. + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] + struct PointCloud { + header: Header, + points: Vec, + } + + impl CdrSerialize for PointCloud { + fn cdr_serialize( + &self, + w: &mut CdrWriter<'_, BO, B>, + ) { + self.header.cdr_serialize(w); + // Vec: element-by-element (Point3d: CdrSerialize) + w.write_sequence_length(self.points.len()); + for pt in &self.points { + pt.cdr_serialize(w); + } + } + } + + impl CdrDeserialize for PointCloud { + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + let header = Header::cdr_deserialize(r)?; + let n = r.read_sequence_length()?; + let mut points = Vec::with_capacity(n); + for _ in 0..n { + points.push(Point3d::cdr_deserialize(r)?); + } + Ok(Self { header, points }) + } + } + + impl CdrSerializedSize for PointCloud { + fn cdr_serialized_size(&self, pos: usize) -> usize { + let p = self.header.cdr_serialized_size(pos); + // sequence length u32 (4-byte aligned) + let p = p + ((4 - p % 4) % 4) + 4; + let mut p = p; + for pt in &self.points { + p = pt.cdr_serialized_size(p); + } + p + } + } + + // ── Helpers ─────────────────────────────────────────────────────────────── + + fn serde_bytes(value: &T) -> Vec { + SerdeCdrSerdes::::serialize(value) + } + + fn fast_bytes(value: &T) -> Vec { + NativeCdrSerdes::::serialize(value) + } + + fn fast_deserialize(bytes: &[u8]) -> T { + NativeCdrSerdes::::deserialize(bytes).expect("NativeCdrSerdes::deserialize failed") + } + + // ── Tests ───────────────────────────────────────────────────────────────── + + #[test] + fn header_byte_identical_to_serde() { + let msg = Header { + seq: 42, + frame_id: "base_link".to_string(), + }; + assert_eq!(serde_bytes(&msg), fast_bytes(&msg)); + } + + #[test] + fn header_fast_roundtrip() { + let msg = Header { + seq: 99, + frame_id: "map".to_string(), + }; + let bytes = fast_bytes(&msg); + let decoded: Header = fast_deserialize(&bytes); + assert_eq!(msg, decoded); + } + + #[test] + #[allow(clippy::approx_constant)] + fn point3d_byte_identical_to_serde() { + let pt = Point3d { + x: 1.0, + y: 2.5, + z: -3.14, + }; + assert_eq!(serde_bytes(&pt), fast_bytes(&pt)); + } + + #[test] + #[allow(clippy::approx_constant)] + fn point3d_fast_roundtrip() { + let pt = Point3d { + x: 1.0, + y: 2.5, + z: -3.14, + }; + let bytes = fast_bytes(&pt); + let decoded: Point3d = fast_deserialize(&bytes); + assert_eq!(pt, decoded); + } + + #[test] + fn pointcloud_byte_identical_to_serde() { + let msg = PointCloud { + header: Header { + seq: 1, + frame_id: "lidar".to_string(), + }, + points: vec![ + Point3d { + x: 0.0, + y: 0.0, + z: 0.0, + }, + Point3d { + x: 1.0, + y: 2.0, + z: 3.0, + }, + Point3d { + x: -1.0, + y: -2.0, + z: -3.0, + }, + ], + }; + assert_eq!(serde_bytes(&msg), fast_bytes(&msg)); + } + + #[test] + fn pointcloud_fast_roundtrip() { + let msg = PointCloud { + header: Header { + seq: 7, + frame_id: "camera".to_string(), + }, + points: (0..100) + .map(|i| Point3d { + x: i as f64, + y: (i * 2) as f64, + z: (i * 3) as f64, + }) + .collect(), + }; + let bytes = fast_bytes(&msg); + let decoded: PointCloud = fast_deserialize(&bytes); + assert_eq!(msg, decoded); + } + + #[test] + fn empty_sequence_roundtrip() { + let msg = PointCloud { + header: Header { + seq: 0, + frame_id: String::new(), + }, + points: vec![], + }; + let bytes = fast_bytes(&msg); + let decoded: PointCloud = fast_deserialize(&bytes); + assert_eq!(msg, decoded); + } + + #[test] + fn size_hint_matches_actual() { + let msg = PointCloud { + header: Header { + seq: 1, + frame_id: "test".to_string(), + }, + points: vec![ + Point3d { + x: 1.0, + y: 2.0, + z: 3.0 + }; + 10 + ], + }; + let hint = msg.cdr_serialized_size(0) + 4; + let bytes = fast_bytes(&msg); + // The hint should be >= actual payload size + assert!( + hint >= bytes.len() - 4, + "hint={hint} bytes.len()={}", + bytes.len() + ); + } +} diff --git a/crates/ros-z/src/node.rs b/crates/ros-z/src/node.rs index 7a77e7091..2ad11dee6 100644 --- a/crates/ros-z/src/node.rs +++ b/crates/ros-z/src/node.rs @@ -12,7 +12,7 @@ use crate::{ action::{client::ZActionClientBuilder, server::ZActionServerBuilder}, context::{GlobalCounter, RemapRules}, dynamic::{ - DynamicCdrSerdes, DynamicMessage, MessageSchema, TypeDescriptionClient, + DynamicMessage, DynamicSerdeCdrSerdes, MessageSchema, TypeDescriptionClient, TypeDescriptionService, }, entity::*, @@ -208,7 +208,7 @@ impl ZNode { /// - Absolute topics (starting with '/') are used as-is /// - Private topics (starting with '~') are expanded to /// /// - Relative topics are expanded to // - pub fn create_pub(&self, topic: &str) -> ZPubBuilder + pub fn create_pub(&self, topic: &str) -> ZPubBuilder where T: ZMessage + WithTypeInfo, { @@ -221,7 +221,7 @@ impl ZNode { &self, topic: &str, type_info: Option, - ) -> ZPubBuilder + ) -> ZPubBuilder where T: ZMessage, { @@ -255,9 +255,9 @@ impl ZNode { /// - Absolute topics (starting with '/') are used as-is /// - Private topics (starting with '~') are expanded to /// /// - Relative topics are expanded to // - pub fn create_sub(&self, topic: &str) -> ZSubBuilder + pub fn create_sub(&self, topic: &str) -> ZSubBuilder where - T: WithTypeInfo, + T: ZMessage + WithTypeInfo, { debug!("[NOD] Creating subscriber: topic={}", topic); self.create_sub_impl(topic, Some(T::type_info())) @@ -268,7 +268,10 @@ impl ZNode { &self, topic: &str, type_info: Option, - ) -> ZSubBuilder { + ) -> ZSubBuilder + where + T: ZMessage, + { // Note: Topic qualification happens in ZSubBuilder::build() // to allow error handling in the Result type let entity = EndpointEntity { @@ -440,7 +443,7 @@ impl ZNode { &self, topic: &str, schema: Arc, - ) -> Result> { + ) -> Result> { // Register schema with type description service if enabled if let Some(service) = &self.type_desc_service { if let Err(e) = service.register_schema(schema.clone()) { @@ -490,7 +493,7 @@ impl ZNode { // Build the publisher self.create_pub_impl::(topic, type_info) - .with_serdes::() + .with_serdes::() .with_dyn_schema(schema) .build() } @@ -531,7 +534,7 @@ impl ZNode { topic: &str, discovery_timeout: Duration, ) -> Result<( - ZSub, + ZSub, Arc, )> { debug!( @@ -580,7 +583,7 @@ impl ZNode { // Build the subscriber with the discovered schema let subscriber = self .create_sub_impl::(topic, type_info) - .with_serdes::() + .with_serdes::() .with_dyn_schema(schema.clone()) .build()?; @@ -611,7 +614,7 @@ impl ZNode { &self, topic: &str, schema: Arc, - ) -> Result> { + ) -> Result> { // Create TypeInfo from schema for proper key expression matching // Convert ROS 2 canonical name to DDS name // "std_msgs/msg/String" → "std_msgs::msg::dds_::String_" @@ -646,7 +649,7 @@ impl ZNode { // Build the subscriber with proper type info self.create_sub_impl::(topic, type_info) - .with_serdes::() + .with_serdes::() .with_dyn_schema(schema) .build() } diff --git a/crates/ros-z/src/pubsub.rs b/crates/ros-z/src/pubsub.rs index 92d83038e..ded868c23 100644 --- a/crates/ros-z/src/pubsub.rs +++ b/crates/ros-z/src/pubsub.rs @@ -16,7 +16,7 @@ use crate::impl_with_type_info; use crate::queue::BoundedQueue; use crate::topic_name; -use crate::msg::{CdrSerdes, ZDeserializer, ZMessage, ZSerializer}; +use crate::msg::{SerdeCdrSerdes, ZDeserializer, ZMessage, ZSerializer}; use crate::qos::QosProfile; use ros_z_protocol::qos::{QosDurability, QosHistory, QosReliability}; use std::sync::Mutex; @@ -54,7 +54,7 @@ impl std::fmt::Debug for ZPub { } #[derive(Debug)] -pub struct ZPubBuilder> { +pub struct ZPubBuilder> { pub entity: EndpointEntity, pub session: Arc, pub graph: Arc, @@ -191,7 +191,7 @@ impl ZPubBuilder { /// ```ignore /// let publisher = node /// .create_pub_impl::("topic", None) - /// .with_serdes::() + /// .with_serdes::() /// .with_dyn_schema(schema) /// .build()?; /// ``` @@ -546,7 +546,7 @@ where } // Specialized implementation for DynamicMessage publisher -impl ZPub { +impl ZPub { /// Get the dynamic schema used by this publisher. /// /// Returns `None` if the publisher was not created with `.with_dyn_schema()`. @@ -555,7 +555,7 @@ impl ZPub { } } -pub struct ZSubBuilder> { +pub struct ZSubBuilder> { pub entity: EndpointEntity, pub session: Arc, pub(crate) keyexpr_format: ros_z_protocol::KeyExprFormat, @@ -636,7 +636,7 @@ where /// Set the dynamic message schema for runtime-typed messages. /// - /// This is required when using `DynamicMessage` with `DynamicCdrSerdes`. + /// This is required when using `DynamicMessage` with `DynamicSerdeCdrSerdes`. /// The schema will be used to deserialize incoming messages. /// /// # Example @@ -644,7 +644,7 @@ where /// ```ignore /// let subscriber = node /// .create_sub::("/topic") - /// .with_serdes::() + /// .with_serdes::() /// .with_dyn_schema(schema) /// .build()?; /// ``` @@ -866,7 +866,7 @@ pub struct ZSub { _lv_token: LivelinessToken, events_mgr: Arc>, /// Schema for dynamic message deserialization. - /// Required when using `DynamicMessage` with `DynamicCdrSerdes`. + /// Required when using `DynamicMessage` with `DynamicSerdeCdrSerdes`. pub dyn_schema: Option>, /// Expected encoding for validation. pub expected_encoding: Option, @@ -970,7 +970,7 @@ where } // Specialized implementation for DynamicMessage -impl ZSub { +impl ZSub { /// Receive and deserialize the next dynamic message. /// /// This method requires that the subscriber was built with `.with_dyn_schema()`. @@ -1003,7 +1003,7 @@ impl ZSub { let payload = sample.payload().to_bytes(); - let result = crate::dynamic::DynamicCdrSerdes::deserialize((&payload, schema)) + let result = crate::dynamic::DynamicSerdeCdrSerdes::deserialize((&payload, schema)) .map_err(|e| zenoh::Error::from(e.to_string())); Some(result) } diff --git a/crates/ros-z/src/service.rs b/crates/ros-z/src/service.rs index 4f632bf41..364e4b434 100644 --- a/crates/ros-z/src/service.rs +++ b/crates/ros-z/src/service.rs @@ -28,7 +28,7 @@ use crate::{ common::DataHandler, entity::EndpointEntity, impl_with_type_info, - msg::{CdrSerdes, ZDeserializer, ZMessage, ZService}, + msg::{SerdeCdrSerdes, ZDeserializer, ZMessage, ZService}, qos::QosHistory, queue::BoundedQueue, }; @@ -173,7 +173,9 @@ where // For ROS-Z pub fn take_response(&self) -> Result where - for<'c> T::Response: ZMessage> + Deserialize<'c>, + T::Response: ZMessage, + for<'a> ::Serdes: + ZDeserializer = &'a [u8]>, { let sample = self.take_sample()?; let msg = ::deserialize(&sample.payload().to_bytes()) diff --git a/crates/ros-z/src/zbuf.rs b/crates/ros-z/src/zbuf.rs index c7829e6cf..ea990e69a 100644 --- a/crates/ros-z/src/zbuf.rs +++ b/crates/ros-z/src/zbuf.rs @@ -6,10 +6,11 @@ //! The wrapper uses `serialize_bytes()` instead of `serialize_seq()` for better performance //! with large byte arrays, which is critical for messages like sensor images. +use ros_z_cdr::{CdrBuffer, CdrDeserialize, CdrReader, CdrSerialize, CdrSerializedSize, CdrWriter}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use std::fmt; use zenoh_buffers::ZBuf as ZenohZBuf; -use zenoh_buffers::buffer::SplitBuffer; +use zenoh_buffers::buffer::{Buffer, SplitBuffer}; /// ROS-Z wrapper around Zenoh's ZBuf with optimized serde. /// @@ -104,7 +105,61 @@ impl std::ops::DerefMut for ZBuf { } } -// Optimized serde implementation +// ── CdrSerialize / CdrDeserialize / CdrSerializedSize ──────────────────────── +// +// CDR encoding for ZBuf: u32 sequence length prefix + raw bytes (like Vec). +// This matches the ROS CDR wire format for `byte[]` / `uint8[]` fields. + +impl CdrSerialize for ZBuf { + #[inline] + fn cdr_serialize(&self, w: &mut CdrWriter<'_, BO, B>) { + let bytes = self.0.contiguous(); + // write_bytes writes the u32 length prefix followed by the raw bytes + w.write_bytes(bytes.as_ref()); + } +} + +impl CdrDeserialize for ZBuf { + #[inline] + fn cdr_deserialize<'de, BO: byteorder::ByteOrder>( + r: &mut CdrReader<'de, BO>, + ) -> ros_z_cdr::Result { + let count = r.read_sequence_length()?; + let bytes = r.read_bytes(count)?; + // Try zero-copy: check if the bytes sit inside a source ZBuf. + let zbuf = ros_z_cdr::ZBUF_DESER_SOURCE.with(|cell| { + let borrow = cell.borrow(); + let source = borrow.as_ref()?; + let v_start = bytes.as_ptr() as usize; + let v_end = v_start + bytes.len(); + for zslice in source.zslices() { + let s_start = zslice.as_slice().as_ptr() as usize; + let s_end = s_start + zslice.len(); + if v_start >= s_start && v_end <= s_end { + let offset = v_start - s_start; + let sub = zslice.subslice(offset..offset + bytes.len())?; + let mut z = ZenohZBuf::default(); + z.push_zslice(sub); + return Some(z); + } + } + None + }); + let inner = zbuf.unwrap_or_else(|| ZenohZBuf::from(bytes.to_vec())); + Ok(ZBuf(inner)) + } +} + +impl CdrSerializedSize for ZBuf { + #[inline] + fn cdr_serialized_size(&self, pos: usize) -> usize { + // u32 length prefix (4-byte aligned) + byte contents + let after_len = pos + (4 - pos % 4) % 4 + 4; + after_len + self.0.len() + } +} + +// ── Optimized serde implementation ────────────────────────────────────────── impl Serialize for ZBuf { fn serialize(&self, serializer: S) -> Result where diff --git a/crates/ros-z/tests/action/communication.rs b/crates/ros-z/tests/action/communication.rs index 3229310a4..1679bceb6 100644 --- a/crates/ros-z/tests/action/communication.rs +++ b/crates/ros-z/tests/action/communication.rs @@ -149,10 +149,9 @@ mod tests { }; // Respond to the cancel request - let response_bytes = - ros_z::msg::CdrSerdes::::serialize( - &cancel_resp, - ); + let response_bytes = ros_z::msg::SerdeCdrSerdes::< + ros_z::action::messages::CancelGoalResponse, + >::serialize(&cancel_resp); response_tx .reply(response_tx.key_expr().clone(), response_bytes) .wait()?; diff --git a/crates/ros-z/tests/pubsub.rs b/crates/ros-z/tests/pubsub.rs index b71957e6d..d4bcabab9 100644 --- a/crates/ros-z/tests/pubsub.rs +++ b/crates/ros-z/tests/pubsub.rs @@ -24,6 +24,10 @@ impl MessageTypeInfo for TestMessage { impl ros_z::ros_msg::WithTypeInfo for TestMessage {} +impl ros_z::msg::ZMessage for TestMessage { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn test_basic_pubsub() { let ctx = ZContextBuilder::default() diff --git a/crates/ros-z/tests/service.rs b/crates/ros-z/tests/service.rs index 0fee9cf03..587d7ff37 100644 --- a/crates/ros-z/tests/service.rs +++ b/crates/ros-z/tests/service.rs @@ -26,6 +26,10 @@ impl MessageTypeInfo for AddTwoIntsRequest { impl ros_z::WithTypeInfo for AddTwoIntsRequest {} +impl ros_z::msg::ZMessage for AddTwoIntsRequest { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + // Simple test service response #[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)] struct AddTwoIntsResponse { @@ -44,6 +48,10 @@ impl MessageTypeInfo for AddTwoIntsResponse { impl ros_z::WithTypeInfo for AddTwoIntsResponse {} +impl ros_z::msg::ZMessage for AddTwoIntsResponse { + type Serdes = ros_z::msg::SerdeCdrSerdes; +} + // Service type definition struct AddTwoInts;