foobar

Author: jjeffcaii

rsocket/src/frame/mod.rs

@@ -167,14 +167,14 @@ impl Frame {
         body.map(|it| Frame::new(sid, it, flag))
     }
 
-    pub fn is_followable(&self) -> bool {
+    pub(crate) fn is_followable_or_payload(&self) -> (bool, bool) {
         match &self.body {
-            Body::RequestFNF(_) => true,
-            Body::RequestResponse(_) => true,
-            Body::RequestStream(_) => true,
-            Body::RequestChannel(_) => true,
-            Body::Payload(_) => true,
-            _ => false,
+            Body::RequestFNF(_) => (true, false),
+            Body::RequestResponse(_) => (true, false),
+            Body::RequestStream(_) => (true, false),
+            Body::RequestChannel(_) => (true, false),
+            Body::Payload(_) => (true, true),
+            _ => (false, false),
         }
     }
 

rsocket/src/frame/setup.rs

@@ -35,7 +35,6 @@ impl Writeable for Setup {
             bf.put_u16(b.len() as u16);
             bf.put(b.bytes());
         }
-        // TODO: remove string clone
         bf.put_u8(self.mime_metadata.len() as u8);
         bf.put(Bytes::from(self.mime_metadata.clone()));
         bf.put_u8(self.mime_data.len() as u8);

rsocket/src/frame/utils.rs

@@ -17,7 +17,6 @@ impl PayloadSupport {
         a + b
     }
 
-    // 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_advance(bf);

rsocket/src/lib.rs

@@ -24,11 +24,11 @@
 //! The easiest way to get started is to use RSocket. Do this by enabling TCP transport support.
 //!
 //! ```toml
-//! rsocket_rust = "0.5"
-//! rsocket_rust_transport_tcp = "0.5"
+//! rsocket_rust = "0.5.1"
+//! rsocket_rust_transport_tcp = "0.5.1"
 //!
 //! # If you want to use websocket transport instead.
-//! # rsocket_rust_transport_websocket = "0.5"
+//! # rsocket_rust_transport_websocket = "0.5.1"
 //! ```
 //!
 //! # Examples

rsocket/src/transport/socket.rs

@@ -197,21 +197,27 @@ where
         }
     }
 
-    // TODO: support fragmentation
     async fn join_frame(&self, input: Frame) -> Option<Frame> {
-        if !input.is_followable() {
+        let (is_follow, is_payload) = input.is_followable_or_payload();
+        if !is_follow {
             return Some(input);
         }
         let sid = input.get_stream_id();
         let mut joiners = self.joiners.lock().await;
 
         if input.get_flag() & frame::FLAG_FOLLOW != 0 {
+            // TODO: check conflict
             (*joiners)
                 .entry(sid)
                 .or_insert_with(Joiner::new)
                 .push(input);
             return None;
         }
+
+        if !is_payload {
+            return Some(input);
+        }
+
         match (*joiners).remove(&sid) {
             None => Some(input),
             Some(mut joiner) => {
@@ -266,6 +272,10 @@ where
 
     #[inline]
     async fn on_error(&self, sid: u32, flag: u16, input: frame::Error) {
+        {
+            let mut joiners = self.joiners.lock().await;
+            (*joiners).remove(&sid);
+        }
         // pick handler
         let mut handlers = self.handlers.lock().await;
         if let Some(handler) = (*handlers).remove(&sid) {
@@ -282,6 +292,10 @@ where
 
     #[inline]
     async fn on_cancel(&self, sid: u32, _flag: u16) {
+        {
+            let mut joiners = self.joiners.lock().await;
+            (*joiners).remove(&sid);
+        }
         let mut handlers = self.handlers.lock().await;
         if let Some(handler) = (*handlers).remove(&sid) {
             let e = Err(RSocketError::from(ErrorKind::Cancelled()));
@@ -343,32 +357,6 @@ where
             }
             Entry::Vacant(v) => warn!("invalid payload id {}: no such request!", sid),
         }
-
-        // match (*handlers).remove(&sid).unwrap() {
-        //     Handler::ReqRR(sender) => sender.send(Ok(input)).unwrap(),
-        //     Handler::ResRR(c) => unreachable!(),
-        //     Handler::ReqRS(sender) => {
-        //         if flag & frame::FLAG_NEXT != 0 {
-        //             sender
-        //                 .unbounded_send(Ok(input))
-        //                 .expect("Send payload response failed.");
-        //         }
-        //         if flag & frame::FLAG_COMPLETE == 0 {
-        //             (*handlers).insert(sid, Handler::ReqRS(sender));
-        //         }
-        //     }
-        //     Handler::ReqRC(sender) => {
-        //         // TODO: support channel
-        //         if flag & frame::FLAG_NEXT != 0 {
-        //             sender
-        //                 .unbounded_send(Ok(input))
-        //                 .expect("Send payload response failed");
-        //         }
-        //         if flag & frame::FLAG_COMPLETE == 0 {
-        //             (*handlers).insert(sid, Handler::ReqRC(sender));
-        //         }
-        //     }
-        // };
     }
 
     #[inline]