feat!: make session an arc-like object

ci/valgrind-check/src/pub_sub/bin/z_pub_sub.rs

@@ -13,7 +13,7 @@
 //
 use std::time::Duration;
 
-use zenoh::{config::Config, key_expr::KeyExpr, prelude::*};
+use zenoh::{config::Config, key_expr::KeyExpr};
 
 #[tokio::main]
 async fn main() {

ci/valgrind-check/src/queryable_get/bin/z_queryable_get.rs

@@ -16,7 +16,6 @@ use std::{convert::TryFrom, time::Duration};
 use zenoh::{
     config::Config,
     key_expr::KeyExpr,
-    prelude::*,
     query::{QueryTarget, Selector},
 };
 

commons/zenoh-codec/src/network/mod.rs

@@ -76,7 +76,9 @@ where
         let header: u8 = self.codec.read(&mut *reader)?;
 
         let codec = Zenoh080Header::new(header);
-        codec.read(&mut *reader)
+        let mut msg: NetworkMessage = codec.read(&mut *reader)?;
+        msg.reliability = self.reliability;
+        Ok(msg)
     }
 }
 

commons/zenoh-codec/src/transport/batch.rs

@@ -150,13 +150,12 @@ where
     fn write(self, writer: &mut W, x: (&NetworkMessage, &FrameHeader)) -> Self::Output {
         let (m, f) = x;
 
-        // @TODO: m.is_reliable() always return true for the time being
-        // if let (Reliability::Reliable, false) | (Reliability::BestEffort, true) =
-        //     (f.reliability, m.is_reliable())
-        // {
-        //     // We are not serializing on the right frame.
-        //     return Err(BatchError::NewFrame);
-        // }
+        if let (Reliability::Reliable, false) | (Reliability::BestEffort, true) =
+            (f.reliability, m.is_reliable())
+        {
+            // We are not serializing on the right frame.
+            return Err(BatchError::NewFrame);
+        }
 
         // Mark the write operation
         let mark = writer.mark();

commons/zenoh-protocol/src/core/mod.rs

@@ -346,12 +346,12 @@ impl TryFrom<u8> for Priority {
 #[repr(u8)]
 pub enum Reliability {
     #[default]
-    BestEffort,
     Reliable,
+    BestEffort,
 }
 
 impl Reliability {
-    pub const DEFAULT: Self = Self::BestEffort;
+    pub const DEFAULT: Self = Self::Reliable;
 
     #[cfg(feature = "test")]
     pub fn rand() -> Self {

commons/zenoh-protocol/src/network/mod.rs

@@ -30,7 +30,7 @@ pub use push::Push;
 pub use request::{AtomicRequestId, Request, RequestId};
 pub use response::{Response, ResponseFinal};
 
-use crate::core::{CongestionControl, Priority};
+use crate::core::{CongestionControl, Priority, Reliability};
 
 pub mod id {
     // WARNING: it's crucial that these IDs do NOT collide with the IDs
@@ -83,6 +83,7 @@ pub enum NetworkBody {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct NetworkMessage {
     pub body: NetworkBody,
+    pub reliability: Reliability,
     #[cfg(feature = "stats")]
     pub size: Option<core::num::NonZeroUsize>,
 }
@@ -109,8 +110,7 @@ impl NetworkMessage {
 
     #[inline]
     pub fn is_reliable(&self) -> bool {
-        // TODO
-        true
+        self.reliability == Reliability::Reliable
     }
 
     #[inline]
@@ -179,6 +179,7 @@ impl From<NetworkBody> for NetworkMessage {
     fn from(body: NetworkBody) -> Self {
         Self {
             body,
+            reliability: Reliability::DEFAULT,
             #[cfg(feature = "stats")]
             size: None,
         }

commons/zenoh-protocol/src/transport/frame.rs

@@ -95,7 +95,8 @@ impl Frame {
         let ext_qos = ext::QoSType::rand();
         let mut payload = vec![];
         for _ in 0..rng.gen_range(1..4) {
-            let m = NetworkMessage::rand();
+            let mut m = NetworkMessage::rand();
+            m.reliability = reliability;
             payload.push(m);
         }
 

examples/examples/z_forward.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 use zenoh_ext::SubscriberForward;
 

examples/examples/z_get_liveliness.rs

@@ -14,7 +14,7 @@
 use std::time::Duration;
 
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_info.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{prelude::*, session::ZenohId};
+use zenoh::session::ZenohId;
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_liveliness.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_pong.rs

@@ -21,7 +21,7 @@ fn main() {
 
     let (config, express) = parse_args();
 
-    let session = zenoh::open(config).wait().unwrap().into_arc();
+    let session = zenoh::open(config).wait().unwrap();
 
     // The key expression to read the data from
     let key_expr_ping = keyexpr::new("test/ping").unwrap();

examples/examples/z_pull.rs

@@ -14,7 +14,7 @@
 use std::time::Duration;
 
 use clap::Parser;
-use zenoh::{handlers::RingChannel, key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{handlers::RingChannel, key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_queryable.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_queryable_shm.rs

@@ -15,7 +15,6 @@ use clap::Parser;
 use zenoh::{
     bytes::ZBytes,
     key_expr::KeyExpr,
-    prelude::*,
     shm::{
         zshm, BlockOn, GarbageCollect, PosixShmProviderBackend, ShmProviderBuilder,
         POSIX_PROTOCOL_ID,

examples/examples/z_storage.rs

@@ -19,7 +19,6 @@ use clap::Parser;
 use futures::select;
 use zenoh::{
     key_expr::{keyexpr, KeyExpr},
-    prelude::*,
     sample::{Sample, SampleKind},
     Config,
 };

examples/examples/z_sub.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, Config};
+use zenoh::{key_expr::KeyExpr, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_sub_liveliness.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 use clap::Parser;
-use zenoh::{key_expr::KeyExpr, prelude::*, sample::SampleKind, Config};
+use zenoh::{key_expr::KeyExpr, sample::SampleKind, Config};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_sub_shm.rs

@@ -14,7 +14,7 @@
 use clap::Parser;
 #[cfg(all(feature = "shared-memory", feature = "unstable"))]
 use zenoh::shm::zshm;
-use zenoh::{bytes::ZBytes, config::Config, key_expr::KeyExpr, prelude::*};
+use zenoh::{bytes::ZBytes, config::Config, key_expr::KeyExpr};
 use zenoh_examples::CommonArgs;
 
 #[tokio::main]

examples/examples/z_sub_thr.rs

@@ -87,9 +87,7 @@ fn main() {
             }
         })
         .wait()
-        .unwrap()
-        // Make the subscriber run in background, until the session is closed.
-        .background();
+        .unwrap();
 
     println!("Press CTRL-C to quit...");
     std::thread::park();

io/zenoh-transport/src/common/batch.rs

@@ -570,7 +570,7 @@ mod tests {
         let mut batch = WBatch::new(config);
 
         let tmsg: TransportMessage = KeepAlive.into();
-        let nmsg: NetworkMessage = Push {
+        let mut nmsg: NetworkMessage = Push {
             wire_expr: WireExpr::empty(),
             ext_qos: ext::QoSType::new(Priority::DEFAULT, CongestionControl::Block, false),
             ext_tstamp: None,
@@ -601,13 +601,15 @@ mod tests {
             sn: 0,
             ext_qos: frame::ext::QoSType::DEFAULT,
         };
+        nmsg.reliability = frame.reliability;
 
         // Serialize with a frame
         batch.encode((&nmsg, &frame)).unwrap();
         assert_ne!(batch.len(), 0);
         nmsgs_in.push(nmsg.clone());
 
         frame.reliability = Reliability::BestEffort;
+        nmsg.reliability = frame.reliability;
         batch.encode((&nmsg, &frame)).unwrap();
         assert_ne!(batch.len(), 0);
         nmsgs_in.push(nmsg.clone());

io/zenoh-transport/src/common/defragmentation.rs

@@ -66,7 +66,11 @@ impl DefragBuffer {
     pub(crate) fn push(&mut self, sn: TransportSn, zslice: ZSlice) -> ZResult<()> {
         if sn != self.sn.get() {
             self.clear();
-            bail!("Expected SN {}, received {}", self.sn.get(), sn)
+            bail!(
+                "Defragmentation SN error: expected SN {}, received {}",
+                self.sn.get(),
+                sn
+            )
         }
 
         let new_len = self.len + zslice.len();

io/zenoh-transport/src/common/pipeline.rs

@@ -30,7 +30,7 @@ use zenoh_codec::{transport::batch::BatchError, WCodec, Zenoh080};
 use zenoh_config::QueueSizeConf;
 use zenoh_core::zlock;
 use zenoh_protocol::{
-    core::{Priority, Reliability},
+    core::Priority,
     network::NetworkMessage,
     transport::{
         fragment::FragmentHeader,
@@ -220,7 +220,7 @@ impl StageIn {
 
         // The Frame
         let frame = FrameHeader {
-            reliability: Reliability::Reliable, // TODO
+            reliability: msg.reliability,
             sn,
             ext_qos: frame::ext::QoSType::new(priority),
         };