Merge branch 'master' into fix-client-path

.github/workflows/CI.yml

@@ -16,7 +16,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
-    - uses: hecrj/setup-rust-action@v1
+    - uses: hecrj/setup-rust-action@v2
       with:
         components: rustfmt
     - run: cargo fmt --all --check
@@ -32,7 +32,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
-    - uses: hecrj/setup-rust-action@v1
+    - uses: hecrj/setup-rust-action@v2
     - name: Install protoc
       uses: taiki-e/install-action@v2
       with:
@@ -69,7 +69,7 @@ jobs:
         os: [ubuntu-latest, macOS-latest, windows-latest]
     steps:
     - uses: actions/checkout@v3
-    - uses: hecrj/setup-rust-action@v1
+    - uses: hecrj/setup-rust-action@v2
     - uses: taiki-e/install-action@cargo-hack
     - name: Install protoc
       uses: taiki-e/install-action@v2
@@ -86,9 +86,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
-    - uses: hecrj/setup-rust-action@v1
+    - uses: hecrj/setup-rust-action@v2
       with:
-        rust-version: "1.64"    # msrv
+        rust-version: "1.65"    # msrv
     - name: Install protoc
       uses: taiki-e/install-action@v2
       with:
@@ -106,7 +106,7 @@ jobs:
         os: [ubuntu-latest, macOS-latest, windows-latest]
     steps:
     - uses: actions/checkout@v3
-    - uses: hecrj/setup-rust-action@v1
+    - uses: hecrj/setup-rust-action@v2
     - name: Install protoc
       uses: taiki-e/install-action@v2
       with:
@@ -124,7 +124,7 @@ jobs:
         os: [ubuntu-latest, macOS-latest, windows-latest]
     steps:
     - uses: actions/checkout@v3
-    - uses: hecrj/setup-rust-action@v1
+    - uses: hecrj/setup-rust-action@v2
     - name: Install protoc
       uses: taiki-e/install-action@v2
       with:

README.md

@@ -45,7 +45,7 @@ For IntelliJ IDEA users, please refer to [this](https://github.com/intellij-rust
 
 ### Rust Version
 
-`tonic`'s MSRV is `1.64`.
+`tonic`'s MSRV is `1.65`.
 
 ```bash
 $ rustup update

codegen/src/main.rs

@@ -1,4 +1,4 @@
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 fn main() {
     // tonic-health
@@ -45,11 +45,11 @@ fn main() {
 }
 
 fn codegen(
-    root_dir: &PathBuf,
+    root_dir: &Path,
     iface_files: &[&str],
     include_dirs: &[&str],
-    out_dir: &PathBuf,
-    file_descriptor_set_path: &PathBuf,
+    out_dir: &Path,
+    file_descriptor_set_path: &Path,
     build_client: bool,
     build_server: bool,
 ) {
@@ -59,12 +59,12 @@ fn codegen(
         .unwrap();
 
     let iface_files: Vec<PathBuf> = iface_files
-        .into_iter()
+        .iter()
         .map(|&path| root_dir.join(path))
         .collect();
 
     let include_dirs: Vec<PathBuf> = include_dirs
-        .into_iter()
+        .iter()
         .map(|&path| root_dir.join(path))
         .collect();
     let out_dir = root_dir.join(out_dir);

examples/src/dynamic/server.rs

@@ -32,10 +32,7 @@ impl Echo for MyEcho {
 }
 
 fn init_echo(args: &[String], builder: &mut RoutesBuilder) {
-    let enabled = args
-        .into_iter()
-        .find(|arg| arg.as_str() == "echo")
-        .is_some();
+    let enabled = args.iter().any(|arg| arg.as_str() == "echo");
     if enabled {
         println!("Adding Echo service...");
         let svc = EchoServer::new(MyEcho::default());
@@ -62,10 +59,7 @@ impl Greeter for MyGreeter {
 }
 
 fn init_greeter(args: &[String], builder: &mut RoutesBuilder) {
-    let enabled = args
-        .into_iter()
-        .find(|arg| arg.as_str() == "greeter")
-        .is_some();
+    let enabled = args.iter().any(|arg| arg.as_str() == "greeter");
 
     if enabled {
         println!("Adding Greeter service...");

examples/src/mock/mock.rs

@@ -23,7 +23,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     tokio::spawn(async move {
         Server::builder()
             .add_service(GreeterServer::new(greeter))
-            .serve_with_incoming(tokio_stream::iter(vec![Ok::<_, std::io::Error>(server)]))
+            .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
             .await
     });
 

examples/src/richer-error/client.rs

@@ -46,7 +46,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     };
 
-    println!(" Successfull response received.\n\n {:?}\n", response);
+    println!(" Successful response received.\n\n {:?}\n", response);
 
     Ok(())
 }

examples/src/richer-error/client_vec.rs

@@ -52,7 +52,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     };
 
-    println!(" Successfull response received.\n\n {:?}\n", response);
+    println!(" Successful response received.\n\n {:?}\n", response);
 
     Ok(())
 }

interop/src/client.rs

@@ -207,7 +207,7 @@ pub async fn ping_pong(client: &mut TestClient, assertions: &mut Vec<TestAsserti
 }
 
 pub async fn empty_stream(client: &mut TestClient, assertions: &mut Vec<TestAssertion>) {
-    let stream = tokio_stream::iter(Vec::new());
+    let stream = tokio_stream::empty();
     let result = client.full_duplex_call(Request::new(stream)).await;
 
     assertions.push(test_assert!(
@@ -270,7 +270,7 @@ pub async fn status_code_and_message(client: &mut TestClient, assertions: &mut V
     let result = client.unary_call(Request::new(simple_req)).await;
     validate_response(result, assertions);
 
-    let stream = tokio_stream::iter(vec![duplex_req]);
+    let stream = tokio_stream::once(duplex_req);
     let result = match client.full_duplex_call(Request::new(stream)).await {
         Ok(response) => {
             let stream = response.into_inner();
@@ -356,7 +356,7 @@ pub async fn custom_metadata(client: &mut TestClient, assertions: &mut Vec<TestA
     req_unary.metadata_mut().insert(key1, value1.clone());
     req_unary.metadata_mut().insert_bin(key2, value2.clone());
 
-    let stream = tokio_stream::iter(vec![make_ping_pong_request(0)]);
+    let stream = tokio_stream::once(make_ping_pong_request(0));
     let mut req_stream = Request::new(stream);
     req_stream.metadata_mut().insert(key1, value1.clone());
     req_stream.metadata_mut().insert_bin(key2, value2.clone());

interop/src/server.rs

@@ -7,7 +7,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio_stream::StreamExt;
-use tonic::{body::BoxBody, transport::NamedService, Code, Request, Response, Status};
+use tonic::{body::BoxBody, server::NamedService, Code, Request, Response, Status};
 use tower::Service;
 
 pub use pb::test_service_server::TestServiceServer;
@@ -115,7 +115,7 @@ impl pb::test_service_server::TestService for TestService {
                 return Err(status);
             }
 
-            let single_message = tokio_stream::iter(vec![Ok(first_msg)]);
+            let single_message = tokio_stream::once(Ok(first_msg));
             let mut stream = single_message.chain(stream);
 
             let stream = try_stream! {

tests/compression/Cargo.toml

@@ -11,11 +11,12 @@ bytes = "1"
 http = "0.2"
 http-body = "0.4"
 hyper = "0.14.3"
+paste = "1.0.12"
 pin-project = "1.0"
 prost = "0.12"
 tokio = {version = "1.0", features = ["macros", "rt-multi-thread", "net"]}
 tokio-stream = "0.1"
-tonic = {path = "../../tonic", features = ["gzip"]}
+tonic = {path = "../../tonic", features = ["gzip", "zstd"]}
 tower = {version = "0.4", features = []}
 tower-http = {version = "0.4", features = ["map-response-body", "map-request-body"]}
 

tests/compression/src/bidirectional_stream.rs

@@ -1,20 +1,45 @@
 use super::*;
+use http_body::Body;
 use tonic::codec::CompressionEncoding;
 
-#[tokio::test(flavor = "multi_thread")]
-async fn client_enabled_server_enabled() {
+util::parametrized_tests! {
+    client_enabled_server_enabled,
+    zstd: CompressionEncoding::Zstd,
+    gzip: CompressionEncoding::Gzip,
+}
+
+#[allow(dead_code)]
+async fn client_enabled_server_enabled(encoding: CompressionEncoding) {
     let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
 
     let svc = test_server::TestServer::new(Svc::default())
-        .accept_compressed(CompressionEncoding::Gzip)
-        .send_compressed(CompressionEncoding::Gzip);
+        .accept_compressed(encoding)
+        .send_compressed(encoding);
 
     let request_bytes_counter = Arc::new(AtomicUsize::new(0));
     let response_bytes_counter = Arc::new(AtomicUsize::new(0));
 
-    fn assert_right_encoding<B>(req: http::Request<B>) -> http::Request<B> {
-        assert_eq!(req.headers().get("grpc-encoding").unwrap(), "gzip");
-        req
+    #[derive(Clone)]
+    pub struct AssertRightEncoding {
+        encoding: CompressionEncoding,
+    }
+
+    #[allow(dead_code)]
+    impl AssertRightEncoding {
+        pub fn new(encoding: CompressionEncoding) -> Self {
+            Self { encoding }
+        }
+
+        pub fn call<B: Body>(self, req: http::Request<B>) -> http::Request<B> {
+            let expected = match self.encoding {
+                CompressionEncoding::Gzip => "gzip",
+                CompressionEncoding::Zstd => "zstd",
+                _ => panic!("unexpected encoding {:?}", self.encoding),
+            };
+            assert_eq!(req.headers().get("grpc-encoding").unwrap(), expected);
+
+            req
+        }
     }
 
     tokio::spawn({
@@ -24,7 +49,9 @@ async fn client_enabled_server_enabled() {
             Server::builder()
                 .layer(
                     ServiceBuilder::new()
-                        .map_request(assert_right_encoding)
+                        .map_request(move |req| {
+                            AssertRightEncoding::new(encoding).clone().call(req)
+                        })
                         .layer(measure_request_body_size_layer(
                             request_bytes_counter.clone(),
                         ))
@@ -37,15 +64,15 @@ async fn client_enabled_server_enabled() {
                         .into_inner(),
                 )
                 .add_service(svc)
-                .serve_with_incoming(tokio_stream::iter(vec![Ok::<_, std::io::Error>(server)]))
+                .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
                 .await
                 .unwrap();
         }
     });
 
     let mut client = test_client::TestClient::new(mock_io_channel(client).await)
-        .send_compressed(CompressionEncoding::Gzip)
-        .accept_compressed(CompressionEncoding::Gzip);
+        .send_compressed(encoding)
+        .accept_compressed(encoding);
 
     let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec();
     let stream = tokio_stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]);
@@ -56,7 +83,12 @@ async fn client_enabled_server_enabled() {
         .await
         .unwrap();
 
-    assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip");
+    let expected = match encoding {
+        CompressionEncoding::Gzip => "gzip",
+        CompressionEncoding::Zstd => "zstd",
+        _ => panic!("unexpected encoding {:?}", encoding),
+    };
+    assert_eq!(res.metadata().get("grpc-encoding").unwrap(), expected);
 
     let mut stream: Streaming<SomeData> = res.into_inner();