zenoh_codec/network/
push.rs

1//
2// Copyright (c) 2022 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    reader::{DidntRead, Reader},
16    writer::{DidntWrite, Writer},
17};
18use zenoh_protocol::{
19    common::{iext, imsg},
20    core::WireExpr,
21    network::{
22        id,
23        push::{ext, flag},
24        Mapping, Push,
25    },
26    zenoh::PushBody,
27};
28
29use crate::{common::extension, RCodec, WCodec, Zenoh080, Zenoh080Condition, Zenoh080Header};
30
31impl<W> WCodec<&Push, &mut W> for Zenoh080
32where
33    W: Writer,
34{
35    type Output = Result<(), DidntWrite>;
36
37    #[inline(always)]
38    fn write(self, writer: &mut W, x: &Push) -> Self::Output {
39        let Push {
40            wire_expr,
41            ext_qos,
42            ext_tstamp,
43            ext_nodeid,
44            payload,
45        } = x;
46
47        // Header
48        let mut header = id::PUSH;
49        let mut n_exts = ((ext_qos != &ext::QoSType::DEFAULT) as u8)
50            + (ext_tstamp.is_some() as u8)
51            + ((ext_nodeid != &ext::NodeIdType::DEFAULT) as u8);
52        if n_exts != 0 {
53            header |= flag::Z;
54        }
55        if wire_expr.mapping != Mapping::DEFAULT {
56            header |= flag::M;
57        }
58        if wire_expr.has_suffix() {
59            header |= flag::N;
60        }
61        self.write(&mut *writer, header)?;
62
63        // Body
64        self.write(&mut *writer, wire_expr)?;
65
66        // Extensions
67        if ext_qos != &ext::QoSType::DEFAULT {
68            n_exts -= 1;
69            self.write(&mut *writer, (*ext_qos, n_exts != 0))?;
70        }
71        if let Some(ts) = ext_tstamp.as_ref() {
72            n_exts -= 1;
73            self.write(&mut *writer, (ts, n_exts != 0))?;
74        }
75        if ext_nodeid != &ext::NodeIdType::DEFAULT {
76            n_exts -= 1;
77            self.write(&mut *writer, (*ext_nodeid, n_exts != 0))?;
78        }
79        // Payload
80        self.write(&mut *writer, payload)?;
81
82        Ok(())
83    }
84}
85
86impl<R> RCodec<Push, &mut R> for Zenoh080
87where
88    R: Reader,
89{
90    type Error = DidntRead;
91
92    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
93        let header: u8 = self.read(&mut *reader)?;
94        let codec = Zenoh080Header::new(header);
95        codec.read(reader)
96    }
97}
98
99impl<R> RCodec<Push, &mut R> for Zenoh080Header
100where
101    R: Reader,
102{
103    type Error = DidntRead;
104
105    #[inline(always)]
106    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
107        if imsg::mid(self.header) != id::PUSH {
108            return Err(DidntRead);
109        }
110
111        // Body
112        let ccond = Zenoh080Condition::new(imsg::has_flag(self.header, flag::N));
113        let mut wire_expr: WireExpr<'static> = ccond.read(&mut *reader)?;
114        wire_expr.mapping = if imsg::has_flag(self.header, flag::M) {
115            Mapping::Sender
116        } else {
117            Mapping::Receiver
118        };
119
120        // Extensions
121        let mut ext_qos = ext::QoSType::DEFAULT;
122        let mut ext_tstamp = None;
123        let mut ext_nodeid = ext::NodeIdType::DEFAULT;
124
125        let mut has_ext = imsg::has_flag(self.header, flag::Z);
126        while has_ext {
127            let ext: u8 = self.codec.read(&mut *reader)?;
128            let eodec = Zenoh080Header::new(ext);
129            match iext::eid(ext) {
130                ext::QoS::ID => {
131                    let (q, ext): (ext::QoSType, bool) = eodec.read(&mut *reader)?;
132                    ext_qos = q;
133                    has_ext = ext;
134                }
135                ext::Timestamp::ID => {
136                    let (t, ext): (ext::TimestampType, bool) = eodec.read(&mut *reader)?;
137                    ext_tstamp = Some(t);
138                    has_ext = ext;
139                }
140                ext::NodeId::ID => {
141                    let (nid, ext): (ext::NodeIdType, bool) = eodec.read(&mut *reader)?;
142                    ext_nodeid = nid;
143                    has_ext = ext;
144                }
145                _ => {
146                    has_ext = extension::skip(reader, "Push", ext)?;
147                }
148            }
149        }
150
151        // Payload
152        let payload: PushBody = self.codec.read(&mut *reader)?;
153
154        Ok(Push {
155            wire_expr,
156            payload,
157            ext_qos,
158            ext_tstamp,
159            ext_nodeid,
160        })
161    }
162}
zenoh_codec/network/push.rs

zenoh_codec/network/
push.rs
