diff --git a/Cargo.toml b/Cargo.toml index 76604db..9cc7739 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,5 +4,6 @@ members = [ "tonic-macros", "tonic-build", "tonic-examples", + "tonic-interop", "tower-h2" ] diff --git a/tonic-interop/Cargo.toml b/tonic-interop/Cargo.toml new file mode 100644 index 0000000..db0bbbd --- /dev/null +++ b/tonic-interop/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "tonic-interop" +version = "0.1.0" +authors = ["Lucio Franco "] +edition = "2018" + +[[bin]] +name = "client" +path = "src/bin/client.rs" + +[[bin]] +name = "server" +path = "src/bin/server.rs" + +[dependencies] +tokio = "=0.2.0-alpha.1" +tonic = { path = "../tonic" } +prost = "0.5" +prost-derive = "0.5" +bytes = "0.4" +tower-h2 = { path = "../tower-h2" } +http = "0.1" + +console = "0.7" +clap = "2.0" + +[build-dependencies] +tonic-build = { path = "../tonic-build" } diff --git a/tonic-interop/build.rs b/tonic-interop/build.rs new file mode 100644 index 0000000..5166d97 --- /dev/null +++ b/tonic-interop/build.rs @@ -0,0 +1,11 @@ +fn main() { + let files = &["proto/grpc/testing/test.proto"]; + let dirs = &["proto/grpc/testing"]; + + tonic_build::compile_protos(files, dirs).unwrap(); + + // prevent needing to rebuild if files (or deps) haven't changed + for file in files { + println!("cargo:rerun-if-changed={}", file); + } +} diff --git a/tonic-interop/proto/grpc/testing/empty.proto b/tonic-interop/proto/grpc/testing/empty.proto new file mode 100644 index 0000000..eb84e13 --- /dev/null +++ b/tonic-interop/proto/grpc/testing/empty.proto @@ -0,0 +1,27 @@ +// Copyright 2015 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package grpc.testing; + +// An empty message that you can re-use to avoid defining duplicated empty +// messages in your project. A typical example is to use it as argument or the +// return value of a service API. For instance: +// +// service Foo { +// rpc Bar (grpc.testing.Empty) returns (grpc.testing.Empty) { }; +// }; +// +message Empty {} \ No newline at end of file diff --git a/tonic-interop/proto/grpc/testing/messages.proto b/tonic-interop/proto/grpc/testing/messages.proto new file mode 100644 index 0000000..722dfd1 --- /dev/null +++ b/tonic-interop/proto/grpc/testing/messages.proto @@ -0,0 +1,168 @@ +// Copyright 2015-2016 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Message definitions to be used by integration test service definitions. + +syntax = "proto3"; + +package grpc.testing; + +// TODO(dgq): Go back to using well-known types once +// https://github.com/grpc/grpc/issues/6980 has been fixed. +// import "google/protobuf/wrappers.proto"; +message BoolValue { + // The bool value. + bool value = 1; +} + +// DEPRECATED, don't use. To be removed shortly. +// The type of payload that should be returned. +enum PayloadType { + // Compressable text format. + COMPRESSABLE = 0; +} + +// A block of data, to simply increase gRPC message size. +message Payload { + // DEPRECATED, don't use. To be removed shortly. + // The type of data in body. + PayloadType type = 1; + // Primary contents of payload. + bytes body = 2; +} + +// A protobuf representation for grpc status. This is used by test +// clients to specify a status that the server should attempt to return. +message EchoStatus { + int32 code = 1; + string message = 2; +} + +// Unary request. +message SimpleRequest { + // DEPRECATED, don't use. To be removed shortly. + // Desired payload type in the response from the server. + // If response_type is RANDOM, server randomly chooses one from other formats. + PayloadType response_type = 1; + + // Desired payload size in the response from the server. + int32 response_size = 2; + + // Optional input payload sent along with the request. + Payload payload = 3; + + // Whether SimpleResponse should include username. + bool fill_username = 4; + + // Whether SimpleResponse should include OAuth scope. + bool fill_oauth_scope = 5; + + // Whether to request the server to compress the response. This field is + // "nullable" in order to interoperate seamlessly with clients not able to + // implement the full compression tests by introspecting the call to verify + // the response's compression status. + BoolValue response_compressed = 6; + + // Whether server should return a given status + EchoStatus response_status = 7; + + // Whether the server should expect this request to be compressed. + BoolValue expect_compressed = 8; +} + +// Unary response, as configured by the request. +message SimpleResponse { + // Payload to increase message size. + Payload payload = 1; + // The user the request came from, for verifying authentication was + // successful when the client expected it. + string username = 2; + // OAuth scope. + string oauth_scope = 3; +} + +// Client-streaming request. +message StreamingInputCallRequest { + // Optional input payload sent along with the request. + Payload payload = 1; + + // Whether the server should expect this request to be compressed. This field + // is "nullable" in order to interoperate seamlessly with servers not able to + // implement the full compression tests by introspecting the call to verify + // the request's compression status. + BoolValue expect_compressed = 2; + + // Not expecting any payload from the response. +} + +// Client-streaming response. +message StreamingInputCallResponse { + // Aggregated size of payloads received from the client. + int32 aggregated_payload_size = 1; +} + +// Configuration for a particular response. +message ResponseParameters { + // Desired payload sizes in responses from the server. + int32 size = 1; + + // Desired interval between consecutive responses in the response stream in + // microseconds. + int32 interval_us = 2; + + // Whether to request the server to compress the response. This field is + // "nullable" in order to interoperate seamlessly with clients not able to + // implement the full compression tests by introspecting the call to verify + // the response's compression status. + BoolValue compressed = 3; +} + +// Server-streaming request. +message StreamingOutputCallRequest { + // DEPRECATED, don't use. To be removed shortly. + // Desired payload type in the response from the server. + // If response_type is RANDOM, the payload from each response in the stream + // might be of different types. This is to simulate a mixed type of payload + // stream. + PayloadType response_type = 1; + + // Configuration for each expected response message. + repeated ResponseParameters response_parameters = 2; + + // Optional input payload sent along with the request. + Payload payload = 3; + + // Whether server should return a given status + EchoStatus response_status = 7; +} + +// Server-streaming response, as configured by the request and parameters. +message StreamingOutputCallResponse { + // Payload to increase response size. + Payload payload = 1; +} + +// For reconnect interop test only. +// Client tells server what reconnection parameters it used. +message ReconnectParams { + int32 max_reconnect_backoff_ms = 1; +} + +// For reconnect interop test only. +// Server tells client whether its reconnects are following the spec and the +// reconnect backoffs it saw. +message ReconnectInfo { + bool passed = 1; + repeated int32 backoff_ms = 2; +} diff --git a/tonic-interop/proto/grpc/testing/test.proto b/tonic-interop/proto/grpc/testing/test.proto new file mode 100644 index 0000000..830d68e --- /dev/null +++ b/tonic-interop/proto/grpc/testing/test.proto @@ -0,0 +1,78 @@ +// Copyright 2015-2016 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// An integration test service that covers all the method signature permutations +// of unary/streaming requests/responses. + +syntax = "proto3"; + +import "empty.proto"; +import "messages.proto"; + +package grpc.testing; + +// A simple service to test the various types of RPCs and experiment with +// performance with various types of payload. +service TestService { + // One empty request followed by one empty response. + rpc EmptyCall(grpc.testing.Empty) returns (grpc.testing.Empty); + + // One request followed by one response. + rpc UnaryCall(SimpleRequest) returns (SimpleResponse); + + // One request followed by one response. Response has cache control + // headers set such that a caching HTTP proxy (such as GFE) can + // satisfy subsequent requests. + rpc CacheableUnaryCall(SimpleRequest) returns (SimpleResponse); + + // One request followed by a sequence of responses (streamed download). + // The server returns the payload with client desired type and sizes. + rpc StreamingOutputCall(StreamingOutputCallRequest) + returns (stream StreamingOutputCallResponse); + + // A sequence of requests followed by one response (streamed upload). + // The server returns the aggregated size of client payload as the result. + rpc StreamingInputCall(stream StreamingInputCallRequest) + returns (StreamingInputCallResponse); + + // A sequence of requests with each request served by the server immediately. + // As one request could lead to multiple responses, this interface + // demonstrates the idea of full duplexing. + rpc FullDuplexCall(stream StreamingOutputCallRequest) + returns (stream StreamingOutputCallResponse); + + // A sequence of requests followed by a sequence of responses. + // The server buffers all the client requests and then serves them in order. A + // stream of responses are returned to the client when the server starts with + // first request. + rpc HalfDuplexCall(stream StreamingOutputCallRequest) + returns (stream StreamingOutputCallResponse); + + // The test server will not implement this method. It will be used + // to test the behavior when clients call unimplemented methods. + rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty); +} + +// A simple service NOT implemented at servers so clients can test for +// that case. +service UnimplementedService { + // A call that no server should implement + rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty); +} + +// A service used to control reconnect server. +service ReconnectService { + rpc Start(grpc.testing.ReconnectParams) returns (grpc.testing.Empty); + rpc Stop(grpc.testing.Empty) returns (grpc.testing.ReconnectInfo); +} \ No newline at end of file diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs new file mode 100644 index 0000000..f1cd863 --- /dev/null +++ b/tonic-interop/src/bin/client.rs @@ -0,0 +1,79 @@ +#![feature(async_await)] + +use clap::{arg_enum, App, Arg, values_t}; +use tonic_interop::client; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let matches = App::new("My Super Program") + .version("1.0") + .about("Does awesome things") + .arg( + Arg::with_name("test_case") + .long("test_case") + .value_name("TESTCASE") + .help( + "The name of the test case to execute. For example, + \"empty_unary\".", + ) + .possible_values(&Testcase::variants()) + .default_value("large_unary") + .takes_value(true) + .min_values(1) + .use_delimiter(true), + ) + .get_matches(); + + let test_cases = values_t!(matches, "test_case", Testcase).unwrap_or_else(|e| e.exit()); + + let addr = "127.0.0.1:10000".parse()?; + + let mut client = client::create(addr).await?; + + for test_case in test_cases { + println!("{:?}:", test_case); + let mut test_results = Vec::new(); + + match test_case { + Testcase::empty_unary => client::unary_call(&mut client, &mut test_results).await, + _ => unimplemented!(), + } + + for result in test_results { + println!(" {}", result); + } + } + + Ok(()) +} + +arg_enum! { + #[derive(Debug, Copy, Clone)] + #[allow(non_camel_case_types)] + enum Testcase { + empty_unary, + cacheable_unary, + large_unary, + client_compressed_unary, + server_compressed_unary, + client_streaming, + client_compressed_streaming, + server_streaming, + server_compressed_streaming, + ping_pong, + empty_stream, + compute_engine_creds, + jwt_token_creds, + oauth2_auth_token, + per_rpc_creds, + custom_metadata, + status_code_and_message, + special_status_message, + unimplemented_method, + unimplemented_service, + cancel_after_begin, + cancel_after_first_response, + timeout_on_sleeping_server, + concurrent_large_unary + } +} diff --git a/tonic-interop/src/bin/server.rs b/tonic-interop/src/bin/server.rs new file mode 100644 index 0000000..276fc9b --- /dev/null +++ b/tonic-interop/src/bin/server.rs @@ -0,0 +1,49 @@ +#![feature(async_await)] + +use std::time::Duration; +use tokio::{net::TcpListener, timer::Delay}; +use tonic::{Request, Response, Status}; +use tower_h2::Server; + +pub mod pb { + #![allow(dead_code)] + #![allow(unused_imports)] + include!(concat!(env!("OUT_DIR"), "/grpc.testing.rs")); + tonic::client!(service = "grpc.testing.TestService", proto = "self"); +} + +use pb::*; + +#[derive(Default, Clone)] +pub struct TestService { + data: String, +} + +#[tonic::server(service = "grpc.testing.TestService", proto = "pb")] +impl TestService { + pub async fn empty_call(&self, request: Request) -> Result, Status> { + println!("REQUEST={:?}", request); + Ok(Response::new(Empty {})) + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let addr = "[::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 { + if let Err(e) = sock.set_nodelay(true) { + return Err(e.into()); + } + + if let Err(e) = server.serve(sock).await { + println!("H2 ERROR: {}", e); + } + } + + Ok(()) +} diff --git a/tonic-interop/src/client.rs b/tonic-interop/src/client.rs new file mode 100644 index 0000000..31612e6 --- /dev/null +++ b/tonic-interop/src/client.rs @@ -0,0 +1,39 @@ +use crate::{pb::*, test_assert, TestAssertion}; +use std::net::SocketAddr; +use tokio::net::TcpStream; +use tonic::Request; +use tower_h2::{add_origin::AddOrigin, Connection}; + +pub type Client = TestServiceClient>>; + +tonic::client!(service = "grpc.testing.TestService", proto = "crate::pb"); + +pub async fn create(addr: SocketAddr) -> 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(TestServiceClient::new(svc)) +} + +pub async fn unary_call(client: &mut Client, assertions: &mut Vec) { + let result = client.empty_call(Request::new(Empty {})).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!( + "body must not be null", + body == Empty {}, + format!("body={:?}", body) + )); + } +} diff --git a/tonic-interop/src/lib.rs b/tonic-interop/src/lib.rs new file mode 100644 index 0000000..16010ce --- /dev/null +++ b/tonic-interop/src/lib.rs @@ -0,0 +1,90 @@ +#![feature(async_await)] + +pub mod client; + +pub mod pb { + #![allow(dead_code)] + #![allow(unused_imports)] + include!(concat!(env!("OUT_DIR"), "/grpc.testing.rs")); +} + +use std::fmt; + +#[derive(Debug)] +pub enum TestAssertion { + Passed { + description: &'static str, + }, + Failed { + description: &'static str, + expression: &'static str, + why: Option, + }, +} + +impl fmt::Display for TestAssertion { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + use console::{style, Emoji}; + match *self { + TestAssertion::Passed { ref description } => write!( + f, + "{check} {desc}", + check = style(Emoji("✔", "+")).green(), + desc = style(description).green(), + ), + TestAssertion::Failed { + ref description, + ref expression, + why: Some(ref why), + } => write!( + f, + "{check} {desc}\n in `{exp}`: {why}", + check = style(Emoji("✖", "x")).red(), + desc = style(description).red(), + exp = style(expression).red(), + why = style(why).red(), + ), + TestAssertion::Failed { + ref description, + ref expression, + why: None, + } => write!( + f, + "{check} {desc}\n in `{exp}`", + check = style(Emoji("✖", "x")).red(), + desc = style(description).red(), + exp = style(expression).red(), + ), + } + } +} + +#[macro_export] +macro_rules! test_assert { + ($description:expr, $assertion:expr) => { + if $assertion { + crate::TestAssertion::Passed { + description: $description, + } + } else { + TestAssertion::Failed { + description: $description, + expression: stringify!($assertion), + why: None, + } + } + }; + ($description:expr, $assertion:expr, $why:expr) => { + if $assertion { + crate::TestAssertion::Passed { + description: $description, + } + } else { + crate::TestAssertion::Failed { + description: $description, + expression: stringify!($assertion), + why: Some($why), + } + } + }; +} diff --git a/tonic-macros/src/lib.rs b/tonic-macros/src/lib.rs index 5dc375d..3568bff 100644 --- a/tonic-macros/src/lib.rs +++ b/tonic-macros/src/lib.rs @@ -20,6 +20,8 @@ pub fn client(attr: TokenStream) -> TokenStream { let methods = client::generate(service, proto_path); let output = quote! { + use tonic::_codegen::*; + pub struct #service_ident { inner: tonic::client::Grpc, } diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 1596dd1..0ac49dd 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -1,6 +1,6 @@ use crate::{ body::{Body, BoxBody}, - codec::{decode_response, decode_empty, encode, Codec, Streaming}, + codec::{decode_empty, decode_response, encode, Codec, Streaming}, Code, GrpcService, Request, Response, Status, }; use futures_core::Stream; diff --git a/tonic/src/codec/prost.rs b/tonic/src/codec/prost.rs index 177a582..14af51d 100644 --- a/tonic/src/codec/prost.rs +++ b/tonic/src/codec/prost.rs @@ -27,7 +27,7 @@ where type Encoder = ProstEncoder; type Decoder = ProstDecoder; - const CONTENT_TYPE: &'static str = "application/groc+proto"; + const CONTENT_TYPE: &'static str = "application/grpc+proto"; fn encoder(&mut self) -> Self::Encoder { ProstEncoder(PhantomData) diff --git a/tower-h2/src/flush.rs b/tower-h2/src/flush.rs index 55c815c..87c4d39 100644 --- a/tower-h2/src/flush.rs +++ b/tower-h2/src/flush.rs @@ -17,6 +17,7 @@ where state: FlushState, } +#[derive(Debug)] enum FlushState { Data, Trailers,