diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs index 166423c..101e452 100644 --- a/tonic-interop/src/bin/client.rs +++ b/tonic-interop/src/bin/client.rs @@ -29,6 +29,7 @@ async fn main() -> Result<(), Box> { let addr = "127.0.0.1:10000".parse()?; let mut client = client::create(addr).await?; + let mut unimplemented_client = client::create_unimplemented(addr).await?; for test_case in test_cases { println!("{:?}:", test_case); @@ -40,13 +41,24 @@ async fn main() -> Result<(), Box> { 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, + Testcase::status_code_and_message => { + client::status_code_and_message(&mut client, &mut test_results).await + } + Testcase::special_status_message => { + client::special_status_message(&mut client, &mut test_results).await + } + Testcase::unimplemented_method => { + client::unimplemented_method(&mut client, &mut test_results).await + } + Testcase::unimplemented_service => { + client::unimplemented_service(&mut unimplemented_client, &mut test_results).await + } + Testcase::custom_metadata => client::custom_metadata(&mut client, &mut test_results).await, _ => unimplemented!(), } diff --git a/tonic-interop/src/client.rs b/tonic-interop/src/client.rs index c539040..7e589ae 100644 --- a/tonic-interop/src/client.rs +++ b/tonic-interop/src/client.rs @@ -2,20 +2,25 @@ use crate::{pb::*, test_assert, TestAssertion}; use futures_util::{future, stream, SinkExt, StreamExt}; use std::net::SocketAddr; use tokio::{net::TcpStream, sync::mpsc}; -use tonic::{Request, Response}; +use tonic::{metadata::MetadataValue, Code, Request, Response, Status}; use tower_h2::{add_origin::AddOrigin, Connection}; pub type Client = TestServiceClient>>; +pub type UnimplementedClient = UnimplementedServiceClient>>; tonic::client!(service = "grpc.testing.TestService", proto = "crate::pb"); +tonic::client!( + service = "grpc.testing.UnimplementedService", + 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"; +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> { let io = TcpStream::connect(&addr).await?; @@ -28,6 +33,19 @@ pub async fn create(addr: SocketAddr) -> Result Result> { + let io = TcpStream::connect(&addr).await?; + + let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); + + let svc = Connection::handshake(io).await?; + let svc = AddOrigin::new(svc, origin); + + Ok(UnimplementedServiceClient::new(svc)) +} + pub async fn empty_unary(client: &mut Client, assertions: &mut Vec) { let result = client.empty_call(Request::new(Empty {})).await; @@ -164,16 +182,6 @@ pub async fn server_streaming(client: &mut Client, assertions: &mut Vec) { - 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(); @@ -250,3 +258,184 @@ pub async fn empty_stream(client: &mut Client, assertions: &mut Vec) { + fn validate_response(result: Result, assertions: &mut Vec) + where + T: std::fmt::Debug, + { + assertions.push(test_assert!( + "call must fail with unknown status code", + match &result { + Err(status) => status.code() == Code::Unknown, + _ => false, + }, + format!("result={:?}", result) + )); + + assertions.push(test_assert!( + "call must respsond with expected status message", + match &result { + Err(status) => status.message() == TEST_STATUS_MESSAGE, + _ => false, + }, + format!("result={:?}", result) + )); + } + + let simple_req = SimpleRequest { + response_status: Some(EchoStatus { + code: 2, + message: TEST_STATUS_MESSAGE.to_string(), + ..Default::default() + }), + ..Default::default() + }; + + let duplex_req = StreamingOutputCallRequest { + response_status: Some(EchoStatus { + code: 2, + message: TEST_STATUS_MESSAGE.to_string(), + ..Default::default() + }), + ..Default::default() + }; + + let result = client.unary_call(Request::new(simple_req)).await; + validate_response(result, assertions); + + let stream = stream::iter(vec![Ok(duplex_req)]); + let result = match client.full_duplex_call(Request::new(stream)).await { + Ok(response) => { + let stream = response.into_inner(); + let responses = stream.collect::>().await; + Ok(responses) + } + Err(e) => Err(e), + }; + + validate_response(result, assertions); +} + +pub async fn special_status_message(client: &mut Client, assertions: &mut Vec) { + let req = SimpleRequest { + response_status: Some(EchoStatus { + code: 2, + message: SPECIAL_TEST_STATUS_MESSAGE.to_string(), + ..Default::default() + }), + ..Default::default() + }; + + let result = client.unary_call(Request::new(req)).await; + + assertions.push(test_assert!( + "call must fail with unknown status code", + match &result { + Err(status) => status.code() == Code::Unknown, + _ => false, + }, + format!("result={:?}", result) + )); + + assertions.push(test_assert!( + "call must respsond with expected status message", + match &result { + Err(status) => status.message() == SPECIAL_TEST_STATUS_MESSAGE, + _ => false, + }, + format!("result={:?}", result) + )); +} + +pub async fn unimplemented_method(client: &mut Client, assertions: &mut Vec) { + let result = client.unimplemented_call(Request::new(Empty {})).await; + assertions.push(test_assert!( + "call must fail with unimplemented status code", + match &result { + Err(status) => status.code() == Code::Unimplemented, + _ => false, + }, + format!("result={:?}", result) + )); +} + +pub async fn unimplemented_service( + client: &mut UnimplementedClient, + assertions: &mut Vec, +) { + let result = client.unimplemented_call(Request::new(Empty {})).await; + assertions.push(test_assert!( + "call must fail with unimplemented status code", + match &result { + Err(status) => status.code() == Code::Unimplemented, + _ => false, + }, + format!("result={:?}", result) + )); +} + +pub async fn custom_metadata(client: &mut Client, assertions: &mut Vec) { + let key1 = "x-grpc-test-echo-initial"; + let value1 = MetadataValue::from_str("test_initial_metadata_value").unwrap(); + let key2 = "x-grpc-test-echo-trailing-bin"; + let value2 = MetadataValue::from_bytes(&[0xab, 0xab, 0xab]); + + let req = SimpleRequest { + response_type: PayloadType::Compressable as i32, + response_size: LARGE_RSP_SIZE, + payload: Some(crate::client_payload(LARGE_REQ_SIZE)), + ..Default::default() + }; + let mut req_unary = Request::new(req); + req_unary.metadata_mut().insert(key1, value1.clone()); + req_unary.metadata_mut().insert_bin(key2, value2.clone()); + + // TODO: custom metadata for fullduplex + let stream = stream::iter(vec![Ok(make_ping_pong_request(0))]); + let mut req_stream = Request::new(stream); + req_stream.metadata_mut().insert(key1, value1.clone()); + req_stream.metadata_mut().insert_bin(key2, value2.clone()); + + // let response = client + // .unary_call(req_unary) + // .await + // .expect("call should pass."); + + // assertions.push(test_assert!( + // "metadata string must match in unary", + // response.metadata().get(key1) == Some(&value1), + // format!("result={:?}", response.metadata().get(key1)) + // )); + // assertions.push(test_assert!( + // "metadata bin must match in unary", + // response.metadata().get_bin(key2) == Some(&value2), + // format!("result={:?}", response.metadata().get_bin(key1)) + // )); + + let response = client + .full_duplex_call(req_stream) + .await + .expect("call should pass."); + + assertions.push(test_assert!( + "metadata string must match in unary", + response.metadata().get(key1) == Some(&value1), + format!("result={:?}", response.metadata().get(key1)) + )); + assertions.push(test_assert!( + "metadata bin must match in unary", + response.metadata().get_bin(key2) == Some(&value2), + format!("result={:?}", response.metadata().get_bin(key1)) + )); +} + +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() + } +} diff --git a/tonic-interop/test.sh b/tonic-interop/test.sh new file mode 100755 index 0000000..43ef7af --- /dev/null +++ b/tonic-interop/test.sh @@ -0,0 +1,6 @@ + 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 + +# DISABLED: ,custom_metadata diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index b73e635..9ed8e02 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -81,6 +81,7 @@ enum State { ReadBody { compression: bool, len: usize }, } +#[derive(Debug)] enum Direction { Request, Response(StatusCode), @@ -103,7 +104,11 @@ where let mut state = State::ReadHeader; loop { - if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state)? { + if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state)? { + // TODO: implement the ability to poll trailers when we _know_ that + // the comnsumer of this stream will only poll for the first message. + // This means we skip the poll_trailers step. + yield item; } @@ -142,9 +147,7 @@ where if let Direction::Response(status) = direction { let trailer = future::poll_fn(|cx| unsafe { std::pin::Pin::new_unchecked(&mut source) }.poll_trailers(cx)); let trailer = match trailer.await { - Ok(trailer) => { - crate::status::infer_grpc_status(trailer, status)?; - }, + Ok(trailer) => crate::status::infer_grpc_status(trailer, status)?, Err(e) => { let err = e.into(); debug!("decoder inner trailers error: {:?}", err);