feat(request_response): handle empty response correctly

jjeffcaii · jjeffcaii · commit 5e4b5b2eb650 · 2020-12-31T00:06:12.000+08:00
diff --git a/rsocket-messaging/src/requester.rs b/rsocket-messaging/src/requester.rs
@@ -15,7 +15,7 @@ use url::Url;
 type FnMetadata = Box<dyn FnMut() -> Result<(MimeType, Vec<u8>)>>;
 type FnData = Box<dyn FnMut(&MimeType) -> Result<Vec<u8>>>;
 type PreflightResult = Result<(Payload, MimeType, Arc<Box<dyn RSocket>>)>;
-type UnpackerResult = Result<(MimeType, Payload)>;
+type UnpackerResult = Result<(MimeType, Option<Payload>)>;
 type UnpackersResult = Result<(MimeType, Flux<Result<Payload>>)>;
 
 enum TransportKind {
@@ -352,8 +352,11 @@ impl Unpacker {
         T: Sized + DeserializeOwned,
     {
         let (mime_type, inner) = self.inner?;
-        match inner.data() {
-            Some(raw) => do_unmarshal(&mime_type, raw),
+        match inner {
+            Some(it) => match it.data() {
+                Some(raw) => do_unmarshal(&mime_type, raw),
+                None => Ok(None),
+            },
             None => Ok(None),
         }
     }
diff --git a/rsocket-transport-wasm/src/misc.rs b/rsocket-transport-wasm/src/misc.rs
@@ -73,10 +73,11 @@ impl JsClient {
         let request: JsPayload = request.into_serde().unwrap();
         future_to_promise(async move {
             match inner.request_response(request.into()).await {
-                Ok(v) => {
+                Ok(Some(v)) => {
                     let jp = JsPayload::from(v);
                     Ok((&jp).into())
                 }
+                Ok(None) => Ok(JsValue::UNDEFINED),
                 Err(e) => Err(JsValue::from(&format!("{:?}", e))),
             }
         })
diff --git a/rsocket/src/core/client.rs b/rsocket/src/core/client.rs
@@ -217,7 +217,7 @@ impl RSocket for Client {
         self.socket.fire_and_forget(req).await
     }
 
-    async fn request_response(&self, req: Payload) -> Result<Payload> {
+    async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {
         self.socket.request_response(req).await
     }
 
diff --git a/rsocket/src/spi.rs b/rsocket/src/spi.rs
@@ -18,7 +18,7 @@ pub type Flux<T> = Pin<Box<dyn Send + Stream<Item = T>>>;
 pub trait RSocket: Sync + Send {
     async fn metadata_push(&self, req: Payload) -> Result<()>;
     async fn fire_and_forget(&self, req: Payload) -> Result<()>;
-    async fn request_response(&self, req: Payload) -> Result<Payload>;
+    async fn request_response(&self, req: Payload) -> Result<Option<Payload>>;
     fn request_stream(&self, req: Payload) -> Flux<Result<Payload>>;
     fn request_channel(&self, reqs: Flux<Result<Payload>>) -> Flux<Result<Payload>>;
 }
diff --git a/rsocket/src/transport/socket.rs b/rsocket/src/transport/socket.rs
@@ -36,7 +36,7 @@ struct Responder {
 
 #[derive(Debug)]
 enum Handler {
-    ReqRR(oneshot::Sender<Result<Payload>>),
+    ReqRR(oneshot::Sender<Result<Option<Payload>>>),
     ResRR(Counter),
     ReqRS(mpsc::Sender<Result<Payload>>),
     ReqRC(mpsc::Sender<Result<Payload>>),
@@ -261,12 +261,12 @@ impl DuplexSocket {
         // pick handler
         if let Some((_, handler)) = self.handlers.remove(&sid) {
             let desc = input.get_data_utf8().unwrap().to_owned();
-            let e: Result<_> = Err(RSocketError::must_new_from_code(input.get_code(), desc).into());
+            let e = RSocketError::must_new_from_code(input.get_code(), desc);
             match handler {
-                Handler::ReqRR(tx) => tx.send(e).expect("Send RR failed"),
+                Handler::ReqRR(tx) => tx.send(Err(e.into())).expect("Send RR failed"),
                 Handler::ResRR(_) => unreachable!(),
-                Handler::ReqRS(tx) => tx.send(e).await.expect("Send RS failed"),
-                Handler::ReqRC(tx) => tx.send(e).await.expect("Send RC failed"),
+                Handler::ReqRS(tx) => tx.send(Err(e.into())).await.expect("Send RS failed"),
+                Handler::ReqRC(tx) => tx.send(Err(e.into())).await.expect("Send RC failed"),
             }
         }
     }
@@ -303,7 +303,11 @@ impl DuplexSocket {
                 match o.get() {
                     Handler::ReqRR(_) => match o.remove() {
                         Handler::ReqRR(sender) => {
-                            sender.send(Ok(input)).unwrap();
+                            if flag & Frame::FLAG_NEXT != 0 {
+                                sender.send(Ok(Some(input))).unwrap();
+                            } else {
+                                sender.send(Ok(None)).unwrap();
+                            }
                         }
                         _ => unreachable!(),
                     },
@@ -394,7 +398,7 @@ impl DuplexSocket {
             canceller.send(sid).await.expect("Send canceller failed");
 
             match result {
-                Ok(res) => {
+                Ok(Some(res)) => {
                     Self::try_send_payload(
                         &splitter,
                         &mut tx,
@@ -404,6 +408,9 @@ impl DuplexSocket {
                     )
                     .await;
                 }
+                Ok(None) => {
+                    Self::try_send_complete(&mut tx, sid, Frame::FLAG_COMPLETE).await;
+                }
                 Err(e) => {
                     let sending = frame::Error::builder(sid, 0)
                         .set_code(error::ERR_APPLICATION)
@@ -570,6 +577,14 @@ impl DuplexSocket {
         }
     }
 
+    #[inline]
+    async fn try_send_complete(tx: &mut mpsc::Sender<Frame>, sid: u32, flag: u16) {
+        let sending = frame::Payload::builder(sid, flag).build();
+        if let Err(e) = tx.send(sending).await {
+            error!("respond failed: {}", e);
+        }
+    }
+
     #[inline]
     async fn try_send_payload(
         splitter: &Option<Splitter>,
@@ -697,8 +712,8 @@ impl RSocket for DuplexSocket {
         Ok(())
     }
 
-    async fn request_response(&self, req: Payload) -> Result<Payload> {
-        let (tx, rx) = oneshot::channel::<Result<Payload>>();
+    async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {
+        let (tx, rx) = oneshot::channel::<Result<Option<Payload>>>();
         let sid = self.seq.next();
         let handlers = self.handlers.clone();
         let sender = self.tx.clone();
@@ -911,7 +926,7 @@ impl RSocket for Responder {
         (*inner).fire_and_forget(req).await
     }
 
-    async fn request_response(&self, req: Payload) -> Result<Payload> {
+    async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {
         let inner = self.inner.read().await;
         (*inner).request_response(req).await
     }
diff --git a/rsocket/src/utils.rs b/rsocket/src/utils.rs
@@ -26,9 +26,9 @@ impl RSocket for EchoRSocket {
         Ok(())
     }
 
-    async fn request_response(&self, req: Payload) -> Result<Payload> {
+    async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {
         info!("{:?}", req);
-        Ok(req)
+        Ok(Some(req))
     }
 
     fn request_stream(&self, req: Payload) -> Flux<Result<Payload>> {
@@ -67,7 +67,7 @@ impl RSocket for EmptyRSocket {
         Err(RSocketError::ApplicationException("UNIMPLEMENT".into()).into())
     }
 
-    async fn request_response(&self, _req: Payload) -> Result<Payload> {
+    async fn request_response(&self, _req: Payload) -> Result<Option<Payload>> {
         Err(RSocketError::ApplicationException("UNIMPLEMENT".into()).into())
     }
 

Original file line number	Diff line number	Diff line change
`@@ -73,10 +73,11 @@ impl JsClient {`
`73`	`73`	`let request: JsPayload = request.into_serde().unwrap();`
`74`	`74`	`future_to_promise(async move {`
`75`	`75`	`match inner.request_response(request.into()).await {`
`76`		`- Ok(v) => {`
	`76`	`+ Ok(Some(v)) => {`
`77`	`77`	`let jp = JsPayload::from(v);`
`78`	`78`	`Ok((&jp).into())`
`79`	`79`	`}`
	`80`	`+ Ok(None) => Ok(JsValue::UNDEFINED),`
`80`	`81`	`Err(e) => Err(JsValue::from(&format!("{:?}", e))),`
`81`	`82`	`}`
`82`	`83`	`})`
Original file line number	Diff line number	Diff line change
`@@ -217,7 +217,7 @@ impl RSocket for Client {`
`217`	`217`	`self.socket.fire_and_forget(req).await`
`218`	`218`	`}`
`219`	`219`
`220`		`- async fn request_response(&self, req: Payload) -> Result<Payload> {`
	`220`	`+ async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {`
`221`	`221`	`self.socket.request_response(req).await`
`222`	`222`	`}`
`223`	`223`
Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,7 @@ pub type Flux<T> = Pin<Box<dyn Send + Stream<Item = T>>>;`
`18`	`18`	`pub trait RSocket: Sync + Send {`
`19`	`19`	`async fn metadata_push(&self, req: Payload) -> Result<()>;`
`20`	`20`	`async fn fire_and_forget(&self, req: Payload) -> Result<()>;`
`21`		`- async fn request_response(&self, req: Payload) -> Result<Payload>;`
	`21`	`+ async fn request_response(&self, req: Payload) -> Result<Option<Payload>>;`
`22`	`22`	`fn request_stream(&self, req: Payload) -> Flux<Result<Payload>>;`
`23`	`23`	`fn request_channel(&self, reqs: Flux<Result<Payload>>) -> Flux<Result<Payload>>;`
`24`	`24`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,9 +26,9 @@ impl RSocket for EchoRSocket {`
`26`	`26`	`Ok(())`
`27`	`27`	`}`
`28`	`28`
`29`		`- async fn request_response(&self, req: Payload) -> Result<Payload> {`
	`29`	`+ async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {`
`30`	`30`	`info!("{:?}", req);`
`31`		`- Ok(req)`
	`31`	`+ Ok(Some(req))`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`fn request_stream(&self, req: Payload) -> Flux<Result<Payload>> {`
`@@ -67,7 +67,7 @@ impl RSocket for EmptyRSocket {`
`67`	`67`	`Err(RSocketError::ApplicationException("UNIMPLEMENT".into()).into())`
`68`	`68`	`}`
`69`	`69`
`70`		`- async fn request_response(&self, _req: Payload) -> Result<Payload> {`
	`70`	`+ async fn request_response(&self, _req: Payload) -> Result<Option<Payload>> {`
`71`	`71`	`Err(RSocketError::ApplicationException("UNIMPLEMENT".into()).into())`
`72`	`72`	`}`
`73`	`73`