Support flux result for messaging requester.

Author: jjeffcaii

rsocket-messaging/src/requester.rs

@@ -19,6 +19,8 @@ use url::Url;
 type FnMetadata = Box<dyn FnMut() -> Result<(MimeType, Vec<u8>), Box<dyn Error>>>;
 type FnData = Box<dyn FnMut(&MimeType) -> Result<Vec<u8>, Box<dyn Error>>>;
 type PreflightResult = Result<(Payload, MimeType, Arc<Box<dyn RSocket>>), Box<dyn Error>>;
+type UnpackerResult = Result<(MimeType, Payload), RSocketError>;
+type UnpackersResult = Result<(MimeType, Flux<Result<Payload, RSocketError>>), Box<dyn Error>>;
 
 enum TransportKind {
     TCP(String, u16),
@@ -44,6 +46,14 @@ pub struct RequesterBuilder {
     tp: Option<TransportKind>,
 }
 
+pub struct Unpackers {
+    inner: UnpackersResult,
+}
+
+pub struct Unpacker {
+    inner: UnpackerResult,
+}
+
 impl Default for RequesterBuilder {
     fn default() -> Self {
         Self {
@@ -263,6 +273,16 @@ impl RequestSpec {
         self
     }
 
+    pub async fn retrieve(self) -> Result<(), Box<dyn Error>> {
+        match self.preflight() {
+            Ok((req, _mime_type, rsocket)) => {
+                rsocket.fire_and_forget(req).await;
+                Ok(())
+            }
+            Err(e) => Err(e),
+        }
+    }
+
     pub async fn retrieve_mono(self) -> Unpacker {
         match self.preflight() {
             Ok((req, mime_type, rsocket)) => {
@@ -284,6 +304,18 @@ impl RequestSpec {
         }
     }
 
+    pub fn retrieve_flux(self) -> Unpackers {
+        match self.preflight() {
+            Ok((req, mime_type, rsocket)) => {
+                let results = rsocket.request_stream(req);
+                Unpackers {
+                    inner: Ok((mime_type, results)),
+                }
+            }
+            Err(e) => Unpackers { inner: Err(e) },
+        }
+    }
+
     #[inline]
     fn preflight(self) -> PreflightResult {
         let mut b = BytesMut::new();
@@ -304,36 +336,83 @@ impl RequestSpec {
     }
 }
 
-pub struct Unpacker {
-    inner: Result<(MimeType, Payload), RSocketError>,
+impl Unpackers {
+    pub async fn block<T>(self) -> Result<Vec<T>, Box<dyn Error>>
+    where
+        T: Sized + DeserializeOwned,
+    {
+        let (mime_type, mut results) = self.inner?;
+        let mut res = Vec::new();
+        while let Some(next) = results.next().await {
+            match next {
+                Ok(v) => {
+                    if let Some(data) = v.data() {
+                        let t = do_unmarshal::<T>(&mime_type, data)?;
+                        if let Some(t) = t {
+                            res.push(t);
+                        }
+                    }
+                }
+                Err(e) => return Err(format!("{}", e).into()),
+            }
+        }
+        Ok(res)
+    }
+    pub async fn foreach<T>(self, callback: impl Fn(T)) -> Result<(), Box<dyn Error>>
+    where
+        T: Sized + DeserializeOwned,
+    {
+        let (mime_type, mut results) = self.inner?;
+        while let Some(next) = results.next().await {
+            match next {
+                Ok(v) => {
+                    if let Some(data) = v.data() {
+                        let t = do_unmarshal::<T>(&mime_type, data)?;
+                        if let Some(t) = t {
+                            callback(t);
+                        }
+                    }
+                }
+                Err(e) => return Err(format!("{}", e).into()),
+            }
+        }
+        Ok(())
+    }
 }
 
 impl Unpacker {
-    pub fn block<T>(&self) -> Result<Option<T>, Box<dyn Error>>
+    pub fn block<T>(self) -> Result<Option<T>, Box<dyn Error>>
     where
         T: Sized + DeserializeOwned,
     {
-        match &self.inner {
+        match self.inner {
             Ok((mime_type, inner)) => match inner.data() {
                 // TODO: support more mime types.
-                Some(raw) => match *mime_type {
-                    MimeType::APPLICATION_JSON => {
-                        let t = unmarshal(misc::json(), &raw.as_ref())?;
-                        Ok(Some(t))
-                    }
-                    MimeType::APPLICATION_CBOR => {
-                        let t = unmarshal(misc::cbor(), &raw.as_ref())?;
-                        Ok(Some(t))
-                    }
-                    _ => Err("unsupported mime type!".into()),
-                },
+                Some(raw) => do_unmarshal(&mime_type, raw),
                 None => Ok(None),
             },
             Err(e) => Err(format!("{}", e).into()),
         }
     }
 }
 
+fn do_unmarshal<T>(mime_type: &MimeType, raw: &Bytes) -> Result<Option<T>, Box<dyn Error>>
+where
+    T: Sized + DeserializeOwned,
+{
+    match *mime_type {
+        MimeType::APPLICATION_JSON => {
+            let t = unmarshal(misc::json(), &raw.as_ref())?;
+            Ok(Some(t))
+        }
+        MimeType::APPLICATION_CBOR => {
+            let t = unmarshal(misc::cbor(), &raw.as_ref())?;
+            Ok(Some(t))
+        }
+        _ => Err("unsupported mime type!".into()),
+    }
+}
+
 fn do_marshal<T>(mime_type: &MimeType, data: &T) -> Result<Vec<u8>, Box<dyn Error>>
 where
     T: Sized + Serialize,

rsocket-test/tests/test_messaging.rs

@@ -19,7 +19,7 @@ pub struct Token {
     access: String,
 }
 
-#[derive(Serialize, Deserialize, Debug, Default)]
+#[derive(Serialize, Deserialize, Debug, Default, Clone)]
 pub struct Student {
     id: i64,
     name: String,
@@ -56,21 +56,45 @@ async fn test_messaging() {
         .await
         .expect("Connect failed!");
 
-    let post = Student {
-        id: 1234,
-        name: "Jeffsky".to_owned(),
-        birth: "2020-01-01".to_owned(),
-    };
-
+    // TEST MONO BEGIN
     let res: Response<Student> = requester
         .route("student.v1.upsert")
         .metadata(Tracing::default(), "application/json")
         .metadata_raw("foobar", "message/x.rsocket.authentication.bearer.v0")
-        .data(post)
+        .data(next_post())
         .retrieve_mono()
         .await
         .block()
         .expect("Retrieve failed!")
         .expect("Empty result!");
     info!("------> RESPONSE: {:?}", res);
+
+    // TEST FLUX BEGIN
+    let res: Vec<Student> = requester
+        .route("students.v1")
+        .data(next_post())
+        .retrieve_flux()
+        .block()
+        .await
+        .expect("Retrieve failed!");
+    for it in res.iter() {
+        info!("===> NEXT: {:?}", it);
+    }
+    requester
+        .route("students.v1")
+        .data(next_post())
+        .retrieve_flux()
+        .foreach(|it: Student| {
+            info!("===> FOREACH: {:?}", it);
+        })
+        .await
+        .expect("Retrieve failed!");
+}
+
+fn next_post() -> Student {
+    Student {
+        id: 1234,
+        name: "Jeffsky".to_owned(),
+        birth: "2020-01-01".to_owned(),
+    }
 }