ttrpc: upgrade protobuf from 2.x to 3.x

Upgrade dep protobuf of ttrpc from 2.x to 3.x

Signed-off-by: Tim Zhang <[email protected]>

Author: Tim-Zhang

Cargo.toml

@@ -11,7 +11,7 @@ homepage = "https://github.com/containerd/ttrpc-rust"
 description = "A Rust version of ttrpc."
 
 [dependencies]
-protobuf = { version = "2.0" }
+protobuf = { version = "3.1.0" }
 libc = { version = "0.2.59", features = [ "extra_traits" ] }
 nix = "0.23.0"
 log = "0.4"
@@ -26,7 +26,7 @@ futures = { version = "0.3", optional = true }
 tokio-vsock = { version = "0.3.1", optional = true }
 
 [build-dependencies]
-protobuf-codegen-pure = "2.14.0"
+protobuf-codegen = "3.1.0"
 
 [features]
 default = ["sync"]

build.rs

@@ -7,10 +7,16 @@ fn main() {
     let path: PathBuf = [out_dir.clone(), "mod.rs".to_string()].iter().collect();
     fs::write(path, "pub mod ttrpc;").unwrap();
 
-    protobuf_codegen_pure::Codegen::new()
+    let customize = protobuf_codegen::Customize::default()
+        .gen_mod_rs(false)
+        .generate_accessors(true);
+
+    protobuf_codegen::Codegen::new()
+        .pure()
         .out_dir(out_dir)
         .inputs(&["src/ttrpc.proto"])
         .include("src")
+        .customize(customize)
         .run()
         .expect("Codegen failed.");
 }

src/asynchronous/client.rs

@@ -70,7 +70,7 @@ impl Client {
 
         let msg: GenMessage = Message::new_request(stream_id, req)
             .try_into()
-            .map_err(|e: protobuf::error::ProtobufError| Error::Others(e.to_string()))?;
+            .map_err(|e: protobuf::Error| Error::Others(e.to_string()))?;
 
         let (tx, mut rx): (ResultSender, ResultReceiver) = mpsc::channel(100);
 
@@ -100,8 +100,8 @@ impl Client {
         let res = Response::decode(&msg.payload)
             .map_err(err_to_others_err!(e, "Unpack response error "))?;
 
-        let status = res.get_status();
-        if status.get_code() != Code::OK {
+        let status = res.status();
+        if status.code() != Code::OK {
             return Err(Error::RpcStatus((*status).clone()));
         }
 
@@ -119,7 +119,7 @@ impl Client {
 
         let mut msg: GenMessage = Message::new_request(stream_id, req)
             .try_into()
-            .map_err(|e: protobuf::error::ProtobufError| Error::Others(e.to_string()))?;
+            .map_err(|e: protobuf::Error| Error::Others(e.to_string()))?;
 
         if streaming_client {
             msg.header.add_flags(FLAG_REMOTE_OPEN);

src/asynchronous/stream.rs

@@ -454,7 +454,7 @@ impl StreamReceiver {
                 let resp = Response::decode(&msg.payload)
                     .map_err(err_to_others_err!(e, "Decode message failed."))?;
                 if let Some(status) = resp.status.as_ref() {
-                    if status.get_code() != Code::OK {
+                    if status.code() != Code::OK {
                         return Err(Error::RpcStatus((*status).clone()));
                     }
                 }

src/asynchronous/utils.rs

@@ -145,13 +145,15 @@ macro_rules! async_duplex_streamimg_handler {
 #[macro_export]
 macro_rules! async_client_request {
     ($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($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 creq = ttrpc::Request {
+            service: $server.to_string(),
+            method: $method.to_string(),
+            timeout_nano: $ctx.timeout_nano,
+            metadata: ttrpc::context::to_pb($ctx.metadata),
+            payload: Vec::with_capacity($req.compute_size() as usize),
+            ..Default::default()
+        };
+
         {
             let mut s = CodedOutputStream::vec(&mut creq.payload);
             $req.write_to(&mut s)

src/context.rs

@@ -48,7 +48,7 @@ impl Context {
     }
 }
 
-pub fn from_pb(kvs: &protobuf::RepeatedField<KeyValue>) -> HashMap<String, Vec<String>> {
+pub fn from_pb(kvs: &Vec<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) {
@@ -60,8 +60,9 @@ pub fn from_pb(kvs: &protobuf::RepeatedField<KeyValue>) -> HashMap<String, Vec<S
     meta
 }
 
-pub fn to_pb(kvs: HashMap<String, Vec<String>>) -> protobuf::RepeatedField<KeyValue> {
-    let mut meta: protobuf::RepeatedField<KeyValue> = protobuf::RepeatedField::default();
+pub fn to_pb(kvs: HashMap<String, Vec<String>>) -> Vec<KeyValue> {
+    let mut meta = Vec::with_capacity(kvs.len());
+
     for (k, vl) in kvs {
         for v in vl {
             let key = KeyValue {
@@ -72,6 +73,7 @@ pub fn to_pb(kvs: HashMap<String, Vec<String>>) -> protobuf::RepeatedField<KeyVa
             meta.push(key);
         }
     }
+
     meta
 }
 
@@ -83,7 +85,7 @@ mod tests {
     #[test]
     fn test_metadata() {
         // RepeatedField -> HashMap, test from_pb()
-        let mut src: protobuf::RepeatedField<KeyValue> = protobuf::RepeatedField::default();
+        let mut src = Vec::new();
         for i in &[
             ("key1", "value1-1"),
             ("key1", "value1-2"),
@@ -108,8 +110,7 @@ mod tests {
         assert_eq!(dst.get("key3"), None);
 
         // HashMap -> RepeatedField , test to_pb()
-        let src = context::to_pb(dst);
-        let mut kvs = src.into_vec();
+        let mut kvs = context::to_pb(dst);
         kvs.sort_by(|a, b| a.key.partial_cmp(&b.key).unwrap());
 
         assert_eq!(kvs.len(), 3);

src/proto.rs

@@ -215,17 +215,18 @@ pub trait Codec {
 }
 
 impl<M: protobuf::Message> Codec for M {
-    type E = protobuf::error::ProtobufError;
+    type E = protobuf::Error;
 
     fn size(&self) -> u32 {
-        self.compute_size()
+        self.compute_size() as u32
     }
 
     fn encode(&self) -> Result<Vec<u8>, Self::E> {
         let mut buf = vec![0; self.compute_size() as usize];
         let mut s = CodedOutputStream::bytes(&mut buf);
         self.write_to(&mut s)?;
         s.flush()?;
+        drop(s);
         Ok(buf)
     }
 
@@ -380,12 +381,11 @@ mod tests {
         creq.set_service("grpc.TestServices".to_string());
         creq.set_method("Test".to_string());
         creq.set_timeout_nano(20 * 1000 * 1000);
-        let mut meta: protobuf::RepeatedField<KeyValue> = protobuf::RepeatedField::default();
-        meta.push(KeyValue {
+        let meta = vec![KeyValue {
             key: "test_key1".to_string(),
             value: "test_value1".to_string(),
             ..Default::default()
-        });
+        }];
         creq.set_metadata(meta);
         creq.payload = vec![0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x8, 0x9];
         creq

src/sync/client.rs

@@ -231,8 +231,8 @@ impl Client {
         let res =
             Response::decode(&buf).map_err(err_to_others_err!(e, "Unpack response error "))?;
 
-        let status = res.get_status();
-        if status.get_code() != Code::OK {
+        let status = res.status();
+        if status.code() != Code::OK {
             return Err(Error::RpcStatus((*status).clone()));
         }
 

src/sync/utils.rs

@@ -20,6 +20,8 @@ pub fn response_to_channel(
     res.write_to(&mut s).map_err(err_to_others_err!(e, ""))?;
     s.flush().map_err(err_to_others_err!(e, ""))?;
 
+    drop(s);
+
     let mh = MessageHeader {
         length: buf.len() as u32,
         stream_id,
@@ -82,6 +84,8 @@ macro_rules! client_request {
             .map_err(::ttrpc::err_to_others!(e, ""))?;
         s.flush().map_err(::ttrpc::err_to_others!(e, ""))?;
 
+        drop(s);
+
         let res = $self.client.request(creq)?;
         let mut s = CodedInputStream::from_bytes(&res.payload);
         $cres