Address review feedback

tests/compression/src/bidirectional_stream.rs

@@ -50,7 +50,7 @@ async fn client_enabled_server_enabled() {
 
  let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec();
  let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]);
- let req = Request::new(Box::pin(stream));
+ let req = Request::new(stream);
 
  let res = client
  .compress_input_output_bidirectional_stream(req)
@@ -73,6 +73,6 @@ async fn client_enabled_server_enabled() {
  .expect("stream empty")
  .expect("item was error");
 
- assert!(dbg!(request_bytes_counter.load(SeqCst)) < UNCOMPRESSED_MIN_BODY_SIZE);
- assert!(dbg!(response_bytes_counter.load(SeqCst)) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(request_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
 }

tests/compression/src/client_stream.rs

@@ -44,7 +44,7 @@ async fn client_enabled_server_enabled() {
  client.compress_input_client_stream(req).await.unwrap();
 
  let bytes_sent = request_bytes_counter.load(SeqCst);
- assert!(dbg!(bytes_sent) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(bytes_sent < UNCOMPRESSED_MIN_BODY_SIZE);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -90,7 +90,7 @@ async fn client_disabled_server_enabled() {
  client.compress_input_client_stream(req).await.unwrap();
 
  let bytes_sent = request_bytes_counter.load(SeqCst);
- assert!(dbg!(bytes_sent) > UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(bytes_sent > UNCOMPRESSED_MIN_BODY_SIZE);
 }
 
 #[tokio::test(flavor = "multi_thread")]

tests/compression/src/compressing_request.rs

@@ -47,7 +47,7 @@ async fn client_enabled_server_enabled() {
  .await
  .unwrap();
  let bytes_sent = request_bytes_counter.load(SeqCst);
- assert!(dbg!(bytes_sent) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(bytes_sent < UNCOMPRESSED_MIN_BODY_SIZE);
  }
 }
 

tests/compression/src/compressing_response.rs

@@ -304,14 +304,14 @@ async fn disabling_compression_on_response_but_keeping_compression_on_stream() {
  .await
  .expect("stream empty")
  .expect("item was error");
- assert!(dbg!(response_bytes_counter.load(SeqCst)) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
 
  stream
  .next()
  .await
  .expect("stream empty")
  .expect("item was error");
- assert!(dbg!(response_bytes_counter.load(SeqCst)) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
 }
 
 #[tokio::test(flavor = "multi_thread")]

tests/compression/src/server_stream.rs

@@ -45,14 +45,14 @@ async fn client_enabled_server_enabled() {
  .await
  .expect("stream empty")
  .expect("item was error");
- assert!(dbg!(response_bytes_counter.load(SeqCst)) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
 
  stream
  .next()
  .await
  .expect("stream empty")
  .expect("item was error");
- assert!(dbg!(response_bytes_counter.load(SeqCst)) < UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -99,7 +99,7 @@ async fn client_disabled_server_enabled() {
  .await
  .expect("stream empty")
  .expect("item was error");
- assert!(dbg!(response_bytes_counter.load(SeqCst)) > UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) > UNCOMPRESSED_MIN_BODY_SIZE);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -146,5 +146,5 @@ async fn client_enabled_server_disabled() {
  .await
  .expect("stream empty")
  .expect("item was error");
- assert!(dbg!(response_bytes_counter.load(SeqCst)) > UNCOMPRESSED_MIN_BODY_SIZE);
+ assert!(response_bytes_counter.load(SeqCst) > UNCOMPRESSED_MIN_BODY_SIZE);
 }

tonic/Cargo.toml

@@ -39,7 +39,7 @@ tls-roots-common = ["tls"]
 tls-roots = ["tls-roots-common", "rustls-native-certs"]
 tls-webpki-roots = ["tls-roots-common", "webpki-roots"]
 prost = ["prost1", "prost-derive"]
-compression = ["tokio/rt-multi-thread", "flate2"]
+compression = ["flate2"]
 
 # [[bench]]
 # name = "bench_main"

tonic/src/client/grpc.rs

@@ -86,7 +86,7 @@ impl<T> Grpc<T> {
  #[cfg(not(feature = "compression"))]
  pub fn send_gzip(self) -> Self {
  panic!(
- "`send_gzip` called on a server but the `compression` feature is not enabled on tonic"
+ "`send_gzip` called on a client but the `compression` feature is not enabled on tonic"
  );
  }
 

tonic/src/codec/compression.rs

@@ -137,7 +137,7 @@ pub(crate) fn compress(
  );
  let mut out_writer = out_buf.writer();
 
- tokio::task::block_in_place(|| std::io::copy(&mut gzip_encoder, &mut out_writer))?;
+ std::io::copy(&mut gzip_encoder, &mut out_writer)?;
  }
  }
 
@@ -162,7 +162,7 @@ pub(crate) fn decompress(
  let mut gzip_decoder = GzDecoder::new(&compressed_buf[0..len]);
  let mut out_writer = out_buf.writer();
 
- tokio::task::block_in_place(|| std::io::copy(&mut gzip_decoder, &mut out_writer))?;
+ std::io::copy(&mut gzip_decoder, &mut out_writer)?;
  }
  }
 

tonic/src/lib.rs

@@ -30,8 +30,7 @@
 //! - `prost`: Enables the [`prost`] based gRPC [`Codec`] implementation.
 //! - `compression`: Enables compressing requests, responses, and streams. Note
 //! that you must enable the `compression` feature on both `tonic` and
-//! `tonic-build` to use it. Depends on `tokio`'s multi-threaded runtime and
-//! [flate2]. Not enabled by default.
+//! `tonic-build` to use it. Depends on [flate2]. Not enabled by default.
 //!
 //! # Structure
 //!

tonic/src/server/grpc.rs

@@ -267,7 +267,6 @@ where
  S: ClientStreamingService<T::Decode, Response = T::Encode>,
  B: Body + Send + Sync + 'static,
  B::Error: Into<crate::Error> + Send + 'static,
- T: std::fmt::Debug,
  {
  #[cfg(feature = "compression")]
  let accept_encoding = CompressionEncoding::from_accept_encoding_header(