From 90376ee0ee5de9e50e215a22556c01e6f98b53ca Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Thu, 6 Aug 2026 14:10:58 +0800 Subject: [PATCH 1/5] feat(event): raise MessageLost from sequence gaps RMW_EVENT_MESSAGE_LOST was fully plumbed and never raised: the enum, the rmw_event_type mapping, the rmw_message_lost_status_t fill-in and the Attachment::sequence_number on the wire all existed, but nothing emitted it. A subscriber asking for the event got a callback that never fired and a status permanently zero (#292). MessageLossTracker holds the last sequence seen per publisher GID and raises the event when an arrival skips past one. It is the only place hiroz trailed rmw_zenoh_cpp on event coverage. Deliberately not counted: a subscriber dropping its own oldest queued sample at the history depth. That sample arrived and updated the baseline, so it produces no gap -- and upstream draws the line in the same place, logging depth-drops at debug and raising the event only for gaps. Raises via update_shared_event_status, so the callout happens with no lock held (#259/#260); the per-GID map has its own lock and is never held across it. Depends on #260 for that entry point -- it does not exist on main. --- crates/hiroz/src/event.rs | 109 +++++++++++++++++++++++++++++++++++++ crates/hiroz/src/pubsub.rs | 30 ++++++++-- 2 files changed, 135 insertions(+), 4 deletions(-) diff --git a/crates/hiroz/src/event.rs b/crates/hiroz/src/event.rs index 9dea9e7a1..cd978570b 100644 --- a/crates/hiroz/src/event.rs +++ b/crates/hiroz/src/event.rs @@ -445,6 +445,67 @@ pub fn update_shared_event_status( update_shared_event_status_with_policy(events_mgr, event_type, change, 0) } +/// Detects samples lost **in transit** and raises [`ZenohEventType::MessageLost`]. +/// +/// Every sample carries an [`Attachment`] with the publisher's GID and a +/// per-publisher sequence number. Holding the last sequence seen from each +/// publisher makes a gap detectable: receiving `n` when `n - 2` was the last +/// means one sample never arrived. +/// +/// [`Attachment`]: crate::attachment::Attachment +/// +/// # What this does *not* count +/// +/// A subscriber dropping its own oldest queued sample because the queue is at +/// its history depth. That sample **arrived** — it updated the last-seen +/// sequence on the way in — so it produces no gap, and the ROS event does not +/// claim it. `rmw_zenoh_cpp` draws the line in the same place: its depth-drops +/// are a debug log, and only sequence gaps raise `MESSAGE_LOST`. +pub struct MessageLossTracker { + events_mgr: Arc>, + /// Last sequence number seen per publisher GID. + /// + /// Its own lock, and never held across the callout below — raising the + /// event runs user code, which may re-enter this subscriber. + last_seen: Mutex>, +} + +impl MessageLossTracker { + pub fn new(events_mgr: Arc>) -> Self { + Self { + events_mgr, + last_seen: Mutex::new(HashMap::new()), + } + } + + /// Record an arrival, raising the event if it skipped past anything. + pub fn observe(&self, source_gid: GidArray, sequence_number: i64) { + let lost = { + let Ok(mut seen) = self.last_seen.lock() else { + return; + }; + match seen.insert(source_gid, sequence_number) { + // Not the first from this publisher: anything strictly between + // the two never arrived. A non-positive difference means a + // retransmit, a reorder, or a publisher that restarted its + // numbering — none of which is loss, so it reports nothing. + Some(previous) => sequence_number.saturating_sub(previous).saturating_sub(1), + // First sample from this publisher. There is no baseline to + // measure against, and a subscriber that joined late has not + // "lost" the history it was never sent. + None => 0, + } + }; + + if lost > 0 { + // Clamped rather than truncated: the rmw status field is i32, and a + // publisher restart can produce an arbitrarily large apparent jump. + let lost = lost.min(i64::from(i32::MAX)) as i32; + update_shared_event_status(&self.events_mgr, ZenohEventType::MessageLost, lost); + } + } +} + /// [`update_shared_event_status`] with a QoS policy kind. /// /// # Known hazard @@ -556,6 +617,54 @@ mod tests { assert_eq!(status.current_count, 0); } + /// Drive a tracker through a sequence of arrivals and return the total + /// `MessageLost` count it reported. + fn losses_for(arrivals: &[(u8, i64)]) -> i32 { + let mgr = Arc::new(Mutex::new(EventsManager::new(gid(1)))); + let tracker = MessageLossTracker::new(mgr.clone()); + for &(publisher, sn) in arrivals { + tracker.observe(gid(publisher), sn); + } + mgr.lock() + .unwrap() + .take_event_status(ZenohEventType::MessageLost) + .total_count + } + + #[test] + fn message_loss_is_counted_from_sequence_gaps() { + // 0,1,2 contiguous → nothing lost. Then 5 skips 3 and 4. + assert_eq!(losses_for(&[(1, 0), (1, 1), (1, 2), (1, 5)]), 2); + } + + #[test] + fn message_loss_ignores_the_first_sample_from_a_publisher() { + // A late joiner's first sample has no baseline. Reporting `sn` as the + // loss count would make every subscriber that starts late look lossy. + assert_eq!(losses_for(&[(1, 9_000)]), 0); + } + + #[test] + fn message_loss_is_tracked_per_publisher() { + // Interleaved publishers each keep their own baseline; without that, + // alternating 0,0,1,1 reads as a gap on every other sample. + assert_eq!(losses_for(&[(1, 0), (2, 0), (1, 1), (2, 1)]), 0); + } + + #[test] + fn message_loss_ignores_reorder_and_republish() { + // A non-positive difference is a retransmit, a reorder, or a publisher + // that restarted its numbering — none of which is loss. + assert_eq!(losses_for(&[(1, 5), (1, 3), (1, 5), (1, 0)]), 0); + } + + #[test] + fn message_loss_clamps_an_implausible_jump() { + // A restarted publisher can present an arbitrarily large apparent gap; + // the rmw status field is i32, so it must saturate rather than wrap. + assert_eq!(losses_for(&[(1, 0), (1, i64::MAX)]), i32::MAX); + } + #[test] fn test_update_event_status_fires_callback() { let called = Arc::new(Mutex::new(0i32)); diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 5b637d031..1adee5314 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -10,7 +10,7 @@ use crate::Builder; use crate::attachment::{Attachment, GidArray}; use crate::common::DataHandler; use crate::entity::{EndpointEntity, EndpointKind}; -use crate::event::EventsManager; +use crate::event::{EventsManager, MessageLossTracker}; use crate::graph::Graph; use crate::impl_with_type_info; use crate::queue::BoundedQueue; @@ -36,6 +36,22 @@ const SAMPLE_MISS_HEARTBEAT_PERIOD: Duration = Duration::from_millis(500); /// truncation surprises inside zenoh-ext's internal `as_millis()` paths). const TRANSIENT_LOCAL_QUERY_TIMEOUT: Duration = Duration::from_millis(u64::MAX); +/// Feed an arriving sample's attachment to the loss tracker. +/// +/// A sample whose attachment is missing or undecodable is **ignored, not +/// counted**. The sequence number is the only thing that makes loss detectable, +/// and a publisher that does not send one — a plain zenoh peer rather than a +/// hiroz node — must not be reported as lossy just for being unrecognised. +fn observe_loss(loss: &MessageLossTracker, sample: &Sample) { + let Some(bytes) = sample.attachment() else { + return; + }; + let Ok(attachment) = Attachment::try_from(bytes) else { + return; + }; + loss.observe(attachment.source_gid, attachment.sequence_number); +} + fn cache_depth_from_history(history: QosHistory) -> usize { // Mirrors rmw_zenoh_cpp's `QoS::best_available_qos` (`qos.cpp:107`): // a zero-valued depth (the rmw representation of "unspecified" or @@ -802,9 +818,17 @@ where key_expr, self.entity.qos ); + // Built here rather than at `ZSub` construction: the loss tracker needs + // a handle to it, and the tracker is captured by the callback below. + let gid = crate::entity::endpoint_gid(&self.entity) + .expect("local endpoint always has node identity"); + let events_mgr = Arc::new(Mutex::new(EventsManager::new(gid))); + let loss = Arc::new(MessageLossTracker::new(events_mgr.clone())); + // Wrap handler with encoding validation if expected encoding is set let expected_encoding = self.expected_encoding.clone(); let validated_handler = move |sample: Sample| { + observe_loss(&loss, &sample); // Validate encoding if expected encoding is set if let Some(ref expected) = expected_encoding { let encoding_str = sample.encoding().to_string(); @@ -841,8 +865,6 @@ where let sub_builder = apply_transient_local_sub(sub_builder, &self.entity.qos); let inner = sub_builder.wait()?; - let gid = crate::entity::endpoint_gid(&self.entity) - .expect("local endpoint always has node identity"); let lv_ke = self .keyexpr_format .liveliness_key_expr(&self.entity, &self.session.zid())?; @@ -859,7 +881,7 @@ where _inner: inner, _lv_token: lv_token, queue, - events_mgr: Arc::new(Mutex::new(EventsManager::new(gid))), + events_mgr, graph: self.graph, dyn_schema: self.dyn_schema, expected_encoding: self.expected_encoding, From 82b6eac20258f4dc5ea15702e1aede9a142ab3d8 Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Thu, 6 Aug 2026 14:13:41 +0800 Subject: [PATCH 2/5] fix(event): advance the loss baseline only forward A replayed or reordered sample must not move the high-water mark backwards, or the next ordinary sample reads as a gap. This diverges from rmw_zenoh_cpp deliberately. Upstream uses std::abs(sn - last) and rewrites the baseline unconditionally, so on arrivals 5, 3, 6 it reports 1 lost for the replay and 2 more for the sample after it. Every TransientLocal subscriber replays history, so that false positive is reachable rather than theoretical. --- crates/hiroz/src/event.rs | 48 ++++++++++++++++++++++++++++++++------- 1 file changed, 40 insertions(+), 8 deletions(-) diff --git a/crates/hiroz/src/event.rs b/crates/hiroz/src/event.rs index cd978570b..e7b364bbf 100644 --- a/crates/hiroz/src/event.rs +++ b/crates/hiroz/src/event.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; @@ -484,22 +485,42 @@ impl MessageLossTracker { let Ok(mut seen) = self.last_seen.lock() else { return; }; - match seen.insert(source_gid, sequence_number) { - // Not the first from this publisher: anything strictly between - // the two never arrived. A non-positive difference means a - // retransmit, a reorder, or a publisher that restarted its - // numbering — none of which is loss, so it reports nothing. - Some(previous) => sequence_number.saturating_sub(previous).saturating_sub(1), + match seen.entry(source_gid) { // First sample from this publisher. There is no baseline to // measure against, and a subscriber that joined late has not // "lost" the history it was never sent. - None => 0, + Entry::Vacant(slot) => { + slot.insert(sequence_number); + 0 + } + Entry::Occupied(mut slot) => { + let high_water = *slot.get(); + // The baseline only ever moves **forward**. An arrival at or + // below it is a replay, a retransmit or a reorder — not + // loss — and letting it move the baseline backwards would + // make the *next* ordinary sample look like a gap. + // + // This is a deliberate divergence from `rmw_zenoh_cpp`, + // which uses `std::abs(sn - last)` and rewrites the + // baseline unconditionally. On a `TransientLocal` + // subscriber, history replay delivers older sequence + // numbers as a matter of course, so that shape reports + // phantom loss twice per replayed sample. + if sequence_number <= high_water { + 0 + } else { + slot.insert(sequence_number); + sequence_number.saturating_sub(high_water).saturating_sub(1) + } + } } }; if lost > 0 { // Clamped rather than truncated: the rmw status field is i32, and a - // publisher restart can produce an arbitrarily large apparent jump. + // publisher that restarts its numbering can present an arbitrarily + // large apparent jump. (In ROS a restarted endpoint normally gets a + // fresh GID and lands in the vacant arm instead.) let lost = lost.min(i64::from(i32::MAX)) as i32; update_shared_event_status(&self.events_mgr, ZenohEventType::MessageLost, lost); } @@ -658,6 +679,17 @@ mod tests { assert_eq!(losses_for(&[(1, 5), (1, 3), (1, 5), (1, 0)]), 0); } + /// A replayed sample must not make the *next* ordinary one look like a gap. + /// + /// This is the case `rmw_zenoh_cpp` gets wrong: `std::abs(sn - last)` plus + /// an unconditional baseline rewrite reports 1 lost for the replay and 2 + /// more for the sample after it. Every `TransientLocal` subscriber replays + /// history, so it is reachable rather than theoretical. + #[test] + fn message_loss_survives_a_transient_local_replay() { + assert_eq!(losses_for(&[(1, 5), (1, 3), (1, 6)]), 0); + } + #[test] fn message_loss_clamps_an_implausible_jump() { // A restarted publisher can present an arbitrarily large apparent gap; From ac9bfa2db2bc0a4be024468eb4a1c758c989f672 Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Thu, 6 Aug 2026 14:19:51 +0800 Subject: [PATCH 3/5] test(event): pin the MessageLost wiring, not just its arithmetic The unit tests in event.rs exercise MessageLossTracker directly, so deleting the observe_loss(..) call from the subscriber receive path leaves every one of them green. This file fails in that case. Loss is induced deterministically instead of by dropping a packet: the test publishes onto the subscriber's own key expression through the node's session with a hand-built Attachment, so the sequence gap is exact and there is no timing to lose. --- crates/hiroz-tests/tests/message_lost.rs | 187 +++++++++++++++++++++++ 1 file changed, 187 insertions(+) create mode 100644 crates/hiroz-tests/tests/message_lost.rs diff --git a/crates/hiroz-tests/tests/message_lost.rs b/crates/hiroz-tests/tests/message_lost.rs new file mode 100644 index 000000000..18b263803 --- /dev/null +++ b/crates/hiroz-tests/tests/message_lost.rs @@ -0,0 +1,187 @@ +//! `RMW_EVENT_MESSAGE_LOST` must actually be raised by a live subscriber. +//! +//! `event.rs`'s unit tests pin `MessageLossTracker`'s arithmetic. They say +//! nothing about whether anything *calls* it: delete the `observe_loss(..)` line +//! from the subscriber's receive path and every one of them still passes. This +//! file is the detector for that wiring. +//! +//! Loss is induced deterministically rather than by trying to drop a packet. +//! The subscriber's own key expression is published to directly, through the +//! node's zenoh session, with a hand-built [`Attachment`] carrying a chosen +//! sequence number — so the gap is exact and there is no timing to lose. + +mod common; + +use std::{ + sync::{Arc, Mutex}, + thread, + time::{Duration, Instant}, +}; + +use common::{TestRouter, create_hiroz_context_with_endpoint}; +use hiroz::{ + Builder, GidArray, TypeHash, + attachment::Attachment, + event::ZenohEventType, + ros_msg::MessageTypeInfo, +}; +use serde::{Deserialize, Serialize}; +use serial_test::serial; +use zenoh::Wait; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +struct Tick { + counter: u64, +} + +impl MessageTypeInfo for Tick { + fn type_name() -> &'static str { + "test_msgs::msg::dds_::Tick_" + } + fn type_hash() -> TypeHash { + TypeHash::zero() + } +} + +impl hiroz::ros_msg::WithTypeInfo for Tick {} + +impl hiroz::msg::ZMessage for Tick { + type Serdes = hiroz::msg::SerdeCdrSerdes; +} + +fn gid(n: u8) -> GidArray { + let mut g = [0u8; 16]; + g[0] = n; + g +} + +/// Wait until `total_count` for `MessageLost` stops changing, then return it. +fn settled_loss_count(sub_events: &Arc>) -> i32 { + let mut last = -1; + let mut stable_since = Instant::now(); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let now = sub_events + .lock() + .unwrap() + .take_event_status(ZenohEventType::MessageLost) + .total_count; + if now != last { + last = now; + stable_since = Instant::now(); + } else if stable_since.elapsed() >= Duration::from_millis(400) { + return now; + } + assert!(Instant::now() < deadline, "loss count never settled"); + thread::sleep(Duration::from_millis(25)); + } +} + +/// A gap in a publisher's sequence numbers raises `MessageLost` on the +/// subscriber that saw it, with the count of samples that never arrived. +#[test] +#[serial] +fn a_sequence_gap_raises_message_lost() { + const TOPIC: &str = "/message_lost_gap"; + + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("context"); + let node = ctx.create_node("message_lost_node").build().expect("node"); + + let received = Arc::new(Mutex::new(Vec::::new())); + let cb_received = received.clone(); + let sub = node + .create_sub::(TOPIC) + .build_with_callback(move |msg: Tick| { + cb_received.lock().unwrap().push(msg.counter); + }) + .expect("subscriber"); + + // Publish straight onto the subscriber's own key expression, so the + // sequence numbers are ours to choose. + let ke = node + .keyexpr_format() + .topic_key_expr(sub.entity()) + .expect("topic key expr"); + let session = node.session(); + let publisher_gid = gid(42); + + let put = |sn: i64, counter: u64| { + let zbuf = as hiroz::msg::ZSerializer>::serialize_to_zbuf( + &Tick { counter }, + ); + session + .put((*ke).clone(), zenoh::bytes::ZBytes::from(zbuf)) + .attachment(Attachment::new(sn, publisher_gid)) + .wait() + .expect("put"); + }; + + thread::sleep(Duration::from_millis(300)); + + put(0, 0); // baseline — first from this publisher, never counted + put(1, 1); // contiguous + put(5, 5); // 2, 3 and 4 never arrived + + let lost = settled_loss_count(sub.events_mgr()); + + assert_eq!( + lost, 3, + "expected the three skipped sequence numbers to be reported as lost; \ + got {lost}. Zero means the receive path never fed the loss tracker" + ); + assert_eq!( + received.lock().unwrap().len(), + 3, + "all three published samples should still have been delivered — \ + detecting loss must not drop anything" + ); +} + +/// A subscriber that joins late has not "lost" the history it was never sent. +/// +/// Without the first-sample exemption this reports the publisher's sequence +/// number as the loss count, so every late joiner looks catastrophically lossy. +#[test] +#[serial] +fn joining_late_reports_no_loss() { + const TOPIC: &str = "/message_lost_late_join"; + + let router = TestRouter::new(); + let ctx = create_hiroz_context_with_endpoint(router.endpoint()).expect("context"); + let node = ctx + .create_node("message_lost_late_node") + .build() + .expect("node"); + + let sub = node + .create_sub::(TOPIC) + .build_with_callback(|_msg: Tick| {}) + .expect("subscriber"); + + let ke = node + .keyexpr_format() + .topic_key_expr(sub.entity()) + .expect("topic key expr"); + let session = node.session(); + + thread::sleep(Duration::from_millis(300)); + + // First sample this subscriber ever sees from this publisher, and it is + // already well into the publisher's stream. + let zbuf = as hiroz::msg::ZSerializer>::serialize_to_zbuf( + &Tick { counter: 9000 }, + ); + session + .put((*ke).clone(), zenoh::bytes::ZBytes::from(zbuf)) + .attachment(Attachment::new(9000, gid(7))) + .wait() + .expect("put"); + + let lost = settled_loss_count(sub.events_mgr()); + + assert_eq!( + lost, 0, + "a late joiner must not be charged for history it was never sent" + ); +} From 6e66c64438ea77ea5d9d489bbafcfb8f576cee8a Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Thu, 6 Aug 2026 14:24:33 +0800 Subject: [PATCH 4/5] fix(pubsub): expose events_mgr and entity on every ZSub variant Both accessors lived on the ZSub (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. --- crates/hiroz-tests/tests/message_lost.rs | 17 +++++++------- crates/hiroz/src/pubsub.rs | 28 ++++++++++++++++-------- 2 files changed, 28 insertions(+), 17 deletions(-) diff --git a/crates/hiroz-tests/tests/message_lost.rs b/crates/hiroz-tests/tests/message_lost.rs index 18b263803..cae92b79e 100644 --- a/crates/hiroz-tests/tests/message_lost.rs +++ b/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 = >::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>) -> 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 = 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 = 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"); diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 1adee5314..1b59890d9 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -1012,6 +1012,25 @@ impl std::fmt::Debug for ZSub { } } +/// Accessors that do not depend on how the subscriber delivers. +/// +/// These were on the `ZSub` (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 ZSub { + /// The event manager this subscriber raises endpoint events on. + pub fn events_mgr(&self) -> &Arc> { + &self.events_mgr + } + + /// Get a reference to the endpoint entity for this subscriber. + pub fn entity(&self) -> &EndpointEntity { + &self.entity + } +} + impl ZSub where T: ZMessage, @@ -1043,15 +1062,6 @@ where .ok_or_else(|| crate::error::Error::timeout(timeout)) } - pub fn events_mgr(&self) -> &Arc> { - &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) From 5ed2b56b09276a05d94fdfc584c3888c6543c00e Mon Sep 17 00:00:00 2001 From: yuanyuyuan Date: Thu, 6 Aug 2026 14:28:10 +0800 Subject: [PATCH 5/5] fix(pubsub): repeat ZSub's bounds on the generic accessor impl ZSub declares T: ZMessage, S: ZDeserializer on the struct, so a bare impl does not satisfy them. --- crates/hiroz-tests/tests/message_lost.rs | 4 +--- crates/hiroz/src/pubsub.rs | 2 +- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/crates/hiroz-tests/tests/message_lost.rs b/crates/hiroz-tests/tests/message_lost.rs index cae92b79e..621d7f306 100644 --- a/crates/hiroz-tests/tests/message_lost.rs +++ b/crates/hiroz-tests/tests/message_lost.rs @@ -20,9 +20,7 @@ use std::{ use common::{TestRouter, create_hiroz_context_with_endpoint}; use hiroz::{ - Builder, GidArray, TypeHash, - attachment::Attachment, - event::ZenohEventType, + Builder, GidArray, TypeHash, attachment::Attachment, event::ZenohEventType, ros_msg::MessageTypeInfo, }; use serde::{Deserialize, Serialize}; diff --git a/crates/hiroz/src/pubsub.rs b/crates/hiroz/src/pubsub.rs index 1b59890d9..b03bb30b6 100644 --- a/crates/hiroz/src/pubsub.rs +++ b/crates/hiroz/src/pubsub.rs @@ -1019,7 +1019,7 @@ impl std::fmt::Debug for ZSub { /// 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 ZSub { +impl ZSub { /// The event manager this subscriber raises endpoint events on. pub fn events_mgr(&self) -> &Arc> { &self.events_mgr