Merge pull request containerd#76 from liubin/add-metadata-for-ttrpc-context

api: add metadata for server and client

Author: Tim-Zhang

compiler/src/codegen.rs

@@ -212,7 +212,7 @@ impl<'a> MethodGen<'a> {
     // Method signatures
     fn unary(&self, method_name: &str) -> String {
         format!(
-            "{}(&self, req: &{}, timeout_nano: i64) -> {}<{}>",
+            "{}(&self, ctx: ttrpc::context::Context, req: &{}) -> {}<{}>",
             method_name,
             self.input(),
             fq_grpc("Result"),
@@ -222,7 +222,7 @@ impl<'a> MethodGen<'a> {
 
     fn unary_async(&self, method_name: &str) -> String {
         format!(
-            "{}(&mut self, req: &{}, timeout_nano: i64) -> {}<{}>",
+            "{}(&mut self, ctx: ttrpc::context::Context, req: &{}) -> {}<{}>",
             method_name,
             self.input(),
             fq_grpc("Result"),
@@ -314,7 +314,7 @@ impl<'a> MethodGen<'a> {
                         self.output()
                     ));
                     w.write_line(&format!(
-                        "::ttrpc::client_request!(self, req, timeout_nano, \"{}.{}\", \"{}\", cres);",
+                        "::ttrpc::client_request!(self, ctx, req, \"{}.{}\", \"{}\", cres);",
                         self.package_name,
                         self.service_name,
                         &self.proto.get_name(),
@@ -335,7 +335,7 @@ impl<'a> MethodGen<'a> {
                 pub_async_fn(w, &self.unary_async(&method_name), |w| {
                     w.write_line(&format!("let mut cres = {}::new();", self.output()));
                     w.write_line(&format!(
-                        "::ttrpc::async_client_request!(self, req, timeout_nano, \"{}.{}\", \"{}\", cres);",
+                        "::ttrpc::async_client_request!(self, ctx, req, \"{}.{}\", \"{}\", cres);",
                         self.package_name,
                         self.service_name,
                         &self.proto.get_name(),

example/Cargo.lock

@@ -43,3 +43,6 @@ path = "./async-client.rs"
 [build-dependencies]
 ttrpc-codegen = { path = "../ttrpc-codegen"}
 
+[patch.crates-io]
+ttrpc-compiler = { path = "../compiler"}
+

example/Cargo.toml

@@ -7,6 +7,7 @@ mod protocols;
 
 use nix::sys::socket::*;
 use protocols::r#async::{agent, agent_ttrpc, health, health_ttrpc};
+use ttrpc::context::{self, Context};
 use ttrpc::r#async::Client;
 
 #[tokio::main(flavor = "current_thread")]
@@ -44,7 +45,7 @@ async fn main() {
         println!(
             "Green Thread 1 - {} -> {:?} ended: {:?}",
             "health.check()",
-            thc.check(&req, 0).await,
+            thc.check(default_ctx(), &req).await,
             now.elapsed(),
         );
     });
@@ -57,7 +58,7 @@ async fn main() {
         );
 
         let show = match tac
-            .list_interfaces(&agent::ListInterfacesRequest::new(), 0)
+            .list_interfaces(default_ctx(), &agent::ListInterfacesRequest::new())
             .await
         {
             Err(e) => format!("{:?}", e),
@@ -80,7 +81,7 @@ async fn main() {
         );
 
         let show = match ac
-            .online_cpu_mem(&agent::OnlineCPUMemRequest::new(), 0)
+            .online_cpu_mem(default_ctx(), &agent::OnlineCPUMemRequest::new())
             .await
         {
             Err(e) => format!("{:?}", e),
@@ -101,10 +102,20 @@ async fn main() {
         println!(
             "Green Thread 3 - {} -> {:?} ended: {:?}",
             "health.version()",
-            hc.version(&health::CheckRequest::new(), 0).await,
+            hc.version(default_ctx(), &health::CheckRequest::new())
+                .await,
             now.elapsed()
         );
     });
 
     let _ = tokio::join!(t1, t2, t3);
 }
+
+fn default_ctx() -> Context {
+    let mut ctx = context::with_timeout(0);
+    ctx.add("key-1".to_string(), "value-1-1".to_string());
+    ctx.add("key-1".to_string(), "value-1-2".to_string());
+    ctx.set("key-2".to_string(), vec!["value-2".to_string()]);
+
+    ctx
+}

example/async-client.rs

@@ -39,12 +39,14 @@ impl health_ttrpc::Health for HealthService {
 
         Err(Error::RpcStatus(status))
     }
+
     async fn version(
         &self,
-        _ctx: &::ttrpc::r#async::TtrpcContext,
+        ctx: &::ttrpc::r#async::TtrpcContext,
         req: health::CheckRequest,
     ) -> Result<health::VersionCheckResponse> {
         info!("version {:?}", req);
+        info!("ctx {:?}", ctx);
         let mut rep = health::VersionCheckResponse::new();
         rep.agent_version = "mock.0.1".to_string();
         rep.grpc_version = "0.0.1".to_string();

example/async-server.rs

@@ -17,11 +17,6 @@ fn main() {
         "protocols/protos/oci.proto",
     ];
 
-    // Tell Cargo that if the .proto files changed, to rerun this build script.
-    protos
-        .iter()
-        .for_each(|p| println!("cargo:rerun-if-changed={}", &p));
-
     Codegen::new()
         .out_dir("protocols/sync")
         .inputs(&protos)

example/build.rs

@@ -18,6 +18,7 @@ use nix::sys::socket::*;
 use protocols::sync::{agent, agent_ttrpc, health, health_ttrpc};
 use std::thread;
 use ttrpc::client::Client;
+use ttrpc::context::{self, Context};
 
 fn main() {
     let path = "/tmp/1";
@@ -55,7 +56,7 @@ fn main() {
             "OS Thread {:?} - {} -> {:?} ended: {:?}",
             std::thread::current().id(),
             "health.check()",
-            thc.check(&req, 0),
+            thc.check(default_ctx(), &req),
             now.elapsed(),
         );
     });
@@ -68,7 +69,7 @@ fn main() {
             now.elapsed(),
         );
 
-        let show = match tac.list_interfaces(&agent::ListInterfacesRequest::new(), 0) {
+        let show = match tac.list_interfaces(default_ctx(), &agent::ListInterfacesRequest::new()) {
             Err(e) => format!("{:?}", e),
             Ok(s) => format!("{:?}", s),
         };
@@ -87,7 +88,7 @@ fn main() {
         "agent.online_cpu_mem()",
         now.elapsed()
     );
-    let show = match ac.online_cpu_mem(&agent::OnlineCPUMemRequest::new(), 0) {
+    let show = match ac.online_cpu_mem(default_ctx(), &agent::OnlineCPUMemRequest::new()) {
         Err(e) => format!("{:?}", e),
         Ok(s) => format!("{:?}", s),
     };
@@ -108,10 +109,19 @@ fn main() {
     println!(
         "Main OS Thread - {} -> {:?} ended: {:?}",
         "health.version()",
-        hc.version(&health::CheckRequest::new(), 0),
+        hc.version(default_ctx(), &health::CheckRequest::new()),
         now.elapsed()
     );
 
     t.join().unwrap();
     t2.join().unwrap();
 }
+
+fn default_ctx() -> Context {
+    let mut ctx = context::with_timeout(0);
+    ctx.add("key-1".to_string(), "value-1-1".to_string());
+    ctx.add("key-1".to_string(), "value-1-2".to_string());
+    ctx.set("key-2".to_string(), vec!["value-2".to_string()]);
+
+    ctx
+}

example/client.rs

@@ -38,12 +38,14 @@ impl health_ttrpc::Health for HealthService {
         status.set_message("Just for fun".to_string());
         Err(Error::RpcStatus(status))
     }
+
     fn version(
         &self,
-        _ctx: &::ttrpc::TtrpcContext,
+        ctx: &::ttrpc::TtrpcContext,
         req: health::CheckRequest,
     ) -> Result<health::VersionCheckResponse> {
         info!("version {:?}", req);
+        info!("ctx {:?}", ctx);
         let mut rep = health::VersionCheckResponse::new();
         rep.agent_version = "mock.0.1".to_string();
         rep.grpc_version = "0.0.1".to_string();
@@ -73,7 +75,6 @@ impl agent_ttrpc::AgentService for AgentService {
         i.set_Interfaces(rp);
 
         Ok(i)
-        //Err(::ttrpc::Error::RpcStatus(::ttrpc::get_Status(::ttrpc::Code::NOT_FOUND, "".to_string())))
     }
 }
 

example/server.rs

@@ -12,6 +12,7 @@ use std::sync::Arc;
 use crate::asynchronous::stream::{receive, respond, respond_with_status};
 use crate::asynchronous::unix_incoming::UnixIncoming;
 use crate::common::{self, Domain, MESSAGE_TYPE_REQUEST};
+use crate::context;
 use crate::error::{get_status, Error, Result};
 use crate::r#async::{MethodHandler, TtrpcContext};
 use crate::ttrpc::{Code, Request};
@@ -310,7 +311,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: context::from_pb(&req.metadata) };
 
         match method.handler(ctx, req).await {
             Ok((stream_id, body)) => {

src/asynchronous/server.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;
 
@@ -53,11 +54,13 @@ macro_rules! async_request_handler {
 /// Send request through async client.
 #[macro_export]
 macro_rules! async_client_request {
-    ($self: ident, $req: ident, $timeout_nano: ident, $server: expr, $method: expr, $cres: ident) => {
+    ($self: ident, $ctx: ident, $req: ident, $server: expr, $method: expr, $cres: ident) => {
         let mut creq = ::ttrpc::Request::new();
         creq.set_service($server.to_string());
         creq.set_method($method.to_string());
-        creq.set_timeout_nano($timeout_nano);
+        creq.set_timeout_nano($ctx.timeout_nano);
+        let md = ::ttrpc::context::to_pb($ctx.metadata);
+        creq.set_metadata(md);
         creq.payload.reserve($req.compute_size() as usize);
         {
             let mut s = CodedOutputStream::vec(&mut creq.payload);
@@ -87,6 +90,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>> {