righ_ovsdb/protocol/
codec.rs

1use tokio_util::{
2    bytes::{Buf, BufMut, BytesMut},
3    codec::{Decoder, Encoder},
4};
5
6use super::Message;
7
8#[derive(Debug)]
9enum BufferTag {
10    Obj,
11    Str,
12}
13
14/// The error type for parsing errors encountered by the [Codec].
15#[derive(thiserror::Error, Debug)]
16pub enum CodecError {
17    /// JSON error converting a native struct to wire-protocol.
18    #[error("Error encoding data")]
19    Encode(#[source] serde_json::Error),
20    /// JSON error converting wire-protocol to native struct.
21    #[error("Error decoding data")]
22    Decode(#[source] serde_json::Error),
23    /// Garbled data stream (usually indicates a missing opening brace).
24    #[error("Corrupted data stream: {0}")]
25    DataStreamCorrupted(String),
26    /// Low-level IO error.
27    #[error("Unexpected IO Error")]
28    Io(#[from] std::io::Error),
29}
30
31/// An OVSDB protocol encoder/decoder.
32///
33/// The codec is responsible for converting native objects to wire protocol, and vice versa,
34/// primarily using `serde` and `serde_json`.
35#[derive(Default, Debug)]
36pub struct Codec {
37    data: Vec<u8>,
38    tags: Vec<BufferTag>,
39}
40
41#[allow(missing_docs)]
42impl Codec {
43    /// Instantiate a new Codec.
44    pub fn new() -> Self {
45        Self::default()
46    }
47
48    pub fn try_decode_message(
49        &mut self,
50        src: &[u8],
51    ) -> Result<(Option<Message>, usize), CodecError> {
52        let mut offset = 0;
53
54        while offset < src.len() {
55            match self.tags.last() {
56                Some(BufferTag::Str) => {
57                    if let Some(n) = &src[offset..].iter().position(|&c| c == b'"') {
58                        offset += n + 1;
59                        self.tags.pop();
60                        continue;
61                    } else {
62                        break;
63                    }
64                }
65                Some(BufferTag::Obj) => {
66                    if let Some(n) = &src[offset..]
67                        .iter()
68                        .position(|&c| [b'"', b'{', b'}'].contains(&c))
69                    {
70                        offset += n;
71                        let char = src[offset];
72                        offset += 1;
73                        match &char {
74                            b'"' => self.tags.push(BufferTag::Str),
75                            b'{' => self.tags.push(BufferTag::Obj),
76                            b'}' => {
77                                self.tags.pop();
78                                if self.tags.is_empty() {
79                                    // We have a full object
80                                    self.data.extend_from_slice(src);
81                                    // println!(
82                                    //     "Received: {}",
83                                    //     String::from_utf8(self.data.clone())
84                                    //         .expect("utf8 conversion")
85                                    // );
86                                    let msg: Message =
87                                        match serde_json::from_slice(&self.data.to_vec()) {
88                                            Ok(msg) => msg,
89                                            Err(err) => {
90                                                eprintln!("Failed to decode message: {err}");
91                                                return Err(CodecError::Decode(err));
92                                            }
93                                        };
94                                    self.data.clear();
95                                    return Ok((Some(msg), offset));
96                                }
97                            }
98                            _ => unreachable!(),
99                        }
100                    } else {
101                        break;
102                    }
103                }
104                None => {
105                    if let Some(n) = &src[offset..].iter().position(|&c| c == b'{') {
106                        offset += n + 1;
107                        self.tags.push(BufferTag::Obj);
108                    } else {
109                        return Err(CodecError::DataStreamCorrupted(
110                            "No openening tag found in data stream.".to_string(),
111                        ));
112                    }
113                }
114            }
115        }
116
117        self.data.extend_from_slice(src);
118        Ok((None, src.len()))
119    }
120}
121
122impl Decoder for Codec {
123    type Item = Message;
124    type Error = CodecError;
125
126    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
127        let (res, consume) = self.try_decode_message(src.chunk())?;
128        src.advance(consume);
129        Ok(res)
130    }
131}
132
133impl Encoder<Message> for Codec {
134    type Error = CodecError;
135
136    fn encode(&mut self, item: Message, dst: &mut BytesMut) -> Result<(), Self::Error> {
137        let data = serde_json::to_vec(&item).map_err(CodecError::Encode)?;
138        dst.reserve(data.len());
139        dst.put_slice(&data);
140        println!(
141            "Sent: {}",
142            String::from_utf8(dst.clone().to_vec()).expect("message encode")
143        );
144        Ok(())
145    }
146}
righ_ovsdb/protocol/codec.rs

righ_ovsdb/protocol/
codec.rs
