feat!: remove codegen dependency on compression feature (#1004)

This commit is contained in:
Marcus Griep
2022-06-13 17:48:27 -04:00
committed by GitHub
parent d9f5c75112
commit a585a72293
21 changed files with 264 additions and 388 deletions
+2 -2
View File
@@ -200,7 +200,7 @@ futures = { version = "0.3", default-features = false, features = ["alloc"] }
prost = "0.10" prost = "0.10"
tokio = { version = "1.0", features = [ "rt-multi-thread", "time", "fs", "macros", "net",] } tokio = { version = "1.0", features = [ "rt-multi-thread", "time", "fs", "macros", "net",] }
tokio-stream = { version = "0.1", features = ["net"] } tokio-stream = { version = "0.1", features = ["net"] }
tonic = { path = "../tonic", features = ["tls", "compression"] } tonic = { path = "../tonic", features = ["tls", "gzip"] }
tower = { version = "0.4" } tower = { version = "0.4" }
# Required for routeguide # Required for routeguide
rand = "0.8" rand = "0.8"
@@ -237,4 +237,4 @@ tower-http = { version = "0.3", features = ["add-extension", "util"] }
[build-dependencies] [build-dependencies]
tonic-build = { path = "../tonic-build", features = ["prost", "compression"] } tonic-build = { path = "../tonic-build", features = ["prost"] }
+4 -1
View File
@@ -1,5 +1,6 @@
use hello_world::greeter_client::GreeterClient; use hello_world::greeter_client::GreeterClient;
use hello_world::HelloRequest; use hello_world::HelloRequest;
use tonic::codec::CompressionEncoding;
use tonic::transport::Channel; use tonic::transport::Channel;
pub mod hello_world { pub mod hello_world {
@@ -13,7 +14,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.await .await
.unwrap(); .unwrap();
let mut client = GreeterClient::new(channel).send_gzip().accept_gzip(); let mut client = GreeterClient::new(channel)
.send_compressed(CompressionEncoding::Gzip)
.accept_compressed(CompressionEncoding::Gzip);
let request = tonic::Request::new(HelloRequest { let request = tonic::Request::new(HelloRequest {
name: "Tonic".into(), name: "Tonic".into(),
+4 -1
View File
@@ -2,6 +2,7 @@ use tonic::{transport::Server, Request, Response, Status};
use hello_world::greeter_server::{Greeter, GreeterServer}; use hello_world::greeter_server::{Greeter, GreeterServer};
use hello_world::{HelloReply, HelloRequest}; use hello_world::{HelloReply, HelloRequest};
use tonic::codec::CompressionEncoding;
pub mod hello_world { pub mod hello_world {
tonic::include_proto!("helloworld"); tonic::include_proto!("helloworld");
@@ -32,7 +33,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("GreeterServer listening on {}", addr); println!("GreeterServer listening on {}", addr);
let service = GreeterServer::new(greeter).send_gzip().accept_gzip(); let service = GreeterServer::new(greeter)
.send_compressed(CompressionEncoding::Gzip)
.accept_compressed(CompressionEncoding::Gzip);
Server::builder().add_service(service).serve(addr).await?; Server::builder().add_service(service).serve(addr).await?;
+2 -2
View File
@@ -16,9 +16,9 @@ pin-project = "1.0"
prost = "0.10" prost = "0.10"
tokio = {version = "1.0", features = ["macros", "rt-multi-thread", "net"]} tokio = {version = "1.0", features = ["macros", "rt-multi-thread", "net"]}
tokio-stream = {version = "0.1.5", features = ["net"]} tokio-stream = {version = "0.1.5", features = ["net"]}
tonic = {path = "../../tonic", features = ["compression"]} tonic = {path = "../../tonic", features = ["gzip"]}
tower = {version = "0.4", features = []} tower = {version = "0.4", features = []}
tower-http = {version = "0.3", features = ["map-response-body", "map-request-body"]} tower-http = {version = "0.3", features = ["map-response-body", "map-request-body"]}
[build-dependencies] [build-dependencies]
tonic-build = {path = "../../tonic-build", features = ["compression"]} tonic-build = {path = "../../tonic-build" }
@@ -1,12 +1,13 @@
use super::*; use super::*;
use tonic::codec::CompressionEncoding;
#[tokio::test(flavor = "multi_thread")] #[tokio::test(flavor = "multi_thread")]
async fn client_enabled_server_enabled() { async fn client_enabled_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()) let svc = test_server::TestServer::new(Svc::default())
.accept_gzip() .accept_compressed(CompressionEncoding::Gzip)
.send_gzip(); .send_compressed(CompressionEncoding::Gzip);
let request_bytes_counter = Arc::new(AtomicUsize::new(0)); let request_bytes_counter = Arc::new(AtomicUsize::new(0));
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -43,8 +44,8 @@ async fn client_enabled_server_enabled() {
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await) let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.send_gzip() .send_compressed(CompressionEncoding::Gzip)
.accept_gzip(); .accept_compressed(CompressionEncoding::Gzip);
let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(); let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec();
let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]); let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]);
+13 -6
View File
@@ -1,11 +1,13 @@
use super::*; use super::*;
use http_body::Body as _; use http_body::Body as _;
use tonic::codec::CompressionEncoding;
#[tokio::test(flavor = "multi_thread")] #[tokio::test(flavor = "multi_thread")]
async fn client_enabled_server_enabled() { async fn client_enabled_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).accept_gzip(); let svc =
test_server::TestServer::new(Svc::default()).accept_compressed(CompressionEncoding::Gzip);
let request_bytes_counter = Arc::new(AtomicUsize::new(0)); let request_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -33,7 +35,8 @@ async fn client_enabled_server_enabled() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).send_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.send_compressed(CompressionEncoding::Gzip);
let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(); let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec();
let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]); let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]);
@@ -49,7 +52,8 @@ async fn client_enabled_server_enabled() {
async fn client_disabled_server_enabled() { async fn client_disabled_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).accept_gzip(); let svc =
test_server::TestServer::new(Svc::default()).accept_compressed(CompressionEncoding::Gzip);
let request_bytes_counter = Arc::new(AtomicUsize::new(0)); let request_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -103,7 +107,8 @@ async fn client_enabled_server_disabled() {
.unwrap(); .unwrap();
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).send_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.send_compressed(CompressionEncoding::Gzip);
let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(); let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec();
let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]); let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]);
@@ -122,7 +127,8 @@ async fn client_enabled_server_disabled() {
async fn compressing_response_from_client_stream() { async fn compressing_response_from_client_stream() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).send_gzip(); let svc =
test_server::TestServer::new(Svc::default()).send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -147,7 +153,8 @@ async fn compressing_response_from_client_stream() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let stream = futures::stream::iter(vec![]); let stream = futures::stream::iter(vec![]);
let req = Request::new(Box::pin(stream)); let req = Request::new(Box::pin(stream));
+10 -5
View File
@@ -1,11 +1,13 @@
use super::*; use super::*;
use http_body::Body as _; use http_body::Body as _;
use tonic::codec::CompressionEncoding;
#[tokio::test(flavor = "multi_thread")] #[tokio::test(flavor = "multi_thread")]
async fn client_enabled_server_enabled() { async fn client_enabled_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).accept_gzip(); let svc =
test_server::TestServer::new(Svc::default()).accept_compressed(CompressionEncoding::Gzip);
let request_bytes_counter = Arc::new(AtomicUsize::new(0)); let request_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -35,7 +37,8 @@ async fn client_enabled_server_enabled() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).send_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.send_compressed(CompressionEncoding::Gzip);
for _ in 0..3 { for _ in 0..3 {
client client
@@ -63,7 +66,8 @@ async fn client_enabled_server_disabled() {
.unwrap(); .unwrap();
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).send_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.send_compressed(CompressionEncoding::Gzip);
let status = client let status = client
.compress_input_unary(SomeData { .compress_input_unary(SomeData {
@@ -88,7 +92,8 @@ async fn client_enabled_server_disabled() {
async fn client_mark_compressed_without_header_server_enabled() { async fn client_mark_compressed_without_header_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).accept_gzip(); let svc =
test_server::TestServer::new(Svc::default()).accept_compressed(CompressionEncoding::Gzip);
tokio::spawn({ tokio::spawn({
async move { async move {
@@ -107,7 +112,7 @@ async fn client_mark_compressed_without_header_server_enabled() {
Ok(req) Ok(req)
}, },
) )
.send_gzip(); .send_compressed(CompressionEncoding::Gzip);
let status = client let status = client
.compress_input_unary(SomeData { .compress_input_unary(SomeData {
+22 -12
View File
@@ -1,4 +1,5 @@
use super::*; use super::*;
use tonic::codec::CompressionEncoding;
#[tokio::test(flavor = "multi_thread")] #[tokio::test(flavor = "multi_thread")]
async fn client_enabled_server_enabled() { async fn client_enabled_server_enabled() {
@@ -31,7 +32,8 @@ async fn client_enabled_server_enabled() {
} }
} }
let svc = test_server::TestServer::new(Svc::default()).send_gzip(); let svc =
test_server::TestServer::new(Svc::default()).send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -57,7 +59,8 @@ async fn client_enabled_server_enabled() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
for _ in 0..3 { for _ in 0..3 {
let res = client.compress_output_unary(()).await.unwrap(); let res = client.compress_output_unary(()).await.unwrap();
@@ -97,7 +100,8 @@ async fn client_enabled_server_disabled() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let res = client.compress_output_unary(()).await.unwrap(); let res = client.compress_output_unary(()).await.unwrap();
@@ -135,7 +139,8 @@ async fn client_disabled() {
} }
} }
let svc = test_server::TestServer::new(Svc::default()).send_gzip(); let svc =
test_server::TestServer::new(Svc::default()).send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -175,7 +180,8 @@ async fn client_disabled() {
async fn server_replying_with_unsupported_encoding() { async fn server_replying_with_unsupported_encoding() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).send_gzip(); let svc =
test_server::TestServer::new(Svc::default()).send_compressed(CompressionEncoding::Gzip);
fn add_weird_content_encoding<B>(mut response: http::Response<B>) -> http::Response<B> { fn add_weird_content_encoding<B>(mut response: http::Response<B>) -> http::Response<B> {
response response
@@ -197,7 +203,8 @@ async fn server_replying_with_unsupported_encoding() {
.unwrap(); .unwrap();
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let status: Status = client.compress_output_unary(()).await.unwrap_err(); let status: Status = client.compress_output_unary(()).await.unwrap_err();
assert_eq!(status.code(), tonic::Code::Unimplemented); assert_eq!(status.code(), tonic::Code::Unimplemented);
@@ -214,7 +221,7 @@ async fn disabling_compression_on_single_response() {
let svc = test_server::TestServer::new(Svc { let svc = test_server::TestServer::new(Svc {
disable_compressing_on_response: true, disable_compressing_on_response: true,
}) })
.send_gzip(); .send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -239,7 +246,8 @@ async fn disabling_compression_on_single_response() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let res = client.compress_output_unary(()).await.unwrap(); let res = client.compress_output_unary(()).await.unwrap();
assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip"); assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip");
@@ -254,7 +262,7 @@ async fn disabling_compression_on_response_but_keeping_compression_on_stream() {
let svc = test_server::TestServer::new(Svc { let svc = test_server::TestServer::new(Svc {
disable_compressing_on_response: true, disable_compressing_on_response: true,
}) })
.send_gzip(); .send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -279,7 +287,8 @@ async fn disabling_compression_on_response_but_keeping_compression_on_stream() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let res = client.compress_output_server_stream(()).await.unwrap(); let res = client.compress_output_server_stream(()).await.unwrap();
@@ -309,7 +318,7 @@ async fn disabling_compression_on_response_from_client_stream() {
let svc = test_server::TestServer::new(Svc { let svc = test_server::TestServer::new(Svc {
disable_compressing_on_response: true, disable_compressing_on_response: true,
}) })
.send_gzip(); .send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -334,7 +343,8 @@ async fn disabling_compression_on_response_from_client_stream() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let stream = futures::stream::iter(vec![]); let stream = futures::stream::iter(vec![]);
let req = Request::new(Box::pin(stream)); let req = Request::new(Box::pin(stream));
+9 -4
View File
@@ -1,11 +1,13 @@
use super::*; use super::*;
use tonic::codec::CompressionEncoding;
use tonic::Streaming; use tonic::Streaming;
#[tokio::test(flavor = "multi_thread")] #[tokio::test(flavor = "multi_thread")]
async fn client_enabled_server_enabled() { async fn client_enabled_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).send_gzip(); let svc =
test_server::TestServer::new(Svc::default()).send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -30,7 +32,8 @@ async fn client_enabled_server_enabled() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let res = client.compress_output_server_stream(()).await.unwrap(); let res = client.compress_output_server_stream(()).await.unwrap();
@@ -57,7 +60,8 @@ async fn client_enabled_server_enabled() {
async fn client_disabled_server_enabled() { async fn client_disabled_server_enabled() {
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10); let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
let svc = test_server::TestServer::new(Svc::default()).send_gzip(); let svc =
test_server::TestServer::new(Svc::default()).send_compressed(CompressionEncoding::Gzip);
let response_bytes_counter = Arc::new(AtomicUsize::new(0)); let response_bytes_counter = Arc::new(AtomicUsize::new(0));
@@ -127,7 +131,8 @@ async fn client_enabled_server_disabled() {
} }
}); });
let mut client = test_client::TestClient::new(mock_io_channel(client).await).accept_gzip(); let mut client = test_client::TestClient::new(mock_io_channel(client).await)
.accept_compressed(CompressionEncoding::Gzip);
let res = client.compress_output_server_stream(()).await.unwrap(); let res = client.compress_output_server_stream(()).await.unwrap();
-1
View File
@@ -22,7 +22,6 @@ quote = "1.0"
syn = "1.0" syn = "1.0"
[features] [features]
compression = []
default = ["transport", "prost"] default = ["transport", "prost"]
prost = ["prost-build"] prost = ["prost-build"]
transport = [] transport = []
+6 -6
View File
@@ -79,20 +79,20 @@ pub fn generate<T: Service>(
#service_ident::new(InterceptedService::new(inner, interceptor)) #service_ident::new(InterceptedService::new(inner, interceptor))
} }
/// Compress requests with `gzip`. /// Compress requests with the given encoding.
/// ///
/// This requires the server to support it otherwise it might respond with an /// This requires the server to support it otherwise it might respond with an
/// error. /// error.
#[must_use] #[must_use]
pub fn send_gzip(mut self) -> Self { pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.inner = self.inner.send_gzip(); self.inner = self.inner.send_compressed(encoding);
self self
} }
/// Enable decompressing responses with `gzip`. /// Enable decompressing responses.
#[must_use] #[must_use]
pub fn accept_gzip(mut self) -> Self { pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.inner = self.inner.accept_gzip(); self.inner = self.inner.accept_compressed(encoding);
self self
} }
+14 -26
View File
@@ -39,32 +39,20 @@ pub fn generate<T: Service>(
let mod_attributes = attributes.for_mod(package); let mod_attributes = attributes.for_mod(package);
let struct_attributes = attributes.for_struct(&path); let struct_attributes = attributes.for_struct(&path);
let compression_enabled = cfg!(feature = "compression"); let configure_compression_methods = quote! {
/// Enable decompressing requests with the given encoding.
#[must_use]
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
self.accept_compression_encodings.enable(encoding);
self
}
let compression_config_ty = if compression_enabled { /// Compress responses with the given encoding, if the client supports it.
quote! { EnabledCompressionEncodings } #[must_use]
} else { pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
quote! { () } self.send_compression_encodings.enable(encoding);
}; self
let configure_compression_methods = if compression_enabled {
quote! {
/// Enable decompressing requests with `gzip`.
#[must_use]
pub fn accept_gzip(mut self) -> Self {
self.accept_compression_encodings.enable_gzip();
self
}
/// Compress responses with `gzip`, if the client supports it.
#[must_use]
pub fn send_gzip(mut self) -> Self {
self.send_compression_encodings.enable_gzip();
self
}
} }
} else {
quote! {}
}; };
quote! { quote! {
@@ -87,8 +75,8 @@ pub fn generate<T: Service>(
#[derive(Debug)] #[derive(Debug)]
pub struct #server_service<T: #server_trait> { pub struct #server_service<T: #server_trait> {
inner: _Inner<T>, inner: _Inner<T>,
accept_compression_encodings: #compression_config_ty, accept_compression_encodings: EnabledCompressionEncodings,
send_compression_encodings: #compression_config_ty, send_compression_encodings: EnabledCompressionEncodings,
} }
struct _Inner<T>(Arc<T>); struct _Inner<T>(Arc<T>);
+1 -1
View File
@@ -24,7 +24,7 @@ version = "0.7.2"
[features] [features]
codegen = ["async-trait"] codegen = ["async-trait"]
compression = ["flate2"] gzip = ["flate2"]
default = ["transport", "codegen", "prost"] default = ["transport", "codegen", "prost"]
prost = ["prost1", "prost-derive"] prost = ["prost1", "prost-derive"]
tls = ["rustls-pemfile", "transport", "tokio-rustls"] tls = ["rustls-pemfile", "transport", "tokio-rustls"]
+29 -71
View File
@@ -1,4 +1,3 @@
#[cfg(feature = "compression")]
use crate::codec::compression::{CompressionEncoding, EnabledCompressionEncodings}; use crate::codec::compression::{CompressionEncoding, EnabledCompressionEncodings};
use crate::{ use crate::{
body::BoxBody, body::BoxBody,
@@ -31,10 +30,8 @@ use std::fmt;
/// [gRPC protocol definition]: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests /// [gRPC protocol definition]: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
pub struct Grpc<T> { pub struct Grpc<T> {
inner: T, inner: T,
#[cfg(feature = "compression")]
/// Which compression encodings does the client accept? /// Which compression encodings does the client accept?
accept_compression_encodings: EnabledCompressionEncodings, accept_compression_encodings: EnabledCompressionEncodings,
#[cfg(feature = "compression")]
/// The compression encoding that will be applied to requests. /// The compression encoding that will be applied to requests.
send_compression_encodings: Option<CompressionEncoding>, send_compression_encodings: Option<CompressionEncoding>,
} }
@@ -44,16 +41,14 @@ impl<T> Grpc<T> {
pub fn new(inner: T) -> Self { pub fn new(inner: T) -> Self {
Self { Self {
inner, inner,
#[cfg(feature = "compression")]
send_compression_encodings: None, send_compression_encodings: None,
#[cfg(feature = "compression")]
accept_compression_encodings: EnabledCompressionEncodings::default(), accept_compression_encodings: EnabledCompressionEncodings::default(),
} }
} }
/// Compress requests with `gzip`. /// Compress requests with the provided encoding.
/// ///
/// Requires the server to accept `gzip` otherwise it might return an error. /// Requires the server to accept the specified encoding, otherwise it might return an error.
/// ///
/// # Example /// # Example
/// ///
@@ -61,10 +56,11 @@ impl<T> Grpc<T> {
/// ///
/// ```rust /// ```rust
/// use tonic::transport::Channel; /// use tonic::transport::Channel;
/// # enum CompressionEncoding { Gzip }
/// # struct TestClient<T>(T); /// # struct TestClient<T>(T);
/// # impl<T> TestClient<T> { /// # impl<T> TestClient<T> {
/// # fn new(channel: T) -> Self { Self(channel) } /// # fn new(channel: T) -> Self { Self(channel) }
/// # fn send_gzip(self) -> Self { self } /// # fn send_compressed(self, _: CompressionEncoding) -> Self { self }
/// # } /// # }
/// ///
/// # async { /// # async {
@@ -73,25 +69,15 @@ impl<T> Grpc<T> {
/// .await /// .await
/// .unwrap(); /// .unwrap();
/// ///
/// let client = TestClient::new(channel).send_gzip(); /// let client = TestClient::new(channel).send_compressed(CompressionEncoding::Gzip);
/// # }; /// # };
/// ``` /// ```
#[cfg(feature = "compression")] pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
#[cfg_attr(docsrs, doc(cfg(feature = "compression")))] self.send_compression_encodings = Some(encoding);
pub fn send_gzip(mut self) -> Self {
self.send_compression_encodings = Some(CompressionEncoding::Gzip);
self self
} }
#[doc(hidden)] /// Enable accepting compressed responses.
#[cfg(not(feature = "compression"))]
pub fn send_gzip(self) -> Self {
panic!(
"`send_gzip` called on a client but the `compression` feature is not enabled on tonic"
);
}
/// Enable accepting `gzip` compressed responses.
/// ///
/// Requires the server to also support sending compressed responses. /// Requires the server to also support sending compressed responses.
/// ///
@@ -101,10 +87,11 @@ impl<T> Grpc<T> {
/// ///
/// ```rust /// ```rust
/// use tonic::transport::Channel; /// use tonic::transport::Channel;
/// # enum CompressionEncoding { Gzip }
/// # struct TestClient<T>(T); /// # struct TestClient<T>(T);
/// # impl<T> TestClient<T> { /// # impl<T> TestClient<T> {
/// # fn new(channel: T) -> Self { Self(channel) } /// # fn new(channel: T) -> Self { Self(channel) }
/// # fn accept_gzip(self) -> Self { self } /// # fn accept_compressed(self, _: CompressionEncoding) -> Self { self }
/// # } /// # }
/// ///
/// # async { /// # async {
@@ -113,22 +100,14 @@ impl<T> Grpc<T> {
/// .await /// .await
/// .unwrap(); /// .unwrap();
/// ///
/// let client = TestClient::new(channel).accept_gzip(); /// let client = TestClient::new(channel).accept_compressed(CompressionEncoding::Gzip);
/// # }; /// # };
/// ``` /// ```
#[cfg(feature = "compression")] pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
#[cfg_attr(docsrs, doc(cfg(feature = "compression")))] self.accept_compression_encodings.enable(encoding);
pub fn accept_gzip(mut self) -> Self {
self.accept_compression_encodings.enable_gzip();
self self
} }
#[doc(hidden)]
#[cfg(not(feature = "compression"))]
pub fn accept_gzip(self) -> Self {
panic!("`accept_gzip` called on a client but the `compression` feature is not enabled on tonic");
}
/// Check if the inner [`GrpcService`] is able to accept a new request. /// Check if the inner [`GrpcService`] is able to accept a new request.
/// ///
/// This will call [`GrpcService::poll_ready`] until it returns ready or /// This will call [`GrpcService::poll_ready`] until it returns ready or
@@ -238,14 +217,7 @@ impl<T> Grpc<T> {
let uri = Uri::from_parts(parts).expect("path_and_query only is valid Uri"); let uri = Uri::from_parts(parts).expect("path_and_query only is valid Uri");
let request = request let request = request
.map(|s| { .map(|s| encode_client(codec.encoder(), s, self.send_compression_encodings))
encode_client(
codec.encoder(),
s,
#[cfg(feature = "compression")]
self.send_compression_encodings,
)
})
.map(BoxBody::new); .map(BoxBody::new);
let mut request = request.into_http( let mut request = request.into_http(
@@ -265,24 +237,21 @@ impl<T> Grpc<T> {
.headers_mut() .headers_mut()
.insert(CONTENT_TYPE, HeaderValue::from_static("application/grpc")); .insert(CONTENT_TYPE, HeaderValue::from_static("application/grpc"));
#[cfg(feature = "compression")] if let Some(encoding) = self.send_compression_encodings {
{ request.headers_mut().insert(
if let Some(encoding) = self.send_compression_encodings { crate::codec::compression::ENCODING_HEADER,
request.headers_mut().insert( encoding.into_header_value(),
crate::codec::compression::ENCODING_HEADER, );
encoding.into_header_value(), }
);
}
if let Some(header_value) = self if let Some(header_value) = self
.accept_compression_encodings .accept_compression_encodings
.into_accept_encoding_header_value() .into_accept_encoding_header_value()
{ {
request.headers_mut().insert( request.headers_mut().insert(
crate::codec::compression::ACCEPT_ENCODING_HEADER, crate::codec::compression::ACCEPT_ENCODING_HEADER,
header_value, header_value,
); );
}
} }
let response = self let response = self
@@ -291,7 +260,6 @@ impl<T> Grpc<T> {
.await .await
.map_err(|err| Status::from_error(err.into()))?; .map_err(|err| Status::from_error(err.into()))?;
#[cfg(feature = "compression")]
let encoding = CompressionEncoding::from_encoding_header( let encoding = CompressionEncoding::from_encoding_header(
response.headers(), response.headers(),
self.accept_compression_encodings, self.accept_compression_encodings,
@@ -314,13 +282,7 @@ impl<T> Grpc<T> {
let response = response.map(|body| { let response = response.map(|body| {
if expect_additional_trailers { if expect_additional_trailers {
Streaming::new_response( Streaming::new_response(codec.decoder(), body, status_code, encoding)
codec.decoder(),
body,
status_code,
#[cfg(feature = "compression")]
encoding,
)
} else { } else {
Streaming::new_empty(codec.decoder(), body) Streaming::new_empty(codec.decoder(), body)
} }
@@ -334,9 +296,7 @@ impl<T: Clone> Clone for Grpc<T> {
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
inner: self.inner.clone(), inner: self.inner.clone(),
#[cfg(feature = "compression")]
send_compression_encodings: self.send_compression_encodings, send_compression_encodings: self.send_compression_encodings,
#[cfg(feature = "compression")]
accept_compression_encodings: self.accept_compression_encodings, accept_compression_encodings: self.accept_compression_encodings,
} }
} }
@@ -348,10 +308,8 @@ impl<T: fmt::Debug> fmt::Debug for Grpc<T> {
f.field("inner", &self.inner); f.field("inner", &self.inner);
#[cfg(feature = "compression")]
f.field("compression_encoding", &self.send_compression_encodings); f.field("compression_encoding", &self.send_compression_encodings);
#[cfg(feature = "compression")]
f.field( f.field(
"accept_compression_encodings", "accept_compression_encodings",
&self.accept_compression_encodings, &self.accept_compression_encodings,
+55 -18
View File
@@ -1,6 +1,7 @@
use super::encode::BUFFER_SIZE; use super::encode::BUFFER_SIZE;
use crate::{metadata::MetadataValue, Status}; use crate::{metadata::MetadataValue, Status};
use bytes::{Buf, BufMut, BytesMut}; use bytes::{Buf, BytesMut};
#[cfg(feature = "gzip")]
use flate2::read::{GzDecoder, GzEncoder}; use flate2::read::{GzDecoder, GzEncoder};
use std::fmt; use std::fmt;
@@ -10,28 +11,44 @@ pub(crate) const ACCEPT_ENCODING_HEADER: &str = "grpc-accept-encoding";
/// Struct used to configure which encodings are enabled on a server or channel. /// Struct used to configure which encodings are enabled on a server or channel.
#[derive(Debug, Default, Clone, Copy)] #[derive(Debug, Default, Clone, Copy)]
pub struct EnabledCompressionEncodings { pub struct EnabledCompressionEncodings {
#[cfg(feature = "gzip")]
pub(crate) gzip: bool, pub(crate) gzip: bool,
} }
impl EnabledCompressionEncodings { impl EnabledCompressionEncodings {
/// Check if `gzip` compression is enabled. /// Check if a [`CompressionEncoding`] is enabled.
pub fn gzip(self) -> bool { pub fn is_enabled(&self, encoding: CompressionEncoding) -> bool {
self.gzip match encoding {
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip => self.gzip,
}
} }
/// Enable `gzip` compression. /// Enable a [`CompressionEncoding`].
pub fn enable_gzip(&mut self) { pub fn enable(&mut self, encoding: CompressionEncoding) {
self.gzip = true; match encoding {
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip => self.gzip = true,
}
} }
pub(crate) fn into_accept_encoding_header_value(self) -> Option<http::HeaderValue> { pub(crate) fn into_accept_encoding_header_value(self) -> Option<http::HeaderValue> {
let Self { gzip } = self; if self.is_gzip_enabled() {
if gzip {
Some(http::HeaderValue::from_static("gzip,identity")) Some(http::HeaderValue::from_static("gzip,identity"))
} else { } else {
None None
} }
} }
#[cfg(feature = "gzip")]
const fn is_gzip_enabled(&self) -> bool {
self.gzip
}
#[cfg(not(feature = "gzip"))]
const fn is_gzip_enabled(&self) -> bool {
false
}
} }
/// The compression encodings Tonic supports. /// The compression encodings Tonic supports.
@@ -39,6 +56,8 @@ impl EnabledCompressionEncodings {
#[non_exhaustive] #[non_exhaustive]
pub enum CompressionEncoding { pub enum CompressionEncoding {
#[allow(missing_docs)] #[allow(missing_docs)]
#[cfg(feature = "gzip")]
#[cfg_attr(docsrs, doc(cfg(feature = "gzip")))]
Gzip, Gzip,
} }
@@ -48,13 +67,16 @@ impl CompressionEncoding {
map: &http::HeaderMap, map: &http::HeaderMap,
enabled_encodings: EnabledCompressionEncodings, enabled_encodings: EnabledCompressionEncodings,
) -> Option<Self> { ) -> Option<Self> {
if !enabled_encodings.is_gzip_enabled() {
return None;
}
let header_value = map.get(ACCEPT_ENCODING_HEADER)?; let header_value = map.get(ACCEPT_ENCODING_HEADER)?;
let header_value_str = header_value.to_str().ok()?; let header_value_str = header_value.to_str().ok()?;
let EnabledCompressionEncodings { gzip } = enabled_encodings;
split_by_comma(header_value_str).find_map(|value| match value { split_by_comma(header_value_str).find_map(|value| match value {
"gzip" if gzip => Some(CompressionEncoding::Gzip), #[cfg(feature = "gzip")]
"gzip" => Some(CompressionEncoding::Gzip),
_ => None, _ => None,
}) })
} }
@@ -76,10 +98,11 @@ impl CompressionEncoding {
return Ok(None); return Ok(None);
}; };
let EnabledCompressionEncodings { gzip } = enabled_encodings;
match header_value_str { match header_value_str {
"gzip" if gzip => Ok(Some(CompressionEncoding::Gzip)), #[cfg(feature = "gzip")]
"gzip" if enabled_encodings.is_enabled(CompressionEncoding::Gzip) => {
Ok(Some(CompressionEncoding::Gzip))
}
"identity" => Ok(None), "identity" => Ok(None),
other => { other => {
let mut status = Status::unimplemented(format!( let mut status = Status::unimplemented(format!(
@@ -102,14 +125,24 @@ impl CompressionEncoding {
pub(crate) fn into_header_value(self) -> http::HeaderValue { pub(crate) fn into_header_value(self) -> http::HeaderValue {
match self { match self {
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip => http::HeaderValue::from_static("gzip"), CompressionEncoding::Gzip => http::HeaderValue::from_static("gzip"),
} }
} }
pub(crate) fn encodings() -> &'static [Self] {
&[
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip,
]
}
} }
impl fmt::Display for CompressionEncoding { impl fmt::Display for CompressionEncoding {
#[allow(unused_variables)]
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self { match *self {
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip => write!(f, "gzip"), CompressionEncoding::Gzip => write!(f, "gzip"),
} }
} }
@@ -120,6 +153,7 @@ fn split_by_comma(s: &str) -> impl Iterator<Item = &str> {
} }
/// Compress `len` bytes from `decompressed_buf` into `out_buf`. /// Compress `len` bytes from `decompressed_buf` into `out_buf`.
#[allow(unused_variables, unreachable_code)]
pub(crate) fn compress( pub(crate) fn compress(
encoding: CompressionEncoding, encoding: CompressionEncoding,
decompressed_buf: &mut BytesMut, decompressed_buf: &mut BytesMut,
@@ -130,13 +164,14 @@ pub(crate) fn compress(
out_buf.reserve(capacity); out_buf.reserve(capacity);
match encoding { match encoding {
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip => { CompressionEncoding::Gzip => {
let mut gzip_encoder = GzEncoder::new( let mut gzip_encoder = GzEncoder::new(
&decompressed_buf[0..len], &decompressed_buf[0..len],
// FIXME: support customizing the compression level // FIXME: support customizing the compression level
flate2::Compression::new(6), flate2::Compression::new(6),
); );
let mut out_writer = out_buf.writer(); let mut out_writer = bytes::BufMut::writer(out_buf);
std::io::copy(&mut gzip_encoder, &mut out_writer)?; std::io::copy(&mut gzip_encoder, &mut out_writer)?;
} }
@@ -148,6 +183,7 @@ pub(crate) fn compress(
} }
/// Decompress `len` bytes from `compressed_buf` into `out_buf`. /// Decompress `len` bytes from `compressed_buf` into `out_buf`.
#[allow(unused_variables, unreachable_code)]
pub(crate) fn decompress( pub(crate) fn decompress(
encoding: CompressionEncoding, encoding: CompressionEncoding,
compressed_buf: &mut BytesMut, compressed_buf: &mut BytesMut,
@@ -159,9 +195,10 @@ pub(crate) fn decompress(
out_buf.reserve(capacity); out_buf.reserve(capacity);
match encoding { match encoding {
#[cfg(feature = "gzip")]
CompressionEncoding::Gzip => { CompressionEncoding::Gzip => {
let mut gzip_decoder = GzDecoder::new(&compressed_buf[0..len]); let mut gzip_decoder = GzDecoder::new(&compressed_buf[0..len]);
let mut out_writer = out_buf.writer(); let mut out_writer = bytes::BufMut::writer(out_buf);
std::io::copy(&mut gzip_decoder, &mut out_writer)?; std::io::copy(&mut gzip_decoder, &mut out_writer)?;
} }
+36 -82
View File
@@ -1,4 +1,3 @@
#[cfg(feature = "compression")]
use super::compression::{decompress, CompressionEncoding}; use super::compression::{decompress, CompressionEncoding};
use super::{DecodeBuf, Decoder, HEADER_SIZE}; use super::{DecodeBuf, Decoder, HEADER_SIZE};
use crate::{body::BoxBody, metadata::MetadataMap, Code, Status}; use crate::{body::BoxBody, metadata::MetadataMap, Code, Status};
@@ -27,18 +26,19 @@ pub struct Streaming<T> {
direction: Direction, direction: Direction,
buf: BytesMut, buf: BytesMut,
trailers: Option<MetadataMap>, trailers: Option<MetadataMap>,
#[cfg(feature = "compression")]
decompress_buf: BytesMut, decompress_buf: BytesMut,
#[cfg(feature = "compression")]
encoding: Option<CompressionEncoding>, encoding: Option<CompressionEncoding>,
} }
impl<T> Unpin for Streaming<T> {} impl<T> Unpin for Streaming<T> {}
#[derive(Debug)] #[derive(Debug, Clone, Copy)]
enum State { enum State {
ReadHeader, ReadHeader,
ReadBody { compression: bool, len: usize }, ReadBody {
compression: Option<CompressionEncoding>,
len: usize,
},
Error, Error,
} }
@@ -54,20 +54,14 @@ impl<T> Streaming<T> {
decoder: D, decoder: D,
body: B, body: B,
status_code: StatusCode, status_code: StatusCode,
#[cfg(feature = "compression")] encoding: Option<CompressionEncoding>, encoding: Option<CompressionEncoding>,
) -> Self ) -> Self
where where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error>, B::Error: Into<crate::Error>,
D: Decoder<Item = T, Error = Status> + Send + 'static, D: Decoder<Item = T, Error = Status> + Send + 'static,
{ {
Self::new( Self::new(decoder, body, Direction::Response(status_code), encoding)
decoder,
body,
Direction::Response(status_code),
#[cfg(feature = "compression")]
encoding,
)
} }
pub(crate) fn new_empty<B, D>(decoder: D, body: B) -> Self pub(crate) fn new_empty<B, D>(decoder: D, body: B) -> Self
@@ -76,40 +70,24 @@ impl<T> Streaming<T> {
B::Error: Into<crate::Error>, B::Error: Into<crate::Error>,
D: Decoder<Item = T, Error = Status> + Send + 'static, D: Decoder<Item = T, Error = Status> + Send + 'static,
{ {
Self::new( Self::new(decoder, body, Direction::EmptyResponse, None)
decoder,
body,
Direction::EmptyResponse,
#[cfg(feature = "compression")]
None,
)
} }
#[doc(hidden)] #[doc(hidden)]
pub fn new_request<B, D>( pub fn new_request<B, D>(decoder: D, body: B, encoding: Option<CompressionEncoding>) -> Self
decoder: D,
body: B,
#[cfg(feature = "compression")] encoding: Option<CompressionEncoding>,
) -> Self
where where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error>, B::Error: Into<crate::Error>,
D: Decoder<Item = T, Error = Status> + Send + 'static, D: Decoder<Item = T, Error = Status> + Send + 'static,
{ {
Self::new( Self::new(decoder, body, Direction::Request, encoding)
decoder,
body,
Direction::Request,
#[cfg(feature = "compression")]
encoding,
)
} }
fn new<B, D>( fn new<B, D>(
decoder: D, decoder: D,
body: B, body: B,
direction: Direction, direction: Direction,
#[cfg(feature = "compression")] encoding: Option<CompressionEncoding>, encoding: Option<CompressionEncoding>,
) -> Self ) -> Self
where where
B: Body + Send + 'static, B: Body + Send + 'static,
@@ -126,9 +104,7 @@ impl<T> Streaming<T> {
direction, direction,
buf: BytesMut::with_capacity(BUFFER_SIZE), buf: BytesMut::with_capacity(BUFFER_SIZE),
trailers: None, trailers: None,
#[cfg(feature = "compression")]
decompress_buf: BytesMut::new(), decompress_buf: BytesMut::new(),
#[cfg(feature = "compression")]
encoding, encoding,
} }
} }
@@ -217,13 +193,12 @@ impl<T> Streaming<T> {
return Ok(None); return Ok(None);
} }
let is_compressed = match self.buf.get_u8() { let compression_encoding = match self.buf.get_u8() {
0 => false, 0 => None,
1 => { 1 => {
#[cfg(feature = "compression")]
{ {
if self.encoding.is_some() { if self.encoding.is_some() {
true self.encoding
} else { } else {
// https://grpc.github.io/grpc/core/md_doc_compression.html // https://grpc.github.io/grpc/core/md_doc_compression.html
// An ill-constructed message with its Compressed-Flag bit set but lacking a grpc-encoding // An ill-constructed message with its Compressed-Flag bit set but lacking a grpc-encoding
@@ -232,13 +207,6 @@ impl<T> Streaming<T> {
return Err(Status::new(Code::Internal, "protocol error: received message with compressed-flag but no grpc-encoding was specified")); return Err(Status::new(Code::Internal, "protocol error: received message with compressed-flag but no grpc-encoding was specified"));
} }
} }
#[cfg(not(feature = "compression"))]
{
return Err(Status::new(
Code::Unimplemented,
"Message compressed, compression support not enabled.".to_string(),
));
}
} }
f => { f => {
trace!("unexpected compression flag"); trace!("unexpected compression flag");
@@ -257,54 +225,40 @@ impl<T> Streaming<T> {
self.buf.reserve(len); self.buf.reserve(len);
self.state = State::ReadBody { self.state = State::ReadBody {
compression: is_compressed, compression: compression_encoding,
len, len,
} }
} }
if let State::ReadBody { len, compression } = &self.state { if let State::ReadBody { len, compression } = self.state {
// if we haven't read enough of the message then return and keep // if we haven't read enough of the message then return and keep
// reading // reading
if self.buf.remaining() < *len || self.buf.len() < *len { if self.buf.remaining() < len || self.buf.len() < len {
return Ok(None); return Ok(None);
} }
let decoding_result = if *compression { let decoding_result = if let Some(encoding) = compression {
#[cfg(feature = "compression")] self.decompress_buf.clear();
if let Err(err) = decompress(encoding, &mut self.buf, &mut self.decompress_buf, len)
{ {
self.decompress_buf.clear(); let message = if let Direction::Response(status) = self.direction {
format!(
if let Err(err) = decompress( "Error decompressing: {}, while receiving response with status: {}",
self.encoding.unwrap_or_else(|| { err, status
// SAFETY: The check while in State::ReadHeader would already have returned Code::Internal )
unreachable!("message was compressed but `Streaming.encoding` was `None`. This is a bug in Tonic. Please file an issue") } else {
}), format!("Error decompressing: {}, while sending request", err)
&mut self.buf, };
&mut self.decompress_buf, return Err(Status::new(Code::Internal, message));
*len,
) {
let message = if let Direction::Response(status) = self.direction {
format!(
"Error decompressing: {}, while receiving response with status: {}",
err, status
)
} else {
format!("Error decompressing: {}, while sending request", err)
};
return Err(Status::new(Code::Internal, message));
}
let decompressed_len = self.decompress_buf.len();
self.decoder.decode(&mut DecodeBuf::new(
&mut self.decompress_buf,
decompressed_len,
))
} }
let decompressed_len = self.decompress_buf.len();
#[cfg(not(feature = "compression"))] self.decoder.decode(&mut DecodeBuf::new(
unreachable!("should not take this branch if compression is disabled") &mut self.decompress_buf,
decompressed_len,
))
} else { } else {
self.decoder self.decoder.decode(&mut DecodeBuf::new(&mut self.buf, len))
.decode(&mut DecodeBuf::new(&mut self.buf, *len))
}; };
return match decoding_result { return match decoding_result {
+27 -44
View File
@@ -1,4 +1,3 @@
#[cfg(feature = "compression")]
use super::compression::{compress, CompressionEncoding, SingleMessageCompressionOverride}; use super::compression::{compress, CompressionEncoding, SingleMessageCompressionOverride};
use super::{EncodeBuf, Encoder, HEADER_SIZE}; use super::{EncodeBuf, Encoder, HEADER_SIZE};
use crate::{Code, Status}; use crate::{Code, Status};
@@ -18,22 +17,14 @@ pub(super) const BUFFER_SIZE: usize = 8 * 1024;
pub(crate) fn encode_server<T, U>( pub(crate) fn encode_server<T, U>(
encoder: T, encoder: T,
source: U, source: U,
#[cfg(feature = "compression")] compression_encoding: Option<CompressionEncoding>, compression_encoding: Option<CompressionEncoding>,
#[cfg(feature = "compression")] compression_override: SingleMessageCompressionOverride, compression_override: SingleMessageCompressionOverride,
) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>> ) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>>
where where
T: Encoder<Error = Status>, T: Encoder<Error = Status>,
U: Stream<Item = Result<T::Item, Status>>, U: Stream<Item = Result<T::Item, Status>>,
{ {
let stream = encode( let stream = encode(encoder, source, compression_encoding, compression_override).into_stream();
encoder,
source,
#[cfg(feature = "compression")]
compression_encoding,
#[cfg(feature = "compression")]
compression_override,
)
.into_stream();
EncodeBody::new_server(stream) EncodeBody::new_server(stream)
} }
@@ -41,7 +32,7 @@ where
pub(crate) fn encode_client<T, U>( pub(crate) fn encode_client<T, U>(
encoder: T, encoder: T,
source: U, source: U,
#[cfg(feature = "compression")] compression_encoding: Option<CompressionEncoding>, compression_encoding: Option<CompressionEncoding>,
) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>> ) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>>
where where
T: Encoder<Error = Status>, T: Encoder<Error = Status>,
@@ -50,9 +41,7 @@ where
let stream = encode( let stream = encode(
encoder, encoder,
source.map(Ok), source.map(Ok),
#[cfg(feature = "compression")]
compression_encoding, compression_encoding,
#[cfg(feature = "compression")]
SingleMessageCompressionOverride::default(), SingleMessageCompressionOverride::default(),
) )
.into_stream(); .into_stream();
@@ -62,8 +51,8 @@ where
fn encode<T, U>( fn encode<T, U>(
mut encoder: T, mut encoder: T,
source: U, source: U,
#[cfg(feature = "compression")] compression_encoding: Option<CompressionEncoding>, compression_encoding: Option<CompressionEncoding>,
#[cfg(feature = "compression")] compression_override: SingleMessageCompressionOverride, compression_override: SingleMessageCompressionOverride,
) -> impl TryStream<Ok = Bytes, Error = Status> ) -> impl TryStream<Ok = Bytes, Error = Status>
where where
T: Encoder<Error = Status>, T: Encoder<Error = Status>,
@@ -72,17 +61,17 @@ where
async_stream::stream! { async_stream::stream! {
let mut buf = BytesMut::with_capacity(BUFFER_SIZE); let mut buf = BytesMut::with_capacity(BUFFER_SIZE);
#[cfg(feature = "compression")] let compression_encoding = if compression_override == SingleMessageCompressionOverride::Disable {
let (compression_enabled_for_stream, mut uncompression_buf) = match compression_encoding { None
Some(CompressionEncoding::Gzip) => (true, BytesMut::with_capacity(BUFFER_SIZE)), } else {
None => (false, BytesMut::new()), compression_encoding
}; };
#[cfg(feature = "compression")] let mut uncompression_buf = if compression_encoding.is_some() {
let compress_item = compression_enabled_for_stream && compression_override == SingleMessageCompressionOverride::Inherit; BytesMut::with_capacity(BUFFER_SIZE)
} else {
#[cfg(not(feature = "compression"))] BytesMut::new()
let compress_item = false; };
futures_util::pin_mut!(source); futures_util::pin_mut!(source);
@@ -94,26 +83,20 @@ where
buf.advance_mut(HEADER_SIZE); buf.advance_mut(HEADER_SIZE);
} }
if compress_item { if let Some(encoding) = compression_encoding {
#[cfg(feature = "compression")] uncompression_buf.clear();
{
uncompression_buf.clear();
encoder.encode(item, &mut EncodeBuf::new(&mut uncompression_buf)) encoder.encode(item, &mut EncodeBuf::new(&mut uncompression_buf))
.map_err(|err| Status::internal(format!("Error encoding: {}", err)))?; .map_err(|err| Status::internal(format!("Error encoding: {}", err)))?;
let uncompressed_len = uncompression_buf.len(); let uncompressed_len = uncompression_buf.len();
compress( compress(
compression_encoding.unwrap(), encoding,
&mut uncompression_buf, &mut uncompression_buf,
&mut buf, &mut buf,
uncompressed_len, uncompressed_len,
).map_err(|err| Status::internal(format!("Error compressing: {}", err)))?; ).map_err(|err| Status::internal(format!("Error compressing: {}", err)))?;
}
#[cfg(not(feature = "compression"))]
unreachable!("compression disabled, should not take this branch");
} else { } else {
encoder.encode(item, &mut EncodeBuf::new(&mut buf)) encoder.encode(item, &mut EncodeBuf::new(&mut buf))
.map_err(|err| Status::internal(format!("Error encoding: {}", err)))?; .map_err(|err| Status::internal(format!("Error encoding: {}", err)))?;
@@ -124,7 +107,7 @@ where
assert!(len <= std::u32::MAX as usize); assert!(len <= std::u32::MAX as usize);
{ {
let mut buf = &mut buf[..HEADER_SIZE]; let mut buf = &mut buf[..HEADER_SIZE];
buf.put_u8(compress_item as u8); buf.put_u8(compression_encoding.is_some() as u8);
buf.put_u32(len as u32); buf.put_u32(len as u32);
} }
-3
View File
@@ -4,7 +4,6 @@
//! and a protobuf codec based on prost. //! and a protobuf codec based on prost.
mod buffer; mod buffer;
#[cfg(feature = "compression")]
pub(crate) mod compression; pub(crate) mod compression;
mod decode; mod decode;
mod encode; mod encode;
@@ -17,8 +16,6 @@ use std::io;
pub(crate) use self::encode::{encode_client, encode_server}; pub(crate) use self::encode::{encode_client, encode_server};
pub use self::buffer::{DecodeBuf, EncodeBuf}; pub use self::buffer::{DecodeBuf, EncodeBuf};
#[cfg(feature = "compression")]
#[cfg_attr(docsrs, doc(cfg(feature = "compression")))]
pub use self::compression::{CompressionEncoding, EnabledCompressionEncodings}; pub use self::compression::{CompressionEncoding, EnabledCompressionEncodings};
pub use self::decode::Streaming; pub use self::decode::Streaming;
#[cfg(feature = "prost")] #[cfg(feature = "prost")]
-1
View File
@@ -10,7 +10,6 @@ pub use std::sync::Arc;
pub use std::task::{Context, Poll}; pub use std::task::{Context, Poll};
pub use tower_service::Service; pub use tower_service::Service;
pub type StdError = Box<dyn std::error::Error + Send + Sync + 'static>; pub type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
#[cfg(feature = "compression")]
pub use crate::codec::{CompressionEncoding, EnabledCompressionEncodings}; pub use crate::codec::{CompressionEncoding, EnabledCompressionEncodings};
pub use crate::service::interceptor::InterceptedService; pub use crate::service::interceptor::InterceptedService;
pub use bytes::Bytes; pub use bytes::Bytes;
+2 -2
View File
@@ -116,8 +116,8 @@ impl<T> Response<T> {
/// **Note**: This only has effect on responses to unary requests and responses to client to /// **Note**: This only has effect on responses to unary requests and responses to client to
/// server streams. Response streams (server to client stream and bidirectional streams) will /// server streams. Response streams (server to client stream and bidirectional streams) will
/// still be compressed according to the configuration of the server. /// still be compressed according to the configuration of the server.
#[cfg(feature = "compression")] #[cfg(feature = "gzip")]
#[cfg_attr(docsrs, doc(cfg(feature = "compression")))] #[cfg_attr(docsrs, doc(cfg(feature = "gzip")))]
pub fn disable_compression(&mut self) { pub fn disable_compression(&mut self) {
self.extensions_mut() self.extensions_mut()
.insert(crate::codec::compression::SingleMessageCompressionOverride::Disable); .insert(crate::codec::compression::SingleMessageCompressionOverride::Disable);
+23 -96
View File
@@ -1,4 +1,3 @@
#[cfg(feature = "compression")]
use crate::codec::compression::{ use crate::codec::compression::{
CompressionEncoding, EnabledCompressionEncodings, SingleMessageCompressionOverride, CompressionEncoding, EnabledCompressionEncodings, SingleMessageCompressionOverride,
}; };
@@ -34,10 +33,8 @@ macro_rules! t {
pub struct Grpc<T> { pub struct Grpc<T> {
codec: T, codec: T,
/// Which compression encodings does the server accept for requests? /// Which compression encodings does the server accept for requests?
#[cfg(feature = "compression")]
accept_compression_encodings: EnabledCompressionEncodings, accept_compression_encodings: EnabledCompressionEncodings,
/// Which compression encodings might the server use for responses. /// Which compression encodings might the server use for responses.
#[cfg(feature = "compression")]
send_compression_encodings: EnabledCompressionEncodings, send_compression_encodings: EnabledCompressionEncodings,
} }
@@ -49,14 +46,12 @@ where
pub fn new(codec: T) -> Self { pub fn new(codec: T) -> Self {
Self { Self {
codec, codec,
#[cfg(feature = "compression")]
accept_compression_encodings: EnabledCompressionEncodings::default(), accept_compression_encodings: EnabledCompressionEncodings::default(),
#[cfg(feature = "compression")]
send_compression_encodings: EnabledCompressionEncodings::default(), send_compression_encodings: EnabledCompressionEncodings::default(),
} }
} }
/// Enable accepting `gzip` compressed requests. /// Enable accepting compressed requests.
/// ///
/// If a request with an unsupported encoding is received the server will respond with /// If a request with an unsupported encoding is received the server will respond with
/// [`Code::UnUnimplemented`](crate::Code). /// [`Code::UnUnimplemented`](crate::Code).
@@ -66,11 +61,12 @@ where
/// The most common way of using this is through a server generated by tonic-build: /// The most common way of using this is through a server generated by tonic-build:
/// ///
/// ```rust /// ```rust
/// # enum CompressionEncoding { Gzip }
/// # struct Svc; /// # struct Svc;
/// # struct ExampleServer<T>(T); /// # struct ExampleServer<T>(T);
/// # impl<T> ExampleServer<T> { /// # impl<T> ExampleServer<T> {
/// # fn new(svc: T) -> Self { Self(svc) } /// # fn new(svc: T) -> Self { Self(svc) }
/// # fn accept_gzip(self) -> Self { self } /// # fn accept_compressed(self, _: CompressionEncoding) -> Self { self }
/// # } /// # }
/// # #[tonic::async_trait] /// # #[tonic::async_trait]
/// # trait Example {} /// # trait Example {}
@@ -80,22 +76,14 @@ where
/// // ... /// // ...
/// } /// }
/// ///
/// let service = ExampleServer::new(Svc).accept_gzip(); /// let service = ExampleServer::new(Svc).accept_compressed(CompressionEncoding::Gzip);
/// ``` /// ```
#[cfg(feature = "compression")] pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
#[cfg_attr(docsrs, doc(cfg(feature = "compression")))] self.accept_compression_encodings.enable(encoding);
pub fn accept_gzip(mut self) -> Self {
self.accept_compression_encodings.enable_gzip();
self self
} }
#[doc(hidden)] /// Enable sending compressed responses.
#[cfg(not(feature = "compression"))]
pub fn accept_gzip(self) -> Self {
panic!("`accept_gzip` called on a server but the `compression` feature is not enabled on tonic");
}
/// Enable sending `gzip` compressed responses.
/// ///
/// Requires the client to also support receiving compressed responses. /// Requires the client to also support receiving compressed responses.
/// ///
@@ -104,11 +92,12 @@ where
/// The most common way of using this is through a server generated by tonic-build: /// The most common way of using this is through a server generated by tonic-build:
/// ///
/// ```rust /// ```rust
/// # enum CompressionEncoding { Gzip }
/// # struct Svc; /// # struct Svc;
/// # struct ExampleServer<T>(T); /// # struct ExampleServer<T>(T);
/// # impl<T> ExampleServer<T> { /// # impl<T> ExampleServer<T> {
/// # fn new(svc: T) -> Self { Self(svc) } /// # fn new(svc: T) -> Self { Self(svc) }
/// # fn send_gzip(self) -> Self { self } /// # fn send_compressed(self, _: CompressionEncoding) -> Self { self }
/// # } /// # }
/// # #[tonic::async_trait] /// # #[tonic::async_trait]
/// # trait Example {} /// # trait Example {}
@@ -118,24 +107,13 @@ where
/// // ... /// // ...
/// } /// }
/// ///
/// let service = ExampleServer::new(Svc).send_gzip(); /// let service = ExampleServer::new(Svc).send_compressed(CompressionEncoding::Gzip);
/// ``` /// ```
#[cfg(feature = "compression")] pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
#[cfg_attr(docsrs, doc(cfg(feature = "compression")))] self.send_compression_encodings.enable(encoding);
pub fn send_gzip(mut self) -> Self {
self.send_compression_encodings.enable_gzip();
self self
} }
#[doc(hidden)]
#[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"
);
}
#[cfg(feature = "compression")]
#[doc(hidden)] #[doc(hidden)]
pub fn apply_compression_config( pub fn apply_compression_config(
self, self,
@@ -144,26 +122,18 @@ where
) -> Self { ) -> Self {
let mut this = self; let mut this = self;
let EnabledCompressionEncodings { gzip: accept_gzip } = accept_encodings; for &encoding in CompressionEncoding::encodings() {
if accept_gzip { if accept_encodings.is_enabled(encoding) {
this = this.accept_gzip(); this = this.accept_compressed(encoding);
} }
if send_encodings.is_enabled(encoding) {
let EnabledCompressionEncodings { gzip: send_gzip } = send_encodings; this = this.send_compressed(encoding);
if send_gzip { }
this = this.send_gzip();
} }
this this
} }
#[cfg(not(feature = "compression"))]
#[doc(hidden)]
#[allow(unused_variables)]
pub fn apply_compression_config(self, accept_encodings: (), send_encodings: ()) -> Self {
self
}
/// Handle a single unary gRPC request. /// Handle a single unary gRPC request.
pub async fn unary<S, B>( pub async fn unary<S, B>(
&mut self, &mut self,
@@ -175,7 +145,6 @@ where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error> + Send, B::Error: Into<crate::Error> + Send,
{ {
#[cfg(feature = "compression")]
let accept_encoding = CompressionEncoding::from_accept_encoding_header( let accept_encoding = CompressionEncoding::from_accept_encoding_header(
req.headers(), req.headers(),
self.send_compression_encodings, self.send_compression_encodings,
@@ -187,9 +156,7 @@ where
return self return self
.map_response::<stream::Once<future::Ready<Result<T::Encode, Status>>>>( .map_response::<stream::Once<future::Ready<Result<T::Encode, Status>>>>(
Err(status), Err(status),
#[cfg(feature = "compression")]
accept_encoding, accept_encoding,
#[cfg(feature = "compression")]
SingleMessageCompressionOverride::default(), SingleMessageCompressionOverride::default(),
); );
} }
@@ -200,16 +167,9 @@ where
.await .await
.map(|r| r.map(|m| stream::once(future::ok(m)))); .map(|r| r.map(|m| stream::once(future::ok(m))));
#[cfg(feature = "compression")]
let compression_override = compression_override_from_response(&response); let compression_override = compression_override_from_response(&response);
self.map_response( self.map_response(response, accept_encoding, compression_override)
response,
#[cfg(feature = "compression")]
accept_encoding,
#[cfg(feature = "compression")]
compression_override,
)
} }
/// Handle a server side streaming request. /// Handle a server side streaming request.
@@ -224,7 +184,6 @@ where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error> + Send, B::Error: Into<crate::Error> + Send,
{ {
#[cfg(feature = "compression")]
let accept_encoding = CompressionEncoding::from_accept_encoding_header( let accept_encoding = CompressionEncoding::from_accept_encoding_header(
req.headers(), req.headers(),
self.send_compression_encodings, self.send_compression_encodings,
@@ -235,9 +194,7 @@ where
Err(status) => { Err(status) => {
return self.map_response::<S::ResponseStream>( return self.map_response::<S::ResponseStream>(
Err(status), Err(status),
#[cfg(feature = "compression")]
accept_encoding, accept_encoding,
#[cfg(feature = "compression")]
SingleMessageCompressionOverride::default(), SingleMessageCompressionOverride::default(),
); );
} }
@@ -247,11 +204,9 @@ where
self.map_response( self.map_response(
response, response,
#[cfg(feature = "compression")]
accept_encoding, accept_encoding,
// disabling compression of individual stream items must be done on // disabling compression of individual stream items must be done on
// the items themselves // the items themselves
#[cfg(feature = "compression")]
SingleMessageCompressionOverride::default(), SingleMessageCompressionOverride::default(),
) )
} }
@@ -267,7 +222,6 @@ where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error> + Send + 'static, B::Error: Into<crate::Error> + Send + 'static,
{ {
#[cfg(feature = "compression")]
let accept_encoding = CompressionEncoding::from_accept_encoding_header( let accept_encoding = CompressionEncoding::from_accept_encoding_header(
req.headers(), req.headers(),
self.send_compression_encodings, self.send_compression_encodings,
@@ -280,16 +234,9 @@ where
.await .await
.map(|r| r.map(|m| stream::once(future::ok(m)))); .map(|r| r.map(|m| stream::once(future::ok(m))));
#[cfg(feature = "compression")]
let compression_override = compression_override_from_response(&response); let compression_override = compression_override_from_response(&response);
self.map_response( self.map_response(response, accept_encoding, compression_override)
response,
#[cfg(feature = "compression")]
accept_encoding,
#[cfg(feature = "compression")]
compression_override,
)
} }
/// Handle a bi-directional streaming gRPC request. /// Handle a bi-directional streaming gRPC request.
@@ -304,7 +251,6 @@ where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error> + Send, B::Error: Into<crate::Error> + Send,
{ {
#[cfg(feature = "compression")]
let accept_encoding = CompressionEncoding::from_accept_encoding_header( let accept_encoding = CompressionEncoding::from_accept_encoding_header(
req.headers(), req.headers(),
self.send_compression_encodings, self.send_compression_encodings,
@@ -316,9 +262,7 @@ where
self.map_response( self.map_response(
response, response,
#[cfg(feature = "compression")]
accept_encoding, accept_encoding,
#[cfg(feature = "compression")]
SingleMessageCompressionOverride::default(), SingleMessageCompressionOverride::default(),
) )
} }
@@ -331,18 +275,13 @@ where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error> + Send, B::Error: Into<crate::Error> + Send,
{ {
#[cfg(feature = "compression")]
let request_compression_encoding = self.request_encoding_if_supported(&request)?; let request_compression_encoding = self.request_encoding_if_supported(&request)?;
let (parts, body) = request.into_parts(); let (parts, body) = request.into_parts();
#[cfg(feature = "compression")]
let stream = let stream =
Streaming::new_request(self.codec.decoder(), body, request_compression_encoding); Streaming::new_request(self.codec.decoder(), body, request_compression_encoding);
#[cfg(not(feature = "compression"))]
let stream = Streaming::new_request(self.codec.decoder(), body);
futures_util::pin_mut!(stream); futures_util::pin_mut!(stream);
let message = stream let message = stream
@@ -367,24 +306,19 @@ where
B: Body + Send + 'static, B: Body + Send + 'static,
B::Error: Into<crate::Error> + Send, B::Error: Into<crate::Error> + Send,
{ {
#[cfg(feature = "compression")]
let encoding = self.request_encoding_if_supported(&request)?; let encoding = self.request_encoding_if_supported(&request)?;
#[cfg(feature = "compression")]
let request = let request =
request.map(|body| Streaming::new_request(self.codec.decoder(), body, encoding)); request.map(|body| Streaming::new_request(self.codec.decoder(), body, encoding));
#[cfg(not(feature = "compression"))]
let request = request.map(|body| Streaming::new_request(self.codec.decoder(), body));
Ok(Request::from_http(request)) Ok(Request::from_http(request))
} }
fn map_response<B>( fn map_response<B>(
&mut self, &mut self,
response: Result<crate::Response<B>, Status>, response: Result<crate::Response<B>, Status>,
#[cfg(feature = "compression")] accept_encoding: Option<CompressionEncoding>, accept_encoding: Option<CompressionEncoding>,
#[cfg(feature = "compression")] compression_override: SingleMessageCompressionOverride, compression_override: SingleMessageCompressionOverride,
) -> http::Response<BoxBody> ) -> http::Response<BoxBody>
where where
B: TryStream<Ok = T::Encode, Error = Status> + Send + 'static, B: TryStream<Ok = T::Encode, Error = Status> + Send + 'static,
@@ -402,7 +336,6 @@ where
http::header::HeaderValue::from_static("application/grpc"), http::header::HeaderValue::from_static("application/grpc"),
); );
#[cfg(feature = "compression")]
if let Some(encoding) = accept_encoding { if let Some(encoding) = accept_encoding {
// Set the content encoding // Set the content encoding
parts.headers.insert( parts.headers.insert(
@@ -414,16 +347,13 @@ where
let body = encode_server( let body = encode_server(
self.codec.encoder(), self.codec.encoder(),
body.into_stream(), body.into_stream(),
#[cfg(feature = "compression")]
accept_encoding, accept_encoding,
#[cfg(feature = "compression")]
compression_override, compression_override,
); );
http::Response::from_parts(parts, BoxBody::new(body)) http::Response::from_parts(parts, BoxBody::new(body))
} }
#[cfg(feature = "compression")]
fn request_encoding_if_supported<B>( fn request_encoding_if_supported<B>(
&self, &self,
request: &http::Request<B>, request: &http::Request<B>,
@@ -441,13 +371,11 @@ impl<T: fmt::Debug> fmt::Debug for Grpc<T> {
f.field("codec", &self.codec); f.field("codec", &self.codec);
#[cfg(feature = "compression")]
f.field( f.field(
"accept_compression_encodings", "accept_compression_encodings",
&self.accept_compression_encodings, &self.accept_compression_encodings,
); );
#[cfg(feature = "compression")]
f.field( f.field(
"send_compression_encodings", "send_compression_encodings",
&self.send_compression_encodings, &self.send_compression_encodings,
@@ -457,7 +385,6 @@ impl<T: fmt::Debug> fmt::Debug for Grpc<T> {
} }
} }
#[cfg(feature = "compression")]
fn compression_override_from_response<B, E>( fn compression_override_from_response<B, E>(
res: &Result<crate::Response<B>, E>, res: &Result<crate::Response<B>, E>,
) -> SingleMessageCompressionOverride { ) -> SingleMessageCompressionOverride {