Update zenoh

eclipse-zenoh · Sep 12, 2024 · f7eae53 · f7eae53
1 parent 4da5247
commit f7eae53
Show file tree

Hide file tree

Showing 6 changed files with 45 additions and 47 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/zenoh-bridge-dds/src/main.rs b/zenoh-bridge-dds/src/main.rs
@@ -21,7 +21,6 @@ use clap::{App, Arg};
 use zenoh::{
     config::{Config, ModeDependentValue},
     internal::{plugins::PluginsManager, runtime::RuntimeBuilder},
-    prelude::*,
     session::ZenohId,
 };
 use zenoh_plugin_dds::DDSPlugin;

diff --git a/zenoh-plugin-dds/src/dds_mgt.rs b/zenoh-plugin-dds/src/dds_mgt.rs
@@ -31,9 +31,8 @@ use tracing::{debug, error, warn};
 use zenoh::{
     bytes::ZBytes,
     key_expr::{KeyExpr, OwnedKeyExpr},
-    prelude::*,
     qos::CongestionControl,
-    Session,
+    Session, Wait,
 };
 
 const MAX_SAMPLES: usize = 32;

diff --git a/zenoh-plugin-dds/src/lib.rs b/zenoh-plugin-dds/src/lib.rs
@@ -48,11 +48,10 @@ use zenoh::{
     },
     key_expr::{keyexpr, KeyExpr, OwnedKeyExpr},
     liveliness::LivelinessToken,
-    prelude::*,
     qos::CongestionControl,
     query::{ConsolidationMode, Query, QueryTarget, Queryable, Selector},
     sample::{Locality, Sample, SampleKind},
-    Result as ZResult, Session,
+    Result as ZResult, Session, Wait,
 };
 use zenoh_ext::{SessionExt, SubscriberBuilderExt};
 use zenoh_plugin_trait::{plugin_long_version, plugin_version, Plugin, PluginControl};
@@ -300,7 +299,7 @@ pub(crate) struct DdsPluginRuntime<'a> {
     // Note: &'a Arc<Session> here to keep the ownership of Session outside this struct
     // and be able to store the publishers/subscribers it creates in this same struct.
     zsession: &'a Arc<Session>,
-    _member: LivelinessToken<'a>,
+    _member: LivelinessToken,
     dp: dds_entity_t,
     // maps of all discovered DDS entities (indexed by DDS key)
     discovered_participants: HashMap<String, DdsParticipant>,
@@ -786,7 +785,7 @@ impl<'a> DdsPluginRuntime<'a> {
         group_subscriber: &Receiver<Sample>,
         dds_disco_rcv: &Receiver<DiscoveryEvent>,
         admin_keyexpr_prefix: OwnedKeyExpr,
-        admin_queryable: &Queryable<'_, flume::Receiver<Query>>,
+        admin_queryable: &Queryable<flume::Receiver<Query>>,
     ) {
         debug!(r#"Run in "local discovery" mode"#);
 
@@ -979,7 +978,7 @@ impl<'a> DdsPluginRuntime<'a> {
         group_subscriber: &Receiver<Sample>,
         dds_disco_rcv: &Receiver<DiscoveryEvent>,
         admin_keyexpr_prefix: OwnedKeyExpr,
-        admin_queryable: &Queryable<'_, flume::Receiver<Query>>,
+        admin_queryable: &Queryable<flume::Receiver<Query>>,
     ) {
         debug!(r#"Run in "forward discovery" mode"#);
 

diff --git a/zenoh-plugin-dds/src/route_dds_zenoh.rs b/zenoh-plugin-dds/src/route_dds_zenoh.rs
@@ -22,7 +22,6 @@ use cyclors::{
 use serde::Serialize;
 use zenoh::{
     key_expr::{keyexpr, KeyExpr, OwnedKeyExpr},
-    prelude::*,
     qos::CongestionControl,
     sample::Locality,
 };
@@ -32,7 +31,7 @@ use crate::{dds_mgt::*, qos_helpers::*, DdsPluginRuntime, KE_PREFIX_PUB_CACHE};
 
 enum ZPublisher<'a> {
     Publisher(KeyExpr<'a>),
-    PublicationCache(PublicationCache<'a>),
+    PublicationCache(PublicationCache),
 }
 
 impl ZPublisher<'_> {
@@ -157,7 +156,12 @@ impl RouteDDSZenoh<'_> {
                 })?;
             ZPublisher::PublicationCache(pub_cache)
         } else {
-            if let Err(e) = plugin.zsession.declare_publisher(declared_ke.clone()).await {
+            if let Err(e) = plugin
+                .zsession
+                .declare_publisher(declared_ke.clone())
+                .reliability(zenoh::pubsub::Reliability::Reliable)
+                .await
+            {
                 tracing::warn!(
                     "Failed to declare publisher for key {} (rid={}): {}",
                     ke,

diff --git a/zenoh-plugin-dds/src/route_zenoh_dds.rs b/zenoh-plugin-dds/src/route_zenoh_dds.rs
@@ -30,11 +30,10 @@ use cyclors::{
 use serde::{Serialize, Serializer};
 use zenoh::{
     key_expr::{keyexpr, KeyExpr, OwnedKeyExpr},
-    prelude::*,
     pubsub::Subscriber,
     query::{ConsolidationMode, QueryTarget, ReplyKeyExpr, Selector},
     sample::{Locality, Sample},
-    Session,
+    Session, Wait,
 };
 use zenoh_ext::{FetchingSubscriber, SubscriberBuilderExt};
 
@@ -46,12 +45,12 @@ use crate::{
 type AtomicDDSEntity = AtomicI32;
 const DDS_ENTITY_NULL: dds_entity_t = 0;
 
-enum ZSubscriber<'a> {
-    Subscriber(Subscriber<'a, ()>),
-    FetchingSubscriber(FetchingSubscriber<'a, ()>),
+enum ZSubscriber {
+    Subscriber(Subscriber<()>),
+    FetchingSubscriber(FetchingSubscriber<()>),
 }
 
-impl ZSubscriber<'_> {
+impl ZSubscriber {
     fn key_expr(&self) -> &KeyExpr<'static> {
         match self {
             ZSubscriber::Subscriber(s) => s.key_expr(),
@@ -69,7 +68,7 @@ pub(crate) struct RouteZenohDDS<'a> {
     zenoh_session: &'a Arc<Session>,
     // the zenoh subscriber receiving data to be re-published by the DDS Writer
     #[serde(skip)]
-    zenoh_subscriber: ZSubscriber<'a>,
+    zenoh_subscriber: ZSubscriber,
     // the DDS topic name for re-publication
     topic_name: String,
     // the DDS topic type
@@ -188,7 +187,6 @@ impl RouteZenohDDS<'_> {
                 .declare_subscriber(ke.clone())
                 .callback(subscriber_callback)
                 .allowed_origin(Locality::Remote) // Allow only remote publications to avoid loops
-                .reliable()
                 .querying()
                 .query_timeout(plugin.config.queries_timeout)
                 .query_selector(query_selector)
@@ -206,7 +204,6 @@ impl RouteZenohDDS<'_> {
                 .declare_subscriber(ke.clone())
                 .callback(subscriber_callback)
                 .allowed_origin(Locality::Remote) // Allow only remote publications to avoid loops
-                .reliable()
                 .await
                 .map_err(|e| {
                     format!(