api: add metadata for ttrpc context

For tracing purpose, it needs to pass metadata like span id to the server.
`ttrpc-rust` should then parse the metadata to service handler by `TtrpcContext`

This commit is mainly a porting of parts of `metadata_test.go`
and `types.go` if go version, and now only working on server side.

Signed-off-by: bin <bin@hyper.sh>

Author: liubin

src/asynchronous/server.rs

@@ -310,7 +310,7 @@ async fn handle_request(
     let path = format!("/{}/{}", req.service, req.method);
     if let Some(x) = methods.get(&path) {
         let method = x;
-        let ctx = TtrpcContext { fd, mh: header };
+        let ctx = TtrpcContext { fd, mh: header, metadata: common::parse_metadata(&req.metadata) };
 
         match method.handler(ctx, req).await {
             Ok((stream_id, body)) => {

src/asynchronous/utils.rs

@@ -8,6 +8,7 @@ use crate::error::{Error, Result};
 use crate::ttrpc::{Request, Response};
 use async_trait::async_trait;
 use protobuf::Message;
+use std::collections::HashMap;
 use std::os::unix::io::{FromRawFd, RawFd};
 use tokio::net::UnixStream;
 
@@ -87,6 +88,7 @@ pub trait MethodHandler {
 pub struct TtrpcContext {
     pub fd: std::os::unix::io::RawFd,
     pub mh: MessageHeader,
+    pub metadata: HashMap<String, Vec<String>>,
 }
 
 pub fn convert_response_to_buf(res: Response) -> Result<Vec<u8>> {

src/common.rs

@@ -8,8 +8,10 @@
 #![allow(unused_macros)]
 
 use crate::error::{Error, Result};
+use crate::ttrpc::KeyValue;
 use nix::fcntl::{fcntl, FcntlArg, FdFlag, OFlag};
 use nix::sys::socket::*;
+use std::collections::HashMap;
 use std::os::unix::io::RawFd;
 use std::str::FromStr;
 
@@ -139,3 +141,46 @@ pub const MESSAGE_LENGTH_MAX: usize = 4 << 20;
 
 pub const MESSAGE_TYPE_REQUEST: u8 = 0x1;
 pub const MESSAGE_TYPE_RESPONSE: u8 = 0x2;
+
+pub fn parse_metadata(kvs: &protobuf::RepeatedField<KeyValue>) -> HashMap<String, Vec<String>> {
+    let mut meta: HashMap<String, Vec<String>> = HashMap::new();
+    for kv in kvs {
+        if let Some(ref mut vl) = meta.get_mut(&kv.key) {
+            vl.push(kv.value.clone());
+        } else {
+            meta.insert(kv.key.clone(), vec![kv.value.clone()]);
+        }
+    }
+    meta
+}
+
+#[cfg(test)]
+mod tests {
+    use super::parse_metadata;
+    use crate::ttrpc::KeyValue;
+
+    #[test]
+    fn test_parse_metadata() {
+        let mut src: protobuf::RepeatedField<KeyValue> = protobuf::RepeatedField::default();
+        for i in &[
+            ("key1", "value1-1"),
+            ("key1", "value1-2"),
+            ("key2", "value2"),
+        ] {
+            let mut key = KeyValue::default();
+            key.key = i.0.to_string();
+            key.value = i.1.to_string();
+            src.push(key);
+        }
+
+        let dst = parse_metadata(&src);
+        assert_eq!(dst.len(), 2);
+
+        assert_eq!(
+            dst.get("key1"),
+            Some(&vec!["value1-1".to_string(), "value1-2".to_string()])
+        );
+        assert_eq!(dst.get("key2"), Some(&vec!["value2".to_string()]));
+        assert_eq!(dst.get("key3"), None);
+    }
+}

src/sync/server.rs

@@ -209,6 +209,7 @@ fn start_method_handler_thread(
                 fd,
                 mh,
                 res_tx: res_tx.clone(),
+                metadata: common::parse_metadata(&req.metadata),
             };
             if let Err(x) = method.handler(ctx, req) {
                 debug!("method handle {} get error {:?}", path, x);

src/sync/utils.rs

@@ -7,6 +7,7 @@ use crate::common::{MessageHeader, MESSAGE_TYPE_RESPONSE};
 use crate::error::{Error, Result};
 use crate::ttrpc::{Request, Response};
 use protobuf::Message;
+use std::collections::HashMap;
 
 /// Response message through a channel.
 /// Eventually  the message will sent to Client.
@@ -94,6 +95,7 @@ pub struct TtrpcContext {
     pub fd: std::os::unix::io::RawFd,
     pub mh: MessageHeader,
     pub res_tx: std::sync::mpsc::Sender<(MessageHeader, Vec<u8>)>,
+    pub metadata: HashMap<String, Vec<String>>,
 }
 
 /// Trait that implements handler which is a proxy to the desired method (sync).

src/ttrpc.proto

@@ -21,6 +21,12 @@ message Request {
 	string method = 2;
 	bytes payload = 3;
 	int64 timeout_nano = 4;
+	repeated KeyValue metadata = 5;
+}
+
+message KeyValue {
+	string key = 1;
+	string value = 2;
 }
 
 // Get from github.com/gogo/protobuf/protobuf/google/protobuf/any.proto