feat: make IPROTO header encode/decode as methods, add encode_from_parts

Author: Кирилл Безуглый

CHANGELOG.md

@@ -10,6 +10,7 @@
 - `ffi::tarantool::box_schema_version` and `ffi::tarantool::box_session_id` functions
 - `network::protocol::SyncIndex::get` method
 - `network::protocol::codec::{LegacyCall, Nop, Prepare, Begin, Commit, Rollback}` variants 
+- `network::protocol::codec::Header::encode_from_parts` function
 
 ### Changed
 
@@ -25,6 +26,8 @@
 - tlua::LuaTable::get_or_create_metatable is deprecated now in favor of tlua::LuaTable::metatable.
 
 ### Breaking changes
+- Replace `network::protocol::codec::{encode_header, decode_header}` functions
+  with `network::protocol::codec::Header::{encode, decode}` methods.
 
 ### Added (picodata)
 

tarantool/src/net_box/inner.rs

@@ -306,7 +306,7 @@ impl ConnInner {
             cur.set_position(0);
         }
 
-        let header = protocol::decode_header(&mut cur)?;
+        let header = protocol::Header::decode(&mut cur)?;
         if header.iproto_type == protocol::IProtoType::Error as u32 {
             let error = protocol::decode_error(&mut cur, &header)?;
             return Err(Error::Remote(error));

tarantool/src/net_box/recv_queue.rs

@@ -169,7 +169,7 @@ impl RecvQueue {
                 let header = {
                     let mut buffer = self.buffer.borrow_mut();
                     buffer.set_position(start as _);
-                    protocol::decode_header(buffer.by_ref())?
+                    protocol::Header::decode(buffer.by_ref())?
                 };
 
                 let sync = header.sync;

tarantool/src/network/protocol/api.rs

@@ -16,7 +16,7 @@ pub trait Request {
 
     #[inline(always)]
     fn encode_header(&self, out: &mut impl Write, sync: SyncIndex) -> Result<(), Error> {
-        codec::encode_header(out, sync, Self::TYPE)
+        codec::Header::encode_from_parts(out, sync, Self::TYPE)
     }
 
     fn encode_body(&self, out: &mut impl Write) -> Result<(), Error>;

tarantool/src/network/protocol/codec.rs

@@ -84,19 +84,6 @@ pub enum IProtoType {
     Error = 1 << 15,
 }
 
-pub fn encode_header(
-    stream: &mut impl Write,
-    sync: SyncIndex,
-    request_type: IProtoType,
-) -> Result<(), Error> {
-    rmp::encode::write_map_len(stream, 2)?;
-    rmp::encode::write_pfix(stream, REQUEST_TYPE)?;
-    rmp::encode::write_pfix(stream, request_type as u8)?;
-    rmp::encode::write_pfix(stream, SYNC)?;
-    rmp::encode::write_uint(stream, sync.0)?;
-    Ok(())
-}
-
 pub fn chap_sha1_auth_data(password: &str, salt: &[u8]) -> Vec<u8> {
     // prepare 'chap-sha1' scramble:
     // salt = base64_decode(encoded_salt);
@@ -389,48 +376,72 @@ pub struct Header {
     pub schema_version: u64,
 }
 
-pub struct Response<T> {
-    pub header: Header,
-    pub payload: T,
-}
-
-pub fn decode_header(stream: &mut (impl Read + Seek)) -> Result<Header, Error> {
-    let mut sync: Option<u64> = None;
-    let mut iproto_type: Option<u32> = None;
-    let mut error_code: u32 = 0;
-    let mut schema_version: Option<u64> = None;
+impl Header {
+    pub fn encode(&self, stream: &mut impl Write) -> Result<(), Error> {
+        rmp::encode::write_map_len(stream, 2)?;
+        rmp::encode::write_pfix(stream, REQUEST_TYPE)?;
+        rmp::encode::write_pfix(stream, self.iproto_type as u8)?;
+        rmp::encode::write_pfix(stream, SYNC)?;
+        rmp::encode::write_uint(stream, self.sync.0)?;
+        Ok(())
+    }
 
-    let map_len = rmp::decode::read_map_len(stream)?;
-    for _ in 0..map_len {
-        let key = rmp::decode::read_pfix(stream)?;
-        match key {
-            REQUEST_TYPE => {
-                let r#type: u32 = rmp::decode::read_int(stream)?;
+    pub fn encode_from_parts(
+        stream: &mut impl Write,
+        sync: SyncIndex,
+        request_type: IProtoType,
+    ) -> Result<(), Error> {
+        rmp::encode::write_map_len(stream, 2)?;
+        rmp::encode::write_pfix(stream, REQUEST_TYPE)?;
+        rmp::encode::write_pfix(stream, request_type as u8)?;
+        rmp::encode::write_pfix(stream, SYNC)?;
+        rmp::encode::write_uint(stream, sync.0)?;
+        Ok(())
+    }
 
-                const IPROTO_TYPE_ERROR: u32 = IProtoType::Error as _;
-                if (r#type & IPROTO_TYPE_ERROR) != 0 {
-                    iproto_type = Some(IPROTO_TYPE_ERROR);
-                    error_code = r#type & !IPROTO_TYPE_ERROR;
-                } else {
-                    iproto_type = Some(r#type);
+    pub fn decode(stream: &mut (impl Read + Seek)) -> Result<Header, Error> {
+        let mut sync: Option<u64> = None;
+        let mut iproto_type: Option<u32> = None;
+        let mut error_code: u32 = 0;
+        let mut schema_version: Option<u64> = None;
+
+        let map_len = rmp::decode::read_map_len(stream)?;
+        for _ in 0..map_len {
+            let key = rmp::decode::read_pfix(stream)?;
+            match key {
+                REQUEST_TYPE => {
+                    let r#type: u32 = rmp::decode::read_int(stream)?;
+
+                    const IPROTO_TYPE_ERROR: u32 = IProtoType::Error as _;
+                    if (r#type & IPROTO_TYPE_ERROR) != 0 {
+                        iproto_type = Some(IPROTO_TYPE_ERROR);
+                        error_code = r#type & !IPROTO_TYPE_ERROR;
+                    } else {
+                        iproto_type = Some(r#type);
+                    }
                 }
+                SYNC => sync = Some(rmp::decode::read_int(stream)?),
+                SCHEMA_VERSION => schema_version = Some(rmp::decode::read_int(stream)?),
+                _ => msgpack::skip_value(stream)?,
             }
-            SYNC => sync = Some(rmp::decode::read_int(stream)?),
-            SCHEMA_VERSION => schema_version = Some(rmp::decode::read_int(stream)?),
-            _ => msgpack::skip_value(stream)?,
         }
-    }
 
-    if sync.is_none() || iproto_type.is_none() || schema_version.is_none() {
-        return Err(io::Error::from(io::ErrorKind::InvalidData).into());
+        if sync.is_none() || iproto_type.is_none() || schema_version.is_none() {
+            return Err(io::Error::from(io::ErrorKind::InvalidData).into());
+        }
+
+        Ok(Header {
+            sync: SyncIndex(sync.unwrap()),
+            iproto_type: iproto_type.unwrap(),
+            error_code,
+            schema_version: schema_version.unwrap(),
+        })
     }
+}
 
-    Ok(Header {
-        sync: SyncIndex(sync.unwrap()),
-        iproto_type: iproto_type.unwrap(),
-        error_code,
-        schema_version: schema_version.unwrap(),
-    })
+pub struct Response<T> {
+    pub header: Header,
+    pub payload: T,
 }
 
 ////////////////////////////////////////////////////////////////////////////////

tarantool/src/network/protocol/mod.rs

@@ -253,7 +253,7 @@ impl Protocol {
                 None
             }
             State::Auth => {
-                let header = codec::decode_header(message)?;
+                let header = codec::Header::decode(message)?;
                 if header.iproto_type == IProtoType::Error as u32 {
                     let error = codec::decode_error(message, &header)?;
                     return Err(error::Error::Remote(error));
@@ -262,7 +262,7 @@ impl Protocol {
                 None
             }
             State::Ready => {
-                let header = codec::decode_header(message)?;
+                let header = codec::Header::decode(message)?;
                 let response;
                 if header.iproto_type == IProtoType::Error as u32 {
                     response = Err(codec::decode_error(message, &header)?);