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            ext_ts_stack,
45            payload,
46        } = x;
47
48        // Header
49        let mut header = id::PUSH;
50        let mut n_exts = ((ext_qos != &ext::QoSType::DEFAULT) as u8)
51            + (ext_tstamp.is_some() as u8)
52            + ((ext_nodeid != &ext::NodeIdType::DEFAULT) as u8)
53            + (ext_ts_stack.is_some() as u8);
54        if n_exts != 0 {
55            header |= flag::Z;
56        }
57        if wire_expr.mapping != Mapping::DEFAULT {
58            header |= flag::M;
59        }
60        if wire_expr.has_suffix() {
61            header |= flag::N;
62        }
63        self.write(&mut *writer, header)?;
64
65        // Body
66        self.write(&mut *writer, wire_expr)?;
67
68        // Extensions
69        if ext_qos != &ext::QoSType::DEFAULT {
70            n_exts -= 1;
71            self.write(&mut *writer, (*ext_qos, n_exts != 0))?;
72        }
73        if let Some(ts) = ext_tstamp.as_ref() {
74            n_exts -= 1;
75            self.write(&mut *writer, (ts, n_exts != 0))?;
76        }
77        if ext_nodeid != &ext::NodeIdType::DEFAULT {
78            n_exts -= 1;
79            self.write(&mut *writer, (*ext_nodeid, n_exts != 0))?;
80        }
81        if let Some(ts_stack) = ext_ts_stack.as_ref() {
82            n_exts -= 1;
83            self.write(&mut *writer, (ts_stack, n_exts != 0))?;
84        }
85        // Payload
86        self.write(&mut *writer, payload)?;
87
88        Ok(())
89    }
90}
91
92impl<R> RCodec<Push, &mut R> for Zenoh080
93where
94    R: Reader,
95{
96    type Error = DidntRead;
97
98    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
99        let header: u8 = self.read(&mut *reader)?;
100        let codec = Zenoh080Header::new(header);
101        codec.read(reader)
102    }
103}
104
105impl<R> RCodec<Push, &mut R> for Zenoh080Header
106where
107    R: Reader,
108{
109    type Error = DidntRead;
110
111    #[inline(always)]
112    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
113        if imsg::mid(self.header) != id::PUSH {
114            return Err(DidntRead);
115        }
116
117        // Body
118        let ccond = Zenoh080Condition::new(imsg::has_flag(self.header, flag::N));
119        let mut wire_expr: WireExpr<'static> = ccond.read(&mut *reader)?;
120        wire_expr.mapping = if imsg::has_flag(self.header, flag::M) {
121            Mapping::Sender
122        } else {
123            Mapping::Receiver
124        };
125
126        // Extensions
127        let mut ext_qos = ext::QoSType::DEFAULT;
128        let mut ext_tstamp = None;
129        let mut ext_nodeid = ext::NodeIdType::DEFAULT;
130        let mut ext_ts_stack = None;
131
132        let mut has_ext = imsg::has_flag(self.header, flag::Z);
133        while has_ext {
134            let ext: u8 = self.codec.read(&mut *reader)?;
135            let eodec = Zenoh080Header::new(ext);
136            match iext::eid(ext) {
137                ext::QoS::ID => {
138                    let (q, ext): (ext::QoSType, bool) = eodec.read(&mut *reader)?;
139                    ext_qos = q;
140                    has_ext = ext;
141                }
142                ext::Timestamp::ID => {
143                    let (t, ext): (ext::TimestampType, bool) = eodec.read(&mut *reader)?;
144                    ext_tstamp = Some(t);
145                    has_ext = ext;
146                }
147                ext::NodeId::ID => {
148                    let (nid, ext): (ext::NodeIdType, bool) = eodec.read(&mut *reader)?;
149                    ext_nodeid = nid;
150                    has_ext = ext;
151                }
152                ext::TsStack::ID => {
153                    let (ts, ext): (ext::TsStackType, bool) = eodec.read(&mut *reader)?;
154                    ext_ts_stack = Some(ts);
155                    has_ext = ext;
156                }
157                _ => {
158                    has_ext = extension::skip(reader, "Push", ext)?;
159                }
160            }
161        }
162
163        // Payload
164        let payload: PushBody = self.codec.read(&mut *reader)?;
165
166        Ok(Push {
167            wire_expr,
168            payload,
169            ext_qos,
170            ext_tstamp,
171            ext_nodeid,
172            ext_ts_stack,
173        })
174    }
175}
zenoh_codec/network/push.rs

zenoh_codec/network/
push.rs
