fix(volo-grpc): use linkedbytes concat to get the full encoded results

Author: Ggiggle

Cargo.lock

@@ -153,8 +153,12 @@ volo = { path = "../volo" }
 volo-grpc = { path = "../volo-grpc", features = ["grpc-web"] }
 volo-thrift = { path = "../volo-thrift", features = ["multiplex"] }
 volo-http = { path = "../volo-http", features = [
-    "client", "server",
-    "json", "query", "form", "cookie",
+    "client",
+    "server",
+    "json",
+    "query",
+    "form",
+    "cookie",
     "http1",
 ] }
 

examples/Cargo.toml

@@ -0,0 +1,84 @@
+syntax = "proto3";
+
+package codegen;
+
+// 枚举
+enum Status {
+    STATUS_UNSPECIFIED = 0;
+    OK = 1;
+    FAIL = 2;
+}
+
+// 基础类型全集
+message BasicTypes {
+    bool bool_ = 1;
+    int32 int32_ = 2;
+    int64 int64_ = 3;
+    uint32 uint32_ = 4;
+    uint64 uint64_ = 5;
+    sint32 sint32_ = 6;
+    sint64 sint64_ = 7;
+    fixed32 fixed32_ = 8;
+    fixed64 fixed64_ = 9;
+    sfixed32 sfixed32_ = 10;
+    sfixed64 sfixed64_ = 11;
+    float float_ = 12;
+    double double_ = 13;
+    string string_ = 14;
+    bytes bytes_ = 15;
+}
+
+// oneof/可选语义
+message WithOneof {
+    oneof value {
+        int32 int_value = 1;
+        string string_value = 2;
+        BasicTypes basic_value = 3;
+    }
+}
+
+// map/repeated
+message Collections {
+    repeated int32 numbers = 1;
+    repeated BasicTypes items = 2;
+    map<string, int32> dict = 3;
+    map<int32, BasicTypes> id_to_item = 4;
+}
+
+// 嵌套与保留字段
+message Parent {
+    reserved 3, 4;
+    reserved "old_field", "legacy";
+
+    message Child {
+        string name = 1;
+        Status status = 2;
+    }
+
+    repeated Child children = 1;
+    Status status = 2;
+}
+
+// 业务请求/响应
+message EchoRequest {
+    string message = 1;
+    BasicTypes basic = 2;
+    Collections coll = 3;
+    WithOneof maybe = 4;
+}
+
+message EchoResponse {
+    string message = 1;
+    Status status = 2;
+    WithOneof echoed = 3;
+}
+
+// 流式与非流式服务
+service TestService {
+    rpc UnaryEcho(EchoRequest) returns (EchoResponse);
+    rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse);
+    rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse);
+    rpc BidiStreamingEcho(stream EchoRequest) returns (stream EchoResponse);
+}
+
+

tests/code-generation/proto/test.proto

@@ -16,6 +16,11 @@ entries:
           path: examples/addressbook.proto
           includes:
             - examples
+      - idl:
+          source: local
+          path: proto/test.proto
+          includes:
+            - proto
   thrift:
     filename: thrift_gen.rs
     protocol: thrift

tests/code-generation/volo.yml

@@ -53,6 +53,7 @@ hyper-util = { workspace = true, features = [
   "server",
   "http2",
 ] }
+linkedbytes.workspace = true
 matchit.workspace = true
 paste.workspace = true
 percent-encoding.workspace = true

volo-grpc/Cargo.toml

@@ -274,23 +274,23 @@ impl CompressionEncoding {
 /// Compress `len` bytes from `src_buf` into `dest_buf`.
 pub(crate) fn compress(
     encoding: CompressionEncoding,
-    src_buf: &mut LinkedBytes,
-    dest_buf: &mut LinkedBytes,
+    src_buf: &mut BytesMut,
+    dest_buf: &mut BytesMut,
 ) -> Result<(), io::Error> {
-    let len = src_buf.bytes().len();
+    let len = src_buf.len();
     let capacity = ((len / BUFFER_SIZE) + 1) * BUFFER_SIZE;
 
     dest_buf.reserve(capacity);
 
     match encoding {
         #[cfg(feature = "gzip")]
         CompressionEncoding::Gzip(Some(config)) => {
-            let mut gz_encoder = GzEncoder::new(&src_buf.bytes()[0..len], config.level);
+            let mut gz_encoder = GzEncoder::new(&src_buf[0..len], config.level);
             io::copy(&mut gz_encoder, &mut dest_buf.writer())?;
         }
         #[cfg(feature = "zlib")]
         CompressionEncoding::Zlib(Some(config)) => {
-            let mut zlib_encoder = ZlibEncoder::new(&src_buf.bytes()[0..len], config.level);
+            let mut zlib_encoder = ZlibEncoder::new(&src_buf[0..len], config.level);
             io::copy(&mut zlib_encoder, &mut dest_buf.writer())?;
         }
         #[cfg(feature = "zstd")]
@@ -302,13 +302,13 @@ pub(crate) fn compress(
                 level as i32
             };
             let mut zstd_encoder = zstd::Encoder::new(dest_buf.writer(), zstd_level)?;
-            io::copy(&mut &src_buf.bytes()[0..len], &mut zstd_encoder)?;
+            io::copy(&mut &src_buf[0..len], &mut zstd_encoder)?;
             zstd_encoder.finish()?;
         }
         _ => {}
     };
 
-    src_buf.bytes_mut().advance(len);
+    src_buf.advance(len);
     Ok(())
 }
 
@@ -349,7 +349,7 @@ pub(crate) fn decompress(
 
 #[cfg(test)]
 mod tests {
-    use bytes::BufMut;
+    use bytes::{BufMut, BytesMut};
     use pilota::LinkedBytes;
 
     #[cfg(feature = "gzip")]
@@ -367,11 +367,11 @@ mod tests {
 
     #[test]
     fn test_consistency_for_compression() {
-        let mut src = LinkedBytes::with_capacity(BUFFER_SIZE);
-        let mut compress_buf = LinkedBytes::new();
-        let mut de_data = LinkedBytes::with_capacity(BUFFER_SIZE);
+        let mut src = BytesMut::with_capacity(BUFFER_SIZE);
+        let mut compress_buf = BytesMut::new();
+        let mut de_data = BytesMut::with_capacity(BUFFER_SIZE);
         let test_data = &b"test compression"[..];
-        src.put(test_data);
+        src.extend_from_slice(test_data);
 
         let encodings = [
             #[cfg(feature = "gzip")]
@@ -390,11 +390,10 @@ mod tests {
         ];
 
         for encoding in encodings {
-            compress_buf.reset();
+            compress_buf.clear();
             compress(encoding, &mut src, &mut compress_buf).expect("compress failed:");
-            decompress(encoding, compress_buf.bytes_mut(), de_data.bytes_mut())
-                .expect("decompress failed:");
-            assert_eq!(test_data, de_data.bytes());
+            decompress(encoding, &mut compress_buf, &mut de_data).expect("decompress failed:");
+            assert_eq!(test_data, de_data.as_ref());
         }
     }
 }