Merge branch 'hyperium:master' into jparris/local_address

.github/workflows/CI.yml

@@ -88,7 +88,7 @@ jobs:
     - uses: actions/checkout@v3
     - uses: hecrj/setup-rust-action@v1
       with:
-        rust-version: "1.64"    # msrv
+        rust-version: "1.65"    # msrv
     - 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

@@ -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/src/bidirectional_stream.rs

@@ -37,7 +37,7 @@ 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();
         }

tests/compression/src/client_stream.rs

@@ -29,7 +29,7 @@ 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();
         }
@@ -75,7 +75,7 @@ async fn client_disabled_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();
         }
@@ -102,7 +102,7 @@ async fn client_enabled_server_disabled() {
     tokio::spawn(async move {
         Server::builder()
             .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();
     });
@@ -147,7 +147,7 @@ async fn compressing_response_from_client_stream() {
                         .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();
         }
@@ -156,8 +156,7 @@ async fn compressing_response_from_client_stream() {
     let mut client = test_client::TestClient::new(mock_io_channel(client).await)
         .accept_compressed(CompressionEncoding::Gzip);
 
-    let stream = tokio_stream::iter(vec![]);
-    let req = Request::new(Box::pin(stream));
+    let req = Request::new(Box::pin(tokio_stream::empty()));
 
     let res = client.compress_output_client_stream(req).await.unwrap();
     assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip");

tests/compression/src/compressing_request.rs

@@ -31,7 +31,7 @@ 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();
         }
@@ -61,7 +61,7 @@ async fn client_enabled_server_disabled() {
     tokio::spawn(async move {
         Server::builder()
             .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();
     });
@@ -99,7 +99,7 @@ async fn client_mark_compressed_without_header_server_enabled() {
         async move {
             Server::builder()
                 .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();
         }

tests/compression/src/compressing_response.rs

@@ -53,7 +53,7 @@ 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();
         }
@@ -94,7 +94,7 @@ async fn client_enabled_server_disabled() {
                         .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();
         }
@@ -160,7 +160,7 @@ async fn client_disabled() {
                         .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();
         }
@@ -198,7 +198,7 @@ async fn server_replying_with_unsupported_encoding() {
                     .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();
     });
@@ -240,7 +240,7 @@ async fn disabling_compression_on_single_response() {
                         .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();
         }
@@ -281,7 +281,7 @@ async fn disabling_compression_on_response_but_keeping_compression_on_stream() {
                         .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();
         }
@@ -337,7 +337,7 @@ async fn disabling_compression_on_response_from_client_stream() {
                         .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();
         }
@@ -346,8 +346,7 @@ async fn disabling_compression_on_response_from_client_stream() {
     let mut client = test_client::TestClient::new(mock_io_channel(client).await)
         .accept_compressed(CompressionEncoding::Gzip);
 
-    let stream = tokio_stream::iter(vec![]);
-    let req = Request::new(Box::pin(stream));
+    let req = Request::new(Box::pin(tokio_stream::empty()));
 
     let res = client.compress_output_client_stream(req).await.unwrap();
     assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip");

tests/compression/src/server_stream.rs

@@ -26,7 +26,7 @@ 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();
         }
@@ -80,7 +80,7 @@ async fn client_disabled_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();
         }
@@ -125,7 +125,7 @@ async fn client_enabled_server_disabled() {
                         .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();
         }

tests/integration_tests/src/lib.rs

@@ -5,6 +5,7 @@ pub mod pb {
 
 pub mod mock {
     use std::{
+        io::IoSlice,
         pin::Pin,
         task::{Context, Poll},
     };
@@ -51,6 +52,18 @@ pub mod mock {
         ) -> Poll<std::io::Result<()>> {
             Pin::new(&mut self.0).poll_shutdown(cx)
         }
+
+        fn poll_write_vectored(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            bufs: &[IoSlice<'_>],
+        ) -> Poll<Result<usize, std::io::Error>> {
+            Pin::new(&mut self.0).poll_write_vectored(cx, bufs)
+        }
+
+        fn is_write_vectored(&self) -> bool {
+            self.0.is_write_vectored()
+        }
     }
 }
 

tests/integration_tests/tests/max_message_size.rs

@@ -148,7 +148,7 @@ async fn response_stream_limit() {
     tokio::spawn(async move {
         Server::builder()
             .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();
     });
@@ -317,7 +317,7 @@ async fn max_message_run(case: &TestCase) -> Result<(), Status> {
     tokio::spawn(async move {
         Server::builder()
             .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();
     });

tests/integration_tests/tests/routes_builder.rs

@@ -43,7 +43,7 @@ async fn multiple_service_using_routes_builder() {
             let output = Output1 {
                 buf: request.into_inner().buf,
             };
-            let stream = tokio_stream::iter(vec![Ok(output)]);
+            let stream = tokio_stream::once(Ok(output));
 
             Ok(Response::new(Box::pin(stream)))
         }

tonic-build/src/prost.rs

@@ -506,7 +506,7 @@ impl Builder {
     /// should be enabled automatically when run from a build script. However,
     /// the method of detection is not completely reliable since the `CARGO`
     /// environment variable can have been set by anything else. If writing the
-    /// instructions to `stdout` is undesireable, you can disable this setting
+    /// instructions to `stdout` is undesirable, you can disable this setting
     /// explicitly.
     pub fn emit_rerun_if_changed(mut self, enable: bool) -> Self {
         self.emit_rerun_if_changed = enable;