From 93eb9989d6dd4b39c247e3a9703e0245d8e769e0 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Wed, 21 Aug 2019 18:50:27 -0400 Subject: [PATCH] Fix encoding, more interop and update nightly version --- rust-toolchain | 2 +- tonic-examples/src/helloworld/client.rs | 2 - tonic-examples/src/helloworld/server.rs | 2 - tonic-examples/src/routeguide/client.rs | 2 - tonic-examples/src/routeguide/server.rs | 2 - tonic-interop/Cargo.toml | 2 + tonic-interop/src/bin/client.rs | 19 +- tonic-interop/src/bin/server.rs | 44 ++++- tonic-interop/src/client.rs | 219 +++++++++++++++++++++++- tonic-interop/src/lib.rs | 31 +++- tonic-macros/src/lib.rs | 1 - tonic/Cargo.toml | 3 +- tonic/src/body.rs | 29 ++-- tonic/src/client/grpc.rs | 11 +- tonic/src/codec/decode.rs | 19 +- tonic/src/codec/encode.rs | 96 ++++++++++- tonic/src/codec/mod.rs | 2 +- tonic/src/lib.rs | 1 - tower-h2/Cargo.toml | 7 +- tower-h2/examples/client.rs | 2 - tower-h2/examples/server.rs | 2 - tower-h2/src/lib.rs | 2 - tower-h2/src/recv_body.rs | 2 +- 23 files changed, 440 insertions(+), 62 deletions(-) diff --git a/rust-toolchain b/rust-toolchain index 040b985..bf867e0 100644 --- a/rust-toolchain +++ b/rust-toolchain @@ -1 +1 @@ -nightly-2019-08-09 +nightly diff --git a/tonic-examples/src/helloworld/client.rs b/tonic-examples/src/helloworld/client.rs index 14ac4d9..9c30b9b 100644 --- a/tonic-examples/src/helloworld/client.rs +++ b/tonic-examples/src/helloworld/client.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - use tokio::net::TcpStream; use tower_h2::{add_origin::AddOrigin, Connection}; diff --git a/tonic-examples/src/helloworld/server.rs b/tonic-examples/src/helloworld/server.rs index b5f7f49..d1aa570 100644 --- a/tonic-examples/src/helloworld/server.rs +++ b/tonic-examples/src/helloworld/server.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - use std::time::Duration; use tokio::{net::TcpListener, timer::Delay}; use tonic::{Request, Response, Status}; diff --git a/tonic-examples/src/routeguide/client.rs b/tonic-examples/src/routeguide/client.rs index 4f11702..95944d5 100644 --- a/tonic-examples/src/routeguide/client.rs +++ b/tonic-examples/src/routeguide/client.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - use futures::TryStreamExt; use route_guide::{Point, RouteNote}; use std::time::{Duration, Instant}; diff --git a/tonic-examples/src/routeguide/server.rs b/tonic-examples/src/routeguide/server.rs index 94209bf..a89fb3c 100644 --- a/tonic-examples/src/routeguide/server.rs +++ b/tonic-examples/src/routeguide/server.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - mod data; use futures::{Stream, StreamExt}; diff --git a/tonic-interop/Cargo.toml b/tonic-interop/Cargo.toml index db0bbbd..f6d6b53 100644 --- a/tonic-interop/Cargo.toml +++ b/tonic-interop/Cargo.toml @@ -20,9 +20,11 @@ prost-derive = "0.5" bytes = "0.4" tower-h2 = { path = "../tower-h2" } http = "0.1" +futures-util-preview = "=0.3.0-alpha.17" console = "0.7" clap = "2.0" +pretty_env_logger = "0.3" [build-dependencies] tonic-build = { path = "../tonic-build" } diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs index f1cd863..166423c 100644 --- a/tonic-interop/src/bin/client.rs +++ b/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> { + 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> { 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!(), } diff --git a/tonic-interop/src/bin/server.rs b/tonic-interop/src/bin/server.rs index 276fc9b..d90066f 100644 --- a/tonic-interop/src/bin/server.rs +++ b/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) -> Result, Status> { - println!("REQUEST={:?}", request); + println!("empty_call; REQUEST={:?}", request); Ok(Response::new(Empty {})) } + + pub async fn unary_call( + &self, + request: Request, + ) -> Result, 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> { - 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()); } diff --git a/tonic-interop/src/client.rs b/tonic-interop/src/client.rs index 31612e6..c539040 100644 --- a/tonic-interop/src/client.rs +++ b/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>>; 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> { let io = TcpStream::connect(&addr).await?; @@ -19,7 +28,7 @@ pub async fn create(addr: SocketAddr) -> Result) { +pub async fn empty_unary(client: &mut Client, assertions: &mut Vec) { 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 )); } } + +pub async fn large_unary(client: &mut Client, assertions: &mut Vec) { + 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) { +// 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) { + 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) { + 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::>() + .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) { + 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) { + 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::>().await; + + assertions.push(test_assert!( + "there should be no responses", + responses.len() == 0, + format!("responses.len()={:?}", responses.len()) + )); + } +} diff --git a/tonic-interop/src/lib.rs b/tonic-interop/src/lib.rs index 16010ce..03b64ed 100644 --- a/tonic-interop/src/lib.rs +++ b/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) -> Vec { + responses.iter().map(&response_length).collect() +} #[derive(Debug)] pub enum TestAssertion { diff --git a/tonic-macros/src/lib.rs b/tonic-macros/src/lib.rs index 3568bff..e5147b4 100644 --- a/tonic-macros/src/lib.rs +++ b/tonic-macros/src/lib.rs @@ -1,4 +1,3 @@ -#![feature(async_await)] #![recursion_limit = "256"] extern crate proc_macro; diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 51efb55..a0827f2 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -19,7 +19,8 @@ percent-encoding = "1.0.1" tower-service = { git = "https://github.com/tower-rs/tower", branch = "std-future" } tokio-codec = "=0.2.0-alpha.1" async-stream = "0.1.0" -http-body = { git = "https://github.com/hyperium/http-body" } +# http-body = { git = "https://github.com/hyperium/http-body" } +http-body = { path = "../../http-body" } pin-project = "0.4.0-alpha.2" [dev-dependencies] diff --git a/tonic/src/body.rs b/tonic/src/body.rs index 5b56d7b..98fe907 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -1,4 +1,4 @@ -use crate::{Code, Error, Status}; +use crate::{Error, Status}; use bytes::{Buf, Bytes, IntoBuf}; use futures_core::Stream; use futures_util::{ready, TryStreamExt}; @@ -14,7 +14,7 @@ pub trait Body: sealed::Sealed { type Data: Buf; type Error: Into; - fn is_end_stream(self: Pin<&mut Self>) -> bool; + fn is_end_stream(&self) -> bool; fn poll_data( self: Pin<&mut Self>, @@ -35,7 +35,7 @@ where type Data = T::Data; type Error = T::Error; - fn is_end_stream(self: Pin<&mut Self>) -> bool { + fn is_end_stream(&self) -> bool { HttpBody::is_end_stream(self) } @@ -93,8 +93,8 @@ impl HttpBody for BoxBody { type Data = BytesBuf; type Error = Status; - fn is_end_stream(mut self: Pin<&mut Self>) -> bool { - HttpBody::is_end_stream(self.inner.as_mut()) + fn is_end_stream(&self) -> bool { + HttpBody::is_end_stream(&self.inner) } fn poll_data( @@ -136,6 +136,10 @@ where type Data = BytesBuf; type Error = Status; + fn is_end_stream(&self) -> bool { + false + } + fn poll_data( self: Pin<&mut Self>, cx: &mut Context<'_>, @@ -155,13 +159,14 @@ where self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll, Status>> { - let self_proj = self.project(); - let status = if let Some(status) = self_proj.error.take() { - status - } else { - Status::new(Code::Ok, "") - }; + // let self_proj = self.project(); + // let status = if let Some(status) = self_proj.error.take() { + // status + // } else { + // Status::new(Code::Ok, "") + // }; - Poll::Ready(Ok(Some(status.to_header_map()?))) + // Poll::Ready(Ok(Some(status.to_header_map()?))) + Poll::Ready(Ok(None)) } } diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index 0ac49dd..5d1d21f 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -1,6 +1,6 @@ use crate::{ body::{Body, BoxBody}, - codec::{decode_empty, decode_response, encode, Codec, Streaming}, + codec::{decode_empty, decode_response, encode, Codec, EncodeBody, Streaming}, Code, GrpcService, Request, Response, Status, }; use futures_core::Stream; @@ -115,7 +115,8 @@ impl Grpc { let request = request .map(|s| encode(codec.encoder(), Box::pin(s)).into_stream()) - .map(BoxBody::from_stream); + .map(EncodeBody::new_client) + .map(BoxBody::map_from); let mut request = request.into_http(uri); @@ -139,14 +140,16 @@ impl Grpc { let status_code = response.status(); let trailers_only_status = Status::from_header_map(response.headers()); + // We do not need to check for trailers if the `grpc-status` header is present + // with a valid code. let expect_additional_trailers = if let Some(status) = trailers_only_status { if status.code() != Code::Ok { return Err(status); } - true - } else { false + } else { + true }; let response = response diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index cfdfa0c..b73e635 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -4,7 +4,9 @@ use futures_core::{Stream, TryStream}; use futures_util::future; use http::StatusCode; use http_body::Body; +use std::fmt; use std::pin::Pin; +use std::task::{Context, Poll}; use tokio_codec::Decoder; use tracing::{debug, trace}; @@ -59,7 +61,6 @@ impl Streaming { } } -use std::task::{Context, Poll}; impl Stream for Streaming { type Item = Result; @@ -68,6 +69,12 @@ impl Stream for Streaming { } } +impl fmt::Debug for Streaming { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Streaming") + } +} + #[derive(Debug)] enum State { ReadHeader, @@ -92,7 +99,7 @@ where B::Error: Into, { async_stream::try_stream! { - let mut buf = BytesMut::with_capacity(1024 * 1024); + let mut buf = BytesMut::with_capacity(1024 * 1024 * 1024); let mut state = State::ReadHeader; loop { @@ -102,7 +109,9 @@ where // FIXME: Figure out how to verify that this is safe let chunk = match future::poll_fn(|cx| unsafe { std::pin::Pin::new_unchecked(&mut source) }.poll_data(cx)).await { - Some(Ok(d)) => Some(d), + Some(Ok(d)) => { + Some(d) + }, Some(Err(e)) => { let err = e.into(); debug!("decoder inner stream error: {:?}", err); @@ -116,7 +125,9 @@ where if let Some(data) = chunk { buf.put(data); } else { - if buf.has_remaining_mut() { + // FIXME: get BytesMut to impl `Buf` directlty? + let buf1 = (&buf[..]).into_buf(); + if buf1.has_remaining() { trace!("unexpected EOF decoding stream"); Err(Status::new( Code::Internal, diff --git a/tonic/src/codec/encode.rs b/tonic/src/codec/encode.rs index e26a327..edd5b2a 100644 --- a/tonic/src/codec/encode.rs +++ b/tonic/src/codec/encode.rs @@ -1,7 +1,12 @@ -use crate::{body::BytesBuf, Status}; +use crate::{body::BytesBuf, Code, Status}; use bytes::{BufMut, BytesMut, IntoBuf}; use futures_core::{Stream, TryStream}; -use futures_util::StreamExt; +use futures_util::{ready, StreamExt, TryStreamExt}; +use http::HeaderMap; +use http_body::Body; +use pin_project::pin_project; +use std::pin::Pin; +use std::task::{Context, Poll}; use tokio_codec::Encoder; pub fn encode(mut encoder: T, source: U) -> impl TryStream @@ -10,7 +15,7 @@ where U: Stream>, { async_stream::stream! { - let mut buf = BytesMut::with_capacity(1024); + let mut buf = BytesMut::with_capacity(1024 * 1024); futures_util::pin_mut!(source); loop { @@ -39,3 +44,88 @@ where } } } + +#[derive(Debug)] +enum Role { + Client, + Server, +} + +#[pin_project] +#[derive(Debug)] +pub struct EncodeBody { + #[pin] + inner: S, + error: Option, + role: Role, +} + +impl EncodeBody +where + S: Stream>, +{ + pub fn new_client(inner: S) -> Self { + Self { + inner, + error: None, + role: Role::Client, + } + } + + pub fn new_server(inner: S) -> Self { + Self { + inner, + error: None, + role: Role::Server, + } + } +} + +impl Body for EncodeBody +where + S: Stream>, +{ + type Data = BytesBuf; + type Error = Status; + + fn is_end_stream(&self) -> bool { + false + } + + fn poll_data( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll>> { + let mut self_proj = self.project(); + match ready!(self_proj.inner.try_poll_next_unpin(cx)) { + Some(Ok(d)) => Some(Ok(d)).into(), + Some(Err(status)) => match self_proj.role { + Role::Client => Some(Err(status)).into(), + Role::Server => { + *self_proj.error = Some(status); + None.into() + } + }, + None => None.into(), + } + } + + fn poll_trailers( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + ) -> Poll, Status>> { + match self.role { + Role::Client => Poll::Ready(Ok(None)), + Role::Server => { + let self_proj = self.project(); + let status = if let Some(status) = self_proj.error.take() { + status + } else { + Status::new(Code::Ok, "") + }; + + Poll::Ready(Ok(Some(status.to_header_map()?))) + } + } + } +} diff --git a/tonic/src/codec/mod.rs b/tonic/src/codec/mod.rs index 0337786..afd374a 100644 --- a/tonic/src/codec/mod.rs +++ b/tonic/src/codec/mod.rs @@ -3,7 +3,7 @@ mod encode; mod prost; pub use self::decode::{decode_empty, decode_request, decode_response, Streaming}; -pub use self::encode::encode; +pub use self::encode::{encode, EncodeBody}; pub use self::prost::ProstCodec; use crate::Status; diff --git a/tonic/src/lib.rs b/tonic/src/lib.rs index da4a27c..8a4abce 100644 --- a/tonic/src/lib.rs +++ b/tonic/src/lib.rs @@ -1,4 +1,3 @@ -#![feature(async_await)] #![recursion_limit = "512"] //! gRPC implementation diff --git a/tower-h2/Cargo.toml b/tower-h2/Cargo.toml index fe66545..a496803 100644 --- a/tower-h2/Cargo.toml +++ b/tower-h2/Cargo.toml @@ -12,9 +12,12 @@ tokio-io = "0.2.0-alpha.1" tokio-executor = "0.2.0-alpha.1" tower-service = { git = "http://github.com/tower-rs/tower", branch = "std-future" } tower-util = { git = "http://github.com/tower-rs/tower", branch = "std-future" } -h2 = { git = "https://github.com/LucioFranco/h2", branch = "lucio/tower-h2-hack" } +# h2 = { git = "https://github.com/LucioFranco/h2", branch = "lucio/tower-h2-hack" } +# h2 = { git = "https://github.com/hyperium/h2" } +h2 = { path = "../../h2" } http = "0.1" -http-body = { git = "https://github.com/hyperium/http-body" } +# http-body = { git = "https://github.com/hyperium/http-body" } +http-body = { path = "../../http-body" } log = "0.4" [dev-dependencies] diff --git a/tower-h2/examples/client.rs b/tower-h2/examples/client.rs index 8c1985c..9a5264c 100644 --- a/tower-h2/examples/client.rs +++ b/tower-h2/examples/client.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - use http::Request; use std::pin::Pin; use std::task::{Context, Poll}; diff --git a/tower-h2/examples/server.rs b/tower-h2/examples/server.rs index 3eaa445..9041020 100644 --- a/tower-h2/examples/server.rs +++ b/tower-h2/examples/server.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - use futures_util::future; use http::{Request, Response}; use std::pin::Pin; diff --git a/tower-h2/src/lib.rs b/tower-h2/src/lib.rs index cdff648..28f5bb5 100644 --- a/tower-h2/src/lib.rs +++ b/tower-h2/src/lib.rs @@ -1,5 +1,3 @@ -#![feature(async_await)] - #[macro_use] extern crate log; diff --git a/tower-h2/src/recv_body.rs b/tower-h2/src/recv_body.rs index 0b7e6c5..1f6067b 100644 --- a/tower-h2/src/recv_body.rs +++ b/tower-h2/src/recv_body.rs @@ -34,7 +34,7 @@ impl Body for RecvBody { type Data = Data; type Error = h2::Error; - fn is_end_stream(self: Pin<&mut Self>) -> bool { + fn is_end_stream(&self) -> bool { self.inner.is_end_stream() }