feat: add start/stop extension markers for fragment chain

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

@@ -39,6 +39,8 @@ where
             more,
             sn,
             ext_qos,
+            ext_start,
+            ext_stop,
         } = x;
 
         // Header
@@ -49,7 +51,10 @@ where
         if *more {
             header |= flag::M;
         }
-        if ext_qos != &ext::QoSType::DEFAULT {
+        let mut n_exts = (ext_qos != &ext::QoSType::DEFAULT) as u8
+            + ext_start.is_some() as u8
+            + ext_stop.is_some() as u8;
+        if n_exts != 0 {
             header |= flag::Z;
         }
         self.write(&mut *writer, header)?;
@@ -59,7 +64,16 @@ where
 
         // Extensions
         if ext_qos != &ext::QoSType::DEFAULT {
-            self.write(&mut *writer, (*ext_qos, false))?;
+            n_exts -= 1;
+            self.write(&mut *writer, (*ext_qos, n_exts != 0))?;
+        }
+        if let Some(start) = ext_start {
+            n_exts -= 1;
+            self.write(&mut *writer, (start, n_exts != 0))?
+        }
+        if let Some(stop) = ext_stop {
+            n_exts -= 1;
+            self.write(&mut *writer, (stop, n_exts != 0))?
         }
 
         Ok(())
@@ -99,6 +113,8 @@ where
 
         // Extensions
         let mut ext_qos = ext::QoSType::DEFAULT;
+        let mut ext_start = None;
+        let mut ext_stop = None;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -110,6 +126,16 @@ where
                     ext_qos = q;
                     has_ext = ext;
                 }
+                ext::Start::ID => {
+                    let (start, ext): (ext::Start, bool) = eodec.read(&mut *reader)?;
+                    ext_start = Some(start);
+                    has_ext = ext;
+                }
+                ext::Stop::ID => {
+                    let (stop, ext): (ext::Stop, bool) = eodec.read(&mut *reader)?;
+                    ext_stop = Some(stop);
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "Fragment", ext)?;
                 }
@@ -121,6 +147,8 @@ where
             more,
             sn,
             ext_qos,
+            ext_start,
+            ext_stop,
         })
     }
 }
@@ -139,6 +167,8 @@ where
             sn,
             payload,
             ext_qos,
+            ext_start,
+            ext_stop,
         } = x;
 
         // Header
@@ -147,6 +177,8 @@ where
             more: *more,
             sn: *sn,
             ext_qos: *ext_qos,
+            ext_start: *ext_start,
+            ext_stop: *ext_stop,
         };
         self.write(&mut *writer, &header)?;
 
@@ -185,6 +217,8 @@ where
             more: header.more,
             sn: header.sn,
             ext_qos: header.ext_qos,
+            ext_start: header.ext_start,
+            ext_stop: header.ext_stop,
             payload,
         })
     }

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

@@ -52,6 +52,7 @@ where
             ext_mlink,
             ext_lowlatency,
             ext_compression,
+            ext_patch,
         } = x;
 
         // Header
@@ -64,7 +65,8 @@ where
             + (ext_auth.is_some() as u8)
             + (ext_mlink.is_some() as u8)
             + (ext_lowlatency.is_some() as u8)
-            + (ext_compression.is_some() as u8);
+            + (ext_compression.is_some() as u8)
+            + (*ext_patch != ext::PatchType::DEFAULT) as u8;
 
         #[cfg(feature = "shared-memory")]
         {
@@ -125,6 +127,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (compression, n_exts != 0))?;
         }
+        if *ext_patch != ext::PatchType::DEFAULT {
+            n_exts -= 1;
+            self.write(&mut *writer, (*ext_patch, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -186,6 +192,7 @@ where
         let mut ext_mlink = None;
         let mut ext_lowlatency = None;
         let mut ext_compression = None;
+        let mut ext_patch = ext::PatchType::DEFAULT;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -228,6 +235,11 @@ where
                     ext_compression = Some(q);
                     has_ext = ext;
                 }
+                ext::Patch::ID => {
+                    let (p, ext): (ext::PatchType, bool) = eodec.read(&mut *reader)?;
+                    ext_patch = p;
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "InitSyn", ext)?;
                 }
@@ -248,6 +260,7 @@ where
             ext_mlink,
             ext_lowlatency,
             ext_compression,
+            ext_patch,
         })
     }
 }
@@ -275,6 +288,7 @@ where
             ext_mlink,
             ext_lowlatency,
             ext_compression,
+            ext_patch,
         } = x;
 
         // Header
@@ -287,7 +301,8 @@ where
             + (ext_auth.is_some() as u8)
             + (ext_mlink.is_some() as u8)
             + (ext_lowlatency.is_some() as u8)
-            + (ext_compression.is_some() as u8);
+            + (ext_compression.is_some() as u8)
+            + (*ext_patch != ext::PatchType::DEFAULT) as u8;
 
         #[cfg(feature = "shared-memory")]
         {
@@ -351,6 +366,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (compression, n_exts != 0))?;
         }
+        if *ext_patch != ext::PatchType::DEFAULT {
+            n_exts -= 1;
+            self.write(&mut *writer, (*ext_patch, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -415,6 +434,7 @@ where
         let mut ext_mlink = None;
         let mut ext_lowlatency = None;
         let mut ext_compression = None;
+        let mut ext_patch = ext::PatchType::DEFAULT;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -457,6 +477,11 @@ where
                     ext_compression = Some(q);
                     has_ext = ext;
                 }
+                ext::Patch::ID => {
+                    let (p, ext): (ext::PatchType, bool) = eodec.read(&mut *reader)?;
+                    ext_patch = p;
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "InitAck", ext)?;
                 }
@@ -478,6 +503,7 @@ where
             ext_mlink,
             ext_lowlatency,
             ext_compression,
+            ext_patch,
         })
     }
 }

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

@@ -150,6 +150,7 @@ where
             next_sn,
             ext_qos,
             ext_shm,
+            ext_patch,
         } = x;
 
         // Header
@@ -160,7 +161,9 @@ where
         if resolution != &Resolution::default() || batch_size != &batch_size::MULTICAST {
             header |= flag::S;
         }
-        let mut n_exts = (ext_qos.is_some() as u8) + (ext_shm.is_some() as u8);
+        let mut n_exts = (ext_qos.is_some() as u8)
+            + (ext_shm.is_some() as u8)
+            + (*ext_patch != ext::PatchType::DEFAULT) as u8;
         if n_exts != 0 {
             header |= flag::Z;
         }
@@ -201,6 +204,10 @@ where
             n_exts -= 1;
             self.write(&mut *writer, (shm, n_exts != 0))?;
         }
+        if *ext_patch != ext::PatchType::DEFAULT {
+            n_exts -= 1;
+            self.write(&mut *writer, (*ext_patch, n_exts != 0))?;
+        }
 
         Ok(())
     }
@@ -264,6 +271,7 @@ where
         // Extensions
         let mut ext_qos = None;
         let mut ext_shm = None;
+        let mut ext_patch = ext::PatchType::DEFAULT;
 
         let mut has_ext = imsg::has_flag(self.header, flag::Z);
         while has_ext {
@@ -280,6 +288,11 @@ where
                     ext_shm = Some(s);
                     has_ext = ext;
                 }
+                ext::Patch::ID => {
+                    let (p, ext): (ext::PatchType, bool) = eodec.read(&mut *reader)?;
+                    ext_patch = p;
+                    has_ext = ext;
+                }
                 _ => {
                     has_ext = extension::skip(reader, "Join", ext)?;
                 }
@@ -296,6 +309,7 @@ where
             next_sn,
             ext_qos,
             ext_shm,
+            ext_patch,
         })
     }
 }

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

@@ -176,3 +176,43 @@ where
         Ok((ext.into(), more))
     }
 }
+
+// Extensions: Patch
+impl<W, const ID: u8> WCodec<(ext::PatchType<{ ID }>, bool), &mut W> for Zenoh080
+where
+    W: Writer,
+{
+    type Output = Result<(), DidntWrite>;
+
+    fn write(self, writer: &mut W, x: (ext::PatchType<{ ID }>, bool)) -> Self::Output {
+        let (x, more) = x;
+        let ext: ZExtZ64<{ ID }> = x.into();
+
+        self.write(&mut *writer, (&ext, more))
+    }
+}
+
+impl<R, const ID: u8> RCodec<(ext::PatchType<{ ID }>, bool), &mut R> for Zenoh080
+where
+    R: Reader,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &mut R) -> Result<(ext::PatchType<{ ID }>, bool), Self::Error> {
+        let header: u8 = self.read(&mut *reader)?;
+        let codec = Zenoh080Header::new(header);
+        codec.read(reader)
+    }
+}
+
+impl<R, const ID: u8> RCodec<(ext::PatchType<{ ID }>, bool), &mut R> for Zenoh080Header
+where
+    R: Reader,
+{
+    type Error = DidntRead;
+
+    fn read(self, reader: &mut R) -> Result<(ext::PatchType<{ ID }>, bool), Self::Error> {
+        let (ext, more): (ZExtZ64<{ ID }>, bool) = self.read(&mut *reader)?;
+        Ok((ext.into(), more))
+    }
+}