Fix encoding, more interop and update nightly version

tonic-interop/src/bin/client.rs

@@ -1,10 +1,10 @@
-#![feature(async_await)]
-
-use clap::{arg_enum, App, Arg, values_t};
+use clap::{arg_enum, values_t, App, Arg};
 use tonic_interop::client;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    pretty_env_logger::init();
+
     let matches = App::new("My Super Program")
         .version("1.0")
         .about("Does awesome things")
@@ -35,7 +35,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let mut test_results = Vec::new();
 
         match test_case {
-            Testcase::empty_unary => client::unary_call(&mut client, &mut test_results).await,
+            Testcase::empty_unary => client::empty_unary(&mut client, &mut test_results).await,
+            Testcase::large_unary => client::large_unary(&mut client, &mut test_results).await,
+            Testcase::client_streaming => {
+                client::client_streaming(&mut client, &mut test_results).await
+            }
+
+            Testcase::server_streaming => {
+                client::server_streaming(&mut client, &mut test_results).await
+            }
+
+            Testcase::ping_pong => client::ping_pong(&mut client, &mut test_results).await,
+            Testcase::empty_stream => client::empty_stream(&mut client, &mut test_results).await,
             _ => unimplemented!(),
         }
 

tonic-interop/src/bin/server.rs

@@ -1,8 +1,5 @@
-#![feature(async_await)]
-
-use std::time::Duration;
-use tokio::{net::TcpListener, timer::Delay};
-use tonic::{Request, Response, Status};
+use tokio::net::TcpListener;
+use tonic::{Code, Request, Response, Status};
 use tower_h2::Server;
 
 pub mod pb {
@@ -22,20 +19,53 @@ pub struct TestService {
 #[tonic::server(service = "grpc.testing.TestService", proto = "pb")]
 impl TestService {
     pub async fn empty_call(&self, request: Request<Empty>) -> Result<Response<Empty>, Status> {
-        println!("REQUEST={:?}", request);
+        println!("empty_call; REQUEST={:?}", request);
         Ok(Response::new(Empty {}))
     }
+
+    pub async fn unary_call(
+        &self,
+        request: Request<SimpleRequest>,
+    ) -> Result<Response<SimpleResponse>, Status> {
+        println!("unary_call; REQUEST={:?}", request);
+
+        let req = request.into_inner();
+
+        if let Some(echo_status) = req.response_status {
+            let status = Status::new(Code::from_i32(echo_status.code), echo_status.message);
+            return Err(status);
+        }
+
+        let res_size = if req.response_size >= 0 {
+            req.response_size as usize
+        } else {
+            let status = Status::new(Code::InvalidArgument, "response_size cannot be negative");
+            return Err(status);
+        };
+
+        let res = pb::SimpleResponse {
+            payload: Some(pb::Payload {
+                body: vec![0; res_size],
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+
+        Ok(Response::new(res))
+    }
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let addr = "[::1]:10000".parse().unwrap();
+    pretty_env_logger::init();
+    let addr = "127.0.0.1:10000".parse().unwrap();
     let mut bind = TcpListener::bind(&addr)?;
 
     let greeter = TestService::default();
     let mut server = Server::new(TestServiceServer::new(greeter), Default::default());
 
     while let Ok((sock, _addr)) = bind.accept().await {
+        println!("new connection");
         if let Err(e) = sock.set_nodelay(true) {
             return Err(e.into());
         }

tonic-interop/src/client.rs

@@ -1,13 +1,22 @@
 use crate::{pb::*, test_assert, TestAssertion};
+use futures_util::{future, stream, SinkExt, StreamExt};
 use std::net::SocketAddr;
-use tokio::net::TcpStream;
-use tonic::Request;
+use tokio::{net::TcpStream, sync::mpsc};
+use tonic::{Request, Response};
 use tower_h2::{add_origin::AddOrigin, Connection};
 
 pub type Client = TestServiceClient<AddOrigin<Connection<tonic::BoxBody>>>;
 
 tonic::client!(service = "grpc.testing.TestService", proto = "crate::pb");
 
+const LARGE_REQ_SIZE: usize = 271828;
+const LARGE_RSP_SIZE: i32 = 314159;
+const REQUEST_LENGTHS: &'static [i32] = &[27182, 8, 1828, 45904];
+const RESPONSE_LENGTHS: &'static [i32] = &[31415, 9, 2653, 58979];
+// const TEST_STATUS_MESSAGE: &'static str = "test status message";
+// const SPECIAL_TEST_STATUS_MESSAGE: &'static str =
+//     "\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
+
 pub async fn create(addr: SocketAddr) -> Result<Client, Box<dyn std::error::Error>> {
     let io = TcpStream::connect(&addr).await?;
 
@@ -19,7 +28,7 @@ pub async fn create(addr: SocketAddr) -> Result<Client, Box<dyn std::error::Erro
     Ok(TestServiceClient::new(svc))
 }
 
-pub async fn unary_call(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+pub async fn empty_unary(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
     let result = client.empty_call(Request::new(Empty {})).await;
 
     assertions.push(test_assert!(
@@ -37,3 +46,207 @@ pub async fn unary_call(client: &mut Client, assertions: &mut Vec<TestAssertion>
         ));
     }
 }
+
+pub async fn large_unary(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+    use std::mem;
+    let payload = crate::client_payload(LARGE_REQ_SIZE);
+    let req = SimpleRequest {
+        response_type: PayloadType::Compressable as i32,
+        response_size: LARGE_RSP_SIZE,
+        payload: Some(payload),
+        ..Default::default()
+    };
+
+    let result = client.unary_call(Request::new(req)).await;
+
+    assertions.push(test_assert!(
+        "call must be successful",
+        result.is_ok(),
+        format!("result={:?}", result)
+    ));
+
+    if let Ok(response) = result {
+        let body = response.into_inner();
+        let payload_len = body.payload.as_ref().map(|p| p.body.len()).unwrap_or(0);
+
+        assertions.push(test_assert!(
+            "body must be 314159 bytes",
+            payload_len == LARGE_RSP_SIZE as usize,
+            format!("mem::size_of_val(&body)={:?}", mem::size_of_val(&body))
+        ));
+    }
+}
+
+// pub async fn cachable_unary(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+//     let payload = Payload {
+//         r#type: PayloadType::Compressable as i32,
+//         body: format!("{:?}", std::time::Instant::now()).into_bytes(),
+//     };
+//     let req = SimpleRequest {
+//         response_type: PayloadType::Compressable as i32,
+//         payload: Some(payload),
+//         ..Default::default()
+//     };
+
+//     client.
+// }
+
+pub async fn client_streaming(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+    let requests = REQUEST_LENGTHS
+        .iter()
+        .map(|len| StreamingInputCallRequest {
+            payload: Some(crate::client_payload(*len as usize)),
+            ..Default::default()
+        })
+        .map(|v| Ok(v));
+
+    let stream = stream::iter(requests);
+
+    let result = client.streaming_input_call(Request::new(stream)).await;
+
+    assertions.push(test_assert!(
+        "call must be successful",
+        result.is_ok(),
+        format!("result={:?}", result)
+    ));
+
+    if let Ok(response) = result {
+        let body = response.into_inner();
+
+        assertions.push(test_assert!(
+            "aggregated payload size must be 74922 bytes",
+            body.aggregated_payload_size == 74922,
+            format!("aggregated_payload_size={:?}", body.aggregated_payload_size)
+        ));
+    }
+}
+
+pub async fn server_streaming(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+    let req = StreamingOutputCallRequest {
+        response_parameters: RESPONSE_LENGTHS
+            .iter()
+            .map(|len| ResponseParameters::with_size(*len))
+            .collect(),
+        ..Default::default()
+    };
+    let req = Request::new(req);
+
+    let result = client.streaming_output_call(req).await;
+
+    assertions.push(test_assert!(
+        "call must be successful",
+        result.is_ok(),
+        format!("result={:?}", result)
+    ));
+
+    if let Ok(response) = result {
+        let responses = response
+            .into_inner()
+            .filter_map(|m| future::ready(m.ok()))
+            .collect::<Vec<_>>()
+            .await;
+        let actual_response_lengths = crate::response_lengths(&responses);
+        let asserts = vec![
+            test_assert!(
+                "there should be four responses",
+                responses.len() == 4,
+                format!("responses.len()={:?}", responses.len())
+            ),
+            test_assert!(
+                "the response payload sizes should match input",
+                RESPONSE_LENGTHS == actual_response_lengths.as_slice(),
+                format!("{:?}={:?}", RESPONSE_LENGTHS, actual_response_lengths)
+            ),
+        ];
+
+        assertions.extend(asserts);
+    }
+}
+
+pub async fn ping_pong(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+    fn make_ping_pong_request(idx: usize) -> StreamingOutputCallRequest {
+        let req_len = REQUEST_LENGTHS[idx];
+        let resp_len = RESPONSE_LENGTHS[idx];
+        StreamingOutputCallRequest {
+            response_parameters: vec![ResponseParameters::with_size(resp_len)],
+            payload: Some(crate::client_payload(req_len as usize)),
+            ..Default::default()
+        }
+    }
+
+    let (mut tx, rx) = mpsc::unbounded_channel();
+    tx.try_send(make_ping_pong_request(0)).unwrap();
+
+    let result = client
+        .full_duplex_call(Request::new(rx.map(|s| Ok(s))))
+        .await;
+
+    assertions.push(test_assert!(
+        "call must be successful",
+        result.is_ok(),
+        format!("result={:?}", result)
+    ));
+
+    if let Ok(mut response) = result.map(Response::into_inner) {
+        let mut responses = Vec::new();
+
+        loop {
+            match response.next().await {
+                Some(result) => {
+                    // TODO: what to do with this result?
+                    responses.push(result.unwrap());
+                    if responses.len() == REQUEST_LENGTHS.len() {
+                        drop(tx);
+                        break;
+                    } else {
+                        tx.send(make_ping_pong_request(responses.len()))
+                            .await
+                            .unwrap();
+                    }
+                }
+                None => {
+                    assertions.push(TestAssertion::Failed {
+                        description:
+                            "server should keep the stream open until the client closes it",
+                        expression: "Stream terminated unexpectedly early",
+                        why: None,
+                    });
+                    break;
+                }
+            }
+        }
+
+        let actual_response_lengths = crate::response_lengths(&responses);
+        assertions.push(test_assert!(
+            "there should be four responses",
+            responses.len() == RESPONSE_LENGTHS.len(),
+            format!("{:?}={:?}", responses.len(), RESPONSE_LENGTHS.len())
+        ));
+        assertions.push(test_assert!(
+            "the response payload sizes should match input",
+            RESPONSE_LENGTHS == actual_response_lengths.as_slice(),
+            format!("{:?}={:?}", RESPONSE_LENGTHS, actual_response_lengths)
+        ));
+    }
+}
+
+pub async fn empty_stream(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+    let stream = stream::iter(Vec::new());
+    let result = client.full_duplex_call(Request::new(stream)).await;
+
+    assertions.push(test_assert!(
+        "call must be successful",
+        result.is_ok(),
+        format!("result={:?}", result)
+    ));
+
+    if let Ok(response) = result.map(Response::into_inner) {
+        let responses = response.collect::<Vec<_>>().await;
+
+        assertions.push(test_assert!(
+            "there should be no responses",
+            responses.len() == 0,
+            format!("responses.len()={:?}", responses.len())
+        ));
+    }
+}

tonic-interop/src/lib.rs

@@ -1,5 +1,3 @@
-#![feature(async_await)]
-
 pub mod client;
 
 pub mod pb {
@@ -8,7 +6,34 @@ pub mod pb {
     include!(concat!(env!("OUT_DIR"), "/grpc.testing.rs"));
 }
 
-use std::fmt;
+use std::{default, fmt, iter};
+
+pub fn client_payload(size: usize) -> pb::Payload {
+    pb::Payload {
+        r#type: default::Default::default(),
+        body: iter::repeat(0u8).take(size).collect(),
+    }
+}
+
+impl pb::ResponseParameters {
+    fn with_size(size: i32) -> Self {
+        pb::ResponseParameters {
+            size,
+            ..Default::default()
+        }
+    }
+}
+
+fn response_length(response: &pb::StreamingOutputCallResponse) -> i32 {
+    match &response.payload {
+        Some(ref payload) => payload.body.len() as i32,
+        None => 0,
+    }
+}
+
+fn response_lengths(responses: &Vec<pb::StreamingOutputCallResponse>) -> Vec<i32> {
+    responses.iter().map(&response_length).collect()
+}
 
 #[derive(Debug)]
 pub enum TestAssertion {