1 use super::*;
2 
3 #[tokio::test(flavor = "multi_thread")]
4 async fn client_enabled_server_enabled() {
5     let (client, server) = tokio::io::duplex(UNCOMPRESSED_MIN_BODY_SIZE * 10);
6 
7     let svc = test_server::TestServer::new(Svc::default())
8         .accept_gzip()
9         .send_gzip();
10 
11     let request_bytes_counter = Arc::new(AtomicUsize::new(0));
12     let response_bytes_counter = Arc::new(AtomicUsize::new(0));
13 
14     fn assert_right_encoding<B>(req: http::Request<B>) -> http::Request<B> {
15         assert_eq!(req.headers().get("grpc-encoding").unwrap(), "gzip");
16         req
17     }
18 
19     tokio::spawn({
20         let request_bytes_counter = request_bytes_counter.clone();
21         let response_bytes_counter = response_bytes_counter.clone();
22         async move {
23             Server::builder()
24                 .layer(
25                     ServiceBuilder::new()
26                         .map_request(assert_right_encoding)
27                         .layer(measure_request_body_size_layer(
28                             request_bytes_counter.clone(),
29                         ))
30                         .layer(MapResponseBodyLayer::new(move |body| {
31                             util::CountBytesBody {
32                                 inner: body,
33                                 counter: response_bytes_counter.clone(),
34                             }
35                         }))
36                         .into_inner(),
37                 )
38                 .add_service(svc)
39                 .serve_with_incoming(futures::stream::iter(vec![Ok::<_, std::io::Error>(
40                     MockStream(server),
41                 )]))
42                 .await
43                 .unwrap();
44         }
45     });
46 
47     let mut client = test_client::TestClient::new(mock_io_channel(client).await)
48         .send_gzip()
49         .accept_gzip();
50 
51     let data = [0_u8; UNCOMPRESSED_MIN_BODY_SIZE].to_vec();
52     let stream = futures::stream::iter(vec![SomeData { data: data.clone() }, SomeData { data }]);
53     let req = Request::new(stream);
54 
55     let res = client
56         .compress_input_output_bidirectional_stream(req)
57         .await
58         .unwrap();
59 
60     assert_eq!(res.metadata().get("grpc-encoding").unwrap(), "gzip");
61 
62     let mut stream: Streaming<SomeData> = res.into_inner();
63 
64     stream
65         .next()
66         .await
67         .expect("stream empty")
68         .expect("item was error");
69 
70     stream
71         .next()
72         .await
73         .expect("stream empty")
74         .expect("item was error");
75 
76     assert!(request_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
77     assert!(response_bytes_counter.load(SeqCst) < UNCOMPRESSED_MIN_BODY_SIZE);
78 }
79