more clean up

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"));

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]

tonic/src/client/grpc.rs

@@ -59,16 +59,14 @@ impl<T> Grpc<T> {
         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<M1, M2, C>(
@@ -127,7 +125,6 @@ impl<T> Grpc<T> {
             .insert(TE, HeaderValue::from_static("trailers"));
 
         // Set the content type
-        // TODO: Don't hard code this here
         let content_type = <C as Codec>::CONTENT_TYPE;
         request
             .headers_mut()
@@ -139,7 +136,8 @@ impl<T> Grpc<T> {
             .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 {

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 {
+// enum Direction {
-    Request,
+//     Request,
-    Response(StatusCode),
+//     Response(StatusCode),
-    EmptyResponse,
+//     EmptyResponse,
-}
+// }
 
 pub fn decode<T, B>(
     mut decoder: T,
@@ -50,14 +50,13 @@ where
     B: Body + 'static,
     B::Error: Into<crate::Error>,
 {
-    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)? {
-            if let Some(item) = decode_chunk(&mut decoder, &mut buf, &mut state).unwrap() {
+                yield item;
-                yield Ok(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;
                 }

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<T, U>(mut encoder: T, mut source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
+pub fn encode<T, U>(mut encoder: T, source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
 where
     T: Encoder<Error = Status>,
-    U: Stream<Item = Result<T::Item, Status>> + Unpin,
+    U: Stream<Item = Result<T::Item, Status>>,
 {
     async_stream::stream! {
         let mut buf = BytesMut::with_capacity(1024);
+        futures_util::pin_mut!(source);
 
         loop {
             match source.next().await {

tonic/src/response.rs

@@ -40,7 +40,14 @@ impl<T> Response<T> {
         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<T>) -> Self {
         let (head, message) = res.into_parts();
         Response {

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 = encode(self.codec.encoder(), body.into_stream()).into_stream();
-                let body = Box::pin(body.into_stream());
-                let body = encode(self.codec.encoder(), body).into_stream();
 
+                // FIXME: try to return impl Trait?
                 let body = Box::pin(body) as BoxStream<BytesBuf>;
                 http::Response::from_parts(parts, body)
             }

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 {

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]

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<dyn std::error::Error>> {

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;
 

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)) {