zenoh_codec/core/
zbuf.rs

1//
2// Copyright (c) 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14use zenoh_buffers::{
15    buffer::Buffer,
16    reader::{DidntRead, Reader},
17    writer::{DidntWrite, Writer},
18    ZBuf,
19};
20
21use crate::{LCodec, RCodec, WCodec, Zenoh080, Zenoh080Bounded};
22
23// ZBuf bounded
24macro_rules! zbuf_impl {
25    ($bound:ty) => {
26        impl LCodec<&ZBuf> for Zenoh080Bounded<$bound> {
27            fn w_len(self, message: &ZBuf) -> usize {
28                message.len()
29            }
30        }
31
32        impl<W> WCodec<&ZBuf, &mut W> for Zenoh080Bounded<$bound>
33        where
34            W: Writer,
35        {
36            type Output = Result<(), DidntWrite>;
37
38            fn write(self, writer: &mut W, x: &ZBuf) -> Self::Output {
39                self.write(&mut *writer, x.len())?;
40                for s in x.zslices() {
41                    writer.write_zslice(s)?;
42                }
43                Ok(())
44            }
45        }
46
47        impl<R> RCodec<ZBuf, &mut R> for Zenoh080Bounded<$bound>
48        where
49            R: Reader,
50        {
51            type Error = DidntRead;
52
53            fn read(self, reader: &mut R) -> Result<ZBuf, Self::Error> {
54                let len: usize = self.read(&mut *reader)?;
55                let mut zbuf = ZBuf::empty();
56                reader.read_zslices(len, |s| zbuf.push_zslice(s))?;
57                Ok(zbuf)
58            }
59        }
60    };
61}
62
63zbuf_impl!(u8);
64zbuf_impl!(u16);
65zbuf_impl!(u32);
66zbuf_impl!(u64);
67zbuf_impl!(usize);
68
69// ZBuf flat
70impl<W> WCodec<&ZBuf, &mut W> for Zenoh080
71where
72    W: Writer,
73{
74    type Output = Result<(), DidntWrite>;
75
76    fn write(self, writer: &mut W, x: &ZBuf) -> Self::Output {
77        let zodec = Zenoh080Bounded::<usize>::new();
78        zodec.write(&mut *writer, x)
79    }
80}
81
82impl<R> RCodec<ZBuf, &mut R> for Zenoh080
83where
84    R: Reader,
85{
86    type Error = DidntRead;
87
88    fn read(self, reader: &mut R) -> Result<ZBuf, Self::Error> {
89        let zodec = Zenoh080Bounded::<usize>::new();
90        zodec.read(&mut *reader)
91    }
92}
93
94impl LCodec<&ZBuf> for Zenoh080 {
95    fn w_len(self, message: &ZBuf) -> usize {
96        let zodec = Zenoh080Bounded::<usize>::new();
97        zodec.w_len(message)
98    }
99}
100
101// ZBuf sliced
102#[cfg(feature = "shared-memory")]
103mod shm {
104    use zenoh_buffers::{ZSlice, ZSliceKind};
105
106    use super::*;
107    use crate::Zenoh080Sliced;
108
109    const RAW: u8 = 0;
110    const SHM_PTR: u8 = 1;
111
112    macro_rules! zbuf_sliced_impl {
113        ($bound:ty) => {
114            impl LCodec<&ZBuf> for Zenoh080Sliced<$bound> {
115                fn w_len(self, message: &ZBuf) -> usize {
116                    if self.is_sliced {
117                        message.zslices().fold(0, |acc, x| acc + 1 + x.len())
118                    } else {
119                        self.codec.w_len(message)
120                    }
121                }
122            }
123
124            impl<W> WCodec<&ZBuf, &mut W> for Zenoh080Sliced<$bound>
125            where
126                W: Writer,
127            {
128                type Output = Result<(), DidntWrite>;
129
130                fn write(self, writer: &mut W, x: &ZBuf) -> Self::Output {
131                    if self.is_sliced {
132                        self.codec.write(&mut *writer, x.zslices().count())?;
133
134                        for zs in x.zslices() {
135                            match zs.kind {
136                                ZSliceKind::Raw => self.codec.write(&mut *writer, RAW)?,
137                                ZSliceKind::ShmPtr => self.codec.write(&mut *writer, SHM_PTR)?,
138                            }
139                            self.codec.write(&mut *writer, zs)?;
140                        }
141                    } else {
142                        self.codec.write(&mut *writer, x)?;
143                    }
144
145                    Ok(())
146                }
147            }
148
149            impl<R> RCodec<ZBuf, &mut R> for Zenoh080Sliced<$bound>
150            where
151                R: Reader,
152            {
153                type Error = DidntRead;
154
155                fn read(self, reader: &mut R) -> Result<ZBuf, Self::Error> {
156                    if self.is_sliced {
157                        let num: usize = self.codec.read(&mut *reader)?;
158                        let mut zbuf = ZBuf::empty();
159                        for _ in 0..num {
160                            let kind: u8 = self.codec.read(&mut *reader)?;
161                            match kind {
162                                RAW => {
163                                    let len: usize = self.codec.read(&mut *reader)?;
164                                    reader.read_zslices(len, |s| zbuf.push_zslice(s))?;
165                                }
166                                SHM_PTR => {
167                                    let mut zslice: ZSlice = self.codec.read(&mut *reader)?;
168                                    zslice.kind = ZSliceKind::ShmPtr;
169                                    zbuf.push_zslice(zslice);
170                                }
171                                _ => return Err(DidntRead),
172                            }
173                        }
174                        Ok(zbuf)
175                    } else {
176                        self.codec.read(&mut *reader)
177                    }
178                }
179            }
180        };
181    }
182
183    zbuf_sliced_impl!(u8);
184    zbuf_sliced_impl!(u16);
185    zbuf_sliced_impl!(u32);
186    zbuf_sliced_impl!(u64);
187    zbuf_sliced_impl!(usize);
188}
zenoh_codec/core/zbuf.rs

zenoh_codec/core/
zbuf.rs
