merge main into protocol changes

Cargo.toml

@@ -112,11 +112,12 @@ libloading = "0.8"
 log = "0.4.17"
 lz4_flex = "0.11"
 nix = { version = "0.27", features = ["fs"] }
-num_cpus = "1.15.0"
+num_cpus = "1.16.0"
 ordered-float = "4.1.1"
 panic-message = "0.3.0"
 paste = "1.0.12"
 petgraph = "0.6.3"
+phf = { version = "0.11.2", features = ["macros"] }
 pnet = "0.34"
 pnet_datalink = "0.34"
 proc-macro2 = "1.0.51"
@@ -139,7 +140,9 @@ secrecy = { version = "0.8.0", features = ["serde", "alloc"] }
 serde = { version = "1.0.154", default-features = false, features = [
   "derive",
 ] } # Default features are disabled due to usage in no_std crates
-serde_json = "1.0.94"
+serde_cbor = "0.11.2"
+serde_json = "1.0.114"
+serde-pickle = "1.1.1"
 serde_yaml = "0.9.19"
 sha3 = "0.10.6"
 shared_memory = "0.12.4"
@@ -157,6 +160,7 @@ tokio-rustls = "0.25.0"
 console-subscriber = "0.2"
 typenum = "1.16.0"
 uhlc = { version = "0.7.0", default-features = false } # Default features are disabled due to usage in no_std crates
+unwrap-infallible = "0.1.5"
 unzip-n = "0.1.2"
 url = "2.3.1"
 urlencoding = "2.1.2"

commons/zenoh-buffers/src/lib.rs

@@ -101,7 +101,8 @@ pub mod buffer {
             let mut slices = self.slices();
             match slices.len() {
                 0 => Cow::Borrowed(b""),
-                1 => Cow::Borrowed(slices.next().unwrap()),
+                // SAFETY: it's safe to use unwrap_unchecked() beacuse we are explicitly checking the length is 1.
+                1 => Cow::Borrowed(unsafe { slices.next().unwrap_unchecked() }),
                 _ => Cow::Owned(slices.fold(Vec::new(), |mut acc, it| {
                     acc.extend(it);
                     acc
@@ -198,6 +199,18 @@ pub mod reader {
         fn rewind(&mut self, mark: Self::Mark) -> bool;
     }
 
+    pub trait AdvanceableReader: Reader {
+        fn skip(&mut self, offset: usize) -> Result<(), DidntRead>;
+        fn backtrack(&mut self, offset: usize) -> Result<(), DidntRead>;
+        fn advance(&mut self, offset: isize) -> Result<(), DidntRead> {
+            if offset > 0 {
+                self.skip(offset as usize)
+            } else {
+                self.backtrack((-offset) as usize)
+            }
+        }
+    }
+
     #[derive(Debug, Clone, Copy)]
     pub struct DidntSiphon;
 

commons/zenoh-buffers/src/zbuf.rs

@@ -15,12 +15,17 @@
 use crate::ZSliceKind;
 use crate::{
     buffer::{Buffer, SplitBuffer},
-    reader::{BacktrackableReader, DidntRead, DidntSiphon, HasReader, Reader, SiphonableReader},
+    reader::{
+        AdvanceableReader, BacktrackableReader, DidntRead, DidntSiphon, HasReader, Reader,
+        SiphonableReader,
+    },
     writer::{BacktrackableWriter, DidntWrite, HasWriter, Writer},
-    ZSlice,
+    ZSlice, ZSliceBuffer,
 };
 use alloc::{sync::Arc, vec::Vec};
 use core::{cmp, iter, mem, num::NonZeroUsize, ops::RangeBounds, ptr};
+#[cfg(feature = "std")]
+use std::io;
 use zenoh_collections::SingleOrVec;
 
 fn get_mut_unchecked<T>(arc: &mut Arc<T>) -> &mut T {
@@ -34,8 +39,10 @@ pub struct ZBuf {
 
 impl ZBuf {
     #[must_use]
-    pub fn empty() -> Self {
-        Self::default()
+    pub const fn empty() -> Self {
+        Self {
+            slices: SingleOrVec::empty(),
+        }
     }
 
     pub fn clear(&mut self) {
@@ -56,6 +63,21 @@ impl ZBuf {
         }
     }
 
+    pub fn to_zslice(&self) -> ZSlice {
+        let mut slices = self.zslices();
+        match self.slices.len() {
+            0 => ZSlice::empty(),
+            // SAFETY: it's safe to use unwrap_unchecked() beacuse we are explicitly checking the length is 1.
+            1 => unsafe { slices.next().unwrap_unchecked().clone() },
+            _ => slices
+                .fold(Vec::new(), |mut acc, it| {
+                    acc.extend(it.as_slice());
+                    acc
+                })
+                .into(),
+        }
+    }
+
     pub fn splice<Range: RangeBounds<usize>>(&mut self, erased: Range, replacement: &[u8]) {
         let start = match erased.start_bound() {
             core::ops::Bound::Included(n) => *n,
@@ -72,6 +94,7 @@ impl ZBuf {
         }
         self.insert(start, replacement);
     }
+
     fn remove(&mut self, mut start: usize, mut end: usize) {
         assert!(start <= end);
         assert!(end <= self.len());
@@ -100,6 +123,7 @@ impl ZBuf {
         let drain_end = end_slice_idx + (end_slice.start >= end_slice.end) as usize;
         self.slices.drain(drain_start..drain_end);
     }
+
     fn insert(&mut self, mut at: usize, slice: &[u8]) {
         if slice.is_empty() {
             return;
@@ -195,17 +219,34 @@ impl PartialEq for ZBuf {
 }
 
 // From impls
+impl From<ZSlice> for ZBuf {
+    fn from(t: ZSlice) -> Self {
+        let mut zbuf = ZBuf::empty();
+        zbuf.push_zslice(t);
+        zbuf
+    }
+}
+
+impl<T> From<Arc<T>> for ZBuf
+where
+    T: ZSliceBuffer + 'static,
+{
+    fn from(t: Arc<T>) -> Self {
+        let zslice: ZSlice = t.into();
+        Self::from(zslice)
+    }
+}
+
 impl<T> From<T> for ZBuf
 where
-    T: Into<ZSlice>,
+    T: ZSliceBuffer + 'static,
 {
     fn from(t: T) -> Self {
-        let mut zbuf = ZBuf::empty();
         let zslice: ZSlice = t.into();
-        zbuf.push_zslice(zslice);
-        zbuf
+        Self::from(zslice)
     }
 }
+
 // Reader
 #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub struct ZBufPos {
@@ -265,7 +306,7 @@ impl<'a> Reader for ZBufReader<'a> {
     }
 
     fn read_exact(&mut self, into: &mut [u8]) -> Result<(), DidntRead> {
-        let len = self.read(into)?;
+        let len = Reader::read(self, into)?;
         if len.get() == into.len() {
             Ok(())
         } else {
@@ -312,7 +353,7 @@ impl<'a> Reader for ZBufReader<'a> {
         match (slice.len() - self.cursor.byte).cmp(&len) {
             cmp::Ordering::Less => {
                 let mut buffer = crate::vec::uninit(len);
-                self.read_exact(&mut buffer)?;
+                Reader::read_exact(self, &mut buffer)?;
                 Ok(buffer.into())
             }
             cmp::Ordering::Equal => {
@@ -383,13 +424,81 @@ impl<'a> SiphonableReader for ZBufReader<'a> {
 }
 
 #[cfg(feature = "std")]
-impl<'a> std::io::Read for ZBufReader<'a> {
-    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+impl<'a> io::Read for ZBufReader<'a> {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         match <Self as Reader>::read(self, buf) {
             Ok(n) => Ok(n.get()),
+            Err(_) => Ok(0),
+        }
+    }
+}
+
+impl<'a> AdvanceableReader for ZBufReader<'a> {
+    fn skip(&mut self, offset: usize) -> Result<(), DidntRead> {
+        let mut remaining_offset = offset;
+        while remaining_offset > 0 {
+            let s = self.inner.slices.get(self.cursor.slice).ok_or(DidntRead)?;
+            let remains_in_current_slice = s.len() - self.cursor.byte;
+            let advance = remaining_offset.min(remains_in_current_slice);
+            remaining_offset -= advance;
+            self.cursor.byte += advance;
+            if self.cursor.byte == s.len() {
+                self.cursor.slice += 1;
+                self.cursor.byte = 0;
+            }
+        }
+        Ok(())
+    }
+
+    fn backtrack(&mut self, offset: usize) -> Result<(), DidntRead> {
+        let mut remaining_offset = offset;
+        while remaining_offset > 0 {
+            let backtrack = remaining_offset.min(self.cursor.byte);
+            remaining_offset -= backtrack;
+            self.cursor.byte -= backtrack;
+            if self.cursor.byte == 0 {
+                if self.cursor.slice == 0 {
+                    break;
+                }
+                self.cursor.slice -= 1;
+                self.cursor.byte = self
+                    .inner
+                    .slices
+                    .get(self.cursor.slice)
+                    .ok_or(DidntRead)?
+                    .len();
+            }
+        }
+        if remaining_offset == 0 {
+            Ok(())
+        } else {
+            Err(DidntRead)
+        }
+    }
+}
+
+#[cfg(feature = "std")]
+impl<'a> io::Seek for ZBufReader<'a> {
+    fn seek(&mut self, pos: io::SeekFrom) -> io::Result<u64> {
+        let current_pos = self
+            .inner
+            .slices()
+            .take(self.cursor.slice)
+            .fold(0, |acc, s| acc + s.len())
+            + self.cursor.byte;
+        let current_pos = i64::try_from(current_pos)
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("{}", e)))?;
+
+        let offset = match pos {
+            std::io::SeekFrom::Start(s) => i64::try_from(s).unwrap_or(i64::MAX) - current_pos,
+            std::io::SeekFrom::Current(s) => s,
+            std::io::SeekFrom::End(s) => self.inner.len() as i64 + s - current_pos,
+        };
+        match self.advance(offset as isize) {
+            Ok(()) => Ok((offset + current_pos) as u64),
             Err(_) => Err(std::io::Error::new(
-                std::io::ErrorKind::UnexpectedEof,
-                "UnexpectedEof",
+                std::io::ErrorKind::InvalidInput,
+                "InvalidInput",
             )),
         }
     }
@@ -609,18 +718,18 @@ impl BacktrackableWriter for ZBufWriter<'_> {
 }
 
 #[cfg(feature = "std")]
-impl<'a> std::io::Write for ZBufWriter<'a> {
-    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+impl<'a> io::Write for ZBufWriter<'a> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         match <Self as Writer>::write(self, buf) {
             Ok(n) => Ok(n.get()),
-            Err(_) => Err(std::io::Error::new(
-                std::io::ErrorKind::UnexpectedEof,
+            Err(_) => Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
                 "UnexpectedEof",
             )),
         }
     }
 
-    fn flush(&mut self) -> std::io::Result<()> {
+    fn flush(&mut self) -> io::Result<()> {
         Ok(())
     }
 }
@@ -663,4 +772,47 @@ mod tests {
 
         assert_eq!(zbuf1, zbuf2);
     }
+
+    #[cfg(feature = "std")]
+    #[test]
+    fn zbuf_seek() {
+        use super::{HasReader, ZBuf};
+        use crate::reader::Reader;
+        use std::io::Seek;
+
+        let mut buf = ZBuf::empty();
+        buf.push_zslice([0u8, 1u8, 2u8, 3u8].into());
+        buf.push_zslice([4u8, 5u8, 6u8, 7u8, 8u8].into());
+        buf.push_zslice([9u8, 10u8, 11u8, 12u8, 13u8, 14u8].into());
+        let mut reader = buf.reader();
+
+        assert_eq!(reader.stream_position().unwrap(), 0);
+        assert_eq!(reader.read_u8().unwrap(), 0);
+        assert_eq!(reader.seek(std::io::SeekFrom::Current(6)).unwrap(), 7);
+        assert_eq!(reader.read_u8().unwrap(), 7);
+        assert_eq!(reader.seek(std::io::SeekFrom::Current(-5)).unwrap(), 3);
+        assert_eq!(reader.read_u8().unwrap(), 3);
+        assert_eq!(reader.seek(std::io::SeekFrom::Current(10)).unwrap(), 14);
+        assert_eq!(reader.read_u8().unwrap(), 14);
+        reader.seek(std::io::SeekFrom::Current(100)).unwrap_err();
+
+        assert_eq!(reader.seek(std::io::SeekFrom::Start(0)).unwrap(), 0);
+        assert_eq!(reader.read_u8().unwrap(), 0);
+        assert_eq!(reader.seek(std::io::SeekFrom::Start(12)).unwrap(), 12);
+        assert_eq!(reader.read_u8().unwrap(), 12);
+        assert_eq!(reader.seek(std::io::SeekFrom::Start(15)).unwrap(), 15);
+        reader.read_u8().unwrap_err();
+        reader.seek(std::io::SeekFrom::Start(100)).unwrap_err();
+
+        assert_eq!(reader.seek(std::io::SeekFrom::End(0)).unwrap(), 15);
+        reader.read_u8().unwrap_err();
+        assert_eq!(reader.seek(std::io::SeekFrom::End(-5)).unwrap(), 10);
+        assert_eq!(reader.read_u8().unwrap(), 10);
+        assert_eq!(reader.seek(std::io::SeekFrom::End(-15)).unwrap(), 0);
+        assert_eq!(reader.read_u8().unwrap(), 0);
+        reader.seek(std::io::SeekFrom::End(-20)).unwrap_err();
+
+        assert_eq!(reader.seek(std::io::SeekFrom::Start(10)).unwrap(), 10);
+        reader.seek(std::io::SeekFrom::Current(-100)).unwrap_err();
+    }
 }