Change messaging RequestSpec to fluent api.

Author: jjeffcaii

rsocket-messaging/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 log = "0.4.8"
 futures = "0.3.5"
 bytes = "0.5.4"
+lazy_static = "1.4.0"
 serde = "1.0.110"
 serde_json = "1.0.53"
 serde_cbor = "0.11.1"

rsocket-messaging/src/lib.rs

@@ -1,7 +1,10 @@
 #[macro_use]
 extern crate log;
+// #[macro_use]
+// extern crate lazy_static;
 
 mod misc;
 mod requester;
 
+pub use misc::{cbor, json, SerDe};
 pub use requester::{RequestSpec, Requester, RequesterBuilder};

rsocket-messaging/src/misc.rs

@@ -1,38 +1,73 @@
-use bytes::{BufMut, BytesMut};
-use rsocket_rust::extension::MimeType;
-use serde::{Deserialize, Serialize};
+use serde::{de::DeserializeOwned, Serialize};
 use std::error::Error;
 
-pub(crate) fn unmarshal<'a, T>(mime_type: &MimeType, raw: &'a [u8]) -> Result<T, Box<dyn Error>>
-where
-    T: Deserialize<'a>,
-{
-    match *mime_type {
-        MimeType::APPLICATION_JSON => Ok(serde_json::from_slice(raw)?),
-        MimeType::APPLICATION_CBOR => Ok(serde_cbor::from_slice(raw)?),
-        _ => panic!(""),
+pub trait SerDe {
+    fn marshal<T>(&self, data: &T) -> Result<Vec<u8>, Box<dyn Error>>
+    where
+        Self: Sized,
+        T: Sized + Serialize;
+
+    fn unmarshal<T>(&self, raw: &[u8]) -> Result<T, Box<dyn Error>>
+    where
+        Self: Sized,
+        T: Sized + DeserializeOwned;
+}
+
+#[derive(Default)]
+struct JsonSerDe {}
+
+impl SerDe for JsonSerDe {
+    fn marshal<T>(&self, data: &T) -> Result<Vec<u8>, Box<dyn Error>>
+    where
+        T: Sized + Serialize,
+    {
+        Ok(serde_json::to_vec(data)?)
     }
+
+    fn unmarshal<T>(&self, raw: &[u8]) -> Result<T, Box<dyn Error>>
+    where
+        T: Sized + DeserializeOwned,
+    {
+        Ok(serde_json::from_slice(raw)?)
+    }
+}
+
+pub fn json() -> impl SerDe {
+    JsonSerDe {}
+}
+
+pub fn cbor() -> impl SerDe {
+    CborSerDe {}
 }
 
-pub(crate) fn marshal<T>(
-    mime_type: &MimeType,
-    bf: &mut BytesMut,
-    data: &T,
-) -> Result<(), Box<dyn Error>>
+struct CborSerDe {}
+
+impl SerDe for CborSerDe {
+    fn marshal<T>(&self, data: &T) -> Result<Vec<u8>, Box<dyn Error>>
+    where
+        T: Sized + Serialize,
+    {
+        Ok(serde_cbor::to_vec(data)?)
+    }
+
+    fn unmarshal<T>(&self, raw: &[u8]) -> Result<T, Box<dyn Error>>
+    where
+        T: Sized + DeserializeOwned,
+    {
+        Ok(serde_cbor::from_slice(raw)?)
+    }
+}
+
+pub(crate) fn marshal<T>(ser: impl SerDe, data: &T) -> Result<Vec<u8>, Box<dyn Error>>
 where
     T: Sized + Serialize,
 {
-    match *mime_type {
-        MimeType::APPLICATION_JSON => {
-            let raw = serde_json::to_vec(data)?;
-            bf.put_slice(&raw[..]);
-            Ok(())
-        }
-        MimeType::APPLICATION_CBOR => {
-            let raw = serde_cbor::to_vec(data)?;
-            bf.put_slice(&raw[..]);
-            Ok(())
-        }
-        _ => panic!(""),
-    }
+    ser.marshal(data)
+}
+
+pub(crate) fn unmarshal<T>(de: impl SerDe, raw: &[u8]) -> Result<T, Box<dyn Error>>
+where
+    T: Sized + DeserializeOwned,
+{
+    de.unmarshal(raw)
 }

rsocket-messaging/src/requester.rs

@@ -1,5 +1,5 @@
-use super::misc::{marshal, unmarshal};
-use bytes::BytesMut;
+use super::misc::{self, marshal, unmarshal};
+use bytes::{Bytes, BytesMut};
 use rsocket_rust::error::RSocketError;
 use rsocket_rust::extension::{
     CompositeMetadata, CompositeMetadataEntry, MimeType, RoutingMetadata,
@@ -8,14 +8,18 @@ use rsocket_rust::prelude::*;
 use rsocket_rust::utils::Writeable;
 use rsocket_rust_transport_tcp::TcpClientTransport;
 use rsocket_rust_transport_websocket::WebsocketClientTransport;
-use serde::{Deserialize, Serialize};
+use serde::{de::DeserializeOwned, Serialize};
 use std::collections::LinkedList;
 use std::error::Error;
 use std::net::SocketAddr;
 use std::result::Result;
 use std::sync::Arc;
 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>>;
+
 enum TransportKind {
     TCP(String, u16),
     WS(String),
@@ -26,10 +30,10 @@ pub struct Requester {
 }
 
 pub struct RequestSpec {
-    data: Option<Vec<u8>>,
     rsocket: Arc<Box<dyn RSocket>>,
     data_mime_type: MimeType,
-    metadatas: LinkedList<(MimeType, Vec<u8>)>,
+    metadatas: LinkedList<FnMetadata>,
+    data: Option<FnData>,
 }
 
 pub struct RequesterBuilder {
@@ -74,14 +78,13 @@ impl RequesterBuilder {
         D: Sized + Serialize,
     {
         // TODO: lazy set
-        let mut bf = BytesMut::new();
         let result = match &self.data_mime_type {
-            Some(m) => marshal(m, &mut bf, data),
-            None => marshal(&MimeType::APPLICATION_JSON, &mut bf, data),
+            Some(m) => do_marshal(m, data),
+            None => do_marshal(&MimeType::APPLICATION_JSON, data),
         };
         match result {
-            Ok(()) => {
-                self.data = Some(bf.to_vec());
+            Ok(raw) => {
+                self.data = Some(raw);
             }
             Err(e) => {
                 error!("marshal failed: {:?}", e);
@@ -96,11 +99,10 @@ impl RequesterBuilder {
         T: Into<MimeType>,
     {
         // TODO: lazy set
-        let mut bf = BytesMut::new();
         let mime_type = mime_type.into();
-        match marshal(&mime_type, &mut bf, metadata) {
-            Ok(()) => {
-                let entry = CompositeMetadataEntry::new(mime_type, bf.freeze());
+        match do_marshal(&mime_type, metadata) {
+            Ok(raw) => {
+                let entry = CompositeMetadataEntry::new(mime_type, Bytes::from(raw));
                 self.metadata.push(entry);
             }
             Err(e) => error!("marshal failed: {:?}", e),
@@ -203,100 +205,143 @@ impl Requester {
         let mut buf = BytesMut::new();
         routing.write_to(&mut buf);
 
-        let mut metadatas: LinkedList<(MimeType, Vec<u8>)> = Default::default();
-        metadatas.push_back((MimeType::MESSAGE_X_RSOCKET_ROUTING_V0, buf.to_vec()));
+        let mut metadatas: LinkedList<FnMetadata> = LinkedList::new();
+        metadatas.push_back(Box::new(move || {
+            Ok((MimeType::MESSAGE_X_RSOCKET_ROUTING_V0, buf.to_vec()))
+        }));
+
         RequestSpec {
-            data: None,
             rsocket: self.rsocket.clone(),
             data_mime_type: MimeType::APPLICATION_JSON,
             metadatas,
+            data: None,
         }
     }
 }
 
 impl RequestSpec {
-    pub fn metadata<T, M>(&mut self, metadata: &T, mime_type: M) -> Result<(), Box<dyn Error>>
+    pub fn metadata<T, M>(mut self, metadata: T, mime_type: M) -> Self
     where
-        T: Sized + Serialize,
+        T: Sized + Serialize + 'static,
         M: Into<MimeType>,
     {
         let mime_type = mime_type.into();
-        let mut b = BytesMut::new();
-        marshal(&mime_type, &mut b, metadata)?;
-        self.metadatas.push_back((mime_type, b.to_vec()));
-        Ok(())
+        let f: FnMetadata = Box::new(move || {
+            let raw = do_marshal(&mime_type, &metadata)?;
+            Ok((mime_type.clone(), raw))
+        });
+        self.metadatas.push_back(f);
+        self
     }
 
-    pub fn metadata_raw<I, M>(&mut self, metadata: I, mime_type: M) -> Result<(), Box<dyn Error>>
+    pub fn metadata_raw<I, M>(mut self, metadata: I, mime_type: M) -> Self
     where
         I: Into<Vec<u8>>,
         M: Into<MimeType>,
     {
+        let mime_type = mime_type.into();
+        let metadata = metadata.into();
         self.metadatas
-            .push_back((mime_type.into(), metadata.into()));
-        Ok(())
+            .push_back(Box::new(move || Ok((mime_type.clone(), metadata.clone()))));
+        self
     }
 
-    pub fn data<T>(&mut self, data: &T) -> Result<(), Box<dyn Error>>
+    pub fn data<T>(mut self, data: T) -> Self
     where
-        T: Sized + Serialize,
+        T: Sized + Serialize + 'static,
     {
-        let mut bf = BytesMut::new();
-        marshal(&self.data_mime_type, &mut bf, data)?;
-        self.data = Some(bf.to_vec());
-        Ok(())
+        self.data = Some(Box::new(move |m| do_marshal(m, &data)));
+        self
     }
 
-    pub fn data_raw<I>(&mut self, data: I) -> Result<(), Box<dyn Error>>
+    pub fn data_raw<I>(mut self, data: I) -> Self
     where
         I: Into<Vec<u8>>,
     {
-        self.data = Some(data.into());
-        Ok(())
+        let data = data.into();
+        self.data = Some(Box::new(move |_m| Ok(data.clone())));
+        self
     }
 
     pub async fn retrieve_mono(self) -> Unpacker {
-        let (req, mime_type, rsocket) = self.preflight();
-        let res = rsocket.request_response(req).await;
-        Unpacker {
-            mime_type,
-            inner: res,
+        match self.preflight() {
+            Ok((req, mime_type, rsocket)) => {
+                let res = rsocket.request_response(req).await;
+                match res {
+                    Ok(v) => Unpacker {
+                        inner: Ok((mime_type, v)),
+                    },
+                    Err(e) => Unpacker { inner: Err(e) },
+                }
+            }
+            Err(e) => {
+                // TODO: better error
+                let msg = format!("{}", e);
+                Unpacker {
+                    inner: Err(RSocketError::from(msg)),
+                }
+            }
         }
     }
 
     #[inline]
-    fn preflight(self) -> (Payload, MimeType, Arc<Box<dyn RSocket>>) {
+    fn preflight(self) -> PreflightResult {
         let mut b = BytesMut::new();
         let mut c = CompositeMetadata::builder();
-        for (mime_type, raw) in self.metadatas.into_iter() {
+
+        for mut b in self.metadatas.into_iter() {
+            let (mime_type, raw) = b()?;
             c = c.push(mime_type, raw);
         }
         c.build().write_to(&mut b);
 
         let mut bu = Payload::builder().set_metadata(b.to_vec());
-        if let Some(raw) = self.data {
+        if let Some(mut gen) = self.data {
+            let raw = gen(&self.data_mime_type)?;
             bu = bu.set_data(raw);
         }
-        (bu.build(), self.data_mime_type, self.rsocket)
+        Ok((bu.build(), self.data_mime_type, self.rsocket))
     }
 }
 
 pub struct Unpacker {
-    mime_type: MimeType,
-    inner: Result<Payload, RSocketError>,
+    inner: Result<(MimeType, Payload), RSocketError>,
 }
 
 impl Unpacker {
-    pub fn block<'a, T>(&'a self) -> Result<Option<T>, Box<dyn Error>>
+    pub fn block<T>(&self) -> Result<Option<T>, Box<dyn Error>>
     where
-        T: Deserialize<'a>,
+        T: Sized + DeserializeOwned,
     {
         match &self.inner {
-            Ok(inner) => match inner.data() {
-                Some(raw) => Ok(Some(unmarshal(&self.mime_type, &raw.as_ref())?)),
+            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()),
+                },
                 None => Ok(None),
             },
             Err(e) => Err(format!("{}", e).into()),
         }
     }
 }
+
+fn do_marshal<T>(mime_type: &MimeType, data: &T) -> Result<Vec<u8>, Box<dyn Error>>
+where
+    T: Sized + Serialize,
+{
+    // TODO: support more mime types
+    match *mime_type {
+        MimeType::APPLICATION_JSON => marshal(misc::json(), data),
+        MimeType::APPLICATION_CBOR => marshal(misc::cbor(), data),
+        _ => Err("unsupported mime type!".into()),
+    }
+}