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 <jjeffcaii@outlook.com>"]
 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);
+ }
+}