feat(tonic): compression support

Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "tests/integration_tests",
     "tests/stream_conflict",
     "tests/root-crate-path",
+    "tests/compression",
     "tonic-web/tests/integration"
 ]
 

tests/compression/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "compression"
+version = "0.1.0"
+authors = ["Lucio Franco <luciofranco14@gmail.com>"]
+edition = "2018"
+publish = false
+license = "MIT"
+
+[dependencies]
+tonic = { path = "../../tonic" }
+prost = "0.7"
+tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "net"] }
+tower = { version = "0.4", features = [] }
+http-body = "0.4"
+http = "0.2"
+tokio-stream = { version = "0.1.5", features = ["net"] }
+tower-http = { version = "0.1", features = ["map-response-body", "map-request-body"] }
+bytes = "1"
+futures = "0.3"
+pin-project = "1.0"
+hyper = "0.14"
+
+[build-dependencies]
+tonic-build = { path = "../../tonic-build" }

tests/compression/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+    tonic_build::compile_protos("proto/test.proto").unwrap();
+}

tests/compression/proto/test.proto

@@ -0,0 +1,18 @@
+syntax = "proto3";
+
+package test;
+
+import "google/protobuf/empty.proto";
+
+service Test {
+  rpc CompressOutputUnary(google.protobuf.Empty) returns (SomeData);
+  rpc CompressInputUnary(SomeData) returns (google.protobuf.Empty);
+  rpc CompressOutputServerStream(google.protobuf.Empty) returns (stream SomeData);
+  rpc CompressInputClientStream(stream SomeData) returns (google.protobuf.Empty);
+  rpc CompressInputOutputBidirectionalStream(stream SomeData) returns (stream SomeData);
+}
+
+message SomeData {
+  // include a bunch of data so there actually is something to compress
+  bytes data = 1;
+}

tests/compression/src/bidirectional_stream.rs

@@ -0,0 +1,76 @@
+use super::*;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_enabled_server_enabled() {
+    let svc = test_server::TestServer::new(Svc).accept_gzip().send_gzip();
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let bytes_sent_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
+    }
+
+    tokio::spawn({
+        let bytes_sent_counter = bytes_sent_counter.clone();
+        async move {
+            Server::builder()
+                .layer(
+                    ServiceBuilder::new()
+                        .map_request(assert_right_encoding)
+                        .layer(measure_request_body_size_layer(bytes_sent_counter.clone()))
+                        .layer(MapResponseBodyLayer::new(move |body| {
+                            util::CountBytesBody {
+                                inner: body,
+                                counter: bytes_sent_counter.clone(),
+                            }
+                        }))
+                        .into_inner(),
+                )
+                .add_service(svc)
+                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+                .await
+                .unwrap();
+        }
+    });
+
+    let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
+        .connect()
+        .await
+        .unwrap();
+
+    let mut client = test_client::TestClient::new(channel)
+        .send_gzip()
+        .accept_gzip();
+
+    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 res = client
+        .compress_input_output_bidirectional_stream(req)
+        .await
+        .unwrap();
+
+    assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip");
+
+    let mut stream: Streaming<SomeData> = res.into_inner();
+
+    stream
+        .next()
+        .await
+        .expect("stream empty")
+        .expect("item was error");
+
+    stream
+        .next()
+        .await
+        .expect("stream empty")
+        .expect("item was error");
+
+    let bytes_sent = bytes_sent_counter.load(Relaxed);
+    assert!(bytes_sent < UNCOMPRESSED_MIN_BODY_SIZE);
+}

tests/compression/src/client_stream.rs

@@ -0,0 +1,133 @@
+use super::*;
+use http_body::Body as _;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_enabled_server_enabled() {
+    let svc = test_server::TestServer::new(Svc).accept_gzip();
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let bytes_sent_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
+    }
+
+    tokio::spawn({
+        let bytes_sent_counter = bytes_sent_counter.clone();
+        async move {
+            Server::builder()
+                .layer(
+                    ServiceBuilder::new()
+                        .map_request(assert_right_encoding)
+                        .layer(measure_request_body_size_layer(bytes_sent_counter.clone()))
+                        .into_inner(),
+                )
+                .add_service(svc)
+                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+                .await
+                .unwrap();
+        }
+    });
+
+    let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
+        .connect()
+        .await
+        .unwrap();
+
+    let mut client = test_client::TestClient::new(channel).send_gzip();
+
+    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));
+
+    client.compress_input_client_stream(req).await.unwrap();
+
+    let bytes_sent = bytes_sent_counter.load(Relaxed);
+    assert!(dbg!(bytes_sent) < UNCOMPRESSED_MIN_BODY_SIZE);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_disabled_server_enabled() {
+    let svc = test_server::TestServer::new(Svc).accept_gzip();
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let bytes_sent_counter = Arc::new(AtomicUsize::new(0));
+
+    fn assert_right_encoding<B>(req: http::Request<B>) -> http::Request<B> {
+        assert!(req.headers().get("grpc-encoding").is_none());
+        req
+    }
+
+    tokio::spawn({
+        let bytes_sent_counter = bytes_sent_counter.clone();
+        async move {
+            Server::builder()
+                .layer(
+                    ServiceBuilder::new()
+                        .map_request(assert_right_encoding)
+                        .layer(measure_request_body_size_layer(bytes_sent_counter.clone()))
+                        .into_inner(),
+                )
+                .add_service(svc)
+                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+                .await
+                .unwrap();
+        }
+    });
+
+    let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
+        .connect()
+        .await
+        .unwrap();
+
+    let mut client = test_client::TestClient::new(channel);
+
+    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));
+
+    client.compress_input_client_stream(req).await.unwrap();
+
+    let bytes_sent = bytes_sent_counter.load(Relaxed);
+    assert!(dbg!(bytes_sent) > UNCOMPRESSED_MIN_BODY_SIZE);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_enabled_server_disabled() {
+    let svc = test_server::TestServer::new(Svc);
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(svc)
+            .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+
+    let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
+        .connect()
+        .await
+        .unwrap();
+
+    let mut client = test_client::TestClient::new(channel).send_gzip();
+
+    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 status = client.compress_input_client_stream(req).await.unwrap_err();
+
+    assert_eq!(status.code(), tonic::Code::Unimplemented);
+    assert_eq!(
+        status.message(),
+        "Content is compressed with `gzip` which isn't supported"
+    );
+}

tests/compression/src/compressing_request.rs

@@ -0,0 +1,97 @@
+use super::*;
+use http_body::Body as _;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_enabled_server_enabled() {
+    let svc = test_server::TestServer::new(Svc).accept_gzip();
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let bytes_sent_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
+    }
+
+    tokio::spawn({
+        let bytes_sent_counter = bytes_sent_counter.clone();
+        async move {
+            Server::builder()
+                .layer(
+                    ServiceBuilder::new()
+                        .layer(
+                            ServiceBuilder::new()
+                                .map_request(assert_right_encoding)
+                                .layer(measure_request_body_size_layer(bytes_sent_counter))
+                                .into_inner(),
+                        )
+                        .into_inner(),
+                )
+                .add_service(svc)
+                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+                .await
+                .unwrap();
+        }
+    });
+
+    let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
+        .connect()
+        .await
+        .unwrap();
+
+    let mut client = test_client::TestClient::new(channel).send_gzip();
+
+    for _ in 0..3 {
+        client
+            .compress_input_unary(SomeData {
+                data: [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(),
+            })
+            .await
+            .unwrap();
+        let bytes_sent = bytes_sent_counter.load(Relaxed);
+        assert!(dbg!(bytes_sent) < UNCOMPRESSED_MIN_BODY_SIZE);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn client_enabled_server_disabled() {
+    let svc = test_server::TestServer::new(Svc);
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(svc)
+            .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
+            .await
+            .unwrap();
+    });
+
+    let channel = Channel::builder(format!("http://{}", addr).parse().unwrap())
+        .connect()
+        .await
+        .unwrap();
+
+    let mut client = test_client::TestClient::new(channel).send_gzip();
+
+    let status = client
+        .compress_input_unary(SomeData {
+            data: [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec(),
+        })
+        .await
+        .unwrap_err();
+
+    assert_eq!(status.code(), tonic::Code::Unimplemented);
+    assert_eq!(
+        status.message(),
+        "Content is compressed with `gzip` which isn't supported"
+    );
+
+    // TODO(david): include header with which encodings are supported as per the spec:
+    //
+    // > The server will then include a grpc-accept-encoding response header which specifies the
+    //   algorithms that the server accepts.
+}