support RoutingMetadata.

Author: jjeffcaii

src/extension.rs

@@ -5,13 +5,94 @@ use crate::result::RSocketResult;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
 const MAX_MIME_LEN: usize = 0x7F;
+const MAX_ROUTING_TAG_LEN: usize = 0xFF;
 
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct CompositeMetadata {
     mime: String,
     payload: Bytes,
 }
 
+#[derive(Debug, Clone)]
+pub struct RoutingMetadata {
+    tags: Vec<String>,
+}
+
+pub struct RoutingMetadataBuilder {
+    inner: RoutingMetadata,
+}
+
+impl RoutingMetadataBuilder {
+    pub fn push_str(self, tag: &str) -> Self {
+        self.push(String::from(tag))
+    }
+
+    pub fn push(mut self, tag: String) -> Self {
+        if tag.len() > MAX_ROUTING_TAG_LEN {
+            panic!("exceeded maximum routing tag length!");
+        }
+        self.inner.tags.push(tag);
+        self
+    }
+    pub fn build(self) -> RoutingMetadata {
+        self.inner
+    }
+}
+
+impl RoutingMetadata {
+    pub fn builder() -> RoutingMetadataBuilder {
+        RoutingMetadataBuilder {
+            inner: RoutingMetadata { tags: vec![] },
+        }
+    }
+
+    pub fn decode(bf: &mut BytesMut) -> RSocketResult<RoutingMetadata> {
+        let mut bu = RoutingMetadata::builder();
+        loop {
+            match Self::decode_once(bf) {
+                Ok(v) => match v {
+                    Some(tag) => bu = bu.push(tag),
+                    None => break,
+                },
+                Err(e) => return Err(e),
+            }
+        }
+        Ok(bu.build())
+    }
+
+    fn decode_once(bf: &mut BytesMut) -> RSocketResult<Option<String>> {
+        if bf.is_empty() {
+            return Ok(None);
+        }
+        let size = bf.get_u8() as usize;
+        if bf.len() < size {
+            return Err(RSocketError::from(ErrorKind::WithDescription(
+                "require more bytes!",
+            )));
+        }
+        let tag = String::from_utf8(bf.split_to(size).to_vec()).unwrap();
+        Ok(Some(tag))
+    }
+
+    pub fn get_tags(&self) -> &Vec<String> {
+        &self.tags
+    }
+
+    pub fn write_to(&self, bf: &mut BytesMut) {
+        for tag in &self.tags {
+            let size = tag.len() as u8;
+            bf.put_u8(size);
+            bf.put_slice(tag.as_bytes());
+        }
+    }
+
+    pub fn bytes(&self) -> Bytes {
+        let mut bf = BytesMut::new();
+        self.write_to(&mut bf);
+        bf.freeze()
+    }
+}
+
 impl CompositeMetadata {
     pub fn new(mime: String, payload: Bytes) -> CompositeMetadata {
         if mime.len() > MAX_MIME_LEN {

src/payload/mod.rs

@@ -1,5 +1,5 @@
 mod default;
 mod setup;
 
-pub use default::{Payload,PayloadBuilder};
-pub use setup::{SetupPayload,SetupPayloadBuilder};
+pub use default::{Payload, PayloadBuilder};
+pub use setup::{SetupPayload, SetupPayloadBuilder};

tests/clients.rs

@@ -1,27 +1,8 @@
 extern crate rsocket_rust;
-
+#[macro_use]
+extern crate log;
 use rsocket_rust::prelude::*;
 
-#[tokio::main]
-#[test]
-async fn test() {
-    let cli = RSocketFactory::connect()
-        .acceptor(|| Box::new(EchoRSocket))
-        .transport(URI::Tcp("127.0.0.1:7878".to_string()))
-        .setup(Payload::from("READY!"))
-        .mime_type("text/plain", "text/plain")
-        .start()
-        .await
-        .unwrap();
-    let req = Payload::builder()
-        .set_data_utf8("Hello World!")
-        .set_metadata_utf8("Rust")
-        .build();
-    let res = cli.request_response(req).await.unwrap();
-    println!("got: {:?}", res);
-    cli.close();
-}
-
 #[tokio::main]
 #[test]
 async fn test_client() {

tests/routing_metadata_test.rs

@@ -0,0 +1,25 @@
+extern crate rsocket_rust;
+
+use bytes::{Bytes, BytesMut};
+use rsocket_rust::extension::RoutingMetadata;
+
+#[test]
+fn routing_metadata_codec() {
+    let m = RoutingMetadata::builder()
+        .push_str("/orders")
+        .push_str("/orders/77778888")
+        .push_str("/users")
+        .push_str("/users/1234")
+        .build();
+
+    let mut bf = BytesMut::new();
+    m.write_to(&mut bf);
+    println!("encode routing metadata: {}", hex::encode(bf.to_vec()));
+    let m2 = RoutingMetadata::decode(&mut bf).unwrap();
+    let tags = m2.get_tags();
+    for tag in tags {
+        println!("decode tag: {}", tag);
+    }
+    assert_eq!(4, tags.len());
+    assert_eq!(m.get_tags(), tags);
+}