diff --git a/crates/hiroz-py/src/node.rs b/crates/hiroz-py/src/node.rs index 6e43eb4dc..e186c6901 100644 --- a/crates/hiroz-py/src/node.rs +++ b/crates/hiroz-py/src/node.rs @@ -16,6 +16,7 @@ use hiroz::node::ZNode; use pyo3::prelude::*; use std::any::Any; use std::sync::Arc; +use zenoh_buffers::buffer::SplitBuffer; /// Try to extract type info from a message class. /// @@ -233,9 +234,25 @@ impl PyZNode { // matching rmw_zenoh_cpp's NodeData::subs_ pattern. The caller does not // need to assign the returned PyZSubscriber to keep the subscription active. let type_name = msg_type_str.clone(); + // Sample-level callback, not `build_with_callback`. The typed form + // would route through `RawBytesCdrSerdes::deserialize`, whose + // `Output` is an owned `RawBytesMessage` and so must `to_vec()` the + // whole payload before this closure runs — a full copy per message, + // scaling with payload size, immediately discarded once msgspec has + // decoded it. Taking the `Sample` lets the decode read straight out + // of the network buffer, and matches what the polling `recv()` path + // in `pubsub.rs` already does. let zsub = sub_builder - .build_with_callback(move |raw_msg: RawBytesMessage| { - let payload = raw_msg.0; + .build_with_sample_callback(move |sample| { + // Same zero-copy setup as `PyZSubscriber::recv`: the ZBuf is + // cheap Arc clones, and publishing it as the deserializer's + // source lets `bytes`-typed fields become sub-ZSlices of the + // received buffer instead of copies. + let payload_zbuf: zenoh_buffers::ZBuf = sample.payload().clone().into(); + hiroz_cdr::ZBUF_DESER_SOURCE.with(|cell| { + *cell.borrow_mut() = Some(payload_zbuf.clone()); + }); + let payload = payload_zbuf.contiguous(); Python::with_gil(|py| { match hiroz_msgs::deserialize_from_cdr(&type_name, py, &payload) { Ok(obj) => { @@ -248,6 +265,9 @@ impl PyZNode { } } }); + hiroz_cdr::ZBUF_DESER_SOURCE.with(|cell| { + *cell.borrow_mut() = None; + }); }) .map_err(|e| e.into_pyerr())?; diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 5b637d031..76d891da2 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -867,6 +867,46 @@ where }) } + /// Build a callback subscriber that receives the whole [`Sample`], undecoded. + /// + /// [`Self::build_with_callback`] hands the callback an owned `S::Output`, and + /// [`ZDeserializer::Output`] carries no lifetime — so a serdes that only + /// forwards bytes must copy the whole message before the callback sees it. + /// Taking the `Sample` instead lets the callback borrow the payload and + /// decode out of the network buffer, and reaches the attachment, encoding and + /// timestamp that the decoded form drops. + /// + /// Otherwise identical to `build_with_callback`, including that the returned + /// [`ZSub`] must be kept alive for the subscription to stay active. + pub fn build_with_sample_callback(self, callback: F) -> Result> + where + F: Fn(Sample) + Send + Sync + 'static, + S: ZDeserializer, + { + let expected_encoding = self.expected_encoding.clone(); + let callback = Arc::new(move |sample: Sample| { + if let Some(ref expected) = expected_encoding { + let encoding_str = sample.encoding().to_string(); + if let Some(received) = + crate::encoding::Encoding::from_zenoh_encoding(&encoding_str) + { + if &received != expected { + tracing::warn!( + "Encoding mismatch: expected {:?}, received {:?}", + expected, + received + ); + } + } else { + tracing::debug!("Unknown encoding format: {}", encoding_str); + } + } + callback(sample); + }); + + self.build_internal(DataHandler::Callback(callback), None) + } + /// Build a subscriber with a callback that processes deserialized messages directly. /// /// This method creates a subscriber that invokes the provided callback for each