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    fn write(self, writer: &mut W, x: &Push) -> Self::Output {
38        let Push {
39            wire_expr,
40            ext_qos,
41            ext_tstamp,
42            ext_nodeid,
43            payload,
44        } = x;
45
46        // Header
47        let mut header = id::PUSH;
48        let mut n_exts = ((ext_qos != &ext::QoSType::DEFAULT) as u8)
49            + (ext_tstamp.is_some() as u8)
50            + ((ext_nodeid != &ext::NodeIdType::DEFAULT) as u8);
51        if n_exts != 0 {
52            header |= flag::Z;
53        }
54        if wire_expr.mapping != Mapping::DEFAULT {
55            header |= flag::M;
56        }
57        if wire_expr.has_suffix() {
58            header |= flag::N;
59        }
60        self.write(&mut *writer, header)?;
61
62        // Body
63        self.write(&mut *writer, wire_expr)?;
64
65        // Extensions
66        if ext_qos != &ext::QoSType::DEFAULT {
67            n_exts -= 1;
68            self.write(&mut *writer, (*ext_qos, n_exts != 0))?;
69        }
70        if let Some(ts) = ext_tstamp.as_ref() {
71            n_exts -= 1;
72            self.write(&mut *writer, (ts, n_exts != 0))?;
73        }
74        if ext_nodeid != &ext::NodeIdType::DEFAULT {
75            n_exts -= 1;
76            self.write(&mut *writer, (*ext_nodeid, n_exts != 0))?;
77        }
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    fn read(self, reader: &mut R) -> Result<Push, Self::Error> {
106        if imsg::mid(self.header) != id::PUSH {
107            return Err(DidntRead);
108        }
109
110        // Body
111        let ccond = Zenoh080Condition::new(imsg::has_flag(self.header, flag::N));
112        let mut wire_expr: WireExpr<'static> = ccond.read(&mut *reader)?;
113        wire_expr.mapping = if imsg::has_flag(self.header, flag::M) {
114            Mapping::Sender
115        } else {
116            Mapping::Receiver
117        };
118
119        // Extensions
120        let mut ext_qos = ext::QoSType::DEFAULT;
121        let mut ext_tstamp = None;
122        let mut ext_nodeid = ext::NodeIdType::DEFAULT;
123
124        let mut has_ext = imsg::has_flag(self.header, flag::Z);
125        while has_ext {
126            let ext: u8 = self.codec.read(&mut *reader)?;
127            let eodec = Zenoh080Header::new(ext);
128            match iext::eid(ext) {
129                ext::QoS::ID => {
130                    let (q, ext): (ext::QoSType, bool) = eodec.read(&mut *reader)?;
131                    ext_qos = q;
132                    has_ext = ext;
133                }
134                ext::Timestamp::ID => {
135                    let (t, ext): (ext::TimestampType, bool) = eodec.read(&mut *reader)?;
136                    ext_tstamp = Some(t);
137                    has_ext = ext;
138                }
139                ext::NodeId::ID => {
140                    let (nid, ext): (ext::NodeIdType, bool) = eodec.read(&mut *reader)?;
141                    ext_nodeid = nid;
142                    has_ext = ext;
143                }
144                _ => {
145                    has_ext = extension::skip(reader, "Push", ext)?;
146                }
147            }
148        }
149
150        // Payload
151        let payload: PushBody = self.codec.read(&mut *reader)?;
152
153        Ok(Push {
154            wire_expr,
155            payload,
156            ext_qos,
157            ext_tstamp,
158            ext_nodeid,
159        })
160    }
161}
zenoh_codec/network/push.rs

zenoh_codec/network/
push.rs
