support server-side request_channel.

jjeffcaii · jjeffcaii · commit 566a7702a1e9 · 2019-12-04T10:39:33.000+08:00
diff --git a/Cargo.toml b/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"
@@ -15,7 +15,7 @@ log = "0.4.8"
 bytes = "0.5.2"
 futures = "0.3.1"
 lazy_static = "1.4.0"
-reactor_rs = {git = "https://github.com/jjeffcaii/reactor-rust", branch = "develop"}
+# reactor_rs = {git = "https://github.com/jjeffcaii/reactor-rust", branch = "develop"}
 
 [dependencies.tokio]
 version = "0.2.2"
diff --git a/README.md b/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
diff --git a/src/spi.rs b/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>>>>;
@@ -24,28 +25,37 @@ 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, reqs: Flux<Payload>) -> Flux<Payload> {
-        info!("echo request_channel");
-        reqs
+    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
     }
 }
 
@@ -70,10 +80,10 @@ impl RSocket for EmptyRSocket {
         Box::pin(future::err(self.must_failed()))
     }
 
-    fn request_stream(&self, req: Payload) -> Flux<Payload> {
+    fn request_stream(&self, _req: Payload) -> Flux<Payload> {
         Box::pin(futures::stream::iter(vec![Err(self.must_failed())]))
     }
-    fn request_channel(&self, mut reqs: Flux<Payload>) -> Flux<Payload> {
+    fn request_channel(&self, _reqs: Flux<Payload>) -> Flux<Payload> {
         Box::pin(futures::stream::iter(vec![Err(self.must_failed())]))
     }
 }
diff --git a/src/transport/misc.rs b/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);
+    }
+}
diff --git a/src/transport/socket.rs b/src/transport/socket.rs
@@ -1,4 +1,4 @@
-use super::misc::StreamID;
+use super::misc::{self, Counter, StreamID};
 use super::spi::{Rx, Transport, Tx};
 use crate::errors::{ErrorKind, RSocketError};
 use crate::frame::{self, Body, Frame};
@@ -12,6 +12,7 @@ use std::env;
 use std::error::Error;
 use std::future::Future;
 use std::pin::Pin;
+use std::ptr;
 use std::sync::{Arc, Mutex, RwLock};
 use tokio::net::TcpListener;
 use tokio::prelude::*;
@@ -33,12 +34,13 @@ struct Responder {
 
 type Single = oneshot::Sender<RSocketResult<Payload>>;
 type Multi = mpsc::UnboundedSender<RSocketResult<Payload>>;
+type MultiReceiver = mpsc::UnboundedReceiver<RSocketResult<Payload>>;
 
 #[derive(Debug)]
 enum Handler {
     Request(Single),
     Stream(Multi),
-    Channel(Multi),
+    Channel((Multi, Counter)),
 }
 
 #[derive(Debug)]
@@ -95,7 +97,7 @@ impl DuplexSocket {
         while let Some(msg) = rx.recv().await {
             let sid = msg.get_stream_id();
             let flag = msg.get_flag();
-            debug!("<--- RCV: {:?}", msg);
+            misc::debug_frame(false, &msg);
             match msg.get_body() {
                 Body::Setup(v) => self.on_setup(&acceptor, sid, flag, SetupPayload::from(v)),
                 Body::Resume(v) => {
@@ -122,6 +124,8 @@ impl DuplexSocket {
                 }
                 Body::RequestChannel(v) => {
                     // TODO: support channel
+                    let input = Payload::from(v);
+                    self.on_request_channel(sid, flag, input).await;
                 }
                 Body::Payload(v) => {
                     let input = Payload::from(v);
@@ -159,25 +163,19 @@ impl DuplexSocket {
             Handler::Request(sender) => sender.send(Ok(input)).unwrap(),
             Handler::Stream(sender) => {
                 if flag & frame::FLAG_NEXT != 0 {
-                    sender.clone().send(Ok(input)).unwrap();
+                    sender.send(Ok(input)).unwrap();
                 }
-                if flag & frame::FLAG_COMPLETE != 0 {
-                    // steam end
-                    drop(sender);
-                } else {
+                if flag & frame::FLAG_COMPLETE == 0 {
                     senders.insert(sid, Handler::Stream(sender));
                 }
             }
-            Handler::Channel(sender) => {
+            Handler::Channel((sender, cdl)) => {
                 // TODO: support channel
                 if flag & frame::FLAG_NEXT != 0 {
-                    sender.clone().send(Ok(input)).unwrap();
+                    sender.send(Ok(input)).unwrap();
                 }
-                if flag & frame::FLAG_COMPLETE != 0 {
-                    // steam end
-                    drop(sender);
-                } else {
-                    senders.insert(sid, Handler::Stream(sender));
+                if flag & frame::FLAG_COMPLETE == 0 {
+                    senders.insert(sid, Handler::Channel((sender, cdl)));
                 }
             }
         };
@@ -256,6 +254,39 @@ impl DuplexSocket {
         });
     }
 
+    #[inline]
+    async fn on_request_channel(&self, sid: u32, flag: u16, first: Payload) {
+        let responder = self.responder.clone();
+        let tx = self.tx.clone();
+        let (sender, receiver) = mpsc::unbounded_channel::<RSocketResult<Payload>>();
+        sender.send(Ok(first)).unwrap();
+        let cdl = Counter::new(2);
+        self.register_handler(sid, Handler::Channel((sender, cdl.clone())));
+        tokio::spawn(async move {
+            // respond client channel
+            let inputs: Flux<Payload> = Box::pin(receiver);
+            let mut outputs = responder.request_channel(inputs);
+            // TODO: support custom RequestN.
+            let request_n = frame::RequestN::builder(sid, 0).build();
+            tx.send(request_n).unwrap();
+
+            while let Some(v) = outputs.next().await {
+                let (d, m) = v.unwrap().split();
+                let mut bu = frame::Payload::builder(sid, frame::FLAG_NEXT);
+                if let Some(b) = d {
+                    bu = bu.set_data(b);
+                }
+                if let Some(b) = m {
+                    bu = bu.set_metadata(b);
+                }
+                let sending = bu.build();
+                tx.send(sending).unwrap();
+            }
+            let complete = frame::Payload::builder(sid, frame::FLAG_COMPLETE).build();
+            tx.send(complete).unwrap();
+        });
+    }
+
     #[inline]
     async fn on_metadata_push(&self, input: Payload) {
         if let Err(e) = self.responder.clone().metadata_push(input).await {
@@ -378,7 +409,8 @@ impl RSocket for DuplexSocket {
         let tx = self.tx.clone();
         // register handler
         let (sender, receiver) = mpsc::unbounded_channel::<RSocketResult<Payload>>();
-        self.register_handler(sid, Handler::Channel(sender));
+        let cdl = Counter::new(2);
+        self.register_handler(sid, Handler::Channel((sender, cdl.clone())));
         tokio::spawn(async move {
             let mut sent: u64 = 0;
             while let Some(it) = reqs.next().await {
@@ -406,6 +438,7 @@ impl RSocket for DuplexSocket {
                 };
                 tx.send(sending).unwrap();
             }
+            cdl.count_down();
             let sending = frame::Payload::builder(sid, frame::FLAG_COMPLETE).build();
             tx.send(sending).unwrap();
         });
diff --git a/src/transport/tcp.rs b/src/transport/tcp.rs
@@ -1,4 +1,5 @@
 use super::codec::RFrameCodec;
+use super::misc;
 use super::spi::{Rx, Tx};
 use crate::frame::Frame;
 use futures::{Sink, SinkExt, Stream, StreamExt};
@@ -30,6 +31,7 @@ pub async fn process(socket: TcpStream, mut inputs: Rx, outputs: Tx) {
   });
   // loop write
   while let Some(it) = inputs.recv().await {
+    misc::debug_frame(true, &it);
     writer.send(it).await.unwrap()
   }
 }
diff --git a/tests/clients.rs b/tests/clients.rs
@@ -17,10 +17,10 @@ async fn test_client() {
         .start()
         .await
         .unwrap();
-    // exec_metadata_push(&cli).await;
-    // exec_fire_and_forget(&cli).await;
-    // exec_request_response(&cli).await;
-    // exec_request_stream(&cli).await;
+    exec_metadata_push(&cli).await;
+    exec_fire_and_forget(&cli).await;
+    exec_request_response(&cli).await;
+    exec_request_stream(&cli).await;
     exec_request_channel(&cli).await;
     cli.close();
 }

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`use super::codec::RFrameCodec;`
	`2`	`+use super::misc;`
`2`	`3`	`use super::spi::{Rx, Tx};`
`3`	`4`	`use crate::frame::Frame;`
`4`	`5`	`use futures::{Sink, SinkExt, Stream, StreamExt};`
`@@ -30,6 +31,7 @@ pub async fn process(socket: TcpStream, mut inputs: Rx, outputs: Tx) {`
`30`	`31`	`});`
`31`	`32`	`// loop write`
`32`	`33`	`while let Some(it) = inputs.recv().await {`
	`34`	`+ misc::debug_frame(true, &it);`
`33`	`35`	`writer.send(it).await.unwrap()`
`34`	`36`	`}`
`35`	`37`	`}`