migrate more duplex socket features.

Author: jjeffcaii

Cargo.toml

@@ -35,6 +35,10 @@ hex = "0.4.0"
 name = "echo"
 path = "examples/echo/main.rs"
 
+[[example]]
+name = "echo_client"
+path = "examples/echo_client/main.rs"
+
 # [[example]]
 # name = "proxy"
 # path = "examples/proxy/main.rs"

examples/echo/main.rs

@@ -2,23 +2,17 @@ extern crate rsocket_rust;
 extern crate tokio;
 #[macro_use]
 extern crate log;
-use futures::{SinkExt, StreamExt};
-use rsocket_rust::frame::Frame;
 use rsocket_rust::prelude::*;
-use rsocket_rust::transport::DuplexSocket;
-use rsocket_rust::transport::{self, Rx, Tx};
 use std::env;
 use std::error::Error;
-use std::sync::Arc;
-use tokio::net::TcpListener;
-use tokio::prelude::*;
-use tokio::sync::mpsc;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     env_logger::builder().init();
+    let addr = env::args().nth(1).unwrap_or("127.0.0.1:7878".to_string());
+
     RSocketFactory::receive()
-        .transport(URI::Tcp("127.0.0.1:7878"))
+        .transport(URI::Tcp(addr))
         .acceptor(|setup, sending_socket| {
             info!("accept setup: {:?}", setup);
             Box::new(EchoRSocket)

examples/echo_client/main.rs

@@ -0,0 +1,21 @@
+extern crate rsocket_rust;
+extern crate tokio;
+#[macro_use]
+extern crate log;
+use rsocket_rust::prelude::*;
+
+#[tokio::main]
+async fn main() {
+    env_logger::builder().init();
+    let setup = Payload::builder()
+        .set_data_utf8("Hello")
+        .set_metadata_utf8("World")
+        .build();
+    let cli = RSocketFactory::connect()
+        .transport(URI::Tcp("127.0.0.1:7878".to_string()))
+        .setup(setup)
+        .start()
+        .await;
+
+    std::thread::sleep_ms(500000);
+}

src/spi.rs

@@ -29,6 +29,7 @@ impl RSocket for EchoRSocket {
     }
 
     fn request_response(&self, req: Payload) -> Single<Payload> {
+        info!("echo request_response: {:?}", req);
         Box::pin(future::ok::<Payload, RSocketError>(req))
     }
 }

src/transport/conn.rs

@@ -37,22 +37,3 @@ impl From<u32> for StreamID {
         StreamID::new(v)
     }
 }
-
-#[derive(Debug)]
-pub(crate) enum Handler {
-    Request(oneshot::Sender<Payload>),
-    Stream(mpsc::Sender<Payload>),
-}
-
-#[derive(Debug)]
-pub(crate) struct Handlers {
-    map: RwLock<HashMap<u32, Handler>>,
-}
-
-impl Handlers {
-    pub(crate) fn new() -> Handlers {
-        Handlers {
-            map: RwLock::new(HashMap::new()),
-        }
-    }
-}

src/transport/misc.rs

@@ -1,8 +1,8 @@
 mod codec;
-mod conn;
 mod misc;
+mod socket;
 mod spi;
 pub(crate) mod tcp;
 pub(crate) use codec::RFrameCodec;
-pub use conn::DuplexSocket;
+pub(crate) use socket::DuplexSocket;
 pub use spi::{Rx, Transport, Tx};