bugfix for frame codec && optimize some codes.

jjeffcaii · jjeffcaii · commit 5f4a9c5f377f · 2019-09-03T16:15:01.000+08:00
diff --git a/README.md b/README.md
@@ -62,7 +62,7 @@ use rsocket_rust::prelude::*;
 #[test]
 fn test_client() {
   let cli = RSocketFactory::connect()
-    .acceptor(Box::new(MockResponder))
+    .acceptor(||Box::new(MockResponder))
     .transport(URI::Tcp("127.0.0.1:7878"))
     .setup(Payload::from("READY!"))
     .mime_type("text/plain", "text/plain")
diff --git a/src/core/socket.rs b/src/core/socket.rs
@@ -2,9 +2,9 @@ extern crate bytes;
 extern crate futures;
 extern crate tokio;
 
-use crate::core::misc::StreamID;
-use crate::core::spi::RSocket;
-use crate::core::{RequestCaller, StreamCaller};
+use super::misc::StreamID;
+use super::spi::RSocket;
+use super::{RequestCaller, StreamCaller};
 use crate::errors::{ErrorKind, RSocketError};
 use crate::frame::{self, Body, Frame};
 use crate::payload::{Payload, SetupPayload};
diff --git a/src/frame/cancel.rs b/src/frame/cancel.rs
@@ -1,4 +1,4 @@
-use crate::frame::{Body, Frame};
+use super::{Body, Frame};
 
 #[derive(Debug, Clone)]
 pub struct Cancel {}
diff --git a/src/frame/error.rs b/src/frame/error.rs
@@ -1,6 +1,6 @@
 extern crate bytes;
 
-use crate::frame::{Body, Frame, Writeable};
+use super::{Body, Frame, Writeable};
 use crate::result::RSocketResult;
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
 
diff --git a/src/frame/keepalive.rs b/src/frame/keepalive.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, Writeable};
+use super::{Body, Frame, Writeable};
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/lease.rs b/src/frame/lease.rs
@@ -1,6 +1,6 @@
 extern crate bytes;
 
-use crate::frame::{Body, Frame, Writeable, FLAG_METADATA};
+use super::{Body, Frame, Writeable, FLAG_METADATA};
 use crate::result::RSocketResult;
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
 
diff --git a/src/frame/metadata_push.rs b/src/frame/metadata_push.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, Writeable};
+use super::{Body, Frame, Writeable};
 use bytes::{BufMut, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/payload.rs b/src/frame/payload.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA};
+use super::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA};
 use bytes::{BufMut, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/request_channel.rs b/src/frame/request_channel.rs
@@ -1,6 +1,6 @@
 extern crate bytes;
 
-use crate::frame::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA, REQUEST_MAX};
+use super::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA, REQUEST_MAX};
 use crate::result::RSocketResult;
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
 
diff --git a/src/frame/request_fnf.rs b/src/frame/request_fnf.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA};
+use super::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA};
 use bytes::{BufMut, ByteOrder, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/request_n.rs b/src/frame/request_n.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, Writeable, REQUEST_MAX};
+use super::{Body, Frame, Writeable, REQUEST_MAX};
 use bytes::{BigEndian, BufMut, ByteOrder, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/request_response.rs b/src/frame/request_response.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA, U24};
+use super::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA, U24};
 use bytes::{BigEndian, BufMut, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/request_stream.rs b/src/frame/request_stream.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA, REQUEST_MAX, U24};
+use super::{Body, Frame, PayloadSupport, Writeable, FLAG_METADATA, REQUEST_MAX, U24};
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/resume.rs b/src/frame/resume.rs
@@ -1,6 +1,6 @@
 extern crate bytes;
 
-use crate::frame::{Body, Frame, Version, Writeable};
+use super::{Body, Frame, Version, Writeable};
 use crate::result::RSocketResult;
 
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
diff --git a/src/frame/resume_ok.rs b/src/frame/resume_ok.rs
@@ -1,7 +1,7 @@
 extern crate bytes;
 
 use crate::result::RSocketResult;
-use crate::frame::{Body, Frame, Writeable};
+use super::{Body, Frame, Writeable};
 use bytes::{BigEndian, BufMut, ByteOrder, BytesMut};
 
 #[derive(Debug, Clone)]
diff --git a/src/frame/setup.rs b/src/frame/setup.rs
@@ -1,8 +1,6 @@
 extern crate bytes;
 
-use crate::frame::{
-  Body, Frame, PayloadSupport, Version, Writeable, FLAG_METADATA, FLAG_RESUME, U24,
-};
+use super::{Body, Frame, PayloadSupport, Version, Writeable, FLAG_METADATA, FLAG_RESUME};
 use crate::mime::MIME_BINARY;
 use crate::result::RSocketResult;
 use bytes::{BigEndian, BufMut, ByteOrder, Bytes, BytesMut};
diff --git a/src/frame/utils.rs b/src/frame/utils.rs
@@ -1,6 +1,6 @@
 extern crate bytes;
 
-use crate::frame::FLAG_METADATA;
+use super::FLAG_METADATA;
 use bytes::{BufMut, Bytes, BytesMut};
 
 pub struct U24 {}
@@ -13,6 +13,14 @@ impl U24 {
   }
 
   pub fn read(bf: &mut BytesMut) -> u32 {
+    let mut n: u32 = 0;
+    n += (bf[0] as u32) << 16;
+    n += (bf[1] as u32) << 8;
+    n += bf[2] as u32;
+    n
+  }
+
+  pub fn read_advance(bf: &mut BytesMut) -> u32 {
     let mut n: u32 = 0;
     let bar = bf.split_to(3);
     n += (bar[0] as u32) << 16;
@@ -40,7 +48,7 @@ impl PayloadSupport {
   // TODO: change to Result
   pub fn read(flag: u16, bf: &mut BytesMut) -> (Option<Bytes>, Option<Bytes>) {
     let m: Option<Bytes> = if flag & FLAG_METADATA != 0 {
-      let n = U24::read(bf);
+      let n = U24::read_advance(bf);
       Some(Bytes::from(bf.split_to(n as usize)))
     } else {
       None
diff --git a/src/transport/codec.rs b/src/transport/codec.rs
@@ -3,7 +3,7 @@ extern crate tokio;
 
 use crate::errors::RSocketError;
 use crate::frame::{Frame, Writeable, U24};
-use bytes::{BufMut, BytesMut};
+use bytes::{BufMut, BytesMut, IntoBuf};
 use std::io;
 use tokio::codec::{Decoder, Encoder};
 
@@ -22,6 +22,7 @@ impl Decoder for FrameCodec {
     if actual < 3 + l {
       return Ok(None);
     }
+    buf.advance(3);
     let mut bb = buf.split_to(l);
     Frame::decode(&mut bb).map(|it| Some(it))
   }
@@ -32,6 +33,7 @@ impl Encoder for FrameCodec {
   type Error = io::Error;
   fn encode(&mut self, item: Frame, buf: &mut BytesMut) -> Result<(), Self::Error> {
     let l = item.len();
+    buf.reserve(3 + (l as usize));
     U24::write(l, buf);
     item.write_to(buf);
     Ok(())
diff --git a/src/transport/tcp.rs b/src/transport/tcp.rs
@@ -1,15 +1,14 @@
 extern crate futures;
 extern crate tokio;
 
+use super::{Context, FrameCodec, Transport};
 use crate::frame::Frame;
-use crate::transport::FrameCodec;
-use crate::transport::{Context, Transport};
 use futures::sync::mpsc;
 use futures::{lazy, Future, Sink, Stream};
 use std::io;
 use std::net::SocketAddr;
 use tokio::codec::Framed;
-use tokio::net::{TcpStream,TcpListener};
+use tokio::net::TcpStream;
 
 pub fn from_addr(addr: &SocketAddr) -> Context {
   let socket = TcpStream::connect(addr).wait().unwrap();
diff --git a/src/x/client.rs b/src/x/client.rs
@@ -1,10 +1,10 @@
 extern crate futures;
 
+use super::URI;
 use crate::core::{Acceptor, DuplexSocket, RSocket};
 use crate::errors::RSocketError;
 use crate::payload::{Payload, SetupPayload, SetupPayloadBuilder};
 use crate::result::RSocketResult;
-use crate::x::URI;
 use futures::{Future, Stream};
 use std::time::Duration;
 
@@ -15,7 +15,7 @@ pub struct Client {
 pub struct ClientBuilder {
   uri: Option<URI>,
   setup: SetupPayloadBuilder,
-  responder: Option<Box<dyn RSocket>>,
+  responder: Option<fn() -> Box<dyn RSocket>>,
 }
 
 impl Client {
@@ -65,9 +65,9 @@ impl ClientBuilder {
   }
 
   pub fn mime_type(mut self, metadata_mime_type: &str, data_mime_type: &str) -> Self {
+    self = self.metadata_mime_type(metadata_mime_type);
+    self = self.data_mime_type(data_mime_type);
     self
-      .metadata_mime_type(metadata_mime_type)
-      .data_mime_type(data_mime_type)
   }
 
   pub fn data_mime_type(mut self, mime_type: &str) -> Self {
@@ -80,7 +80,7 @@ impl ClientBuilder {
     self
   }
 
-  pub fn acceptor(mut self, acceptor: Box<dyn RSocket>) -> Self {
+  pub fn acceptor(mut self, acceptor: fn() -> Box<dyn RSocket>) -> Self {
     self.responder = Some(acceptor);
     self
   }
@@ -92,7 +92,7 @@ impl ClientBuilder {
           let addr = vv.parse().unwrap();
           let mut bu = DuplexSocket::builder();
           if let Some(r) = self.responder {
-            bu = bu.set_acceptor(Acceptor::Direct(r));
+            bu = bu.set_acceptor(Acceptor::Direct(r()));
           }
           let socket = bu.connect(&addr);
           let setup = self.setup.build();
diff --git a/src/x/factory.rs b/src/x/factory.rs
@@ -1,14 +1,12 @@
-use crate::x::{Client,ClientBuilder};
-use crate::x::{ServerBuilder};
+use super::{Client, ClientBuilder, ServerBuilder};
 
-pub struct RSocketFactory{
-}
+pub struct RSocketFactory;
 
-impl<'a> RSocketFactory{
-  pub fn connect() -> ClientBuilder{
+impl RSocketFactory {
+  pub fn connect() -> ClientBuilder {
     Client::builder()
   }
-  pub fn receive() -> ServerBuilder{
+  pub fn receive() -> ServerBuilder {
     ServerBuilder::new()
   }
 }
diff --git a/src/x/server.rs b/src/x/server.rs
@@ -1,15 +1,12 @@
 extern crate futures;
 extern crate tokio;
 
+use super::URI;
 use crate::core::{Acceptor, DuplexSocket, EmptyRSocket, RSocket};
-use crate::errors::RSocketError;
 use crate::payload::SetupPayload;
-use crate::x::URI;
-use futures::future;
 use futures::prelude::*;
-use std::net::SocketAddr;
 use std::sync::Arc;
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpListener;
 
 fn on_setup_noop(_setup: SetupPayload, _socket: Box<dyn RSocket>) -> Box<dyn RSocket> {
   Box::new(EmptyRSocket)
diff --git a/tests/client_test.rs b/tests/client_test.rs
@@ -7,14 +7,14 @@ use rsocket_rust::prelude::*;
 #[test]
 fn test_client() {
   let cli = RSocketFactory::connect()
-    .acceptor(Box::new(MockResponder))
+    .acceptor(|| Box::new(MockResponder))
     .transport(URI::Tcp("127.0.0.1:7878"))
     .setup(Payload::from("READY!"))
     .mime_type("text/plain", "text/plain")
     .start()
     .unwrap();
 
-  for n in 0..10 {
+  for n in 0..10000 {
     let pa = Payload::builder()
       .set_data_utf8("Hello World!")
       .set_metadata_utf8(&format!("#{}", n))

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-use crate::frame::{Body, Frame};`
	`1`	`+use super::{Body, Frame};`
`2`	`2`
`3`	`3`	`#[derive(Debug, Clone)]`
`4`	`4`	`pub struct Cancel {}`