From ef87a558e7c719113d00bba2d6f8fb99d892698f Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Sun, 18 Aug 2019 01:30:26 -0400 Subject: [PATCH] more clean up --- tonic-examples/src/routeguide/client.rs | 2 +- tonic/Cargo.toml | 2 +- tonic/src/client/grpc.rs | 10 ++++------ tonic/src/codec/decode.rs | 25 ++++++++++++------------- tonic/src/codec/encode.rs | 5 +++-- tonic/src/response.rs | 9 ++++++++- tonic/src/server/grpc.rs | 7 +++---- tonic/src/status.rs | 2 +- tower-h2/Cargo.toml | 2 +- tower-h2/examples/client.rs | 2 +- tower-h2/examples/server.rs | 2 +- tower-h2/src/flush.rs | 1 - 12 files changed, 36 insertions(+), 33 deletions(-) diff --git a/tonic-examples/src/routeguide/client.rs b/tonic-examples/src/routeguide/client.rs index 25c2800..4f11702 100644 --- a/tonic-examples/src/routeguide/client.rs +++ b/tonic-examples/src/routeguide/client.rs @@ -1,11 +1,11 @@ #![feature(async_await)] +use futures::TryStreamExt; use route_guide::{Point, RouteNote}; use std::time::{Duration, Instant}; use tokio::{net::TcpStream, timer::Interval}; use tonic::Request; use tower_h2::{add_origin::AddOrigin, Connection}; -use futures::TryStreamExt; mod route_guide { include!(concat!(env!("OUT_DIR"), "/routeguide.rs")); diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index d93cb7e..51efb55 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -19,7 +19,7 @@ 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", branch = "lucio/pin" } +http-body = { git = "https://github.com/hyperium/http-body" } pin-project = "0.4.0-alpha.2" [dev-dependencies] diff --git a/tonic/src/client/grpc.rs b/tonic/src/client/grpc.rs index f5aa01b..f410d2f 100644 --- a/tonic/src/client/grpc.rs +++ b/tonic/src/client/grpc.rs @@ -59,16 +59,14 @@ impl Grpc { M1: Send, M2: Send + Unpin + 'static, { - let response = self.streaming(request, path, codec).await?; + let (parts, mut body) = self.streaming(request, path, codec).await?.into_parts(); - // TODO: use response to parts - let mut body = response.into_inner(); let message = body .try_next() .await? .ok_or(Status::new(Code::Internal, "Missing response message."))?; - Ok(Response::new(message)) + Ok(Response::from_parts(parts, message)) } pub async fn server_streaming( @@ -127,7 +125,6 @@ impl Grpc { .insert(TE, HeaderValue::from_static("trailers")); // Set the content type - // TODO: Don't hard code this here let content_type = ::CONTENT_TYPE; request .headers_mut() @@ -139,7 +136,8 @@ impl Grpc { .await .map_err(|err| Status::from_error(&*(err.into())))?; - let status_code = response.status(); + // TODO: implement decode with status + let _status_code = response.status(); let trailers_only_status = Status::from_header_map(response.headers()); if let Some(status) = trailers_only_status { diff --git a/tonic/src/codec/decode.rs b/tonic/src/codec/decode.rs index f4e8420..2cb2155 100644 --- a/tonic/src/codec/decode.rs +++ b/tonic/src/codec/decode.rs @@ -2,7 +2,7 @@ use crate::{Code, Status}; use bytes::{Buf, BufMut, BytesMut, IntoBuf}; use futures_core::{Stream, TryStream}; use futures_util::future; -use http::StatusCode; +// use http::StatusCode; use http_body::Body; use std::pin::Pin; use tokio_codec::Decoder; @@ -34,11 +34,11 @@ enum State { ReadBody { compression: bool, len: usize }, } -enum Direction { - Request, - Response(StatusCode), - EmptyResponse, -} +// enum Direction { +// Request, +// Response(StatusCode), +// EmptyResponse, +// } pub fn decode( mut decoder: T, @@ -50,14 +50,13 @@ where B: Body + 'static, B::Error: Into, { - async_stream::stream! { + async_stream::try_stream! { let mut buf = BytesMut::with_capacity(1024 * 1024); let mut state = State::ReadHeader; loop { - // TODO: use try_stream! and ? - if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state).unwrap() { - yield Ok(item); + if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state)? { + yield item; } // FIXME: Figure out how to verify that this is safe @@ -67,7 +66,7 @@ where let err = e.into(); debug!("decoder inner stream error: {:?}", err); let status = Status::from_error(&*err); - yield Err(status); + Err(status)?; break; }, None => None, @@ -78,10 +77,10 @@ where } else { if buf.has_remaining_mut() { trace!("unexpected EOF decoding stream"); - yield Err(Status::new( + Err(Status::new( Code::Internal, "Unexpected EOF decoding stream.".to_string(), - )); + ))?; } else { break; } diff --git a/tonic/src/codec/encode.rs b/tonic/src/codec/encode.rs index ff271d9..e26a327 100644 --- a/tonic/src/codec/encode.rs +++ b/tonic/src/codec/encode.rs @@ -4,13 +4,14 @@ use futures_core::{Stream, TryStream}; use futures_util::StreamExt; use tokio_codec::Encoder; -pub fn encode(mut encoder: T, mut source: U) -> impl TryStream +pub fn encode(mut encoder: T, source: U) -> impl TryStream where T: Encoder, - U: Stream> + Unpin, + U: Stream>, { async_stream::stream! { let mut buf = BytesMut::with_capacity(1024); + futures_util::pin_mut!(source); loop { match source.next().await { diff --git a/tonic/src/response.rs b/tonic/src/response.rs index 8905804..63e06a7 100644 --- a/tonic/src/response.rs +++ b/tonic/src/response.rs @@ -40,7 +40,14 @@ impl Response { self.message } - #[allow(dead_code)] + pub(crate) fn into_parts(self) -> (MetadataMap, T) { + (self.metadata, self.message) + } + + pub(crate) fn from_parts(metadata: MetadataMap, message: T) -> Self { + Self { metadata, message } + } + pub(crate) fn from_http(res: http::Response) -> Self { let (head, message) = res.into_parts(); Response { diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index 9f4e448..710575d 100644 --- a/tonic/src/server/grpc.rs +++ b/tonic/src/server/grpc.rs @@ -1,5 +1,5 @@ use crate::{ - body::{BytesBuf, BoxBody}, + body::{BoxBody, BytesBuf}, codec::{decode, encode, Codec, Streaming}, server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService}, Code, Request, Response, Status, @@ -175,10 +175,9 @@ where http::header::HeaderValue::from_static(T::CONTENT_TYPE), ); - // TODO: find way to pin this to the stack instead - let body = Box::pin(body.into_stream()); - let body = encode(self.codec.encoder(), body).into_stream(); + let body = encode(self.codec.encoder(), body.into_stream()).into_stream(); + // FIXME: try to return impl Trait? let body = Box::pin(body) as BoxStream; http::Response::from_parts(parts, body) } diff --git a/tonic/src/status.rs b/tonic/src/status.rs index 55bfca3..49ec386 100644 --- a/tonic/src/status.rs +++ b/tonic/src/status.rs @@ -75,7 +75,7 @@ impl Status { Status::new(code, message) } - // TODO: This should probably be made public eventually. Need to decide on + // FIXME: This should probably be made public eventually. Need to decide on // the exact argument type. #[cfg_attr(not(feature = "h2"), allow(dead_code))] pub(crate) fn from_error(err: &(dyn Error + 'static)) -> Status { diff --git a/tower-h2/Cargo.toml b/tower-h2/Cargo.toml index 026cff9..fe66545 100644 --- a/tower-h2/Cargo.toml +++ b/tower-h2/Cargo.toml @@ -14,7 +14,7 @@ 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" } http = "0.1" -http-body = { git = "https://github.com/hyperium/http-body", branch = "lucio/pin" } +http-body = { git = "https://github.com/hyperium/http-body" } log = "0.4" [dev-dependencies] diff --git a/tower-h2/examples/client.rs b/tower-h2/examples/client.rs index 529fa54..8c1985c 100644 --- a/tower-h2/examples/client.rs +++ b/tower-h2/examples/client.rs @@ -1,10 +1,10 @@ #![feature(async_await)] use http::Request; +use std::pin::Pin; use std::task::{Context, Poll}; use tokio::net::TcpStream; use tower_h2::Connection; -use std::pin::Pin; #[tokio::main] async fn main() -> Result<(), Box> { diff --git a/tower-h2/examples/server.rs b/tower-h2/examples/server.rs index 0d3acf1..3eaa445 100644 --- a/tower-h2/examples/server.rs +++ b/tower-h2/examples/server.rs @@ -2,9 +2,9 @@ use futures_util::future; use http::{Request, Response}; +use std::pin::Pin; use std::task::{Context, Poll}; use tokio::net::TcpListener; -use std::pin::Pin; use tower_h2::{RecvBody, Server}; use tower_service::Service; diff --git a/tower-h2/src/flush.rs b/tower-h2/src/flush.rs index b22f06c..55c815c 100644 --- a/tower-h2/src/flush.rs +++ b/tower-h2/src/flush.rs @@ -90,7 +90,6 @@ where self.h2.reserve_capacity(1); if self.h2.capacity() == 0 { - // TODO: The loop should not be needed once // carllerche/h2#270 is fixed. loop { match ready!(self.h2.poll_capacity(cx)) {