Merge pull request #3 from rsocket/feature/routing_metadata

Feature/routing metadata

Author: jjeffcaii

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsocket_rust"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Jeffsky <[email protected]>"]
 edition = "2018"
 license = "Apache-2.0"
@@ -18,7 +18,7 @@ lazy_static = "1.4.0"
 # reactor_rs = {git = "https://github.com/jjeffcaii/reactor-rust", branch = "develop"}
 
 [dependencies.tokio]
-version = "0.2.1"
+version = "0.2.2"
 default-features = false
 features = ["full"]
 

README.md

@@ -80,12 +80,12 @@ async fn test() {
   - [x] REQUEST_FNF
   - [x] REQUEST_RESPONSE
   - [x] REQUEST_STREAM
-  - [ ] REQUEST_CHANNEL
+  - [x] REQUEST_CHANNEL
 - Transport
   - [x] TCP
   - [ ] Websocket
 - Reactor
   - [ ] ...
 - High Level APIs
-  - [ ] Client
-  - [ ] Server
+  - [x] Client
+  - [x] Server

src/errors.rs

@@ -5,7 +5,7 @@ use std::io;
 #[derive(Debug)]
 pub enum ErrorKind {
   Internal(u32, &'static str),
-  WithDescription(&'static str),
+  WithDescription(String),
   IO(io::Error),
   Cancelled(),
   Send(),
@@ -41,11 +41,18 @@ impl From<ErrorKind> for RSocketError {
     RSocketError { kind }
   }
 }
+impl From<String> for RSocketError {
+  fn from(e: String) -> RSocketError {
+    RSocketError {
+      kind: ErrorKind::WithDescription(e),
+    }
+  }
+}
 
 impl From<&'static str> for RSocketError {
   fn from(e: &'static str) -> RSocketError {
     RSocketError {
-      kind: ErrorKind::WithDescription(e),
+      kind: ErrorKind::WithDescription(String::from(e)),
     }
   }
 }

src/extension.rs

@@ -5,13 +5,92 @@ use crate::result::RSocketResult;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
 const MAX_MIME_LEN: usize = 0x7F;
+const MAX_ROUTING_TAG_LEN: usize = 0xFF;
 
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct CompositeMetadata {
     mime: String,
     payload: Bytes,
 }
 
+#[derive(Debug, Clone)]
+pub struct RoutingMetadata {
+    tags: Vec<String>,
+}
+
+pub struct RoutingMetadataBuilder {
+    inner: RoutingMetadata,
+}
+
+impl RoutingMetadataBuilder {
+    pub fn push_str(self, tag: &str) -> Self {
+        self.push(String::from(tag))
+    }
+
+    pub fn push(mut self, tag: String) -> Self {
+        if tag.len() > MAX_ROUTING_TAG_LEN {
+            panic!("exceeded maximum routing tag length!");
+        }
+        self.inner.tags.push(tag);
+        self
+    }
+    pub fn build(self) -> RoutingMetadata {
+        self.inner
+    }
+}
+
+impl RoutingMetadata {
+    pub fn builder() -> RoutingMetadataBuilder {
+        RoutingMetadataBuilder {
+            inner: RoutingMetadata { tags: vec![] },
+        }
+    }
+
+    pub fn decode(bf: &mut BytesMut) -> RSocketResult<RoutingMetadata> {
+        let mut bu = RoutingMetadata::builder();
+        loop {
+            match Self::decode_once(bf) {
+                Ok(v) => match v {
+                    Some(tag) => bu = bu.push(tag),
+                    None => break,
+                },
+                Err(e) => return Err(e),
+            }
+        }
+        Ok(bu.build())
+    }
+
+    fn decode_once(bf: &mut BytesMut) -> RSocketResult<Option<String>> {
+        if bf.is_empty() {
+            return Ok(None);
+        }
+        let size = bf.get_u8() as usize;
+        if bf.len() < size {
+            return Err(RSocketError::from("require more bytes!"));
+        }
+        let tag = String::from_utf8(bf.split_to(size).to_vec()).unwrap();
+        Ok(Some(tag))
+    }
+
+    pub fn get_tags(&self) -> &Vec<String> {
+        &self.tags
+    }
+
+    pub fn write_to(&self, bf: &mut BytesMut) {
+        for tag in &self.tags {
+            let size = tag.len() as u8;
+            bf.put_u8(size);
+            bf.put_slice(tag.as_bytes());
+        }
+    }
+
+    pub fn bytes(&self) -> Bytes {
+        let mut bf = BytesMut::new();
+        self.write_to(&mut bf);
+        bf.freeze()
+    }
+}
+
 impl CompositeMetadata {
     pub fn new(mime: String, payload: Bytes) -> CompositeMetadata {
         if mime.len() > MAX_MIME_LEN {
@@ -50,24 +129,18 @@ impl CompositeMetadata {
             // Bad
             let mime_len = first as usize;
             if bs.len() < mime_len {
-                return Err(RSocketError::from(ErrorKind::WithDescription(
-                    "bad COMPOSITE_METADATA bytes: missing required bytes!",
-                )));
+                return Err(RSocketError::from("broken COMPOSITE_METADATA bytes!"));
             }
             let front = bs.split_to(mime_len);
             String::from_utf8(front.to_vec()).unwrap()
         };
 
         if bs.len() < 3 {
-            return Err(RSocketError::from(ErrorKind::WithDescription(
-                "bad COMPOSITE_METADATA bytes: missing required bytes!",
-            )));
+            return Err(RSocketError::from("broken COMPOSITE_METADATA bytes!"));
         }
         let payload_size = U24::read_advance(bs) as usize;
         if bs.len() < payload_size {
-            return Err(RSocketError::from(ErrorKind::WithDescription(
-                "bad COMPOSITE_METADATA bytes: missing required bytes!",
-            )));
+            return Err(RSocketError::from("broken COMPOSITE_METADATA bytes!"));
         }
         let p = bs.split_to(payload_size).freeze();
         Ok(Some(CompositeMetadata::new(m, p)))

src/payload/mod.rs

@@ -1,5 +1,5 @@
 mod default;
 mod setup;
 
-pub use default::{Payload,PayloadBuilder};
-pub use setup::{SetupPayload,SetupPayloadBuilder};
+pub use default::{Payload, PayloadBuilder};
+pub use setup::{SetupPayload, SetupPayloadBuilder};

src/spi.rs

@@ -7,6 +7,7 @@ use futures::{Sink, SinkExt, Stream, StreamExt};
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
+use tokio::sync::mpsc;
 
 // TODO: switch to reactor-rust.
 pub type Mono<T> = Pin<Box<dyn Send + Sync + Future<Output = RSocketResult<T>>>>;
@@ -17,32 +18,45 @@ pub trait RSocket: Sync + Send {
     fn fire_and_forget(&self, req: Payload) -> Mono<()>;
     fn request_response(&self, req: Payload) -> Mono<Payload>;
     fn request_stream(&self, req: Payload) -> Flux<Payload>;
-    // fn request_channel(&self, reqs: Flux<Payload>) -> Flux<Payload>;
+    fn request_channel(&self, reqs: Flux<Payload>) -> Flux<Payload>;
 }
 
 pub struct EchoRSocket;
 
 impl RSocket for EchoRSocket {
     fn metadata_push(&self, req: Payload) -> Mono<()> {
-        info!("echo metadata_push: {:?}", req);
+        info!("{:?}", req);
         Box::pin(future::ok::<(), RSocketError>(()))
     }
     fn fire_and_forget(&self, req: Payload) -> Mono<()> {
-        info!("echo fire_and_forget: {:?}", req);
+        info!("{:?}", req);
         Box::pin(future::ok::<(), RSocketError>(()))
     }
     fn request_response(&self, req: Payload) -> Mono<Payload> {
-        info!("echo request_response: {:?}", req);
+        info!("{:?}", req);
         Box::pin(future::ok::<Payload, RSocketError>(req))
     }
     fn request_stream(&self, req: Payload) -> Flux<Payload> {
-        info!("echo request_stream: {:?}", req);
+        info!("{:?}", req);
         Box::pin(futures::stream::iter(vec![
             Ok(req.clone()),
             Ok(req.clone()),
             Ok(req),
         ]))
     }
+    fn request_channel(&self, mut reqs: Flux<Payload>) -> Flux<Payload> {
+        let (sender, receiver) = mpsc::unbounded_channel::<RSocketResult<Payload>>();
+        tokio::spawn(async move {
+            while let Some(it) = reqs.next().await {
+                let pa = it.unwrap();
+                info!("{:?}", pa);
+                sender.send(Ok(pa)).unwrap();
+            }
+        });
+        Box::pin(receiver)
+        // or returns directly
+        // reqs
+    }
 }
 
 pub struct EmptyRSocket;
@@ -66,8 +80,11 @@ impl RSocket for EmptyRSocket {
         Box::pin(future::err(self.must_failed()))
     }
 
-    fn request_stream(&self, req: Payload) -> Flux<Payload> {
-        Box::pin(futures::stream::empty())
+    fn request_stream(&self, _req: Payload) -> Flux<Payload> {
+        Box::pin(futures::stream::iter(vec![Err(self.must_failed())]))
+    }
+    fn request_channel(&self, _reqs: Flux<Payload>) -> Flux<Payload> {
+        Box::pin(futures::stream::iter(vec![Err(self.must_failed())]))
     }
 }
 

src/transport/misc.rs

@@ -9,7 +9,7 @@ use std::{
     collections::HashMap,
     future::Future,
     sync::{
-        atomic::{AtomicU32, Ordering},
+        atomic::{AtomicI64, AtomicU32, Ordering},
         Arc, Mutex, RwLock,
     },
 };
@@ -37,3 +37,30 @@ impl From<u32> for StreamID {
         StreamID::new(v)
     }
 }
+
+#[derive(Debug, Clone)]
+pub(crate) struct Counter {
+    inner: Arc<AtomicI64>,
+}
+
+impl Counter {
+    pub(crate) fn new(value: i64) -> Counter {
+        Counter {
+            inner: Arc::new(AtomicI64::new(value)),
+        }
+    }
+
+    pub(crate) fn count_down(&self) -> i64 {
+        let c = self.inner.clone();
+        c.fetch_add(-1, Ordering::SeqCst)
+    }
+}
+
+#[inline]
+pub(crate) fn debug_frame(snd: bool, f: &frame::Frame) {
+    if snd {
+        debug!("===> SND: {:?}", f);
+    } else {
+        debug!("<=== RCV: {:?}", f);
+    }
+}