From f74bba68580758d7e7bd85e8940e2a7a1aca7240 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Fri, 6 Sep 2019 21:40:59 -0400 Subject: [PATCH] Add server interop tests --- tonic-build/src/lib.rs | 4 ++- tonic-interop/Cargo.toml | 1 + tonic-interop/src/lib.rs | 9 +++++ tonic-interop/src/server.rs | 65 ++++++++++++++++++++++++++++++++----- tonic-interop/test.sh | 17 ++++++++++ 5 files changed, 86 insertions(+), 10 deletions(-) diff --git a/tonic-build/src/lib.rs b/tonic-build/src/lib.rs index baf11b2..c35b382 100644 --- a/tonic-build/src/lib.rs +++ b/tonic-build/src/lib.rs @@ -15,7 +15,9 @@ use proc_macro2::TokenStream; use prost_build::Config; -use std::{io, path, path::Path, process::Command}; +#[cfg(feature = "rustfmt")] +use std::process::Command; +use std::{io, path, path::Path}; mod client; mod service; diff --git a/tonic-interop/Cargo.toml b/tonic-interop/Cargo.toml index 822436a..63a92aa 100644 --- a/tonic-interop/Cargo.toml +++ b/tonic-interop/Cargo.toml @@ -21,6 +21,7 @@ bytes = "0.4" http = "0.1" futures-core-preview = "=0.3.0-alpha.18" futures-util-preview = "=0.3.0-alpha.18" +async-stream = { git = "https://github.com/tokio-rs/async-stream" } console = "0.7" structopt = "0.2" diff --git a/tonic-interop/src/lib.rs b/tonic-interop/src/lib.rs index b4e6e1a..799728f 100644 --- a/tonic-interop/src/lib.rs +++ b/tonic-interop/src/lib.rs @@ -1,3 +1,5 @@ +#![recursion_limit = "256"] + pub mod client; pub mod server; @@ -16,6 +18,13 @@ pub fn client_payload(size: usize) -> pb::Payload { } } +pub fn server_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 { diff --git a/tonic-interop/src/server.rs b/tonic-interop/src/server.rs index ec914a4..e9e3f78 100644 --- a/tonic-interop/src/server.rs +++ b/tonic-interop/src/server.rs @@ -1,5 +1,8 @@ use crate::pb::{self, *}; +use async_stream::try_stream; +use futures_util::TryStreamExt; use std::pin::Pin; +use std::time::{Duration, Instant}; use tonic::{Code, Request, Response, Status}; pub fn create() -> pb::server::TestServiceServer { @@ -21,13 +24,10 @@ type Stream = #[tonic::async_trait] impl pb::server::TestService for TestService { async fn empty_call(&self, _request: Request) -> Result { - println!("empty_call"); Ok(Response::new(Empty {})) } async fn unary_call(&self, request: Request) -> Result { - println!("unary_call"); - let req = request.into_inner(); if let Some(echo_status) = req.response_status { @@ -61,25 +61,72 @@ impl pb::server::TestService for TestService { async fn streaming_output_call( &self, - _: Request, + req: Request, ) -> Result { - unimplemented!() + let StreamingOutputCallRequest { + response_parameters, + .. + } = req.into_inner(); + + let stream = try_stream! { + for param in response_parameters { + let deadline = Instant::now() + Duration::from_micros(param.interval_us as u64); + tokio::timer::delay(deadline).await; + + let payload = crate::server_payload(param.size as usize); + yield StreamingOutputCallResponse { payload: Some(payload) }; + } + }; + + Ok(Response::new( + Box::pin(stream) as Self::StreamingOutputCallStream + )) } async fn streaming_input_call( &self, - _: Streaming, + req: Streaming, ) -> Result { - unimplemented!() + let mut stream = req.into_inner(); + + let mut aggregated_payload_size = 0 as i32; + while let Some(msg) = stream.try_next().await? { + aggregated_payload_size += msg.payload.unwrap().body.len() as i32; + } + + let res = StreamingInputCallResponse { + aggregated_payload_size, + }; + + Ok(Response::new(res)) } type FullDuplexCallStream = Stream; async fn full_duplex_call( &self, - _: Streaming, + req: Streaming, ) -> Result { - unimplemented!() + let mut stream = req.into_inner(); + + let stream = try_stream! { + while let Some(msg) = stream.try_next().await? { + if let Some(echo_status) = msg.response_status { + let status = Status::new(Code::from_i32(echo_status.code), echo_status.message); + Err(status)?; + } + + for param in msg.response_parameters { + let deadline = Instant::now() + Duration::from_micros(param.interval_us as u64); + tokio::timer::delay(deadline).await; + + let payload = crate::server_payload(param.size as usize); + yield StreamingOutputCallResponse { payload: Some(payload) }; + } + } + }; + + Ok(Response::new(Box::pin(stream) as Self::FullDuplexCallStream)) } type HalfDuplexCallStream = Stream; diff --git a/tonic-interop/test.sh b/tonic-interop/test.sh index 2766689..df47e3b 100755 --- a/tonic-interop/test.sh +++ b/tonic-interop/test.sh @@ -29,3 +29,20 @@ trap 'echo ":; killing test server"; kill ${SERVER_PID};' EXIT --test_case=empty_unary,large_unary,client_streaming,server_streaming,ping_pong,\ empty_stream,status_code_and_message,special_status_message,unimplemented_method,\ unimplemented_service,custom_metadata $ARG + +echo ":; killing test server"; kill ${SERVER_PID}; + +# run the test server +./target/debug/server & #$ARG --tls_cert_file $TLS_CRT --tls_key_file $TLS_KEY & +SERVER_PID=$! +echo ":; started tonic test server." + +# trap exits to make sure we kill the server process when the script exits, +# regardless of why (errors, SIGTERM, etc). +trap 'echo ":; killing test server"; kill ${SERVER_PID};' EXIT + +cargo run -p tonic-interop --bin client -- \ +--test_case=empty_unary,large_unary,client_streaming,server_streaming,ping_pong,\ +empty_stream +# status_code_and_message,special_status_message,unimplemented_method,\ +# unimplemented_service,custom_metadata $ARG