fix big bugs for Payload

Author: jjeffcaii

examples/echo.rs

@@ -5,6 +5,7 @@ use clap::{App, Arg, SubCommand};
 use rsocket_rust::prelude::*;
 use rsocket_rust_transport_tcp::{TcpClientTransport, TcpServerTransport};
 use std::error::Error;
+use std::fs;
 
 enum RequestMode {
     FNF,
@@ -154,7 +155,13 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
                 .expect("Connect failed!");
             let mut bu = Payload::builder();
             if let Some(data) = flags.value_of("input") {
-                bu = bu.set_data_utf8(data);
+                if data.starts_with("@") {
+                    let file_content =
+                        fs::read_to_string(&data[1..].to_owned()).expect("Read file failed.");
+                    bu = bu.set_data_utf8(&file_content);
+                } else {
+                    bu = bu.set_data_utf8(data);
+                }
             }
             let req = bu.build();
 

rsocket/src/payload/normal.rs

@@ -122,7 +122,8 @@ impl From<(&'static str, &'static str)> for Payload {
 
 impl From<frame::Payload> for Payload {
     fn from(input: frame::Payload) -> Payload {
-        Payload::new(input.get_data().clone(), input.get_data().clone())
+        let (d, m) = input.split();
+        Payload::new(d, m)
     }
 }
 

rsocket/src/transport/socket.rs

@@ -32,6 +32,7 @@ where
     handlers: Arc<Mutex<HashMap<u32, Handler>>>,
     canceller: Tx<u32>,
     splitter: Option<Splitter>,
+    joiners: Arc<Mutex<HashMap<u32, Joiner>>>,
 }
 
 #[derive(Clone)]
@@ -66,6 +67,7 @@ where
             canceller: canceller_tx,
             responder: Responder::new(),
             handlers: Arc::new(Mutex::new(HashMap::new())),
+            joiners: Arc::new(Mutex::new(HashMap::new())),
             splitter,
         };
 
@@ -188,6 +190,29 @@ where
         }
     }
 
+    async fn join_frame(&self, input: Frame) {
+        let sid = input.get_stream_id();
+        let mut joiners = self.joiners.lock().await;
+        match (*joiners).remove(&sid) {
+            Some(joiner) => {
+                // joiner.push(input);
+            }
+            None => {}
+        }
+    }
+
+    async fn try_join(&self, input: Frame) {
+        let follow = input.get_flag() & frame::FLAG_FOLLOW != 0;
+        match input.get_frame_type() {
+            frame::TYPE_REQUEST_RESPONSE => if follow {},
+            frame::TYPE_REQUEST_STREAM => {}
+            frame::TYPE_REQUEST_FNF => {}
+            frame::TYPE_REQUEST_CHANNEL => {}
+            frame::TYPE_PAYLOAD => {}
+            _ => {}
+        }
+    }
+
     #[inline]
     async fn on_error(&self, sid: u32, flag: u16, input: frame::Error) {
         // pick handler