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    use zenoh_shm::ShmBufInner;
106
107    use super::*;
108    use crate::Zenoh080Sliced;
109
110    const RAW: u8 = 0;
111    const SHM_PTR: u8 = 1;
112
113    macro_rules! zbuf_sliced_impl {
114        ($bound:ty) => {
115            impl LCodec<&ZBuf> for Zenoh080Sliced<$bound> {
116                fn w_len(self, message: &ZBuf) -> usize {
117                    if self.is_sliced {
118                        message.zslices().fold(0, |acc, x| acc + 1 + x.len())
119                    } else {
120                        self.codec.w_len(message)
121                    }
122                }
123            }
124
125            impl<W> WCodec<&ZBuf, &mut W> for Zenoh080Sliced<$bound>
126            where
127                W: Writer,
128            {
129                type Output = Result<(), DidntWrite>;
130
131                fn write(self, writer: &mut W, x: &ZBuf) -> Self::Output {
132                    if self.is_sliced {
133                        self.codec.write(&mut *writer, x.zslices().count())?;
134
135                        for zs in x.zslices() {
136                            match zs.kind {
137                                ZSliceKind::Raw => {
138                                    self.codec.write(&mut *writer, RAW)?;
139                                    self.codec.write(&mut *writer, zs)?;
140                                }
141                                ZSliceKind::ShmPtr => {
142                                    self.codec.write(&mut *writer, SHM_PTR)?;
143                                    let shmb = zs.downcast_ref::<ShmBufInner>().unwrap();
144                                    let mut info = vec![];
145                                    Zenoh080::new().write(&mut &mut info, &shmb.info)?;
146                                    self.codec.write(&mut *writer, &*info)?;
147                                    // Increase the reference count so to keep the ShmBufInner
148                                    // valid until it is received.
149                                    unsafe { shmb.inc_ref_count() };
150                                }
151                            }
152                        }
153                    } else {
154                        self.codec.write(&mut *writer, x)?;
155                    }
156
157                    Ok(())
158                }
159            }
160
161            impl<R> RCodec<ZBuf, &mut R> for Zenoh080Sliced<$bound>
162            where
163                R: Reader,
164            {
165                type Error = DidntRead;
166
167                fn read(self, reader: &mut R) -> Result<ZBuf, Self::Error> {
168                    if self.is_sliced {
169                        let num: usize = self.codec.read(&mut *reader)?;
170                        let mut zbuf = ZBuf::empty();
171                        for _ in 0..num {
172                            let kind: u8 = self.codec.read(&mut *reader)?;
173                            match kind {
174                                RAW => {
175                                    let len: usize = self.codec.read(&mut *reader)?;
176                                    reader.read_zslices(len, |s| zbuf.push_zslice(s))?;
177                                }
178                                SHM_PTR => {
179                                    let mut zslice: ZSlice = self.codec.read(&mut *reader)?;
180                                    zslice.kind = ZSliceKind::ShmPtr;
181                                    zbuf.push_zslice(zslice);
182                                }
183                                _ => return Err(DidntRead),
184                            }
185                        }
186                        Ok(zbuf)
187                    } else {
188                        self.codec.read(&mut *reader)
189                    }
190                }
191            }
192        };
193    }
194
195    zbuf_sliced_impl!(u8);
196    zbuf_sliced_impl!(u16);
197    zbuf_sliced_impl!(u32);
198    zbuf_sliced_impl!(u64);
199    zbuf_sliced_impl!(usize);
200}
zenoh_codec/core/zbuf.rs

zenoh_codec/core/
zbuf.rs
