fix: optimize bytes write action (#24)

jjeffcaii · web-flow · commit 06a22aba8be2 · 2020-11-10T11:08:27.000+08:00
diff --git a/rsocket/src/extension/composite.rs b/rsocket/src/extension/composite.rs
@@ -191,13 +191,13 @@ impl Writeable for CompositeMetadataEntry {
                 let mime_type_len = s.len() as u8;
                 assert!(mime_type_len > 0, "invalid length of MimeType!");
                 bf.put_u8(mime_type_len - 1);
-                bf.put_slice(s.as_ref());
+                bf.extend_from_slice(s.as_ref());
             }
         };
         let metadata_len = self.metadata.len();
         u24::from(metadata_len).write_to(bf);
         if metadata_len > 0 {
-            bf.put(self.metadata.bytes());
+            bf.extend_from_slice(&self.metadata);
         }
     }
 
diff --git a/rsocket/src/frame/error.rs b/rsocket/src/frame/error.rs
@@ -1,5 +1,5 @@
-use super::utils::too_short;
 use super::{Body, Frame};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use std::fmt;
@@ -46,11 +46,11 @@ impl ErrorBuilder {
 impl Error {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<Error> {
         if bf.len() < 4 {
-            too_short(4)
+            Err(RSocketError::InCompleteFrame.into())
         } else {
             let code = bf.get_u32();
             let data: Option<Bytes> = if !bf.is_empty() {
-                Some(Bytes::from(bf.to_vec()))
+                Some(bf.split().freeze())
             } else {
                 None
             };
@@ -85,7 +85,7 @@ impl Writeable for Error {
     fn write_to(&self, bf: &mut BytesMut) {
         bf.put_u32(self.code);
         match &self.data {
-            Some(v) => bf.put(v.bytes()),
+            Some(v) => bf.extend_from_slice(v),
             None => (),
         }
     }
diff --git a/rsocket/src/frame/keepalive.rs b/rsocket/src/frame/keepalive.rs
@@ -1,5 +1,5 @@
-use super::utils::too_short;
 use super::{Body, Frame};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -45,13 +45,13 @@ impl KeepaliveBuilder {
 impl Keepalive {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<Keepalive> {
         if bf.len() < 8 {
-            return too_short(8);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let position = bf.get_u64();
         let data = if bf.is_empty() {
             None
         } else {
-            Some(Bytes::from(bf.to_vec()))
+            Some(bf.split().freeze())
         };
         Ok(Keepalive {
             last_received_position: position,
@@ -83,7 +83,7 @@ impl Writeable for Keepalive {
     fn write_to(&self, bf: &mut BytesMut) {
         bf.put_u64(self.last_received_position);
         match &self.data {
-            Some(v) => bf.put(v.bytes()),
+            Some(v) => bf.extend_from_slice(v),
             None => (),
         }
     }
diff --git a/rsocket/src/frame/lease.rs b/rsocket/src/frame/lease.rs
@@ -1,5 +1,5 @@
-use super::utils::too_short;
 use super::{Body, Frame};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -53,12 +53,12 @@ impl LeaseBuilder {
 impl Lease {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<Lease> {
         if bf.len() < 8 {
-            return too_short(8);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let ttl = bf.get_u32();
         let n = bf.get_u32();
         let m = if flag & Frame::FLAG_METADATA != 0 {
-            Some(Bytes::from(bf.to_vec()))
+            Some(bf.split().freeze())
         } else {
             None
         };
@@ -94,7 +94,7 @@ impl Writeable for Lease {
         bf.put_u32(self.ttl);
         bf.put_u32(self.number_of_requests);
         match &self.metadata {
-            Some(v) => bf.put(v.bytes()),
+            Some(v) => bf.extend_from_slice(v),
             None => (),
         }
     }
diff --git a/rsocket/src/frame/metadata_push.rs b/rsocket/src/frame/metadata_push.rs
@@ -34,8 +34,9 @@ impl MetadataPushBuiler {
 
 impl MetadataPush {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<MetadataPush> {
-        let m = Bytes::from(bf.to_vec());
-        Ok(MetadataPush { metadata: Some(m) })
+        Ok(MetadataPush {
+            metadata: Some(bf.split().freeze()),
+        })
     }
 
     pub fn builder(stream_id: u32, flag: u16) -> MetadataPushBuiler {
@@ -57,7 +58,7 @@ impl MetadataPush {
 impl Writeable for MetadataPush {
     fn write_to(&self, bf: &mut BytesMut) {
         match &self.metadata {
-            Some(v) => bf.put(v.bytes()),
+            Some(v) => bf.extend_from_slice(v),
             None => (),
         }
     }
diff --git a/rsocket/src/frame/request_channel.rs b/rsocket/src/frame/request_channel.rs
@@ -1,5 +1,6 @@
 use super::utils;
 use super::{Body, Frame, REQUEST_MAX};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -68,7 +69,7 @@ impl RequestChannelBuilder {
 impl RequestChannel {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<RequestChannel> {
         if bf.len() < 4 {
-            utils::too_short(4)
+            Err(RSocketError::InCompleteFrame.into())
         } else {
             let initial_request_n = bf.get_u32();
             utils::read_payload(flag, bf).map(move |(metadata, data)| RequestChannel {
diff --git a/rsocket/src/frame/request_n.rs b/rsocket/src/frame/request_n.rs
@@ -1,4 +1,5 @@
 use super::{utils, Body, Frame, REQUEST_MAX};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -35,7 +36,7 @@ impl RequestNBuilder {
 impl RequestN {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<RequestN> {
         if bf.len() < 4 {
-            utils::too_short(4)
+            Err(RSocketError::InCompleteFrame.into())
         } else {
             let n = bf.get_u32();
             Ok(RequestN { n })
diff --git a/rsocket/src/frame/request_stream.rs b/rsocket/src/frame/request_stream.rs
@@ -1,4 +1,5 @@
 use super::{utils, Body, Frame, REQUEST_MAX};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -55,7 +56,7 @@ impl RequestStreamBuilder {
 impl RequestStream {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<RequestStream> {
         if bf.len() < 4 {
-            utils::too_short(4)
+            Err(RSocketError::InCompleteFrame.into())
         } else {
             let initial_request_n = bf.get_u32();
             utils::read_payload(flag, bf).map(move |(metadata, data)| RequestStream {
diff --git a/rsocket/src/frame/resume.rs b/rsocket/src/frame/resume.rs
@@ -1,5 +1,5 @@
-use super::utils::too_short;
 use super::{Body, Frame, Version};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -29,21 +29,21 @@ impl Resume {
 
     pub(crate) fn decode(flag: u16, b: &mut BytesMut) -> crate::Result<Resume> {
         if b.len() < 6 {
-            return too_short(6);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let major = b.get_u16();
         let minor = b.get_u16();
         let token_size = b.get_u16() as usize;
         let token = if token_size > 0 {
             if b.len() < token_size {
-                return too_short(token_size);
+                return Err(RSocketError::InCompleteFrame.into());
             }
             Some(b.split_to(token_size).freeze())
         } else {
             None
         };
         if b.len() < 16 {
-            return too_short(16);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let last_received_server_position = b.get_u64();
         let first_available_client_position = b.get_u64();
@@ -114,7 +114,7 @@ impl Writeable for Resume {
         self.version.write_to(bf);
         if let Some(b) = self.get_token() {
             bf.put_u16(b.len() as u16);
-            bf.put(b.bytes());
+            bf.extend_from_slice(b);
         }
         bf.put_u64(self.get_last_received_server_position());
         bf.put_u64(self.get_first_available_client_position());
diff --git a/rsocket/src/frame/resume_ok.rs b/rsocket/src/frame/resume_ok.rs
@@ -1,5 +1,5 @@
-use super::utils::too_short;
 use super::{Body, Frame};
+use crate::error::RSocketError;
 use crate::utils::Writeable;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -35,7 +35,7 @@ impl ResumeOKBuilder {
 impl ResumeOK {
     pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<ResumeOK> {
         if bf.len() < 8 {
-            too_short(8)
+            Err(RSocketError::InCompleteFrame.into())
         } else {
             Ok(ResumeOK {
                 position: bf.get_u64(),
diff --git a/rsocket/src/frame/setup.rs b/rsocket/src/frame/setup.rs
@@ -1,5 +1,6 @@
-use super::utils::{self, too_short};
+use super::utils;
 use super::{Body, Frame, Version};
+use crate::error::RSocketError;
 use crate::utils::{Writeable, DEFAULT_MIME_TYPE};
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use std::time::Duration;
@@ -26,40 +27,40 @@ impl Setup {
     pub(crate) fn decode(flag: u16, b: &mut BytesMut) -> crate::Result<Setup> {
         // Check minimal length: version(4bytes) + keepalive(4bytes) + lifetime(4bytes)
         if b.len() < 12 {
-            return too_short(12);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let major = b.get_u16();
         let minor = b.get_u16();
         let keepalive = b.get_u32();
         let lifetime = b.get_u32();
         let token: Option<Bytes> = if flag & Frame::FLAG_RESUME != 0 {
             if b.len() < 2 {
-                return too_short(2);
+                return Err(RSocketError::InCompleteFrame.into());
             }
             let token_length = b.get_u16() as usize;
             if b.len() < token_length {
-                return too_short(token_length);
+                return Err(RSocketError::InCompleteFrame.into());
             }
             Some(b.split_to(token_length).freeze())
         } else {
             None
         };
         if b.is_empty() {
-            return too_short(1);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let mut mime_type_length: usize = b[0] as usize;
         b.advance(1);
         if b.len() < mime_type_length {
-            return too_short(mime_type_length);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let mime_metadata = b.split_to(mime_type_length).freeze();
         if b.is_empty() {
-            return too_short(1);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         mime_type_length = b[0] as usize;
         b.advance(1);
         if b.len() < mime_type_length {
-            return too_short(mime_type_length);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let mime_data = b.split_to(mime_type_length).freeze();
         let (metadata, data) = utils::read_payload(flag, b)?;
@@ -143,12 +144,12 @@ impl Writeable for Setup {
         bf.put_u32(self.lifetime);
         if let Some(b) = &self.token {
             bf.put_u16(b.len() as u16);
-            bf.put(b.bytes());
+            bf.extend_from_slice(b);
         }
         bf.put_u8(self.mime_metadata.len() as u8);
-        bf.put(self.mime_metadata.clone());
+        bf.extend_from_slice(&self.mime_metadata);
         bf.put_u8(self.mime_data.len() as u8);
-        bf.put(self.mime_data.clone());
+        bf.extend_from_slice(&self.mime_data);
         utils::write_payload(bf, self.get_metadata(), self.get_data());
     }
 }
diff --git a/rsocket/src/frame/utils.rs b/rsocket/src/frame/utils.rs
@@ -3,13 +3,14 @@ use crate::error::RSocketError;
 use crate::utils::{u24, Writeable};
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
+#[inline]
 pub(crate) fn read_payload(
     flag: u16,
     bf: &mut BytesMut,
 ) -> crate::Result<(Option<Bytes>, Option<Bytes>)> {
     let m: Option<Bytes> = if flag & Frame::FLAG_METADATA != 0 {
         if bf.len() < 3 {
-            return too_short(3);
+            return Err(RSocketError::InCompleteFrame.into());
         }
         let n = u24::read_advance(bf);
         Some(bf.split_to(n.into()).freeze())
@@ -19,7 +20,7 @@ pub(crate) fn read_payload(
     let d: Option<Bytes> = if bf.is_empty() {
         None
     } else {
-        Some(Bytes::from(bf.to_vec()))
+        Some(bf.split().freeze())
     };
     Ok((m, d))
 }
@@ -28,16 +29,13 @@ pub(crate) fn calculate_payload_length(metadata: Option<&Bytes>, data: Option<&B
     metadata.map(|v| 3 + v.len()).unwrap_or(0) + data.map(|v| v.len()).unwrap_or(0)
 }
 
+#[inline]
 pub(crate) fn write_payload(bf: &mut BytesMut, metadata: Option<&Bytes>, data: Option<&Bytes>) {
     if let Some(v) = metadata {
         u24::from(v.len()).write_to(bf);
-        bf.put(v.clone());
+        bf.extend_from_slice(v);
     }
     if let Some(v) = data {
-        bf.put(v.clone())
+        bf.extend_from_slice(v);
     }
 }
-
-pub(crate) fn too_short<T>(n: usize) -> crate::Result<T> {
-    Err(RSocketError::InCompleteFrame.into())
-}

Original file line number	Diff line number	Diff line change
`@@ -191,13 +191,13 @@ impl Writeable for CompositeMetadataEntry {`
`191`	`191`	`let mime_type_len = s.len() as u8;`
`192`	`192`	`assert!(mime_type_len > 0, "invalid length of MimeType!");`
`193`	`193`	`bf.put_u8(mime_type_len - 1);`
`194`		`- bf.put_slice(s.as_ref());`
	`194`	`+ bf.extend_from_slice(s.as_ref());`
`195`	`195`	`}`
`196`	`196`	`};`
`197`	`197`	`let metadata_len = self.metadata.len();`
`198`	`198`	`u24::from(metadata_len).write_to(bf);`
`199`	`199`	`if metadata_len > 0 {`
`200`		`- bf.put(self.metadata.bytes());`
	`200`	`+ bf.extend_from_slice(&self.metadata);`
`201`	`201`	`}`
`202`	`202`	`}`
`203`	`203`
Original file line number	Diff line number	Diff line change
`@@ -34,8 +34,9 @@ impl MetadataPushBuiler {`
`34`	`34`
`35`	`35`	`impl MetadataPush {`
`36`	`36`	`pub(crate) fn decode(flag: u16, bf: &mut BytesMut) -> crate::Result<MetadataPush> {`
`37`		`- let m = Bytes::from(bf.to_vec());`
`38`		`- Ok(MetadataPush { metadata: Some(m) })`
	`37`	`+ Ok(MetadataPush {`
	`38`	`+ metadata: Some(bf.split().freeze()),`
	`39`	`+ })`
`39`	`40`	`}`
`40`	`41`
`41`	`42`	`pub fn builder(stream_id: u32, flag: u16) -> MetadataPushBuiler {`
`@@ -57,7 +58,7 @@ impl MetadataPush {`
`57`	`58`	`impl Writeable for MetadataPush {`
`58`	`59`	`fn write_to(&self, bf: &mut BytesMut) {`
`59`	`60`	`match &self.metadata {`
`60`		`- Some(v) => bf.put(v.bytes()),`
	`61`	`+ Some(v) => bf.extend_from_slice(v),`
`61`	`62`	`None => (),`
`62`	`63`	`}`
`63`	`64`	`}`