change server/client responder handler from fn to Box<Fn>.

jjeffcaii · jjeffcaii · commit a8943a39bc58 · 2020-04-06T19:53:02.000+08:00
diff --git a/README.md b/README.md
@@ -45,12 +45,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     RSocketFactory::receive()
         .transport(TcpServerTransport::from(addr))
-        .acceptor(|setup, _socket| {
+        .acceptor(Box::new(|setup, _socket| {
             info!("accept setup: {:?}", setup);
             Ok(Box::new(EchoRSocket))
             // Or you can reject setup
             // Err(From::from("SETUP_NOT_ALLOW"))
-        })
+        }))
         .on_start(|| info!("+++++++ echo server started! +++++++"))
         .serve()
         .await
@@ -67,7 +67,7 @@ use rsocket_rust_transport_tcp::TcpClientTransport;
 #[test]
 async fn test() {
     let cli = RSocketFactory::connect()
-        .acceptor(|| Box::new(EchoRSocket))
+        .acceptor(Box::new(|| Box::new(EchoRSocket)))
         .transport(TcpClientTransport::from("127.0.0.1:7878"))
         .setup(Payload::from("READY!"))
         .mime_type("text/plain", "text/plain")
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
@@ -21,7 +21,7 @@ path = "../rsocket-transport-tcp"
 path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "0.2.13"
+version = "0.2.16"
 default-features = false
 features = ["full"]
 
diff --git a/examples/cli.rs b/examples/cli.rs
@@ -6,10 +6,10 @@ use std::error::Error;
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let client = RSocketFactory::connect()
         .transport(TcpClientTransport::from("127.0.0.1:7878"))
-        .acceptor(|| {
+        .acceptor(Box::new(|| {
             // Return a responder.
             Box::new(EchoRSocket)
-        })
+        }))
         .start()
         .await
         .expect("Connect failed!");
diff --git a/examples/echo.rs b/examples/echo.rs
@@ -109,12 +109,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
             RSocketFactory::receive()
                 .transport(TcpServerTransport::from(addr))
                 .fragment(mtu)
-                .acceptor(|setup, _socket| {
+                .acceptor(Box::new(|setup, _socket| {
                     info!("accept setup: {:?}", setup);
                     Ok(Box::new(EchoRSocket))
                     // Or you can reject setup
                     // Err(From::from("SETUP_NOT_ALLOW"))
-                })
+                }))
                 .on_start(Box::new(|| info!("+++++++ echo server started! +++++++")))
                 .serve()
                 .await
diff --git a/examples/proxy.rs b/examples/proxy.rs
@@ -12,18 +12,18 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     RSocketFactory::receive()
         .transport(TcpServerTransport::from("127.0.0.1:7979"))
-        .acceptor(|setup, _sending_socket| {
+        .acceptor(Box::new(|setup, _sending_socket| {
             info!("incoming socket: setup={:?}", setup);
             Ok(Box::new(block_on(async move {
                 RSocketFactory::connect()
                     .transport(TcpClientTransport::from("127.0.0.1:7878"))
-                    .acceptor(|| Box::new(EchoRSocket))
+                    .acceptor(Box::new(|| Box::new(EchoRSocket)))
                     .setup(Payload::from("I'm Rust!"))
                     .start()
                     .await
                     .unwrap()
             })))
-        })
+        }))
         .serve()
         .await
 }
diff --git a/rsocket-test/tests/test_clients.rs b/rsocket-test/tests/test_clients.rs
@@ -38,10 +38,10 @@ fn test_websocket() {
     server_runtime.spawn(async move {
         RSocketFactory::receive()
             .transport(WebsocketServerTransport::from(addr))
-            .acceptor(|setup, _socket| {
+            .acceptor(Box::new(|setup, _socket| {
                 info!("accept setup: {:?}", setup);
                 Ok(Box::new(EchoRSocket))
-            })
+            }))
             .on_start(Box::new(|| {
                 info!("+++++++ websocket echo server started! +++++++")
             }))
@@ -55,7 +55,7 @@ fn test_websocket() {
 
     client_runtime.block_on(async {
         let cli = RSocketFactory::connect()
-            .acceptor(|| Box::new(EchoRSocket))
+            .acceptor(Box::new(|| Box::new(EchoRSocket)))
             .transport(WebsocketClientTransport::from(addr))
             .setup(Payload::from("READY!"))
             .mime_type("text/plain", "text/plain")
@@ -86,10 +86,10 @@ fn test_tcp() {
     server_runtime.spawn(async move {
         RSocketFactory::receive()
             .transport(TcpServerTransport::from(addr))
-            .acceptor(|setup, _socket| {
+            .acceptor(Box::new(|setup, _socket| {
                 info!("accept setup: {:?}", setup);
                 Ok(Box::new(EchoRSocket))
-            })
+            }))
             .on_start(Box::new(|| {
                 info!("+++++++ tcp echo server started! +++++++")
             }))
@@ -103,7 +103,7 @@ fn test_tcp() {
 
     client_runtime.block_on(async {
         let cli = RSocketFactory::connect()
-            .acceptor(|| Box::new(EchoRSocket))
+            .acceptor(Box::new(|| Box::new(EchoRSocket)))
             .transport(TcpClientTransport::from(addr))
             .setup(Payload::from("READY!"))
             .mime_type("text/plain", "text/plain")
diff --git a/rsocket-transport-tcp/Cargo.toml b/rsocket-transport-tcp/Cargo.toml
@@ -20,7 +20,7 @@ path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.2.13"
+version = "0.2.16"
 default-features = false
 features = [ "rt-core", "rt-threaded", "tcp", "sync", "stream" ]
 
diff --git a/rsocket-transport-websocket/Cargo.toml b/rsocket-transport-websocket/Cargo.toml
@@ -22,6 +22,6 @@ path = "../rsocket"
 features = ["frame"]
 
 [dependencies.tokio]
-version = "0.2.13"
+version = "0.2.16"
 default-features = false
 features = [ "rt-core", "rt-threaded", "tcp", "sync", "stream" ]
diff --git a/rsocket/Cargo.toml b/rsocket/Cargo.toml
@@ -16,7 +16,7 @@ futures = "0.3.4"
 lazy_static = "1.4.0"
 
 [dependencies.tokio]
-version = "0.2.13"
+version = "0.2.16"
 default-features = false
 features = [ "rt-core", "rt-threaded", "sync", "stream" ]
 
diff --git a/rsocket/README.md b/rsocket/README.md
@@ -36,12 +36,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     RSocketFactory::receive()
         .transport(TcpServerTransport::from(addr))
-        .acceptor(|setup, _socket| {
+        .acceptor(Box::new(|setup, _socket| {
             info!("accept setup: {:?}", setup);
             Ok(Box::new(EchoRSocket))
             // Or you can reject setup
             // Err(From::from("SETUP_NOT_ALLOW"))
-        })
+        }))
         .on_start(|| info!("+++++++ echo server started! +++++++"))
         .serve()
         .await
@@ -58,7 +58,7 @@ use rsocket_rust_transport_tcp::TcpClientTransport;
 #[test]
 async fn test() {
     let cli = RSocketFactory::connect()
-        .acceptor(|| Box::new(EchoRSocket))
+        .acceptor(Box::new(|| Box::new(EchoRSocket)))
         .transport(TcpClientTransport::from("127.0.0.1:7878"))
         .setup(Payload::from("READY!"))
         .mime_type("text/plain", "text/plain")
diff --git a/rsocket/src/lib.rs b/rsocket/src/lib.rs
@@ -44,7 +44,7 @@
 //! async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 //!     RSocketFactory::receive()
 //!         .transport(TcpServerTransport::from("127.0.0.1:7878"))
-//!         .acceptor(|setup, socket| {
+//!         .acceptor(Box::new(|setup, socket| {
 //!             println!("socket establish: setup={:?}", setup);
 //!             tokio::spawn(async move {
 //!                 let req = Payload::builder().set_data_utf8("Hello World!").build();
@@ -54,7 +54,7 @@
 //!             // Return a responder.
 //!             // You can write you own responder by implementing `RSocket` trait.
 //!             Ok(Box::new(EchoRSocket))
-//!         })
+//!         }))
 //!         .on_start(Box::new(|| println!("echo server start success!")))
 //!         .serve()
 //!         .await
@@ -72,10 +72,10 @@
 //! async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 //!     let client = RSocketFactory::connect()
 //!         .transport(TcpClientTransport::from("127.0.0.1:7878"))
-//!         .acceptor(|| {
+//!         .acceptor(Box::new(|| {
 //!             // Return a responder.
 //!             Box::new(EchoRSocket)
-//!         })
+//!         }))
 //!         .start()
 //!         .await
 //!         .expect("Connect failed!");
diff --git a/rsocket/src/spi.rs b/rsocket/src/spi.rs
@@ -1,16 +1,23 @@
 use crate::error::{self, ErrorKind, RSocketError};
 use crate::frame;
-use crate::payload::Payload;
+use crate::payload::{Payload, SetupPayload};
 use crate::utils::RSocketResult;
-
 use futures::future;
 use futures::{Sink, SinkExt, Stream, StreamExt};
+use std::error::Error;
 use std::future::Future;
 use std::pin::Pin;
 use std::result::Result;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 
+pub type ClientResponder = Box<dyn Send + Sync + Fn() -> Box<dyn RSocket>>;
+pub type ServerResponder = Box<
+    dyn Send
+        + Sync
+        + Fn(SetupPayload, Box<dyn RSocket>) -> Result<Box<dyn RSocket>, Box<dyn Error>>,
+>;
+
 pub type Mono<T> = Pin<Box<dyn Send + Sync + Future<Output = T>>>;
 pub type Flux<T> = Pin<Box<dyn Send + Sync + Stream<Item = T>>>;
 
diff --git a/rsocket/src/transport/socket.rs b/rsocket/src/transport/socket.rs
@@ -118,7 +118,7 @@ where
         }
     }
 
-    pub(crate) async fn event_loop(&self, acceptor: Acceptor, mut rx: Rx<Frame>) {
+    pub(crate) async fn event_loop(&self, acceptor: Option<Acceptor>, mut rx: Rx<Frame>) {
         while let Some(msg) = rx.next().await {
             let sid = msg.get_stream_id();
             let flag = msg.get_flag();
@@ -191,6 +191,7 @@ where
     }
 
     async fn join_frame(&self, input: Frame) {
+        // TODO: support fragmentation
         let sid = input.get_stream_id();
         let mut joiners = self.joiners.lock().await;
         match (*joiners).remove(&sid) {
@@ -224,7 +225,7 @@ where
                 Handler::ReqRR(tx) => tx.send(e).expect("Send RR failed"),
                 Handler::ResRR(_) => unreachable!(),
                 Handler::ReqRS(tx) => tx.unbounded_send(e).expect("Send RS failed"),
-                _ => unimplemented!(),
+                Handler::ReqRC(tx) => tx.unbounded_send(e).expect("Send RC failed"),
             }
         }
     }
@@ -287,27 +288,27 @@ where
     #[inline]
     fn on_setup(
         &self,
-        acceptor: &Acceptor,
+        acceptor: &Option<Acceptor>,
         sid: u32,
         flag: u16,
         setup: SetupPayload,
     ) -> Result<(), Box<dyn Error>> {
         match acceptor {
-            Acceptor::Simple(gen) => {
+            None => {
+                self.responder.set(Box::new(EmptyRSocket));
+                Ok(())
+            }
+            Some(Acceptor::Simple(gen)) => {
                 self.responder.set(gen());
                 Ok(())
             }
-            Acceptor::Generate(gen) => match gen(setup, Box::new(self.clone())) {
+            Some(Acceptor::Generate(gen)) => match gen(setup, Box::new(self.clone())) {
                 Ok(it) => {
                     self.responder.set(it);
                     Ok(())
                 }
                 Err(e) => Err(e),
             },
-            Acceptor::Empty() => {
-                self.responder.set(Box::new(EmptyRSocket));
-                Ok(())
-            }
         }
     }
 
diff --git a/rsocket/src/transport/spi.rs b/rsocket/src/transport/spi.rs
@@ -1,7 +1,7 @@
 use crate::error::RSocketError;
 use crate::frame::Frame;
 use crate::payload::SetupPayload;
-use crate::spi::RSocket;
+use crate::spi::{ClientResponder, RSocket, ServerResponder};
 use futures::channel::{mpsc, oneshot};
 use std::error::Error;
 use std::future::Future;
@@ -44,11 +44,8 @@ pub trait ServerTransport {
         Self::Item: ClientTransport + Sized;
 }
 
-pub type FnAcceptorWithSetup =
-    fn(SetupPayload, Box<dyn RSocket>) -> Result<Box<dyn RSocket>, Box<dyn Error>>;
-
+#[derive(Clone)]
 pub(crate) enum Acceptor {
-    Simple(Arc<fn() -> Box<dyn RSocket>>),
-    Generate(Arc<FnAcceptorWithSetup>),
-    Empty(),
+    Simple(Arc<ClientResponder>),
+    Generate(Arc<ServerResponder>),
 }
diff --git a/rsocket/src/x/client.rs b/rsocket/src/x/client.rs
@@ -2,7 +2,7 @@ use crate::error::RSocketError;
 use crate::frame::{self, Frame};
 use crate::payload::{Payload, SetupPayload, SetupPayloadBuilder};
 use crate::runtime::{DefaultSpawner, Spawner};
-use crate::spi::{Flux, Mono, RSocket};
+use crate::spi::{ClientResponder, Flux, Mono, RSocket};
 use crate::transport::{
     self, Acceptor, ClientTransport, DuplexSocket, Rx, RxOnce, Splitter, Tx, TxOnce,
 };
@@ -29,7 +29,7 @@ where
 {
     transport: Option<T>,
     setup: SetupPayloadBuilder,
-    responder: Option<fn() -> Box<dyn RSocket>>,
+    responder: Option<ClientResponder>,
     closer: Option<Box<dyn FnMut() + Send + Sync>>,
     mtu: usize,
 }
@@ -100,7 +100,7 @@ where
         self
     }
 
-    pub fn acceptor(mut self, acceptor: fn() -> Box<dyn RSocket>) -> Self {
+    pub fn acceptor(mut self, acceptor: ClientResponder) -> Self {
         self.responder = Some(acceptor);
         self
     }
@@ -135,9 +135,9 @@ where
         };
         let duplex_socket = DuplexSocket::new(rt, 1, snd_tx.clone(), splitter).await;
         let cloned_duplex_socket = duplex_socket.clone();
-        let acceptor = match self.responder {
-            Some(r) => Acceptor::Simple(Arc::new(r)),
-            None => Acceptor::Empty(),
+        let acceptor: Option<Acceptor> = match self.responder {
+            Some(it) => Some(Acceptor::Simple(Arc::new(it))),
+            None => None,
         };
         let closer = self.closer.take();
         cloned_rt.spawn(async move {
diff --git a/rsocket/src/x/server.rs b/rsocket/src/x/server.rs
