New feature: messaging.

Author: jjeffcaii

Cargo.toml

@@ -1,12 +1,15 @@
 [workspace]
 
 members = [
+# core
 "rsocket",
+# transports
 "rsocket-transport-tcp",
 "rsocket-transport-websocket",
 "rsocket-transport-wasm",
-
-# Internal
+# extra
+"rsocket-messaging",
+# internal
 "examples",
 "rsocket-test",
 ]

examples/Cargo.toml

@@ -21,7 +21,7 @@ path = "../rsocket-transport-tcp"
 path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
-version = "0.2.16"
+version = "0.2.19"
 default-features = false
 features = ["full"]
 

rsocket-messaging/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "rsocket_rust_messaging"
+version = "0.1.0"
+authors = ["Jeffsky <jjeffcaii@outlook.com>"]
+edition = "2018"
+
+[dependencies]
+log = "0.4.8"
+futures = "0.3.4"
+bytes = "0.5.4"
+serde = "1.0.106"
+serde_json = "1.0.52"
+serde_cbor = "0.11.1"
+hex = "0.4.2"
+
+[dependencies.rsocket_rust]
+path = "../rsocket"
+features = ["frame"]

rsocket-messaging/src/lib.rs

@@ -0,0 +1,7 @@
+#[macro_use]
+extern crate log;
+
+mod misc;
+mod requester;
+
+pub use requester::{RequestSpec, Requester};

rsocket-messaging/src/misc.rs

@@ -0,0 +1,38 @@
+use bytes::{BufMut, BytesMut};
+use rsocket_rust::extension::{MimeType, MIME_APPLICATION_CBOR, MIME_APPLICATION_JSON};
+use serde::{Deserialize, 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 {
+        MIME_APPLICATION_JSON => Ok(serde_json::from_slice(raw)?),
+        MIME_APPLICATION_CBOR => Ok(serde_cbor::from_slice(raw)?),
+        _ => panic!(""),
+    }
+}
+
+pub(crate) fn marshal<T>(
+    mime_type: &MimeType,
+    bf: &mut BytesMut,
+    data: &T,
+) -> Result<(), Box<dyn Error>>
+where
+    T: Sized + Serialize,
+{
+    match *mime_type {
+        MIME_APPLICATION_JSON => {
+            let raw = serde_json::to_vec(data)?;
+            bf.put_slice(&raw[..]);
+            Ok(())
+        }
+        MIME_APPLICATION_CBOR => {
+            let raw = serde_cbor::to_vec(data)?;
+            bf.put_slice(&raw[..]);
+            Ok(())
+        }
+        _ => panic!(""),
+    }
+}

rsocket-messaging/src/requester.rs

@@ -0,0 +1,118 @@
+use super::misc::{marshal, unmarshal};
+use bytes::{BufMut, BytesMut};
+use rsocket_rust::error::RSocketError;
+use rsocket_rust::extension::{
+    CompositeMetadata, MimeType, RoutingMetadata, MIME_APPLICATION_JSON,
+    MIME_MESSAGE_X_RSOCKET_ROUTING_V0,
+};
+use rsocket_rust::prelude::*;
+use rsocket_rust::utils::Writeable;
+use serde::{Deserialize, Serialize};
+use std::collections::LinkedList;
+use std::error::Error;
+
+pub struct Requester<S>
+where
+    S: RSocket + Clone,
+{
+    rsocket: S,
+}
+
+pub struct RequestSpec<S>
+where
+    S: RSocket + Clone,
+{
+    data_buf: BytesMut,
+    rsocket: S,
+    data_mime_type: MimeType,
+    metadatas: LinkedList<(MimeType, Vec<u8>)>,
+}
+
+impl<C> Requester<C>
+where
+    C: RSocket + Clone,
+{
+    pub fn new(rsocket: C) -> Requester<C> {
+        Requester { rsocket }
+    }
+
+    pub fn route(&self, route: &str) -> RequestSpec<C> {
+        let routing = RoutingMetadata::builder().push_str(route).build();
+        let mut buf = BytesMut::new();
+        routing.write_to(&mut buf);
+
+        let mut metadatas: LinkedList<(MimeType, Vec<u8>)> = Default::default();
+        metadatas.push_back((MIME_MESSAGE_X_RSOCKET_ROUTING_V0, buf.to_vec()));
+        RequestSpec {
+            data_buf: BytesMut::new(),
+            rsocket: self.rsocket.clone(),
+            data_mime_type: MIME_APPLICATION_JSON,
+            metadatas,
+        }
+    }
+}
+
+impl<C> RequestSpec<C>
+where
+    C: RSocket + Clone,
+{
+    pub fn metadata<T>(&mut self, metadata: &T, mime_type: &str) -> Result<(), Box<dyn Error>>
+    where
+        T: Sized + Serialize,
+    {
+        let mime_type = MimeType::from(mime_type);
+        let mut b = BytesMut::new();
+        marshal(&mime_type, &mut b, metadata)?;
+        self.metadatas.push_back((mime_type, b.to_vec()));
+        Ok(())
+    }
+
+    pub fn data<T>(&mut self, data: &T) -> Result<(), Box<dyn Error>>
+    where
+        T: Sized + Serialize,
+    {
+        marshal(&self.data_mime_type, &mut self.data_buf, data)
+    }
+
+    pub async fn retrieve_mono(&self) -> Unpacker {
+        let req = self.to_req();
+        let res = self.rsocket.request_response(req).await;
+        Unpacker {
+            mime_type: self.data_mime_type.clone(),
+            inner: res,
+        }
+    }
+
+    fn to_req(&self) -> Payload {
+        let mut b = BytesMut::new();
+        let mut c = CompositeMetadata::builder();
+        for (a, b) in self.metadatas.iter() {
+            c = c.push(a.clone(), b);
+        }
+        c.build().write_to(&mut b);
+        Payload::builder()
+            .set_metadata(b.to_vec())
+            .set_data(self.data_buf.to_vec())
+            .build()
+    }
+}
+
+pub struct Unpacker {
+    mime_type: MimeType,
+    inner: Result<Payload, RSocketError>,
+}
+
+impl Unpacker {
+    pub fn block<'a, T>(&'a self) -> Result<Option<T>, Box<dyn Error>>
+    where
+        T: Deserialize<'a>,
+    {
+        match &self.inner {
+            Ok(inner) => match inner.data() {
+                Some(raw) => Ok(Some(unmarshal(&self.mime_type, &raw.as_ref())?)),
+                None => Ok(None),
+            },
+            Err(e) => Err(format!("{}", e).into()),
+        }
+    }
+}

rsocket-test/Cargo.toml

@@ -12,6 +12,8 @@ env_logger = "0.7.1"
 bytes = "0.5.4"
 hex = "0.4.2"
 rand = "0.7.3"
+serde = "1.0.106"
+serde_derive = "1.0.106"
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"
@@ -23,7 +25,10 @@ path = "../rsocket-transport-tcp"
 [dev-dependencies.rsocket_rust_transport_websocket]
 path = "../rsocket-transport-websocket"
 
+[dev-dependencies.rsocket_rust_messaging]
+path = "../rsocket-messaging"
+
 [dev-dependencies.tokio]
-version = "0.2.16"
+version = "0.2.20"
 default-features = false
 features = ["full"]

rsocket-test/tests/test_composite_metadata.rs

@@ -1,20 +1,23 @@
 use bytes::BytesMut;
-use rsocket_rust::extension::{CompositeMetadata, Metadata};
+use rsocket_rust::extension::{self, CompositeMetadata, CompositeMetadataEntry, MimeType};
 use rsocket_rust::utils::Writeable;
 
 #[test]
 fn encode_and_decode_composite_metadata() {
-    let bingo = |metadatas: Vec<&Metadata>| {
+    let bingo = |metadatas: Vec<&CompositeMetadataEntry>| {
         assert_eq!(2, metadatas.len());
-        assert_eq!("text/plain", metadatas[0].get_mime());
-        assert_eq!(b"Hello World!", metadatas[0].get_payload().as_ref());
-        assert_eq!("application/not_well", metadatas[1].get_mime());
-        assert_eq!(b"Not Well!", metadatas[1].get_payload().as_ref());
+        assert_eq!(extension::MIME_TEXT_PLAIN, *metadatas[0].get_mime_type());
+        assert_eq!("Hello World!", metadatas[0].get_metadata_utf8());
+        assert_eq!(
+            MimeType::from("application/not_well"),
+            *metadatas[1].get_mime_type()
+        );
+        assert_eq!(b"Not Well!", metadatas[1].get_metadata().as_ref());
     };
 
     let cm = CompositeMetadata::builder()
-        .push("text/plain", b"Hello World!")
-        .push("application/not_well", "Not Well!")
+        .push(MimeType::from("text/plain"), b"Hello World!")
+        .push(MimeType::from("application/not_well"), "Not Well!")
         .build();
     bingo(cm.iter().collect());
 

rsocket-test/tests/test_messaging.rs

@@ -0,0 +1,52 @@
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate serde_derive;
+
+use rsocket_rust::prelude::*;
+use rsocket_rust_messaging::*;
+use rsocket_rust_transport_tcp::TcpClientTransport;
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct Student {
+    id: i64,
+    name: String,
+    birth: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct Response<T> {
+    code: i32,
+    message: Option<String>,
+    data: T,
+}
+
+#[tokio::main]
+#[test]
+#[ignore]
+async fn test_messaging() {
+    let rsocket = RSocketFactory::connect()
+        .transport(TcpClientTransport::from("tcp://127.0.0.1:7878"))
+        .data_mime_type("application/json")
+        .metadata_mime_type("message/x.rsocket.composite-metadata.v0")
+        .start()
+        .await
+        .expect("Connect failed!");
+    let requester = Requester::new(rsocket);
+
+    let post = Student {
+        id: 1234,
+        name: "Jeffsky".to_owned(),
+        birth: "2020-01-01".to_owned(),
+    };
+    let mut req = requester.route("student.v1.upsert");
+    req.data(&post).unwrap();
+    let res: Response<Student> = req
+        .retrieve_mono()
+        .await
+        .block()
+        .expect("Retrieve failed!")
+        .expect("Empty result!");
+    println!("------> RESPONSE: {:?}", res);
+    ()
+}

rsocket-test/tests/test_mimes.rs

@@ -1,15 +1,12 @@
 extern crate rsocket_rust;
 
-use rsocket_rust::mime::WellKnownMIME;
+use rsocket_rust::extension::{self, MimeType};
 
 #[test]
 fn test_wellknown() {
-    let got = WellKnownMIME::from("application/json");
-    assert_eq!(WellKnownMIME::ApplicationJson, got);
-    WellKnownMIME::foreach(|m| {
-        let mut result = WellKnownMIME::from(m.raw());
-        assert_eq!(m, &result);
-        result = WellKnownMIME::from(format!("{}", m));
-        assert_eq!(m, &result);
-    });
+    let well = MimeType::from("application/json");
+    assert_eq!(extension::MIME_APPLICATION_JSON, well);
+    assert_eq!(0x05, well.as_u8().unwrap());
+    let custom = MimeType::from("application/custom");
+    assert_eq!(MimeType::Normal("application/custom".to_owned()), custom);
 }