Remove pull API and protocol support (#821)

* Remove Pull subscriber * Fix doctest. Remove unused code. * Remove routing code for pull subscriptions * Remove pull mode from DeclareSubscriber * Remove unsupported Put/Del in Request/Response (#839) * Address review comments --------- Co-authored-by: OlivierHecart <[email protected]>
eclipse-zenoh · Mar 19, 2024 · 7300f4c · 7300f4c
1 parent 665c90f
commit 7300f4c
Show file tree

Hide file tree

Showing 34 changed files with 242 additions and 1,079 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/commons/zenoh-codec/src/zenoh/mod.rs b/commons/zenoh-codec/src/zenoh/mod.rs
@@ -13,7 +13,6 @@
 //
 pub mod del;
 pub mod err;
-pub mod pull;
 pub mod put;
 pub mod query;
 pub mod reply;
@@ -81,9 +80,6 @@ where
     fn write(self, writer: &mut W, x: &RequestBody) -> Self::Output {
         match x {
             RequestBody::Query(b) => self.write(&mut *writer, b),
-            RequestBody::Put(b) => self.write(&mut *writer, b),
-            RequestBody::Del(b) => self.write(&mut *writer, b),
-            RequestBody::Pull(b) => self.write(&mut *writer, b),
         }
     }
 }
@@ -100,9 +96,6 @@ where
         let codec = Zenoh080Header::new(header);
         let body = match imsg::mid(codec.header) {
             id::QUERY => RequestBody::Query(codec.read(&mut *reader)?),
-            id::PUT => RequestBody::Put(codec.read(&mut *reader)?),
-            id::DEL => RequestBody::Del(codec.read(&mut *reader)?),
-            id::PULL => RequestBody::Pull(codec.read(&mut *reader)?),
             _ => return Err(DidntRead),
         };
 
@@ -121,7 +114,6 @@ where
         match x {
             ResponseBody::Reply(b) => self.write(&mut *writer, b),
             ResponseBody::Err(b) => self.write(&mut *writer, b),
-            ResponseBody::Put(b) => self.write(&mut *writer, b),
         }
     }
 }
@@ -139,7 +131,6 @@ where
         let body = match imsg::mid(codec.header) {
             id::REPLY => ResponseBody::Reply(codec.read(&mut *reader)?),
             id::ERR => ResponseBody::Err(codec.read(&mut *reader)?),
-            id::PUT => ResponseBody::Put(codec.read(&mut *reader)?),
             _ => return Err(DidntRead),
         };
 

diff --git a/commons/zenoh-codec/src/zenoh/pull.rs b/commons/zenoh-codec/src/zenoh/pull.rs
diff --git a/commons/zenoh-codec/tests/codec.rs b/commons/zenoh-codec/tests/codec.rs
@@ -600,8 +600,3 @@ fn codec_reply() {
 fn codec_err() {
     run!(zenoh::Err, zenoh::Err::rand());
 }
-
-#[test]
-fn codec_pull() {
-    run!(zenoh::Pull, zenoh::Pull::rand());
-}
diff --git a/commons/zenoh-collections/src/ring_buffer.rs b/commons/zenoh-collections/src/ring_buffer.rs
@@ -40,6 +40,15 @@ impl<T> RingBuffer<T> {
         Some(elem)
     }
 
+    #[inline]
+    pub fn push_force(&mut self, elem: T) -> Option<T> {
+        self.push(elem).and_then(|elem| {
+            let ret = self.buffer.pop_front();
+            self.buffer.push_back(elem);
+            ret
+        })
+    }
+
     #[inline]
     pub fn pull(&mut self) -> Option<T> {
         let x = self.buffer.pop_front();

diff --git a/commons/zenoh-protocol/src/network/declare.rs b/commons/zenoh-protocol/src/network/declare.rs
@@ -146,31 +146,6 @@ impl Declare {
     }
 }
 
-#[derive(Debug, Default, Copy, Clone, PartialEq, Eq)]
-#[repr(u8)]
-pub enum Mode {
-    #[default]
-    Push,
-    Pull,
-}
-
-impl Mode {
-    pub const DEFAULT: Self = Self::Push;
-
-    #[cfg(feature = "test")]
-    fn rand() -> Self {
-        use rand::Rng;
-
-        let mut rng = rand::thread_rng();
-
-        if rng.gen_bool(0.5) {
-            Mode::Push
-        } else {
-            Mode::Pull
-        }
-    }
-}
-
 pub mod common {
     use super::*;
 
@@ -320,9 +295,7 @@ pub mod subscriber {
     /// ~  [decl_exts]  ~  if Z==1
     /// +---------------+
     ///
-    /// - if R==1 then the subscription is reliable, else it is best effort
-    /// - if P==1 then the subscription is pull, else it is push
-    ///
+    /// - if R==1 then the subscription is reliable, else it is best effort    ///
     /// ```
     #[derive(Debug, Clone, PartialEq, Eq)]
     pub struct DeclareSubscriber {
@@ -343,34 +316,29 @@ pub mod subscriber {
         /// +-+-+-+-+-+-+-+-+
         /// |Z|0_1|    ID   |
         /// +-+-+-+---------+
-        /// % reserved  |P|R%
+        /// %  reserved   |R%
         /// +---------------+
         ///
         /// - if R==1 then the subscription is reliable, else it is best effort
-        /// - if P==1 then the subscription is pull, else it is push
         /// - rsv:  Reserved
         /// ```        
         #[derive(Debug, Clone, Copy, PartialEq, Eq)]
         pub struct SubscriberInfo {
             pub reliability: Reliability,
-            pub mode: Mode,
         }
 
         impl SubscriberInfo {
             pub const R: u64 = 1;
-            pub const P: u64 = 1 << 1;
 
             pub const DEFAULT: Self = Self {
                 reliability: Reliability::DEFAULT,
-                mode: Mode::DEFAULT,
             };
 
             #[cfg(feature = "test")]
             pub fn rand() -> Self {
                 let reliability = Reliability::rand();
-                let mode = Mode::rand();
 
-                Self { reliability, mode }
+                Self { reliability }
             }
         }
 
@@ -387,12 +355,7 @@ pub mod subscriber {
                 } else {
                     Reliability::BestEffort
                 };
-                let mode = if imsg::has_option(ext.value, SubscriberInfo::P) {
-                    Mode::Pull
-                } else {
-                    Mode::Push
-                };
-                Self { reliability, mode }
+                Self { reliability }
             }
         }
 
@@ -402,9 +365,6 @@ pub mod subscriber {
                 if ext.reliability == Reliability::Reliable {
                     v |= SubscriberInfo::R;
                 }
-                if ext.mode == Mode::Pull {
-                    v |= SubscriberInfo::P;
-                }
                 Info::new(v)
             }
         }

diff --git a/commons/zenoh-protocol/src/zenoh/mod.rs b/commons/zenoh-protocol/src/zenoh/mod.rs
@@ -13,15 +13,13 @@
 //
 pub mod del;
 pub mod err;
-pub mod pull;
 pub mod put;
 pub mod query;
 pub mod reply;
 
 use crate::core::Encoding;
 pub use del::Del;
 pub use err::Err;
-pub use pull::Pull;
 pub use put::Put;
 pub use query::{Consolidation, Query};
 pub use reply::Reply;
@@ -33,7 +31,6 @@ pub mod id {
     pub const QUERY: u8 = 0x03;
     pub const REPLY: u8 = 0x04;
     pub const ERR: u8 = 0x05;
-    pub const PULL: u8 = 0x06;
 }
 
 // DataInfo
@@ -80,9 +77,6 @@ impl From<Del> for PushBody {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum RequestBody {
     Query(Query),
-    Put(Put),
-    Del(Del),
-    Pull(Pull),
 }
 
 impl RequestBody {
@@ -92,11 +86,8 @@ impl RequestBody {
 
         let mut rng = rand::thread_rng();
 
-        match rng.gen_range(0..4) {
+        match rng.gen_range(0..1) {
             0 => RequestBody::Query(Query::rand()),
-            1 => RequestBody::Put(Put::rand()),
-            2 => RequestBody::Del(Del::rand()),
-            3 => RequestBody::Pull(Pull::rand()),
             _ => unreachable!(),
         }
     }
@@ -108,24 +99,11 @@ impl From<Query> for RequestBody {
     }
 }
 
-impl From<Put> for RequestBody {
-    fn from(p: Put) -> RequestBody {
-        RequestBody::Put(p)
-    }
-}
-
-impl From<Del> for RequestBody {
-    fn from(d: Del) -> RequestBody {
-        RequestBody::Del(d)
-    }
-}
-
 // Response
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ResponseBody {
     Reply(Reply),
     Err(Err),
-    Put(Put),
 }
 
 impl ResponseBody {
@@ -134,10 +112,9 @@ impl ResponseBody {
         use rand::Rng;
         let mut rng = rand::thread_rng();
 
-        match rng.gen_range(0..3) {
+        match rng.gen_range(0..2) {
             0 => ResponseBody::Reply(Reply::rand()),
             1 => ResponseBody::Err(Err::rand()),
-            2 => ResponseBody::Put(Put::rand()),
             _ => unreachable!(),
         }
     }