fix(pubsub): expose events_mgr and entity on every ZSub variant

Both accessors lived on the ZSub<T, Sample, S> (queue-mode) impl, so a
callback subscriber could not reach its own events manager -- the handle
the rmw layer needs to install an event callback, and the only way to
observe MessageLost. Neither field has anything to do with the queue.

Moving them to a generic impl is what let the wiring test observe a
callback subscriber's loss counter at all.

Author: YuanYuYuan

crates/hiroz-tests/tests/message_lost.rs

@@ -55,6 +55,13 @@ fn gid(n: u8) -> GidArray {
     g
 }
 
+/// A CDR-encoded `Tick`, ready to hand to `Session::put`.
+fn payload(counter: u64) -> zenoh::bytes::ZBytes {
+    use hiroz::msg::ZSerializer;
+    let zbuf = <hiroz::msg::SerdeCdrSerdes<Tick>>::serialize_to_zbuf(&Tick { counter });
+    zenoh::bytes::ZBytes::from(zbuf)
+}
+
 /// Wait until `total_count` for `MessageLost` stops changing, then return it.
 fn settled_loss_count(sub_events: &Arc<Mutex<hiroz::event::EventsManager>>) -> i32 {
     let mut last = -1;
@@ -107,11 +114,8 @@ fn a_sequence_gap_raises_message_lost() {
     let publisher_gid = gid(42);
 
     let put = |sn: i64, counter: u64| {
-        let zbuf = <hiroz::msg::SerdeCdrSerdes<Tick> as hiroz::msg::ZSerializer>::serialize_to_zbuf(
-            &Tick { counter },
-        );
         session
-            .put((*ke).clone(), zenoh::bytes::ZBytes::from(zbuf))
+            .put((*ke).clone(), payload(counter))
             .attachment(Attachment::new(sn, publisher_gid))
             .wait()
             .expect("put");
@@ -169,11 +173,8 @@ fn joining_late_reports_no_loss() {
 
     // First sample this subscriber ever sees from this publisher, and it is
     // already well into the publisher's stream.
-    let zbuf = <hiroz::msg::SerdeCdrSerdes<Tick> as hiroz::msg::ZSerializer>::serialize_to_zbuf(
-        &Tick { counter: 9000 },
-    );
     session
-        .put((*ke).clone(), zenoh::bytes::ZBytes::from(zbuf))
+        .put((*ke).clone(), payload(9000))
         .attachment(Attachment::new(9000, gid(7)))
         .wait()
         .expect("put");

crates/hiroz/src/pubsub.rs

@@ -1012,6 +1012,25 @@ impl<T: ZMessage, Q, S: ZDeserializer> std::fmt::Debug for ZSub<T, Q, S> {
     }
 }
 
+/// Accessors that do not depend on how the subscriber delivers.
+///
+/// These were on the `ZSub<T, Sample, S>` (queue-mode) impl only, so a callback
+/// subscriber could not reach its own events manager — the handle the rmw layer
+/// needs to install an event callback, and the only way to observe
+/// [`MessageLost`](crate::event::ZenohEventType::MessageLost). Neither field has
+/// anything to do with the queue.
+impl<T, Q, S> ZSub<T, Q, S> {
+    /// The event manager this subscriber raises endpoint events on.
+    pub fn events_mgr(&self) -> &Arc<Mutex<EventsManager>> {
+        &self.events_mgr
+    }
+
+    /// Get a reference to the endpoint entity for this subscriber.
+    pub fn entity(&self) -> &EndpointEntity {
+        &self.entity
+    }
+}
+
 impl<T, S> ZSub<T, Sample, S>
 where
     T: ZMessage,
@@ -1043,15 +1062,6 @@ where
             .ok_or_else(|| crate::error::Error::timeout(timeout))
     }
 
-    pub fn events_mgr(&self) -> &Arc<Mutex<EventsManager>> {
-        &self.events_mgr
-    }
-
-    /// Get a reference to the endpoint entity for this subscriber.
-    pub fn entity(&self) -> &EndpointEntity {
-        &self.entity
-    }
-
     /// Check if there are messages available in the queue
     pub fn is_ready(&self) -> bool {
         self.queue.as_ref().map(|q| !q.is_empty()).unwrap_or(false)