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
80        // Payload
81        self.write(&mut *writer, payload)?;
82
83        Ok(())
84    }
85}
86
87impl<R> RCodec<Push, &mut R> for Zenoh080
88where
89    R: Reader,
90{
91    type Error = DidntRead;
92
93    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
94        let header: u8 = self.read(&mut *reader)?;
95        let codec = Zenoh080Header::new(header);
96        codec.read(reader)
97    }
98}
99
100impl<R> RCodec<Push, &mut R> for Zenoh080Header
101where
102    R: Reader,
103{
104    type Error = DidntRead;
105
106    #[inline(always)]
107    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
108        if imsg::mid(self.header) != id::PUSH {
109            return Err(DidntRead);
110        }
111
112        // Body
113        let ccond = Zenoh080Condition::new(imsg::has_flag(self.header, flag::N));
114        let mut wire_expr: WireExpr<'static> = ccond.read(&mut *reader)?;
115        wire_expr.mapping = if imsg::has_flag(self.header, flag::M) {
116            Mapping::Sender
117        } else {
118            Mapping::Receiver
119        };
120
121        // Extensions
122        let mut ext_qos = ext::QoSType::DEFAULT;
123        let mut ext_tstamp = None;
124        let mut ext_nodeid = ext::NodeIdType::DEFAULT;
125
126        let mut has_ext = imsg::has_flag(self.header, flag::Z);
127        while has_ext {
128            let ext: u8 = self.codec.read(&mut *reader)?;
129            let eodec = Zenoh080Header::new(ext);
130            match iext::eid(ext) {
131                ext::QoS::ID => {
132                    let (q, ext): (ext::QoSType, bool) = eodec.read(&mut *reader)?;
133                    ext_qos = q;
134                    has_ext = ext;
135                }
136                ext::Timestamp::ID => {
137                    let (t, ext): (ext::TimestampType, bool) = eodec.read(&mut *reader)?;
138                    ext_tstamp = Some(t);
139                    has_ext = ext;
140                }
141                ext::NodeId::ID => {
142                    let (nid, ext): (ext::NodeIdType, bool) = eodec.read(&mut *reader)?;
143                    ext_nodeid = nid;
144                    has_ext = ext;
145                }
146                _ => {
147                    has_ext = extension::skip(reader, "Push", ext)?;
148                }
149            }
150        }
151
152        // Payload
153        let payload: PushBody = self.codec.read(&mut *reader)?;
154
155        Ok(Push {
156            wire_expr,
157            payload,
158            ext_qos,
159            ext_tstamp,
160            ext_nodeid,
161        })
162    }
163}
zenoh_codec/network/push.rs

zenoh_codec/network/
push.rs
