David PedersenandGitHub
0583cff80f
* Initial compression support * Support configuring compression on `Server` * Minor clean up * Test that compression is actually happening * Clean up some todos * channels compressing requests * Move compression to be on the codecs * Test sending compressed request to server that doesn't support it * Clean up a bit * Compress server streams * Compress client streams * Bidirectional streaming compression * Handle receiving unsupported encoding * Clean up * Add note to future self * Support disabling compression for individual responses * Add docs * Add compression examples * Disable compression behind feature flag * Add some docs * Make flate2 optional dependency * Fix docs wording * Format * Reply with which encodings are supported * Convert tests to use mocked io * Fix lints * Use separate counters * Don't make a long stream * Address review feedback
79 lines
2.6 KiB
Rust
79 lines
2.6 KiB
Rust
use super::*;
|
|
|
|
#[tokio::test(flavor = "multi_thread")]
|
|
async fn client_enabled_server_enabled() {
|
|
let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
|
|
|
|
let svc = test_server::TestServer::new(Svc::default())
|
|
.accept_gzip()
|
|
.send_gzip();
|
|
|
|
let request_bytes_counter = Arc::new(AtomicUsize::new(0));
|
|
let response_bytes_counter = Arc::new(AtomicUsize::new(0));
|
|
|
|
fn assert_right_encoding<B>(req: http::Request<B>) -> http::Request<B> {
|
|
assert_eq!(req.headers().get("grpc-encoding").unwrap(), "gzip");
|
|
req
|
|
}
|
|
|
|
tokio::spawn({
|
|
let request_bytes_counter = request_bytes_counter.clone();
|
|
let response_bytes_counter = response_bytes_counter.clone();
|
|
async move {
|
|
Server::builder()
|
|
.layer(
|
|
ServiceBuilder::new()
|
|
.map_request(assert_right_encoding)
|
|
.layer(measure_request_body_size_layer(
|
|
request_bytes_counter.clone(),
|
|
))
|
|
.layer(MapResponseBodyLayer::new(move |body| {
|
|
util::CountBytesBody {
|
|
inner: body,
|
|
counter: response_bytes_counter.clone(),
|
|
}
|
|
}))
|
|
.into_inner(),
|
|
)
|
|
.add_service(svc)
|
|
.serve_with_incoming(futures::stream::iter(vec![Ok::<_, std::io::Error>(
|
|
MockStream(server),
|
|
)]))
|
|
.await
|
|
.unwrap();
|
|
}
|
|
});
|
|
|
|
let mut client = test_client::TestClient::new(mock_io_channel(client).await)
|
|
.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(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");
|
|
|
|
assert!(request_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
|
|
assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
|
|
}
|