From a5492a2706d1cb8d0bb5b67dc6b0bffbb417ec81 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Wed, 14 Aug 2019 14:12:12 -0400 Subject: [PATCH] Inital basic not working example --- tonic/Cargo.toml | 6 +++ tonic/examples/server.rs | 100 +++++++++++++++++++++++++++++++++++++++ tonic/src/body.rs | 18 +++++-- tonic/src/codec.rs | 91 +++++++++++++++++++++++++++++++---- tonic/src/server/mod.rs | 83 ++++++++++++++++++++++++-------- tonic/tests/server.rs | 55 +++++++++++++++++++++ 6 files changed, 320 insertions(+), 33 deletions(-) create mode 100644 tonic/examples/server.rs create mode 100644 tonic/tests/server.rs diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 1f69363..4300db4 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -20,3 +20,9 @@ tower-service = { git = "https://github.com/tower-rs/tower", branch = "std-futur tokio-codec = "=0.2.0-alpha.1" async-stream = { path = "../../async-stream/async-stream" } http-body = { git = "https://github.com/hyperium/http-body", branch = "std-future" } + +[dev-dependencies] +tokio = "=0.2.0-alpha.1" +tokio-buf = "=0.2.0-alpha.1" +prost-derive = "0.5" +tower-h2 = { path = "../tower-h2" } diff --git a/tonic/examples/server.rs b/tonic/examples/server.rs new file mode 100644 index 0000000..6b28ced --- /dev/null +++ b/tonic/examples/server.rs @@ -0,0 +1,100 @@ +#![feature(async_await, type_alias_impl_trait)] + +use std::future::Future; +use tonic::{server::{UnaryService, Grpc}, Status, Request, Response, body}; +use tower_service::Service; +use tower_h2::{RecvBody, Server}; +use tokio::net::TcpListener; +use futures_util::future; +use std::task::{Context, Poll}; + +#[derive(Clone, PartialEq, prost::Message)] +pub struct HelloRequest { + #[prost(string, tag = "1")] + pub name: std::string::String, +} +/// The response message containing the greetings +#[derive(Clone, PartialEq, prost::Message)] +pub struct HelloReply { + #[prost(string, tag = "1")] + pub message: std::string::String, +} + +struct SayHello; + +impl UnaryService for SayHello { + type Response = HelloReply; + type Future = impl Future, Status>>; + + fn call(&mut self, request: Request) -> Self::Future { + async move { + println!("REQUEST = {:?}", request); + + let reply = HelloReply { + message: "Zomg, it works!".to_string(), + }; + + Ok(Response::new(reply)) + } + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let addr = "[::1]:50051".parse().unwrap(); + let mut bind = TcpListener::bind(&addr)?; + + let mut server = Server::new(MakeSvc, 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(()) +} + +#[derive(Debug)] +pub struct Svc; + +impl Service> for Svc { + type Response = http::Response; + type Error = tonic::error::Never; + type Future = impl Future>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn call(&mut self, req: http::Request) -> Self::Future { + let fut = async move { + let codec = tonic::codec::ProstCodec::new(); + let mut grpc = Grpc::new(codec); + let response = grpc.unary(SayHello, req).await; + Ok(response) + }; + + Box::pin(fut) + } +} + +pub struct MakeSvc; + +impl Service<()> for MakeSvc { + type Response = Svc; + type Error = std::io::Error; + type Future = future::Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn call(&mut self, _: ()) -> Self::Future { + future::ok(Svc) + } +} diff --git a/tonic/src/body.rs b/tonic/src/body.rs index 2a2ef06..25ba770 100644 --- a/tonic/src/body.rs +++ b/tonic/src/body.rs @@ -1,6 +1,6 @@ use crate::{Code, Status}; use bytes::{Bytes, IntoBuf}; -use futures_core::Stream; +use futures_core::{Stream, TryStream}; use futures_util::{ready, TryStreamExt}; use http::HeaderMap; use http_body::Body; @@ -51,12 +51,22 @@ pub struct BoxAsyncBody { } impl BoxAsyncBody { - pub fn new(inner: S) -> Self + // pub fn new(inner: S) -> Self + // where + // S: Stream> + Send + 'static, + // { + // Self { + // inner: Box::pin(inner), + // error: None, + // } + // } + + pub fn new_try(inner: S) -> Self where - S: Stream> + Send + 'static, + S: TryStream + Send + 'static, { Self { - inner: Box::pin(inner), + inner: Box::pin(inner.into_stream()), error: None, } } diff --git a/tonic/src/codec.rs b/tonic/src/codec.rs index 5087bf3..80c9605 100644 --- a/tonic/src/codec.rs +++ b/tonic/src/codec.rs @@ -1,11 +1,13 @@ #![allow(dead_code)] -use crate::{body::BytesBuf, Status}; +use crate::{body::BytesBuf, Code, Status}; use async_stream::stream; use bytes::{Buf, BufMut, BytesMut, IntoBuf}; use futures_core::TryStream; use futures_util::{future, TryStreamExt}; use http_body::Body; +use prost::Message; +use std::marker::PhantomData; use tokio_codec::{Decoder, Encoder}; use tracing::{debug, trace}; @@ -16,14 +18,13 @@ pub trait Codec { type Encoder: Encoder; type Decoder: Decoder; + const CONTENT_TYPE: &'static str; + fn encoder(&mut self) -> Self::Encoder; fn decoder(&mut self) -> Self::Decoder; } -pub async fn encode( - mut encoder: T, - mut source: U, -) -> impl TryStream +pub fn encode(mut encoder: T, mut source: U) -> impl TryStream where T: Encoder, U: TryStream + Unpin, @@ -79,10 +80,10 @@ where } else { if buf.has_remaining_mut() { trace!("unexpected EOF decoding stream"); - // yield Err(Status::new( - // Code::Internal, - // "Unexpected EOF decoding stream.".to_string(), - // )); + yield Err(Status::new( + Code::Internal, + "Unexpected EOF decoding stream.".to_string(), + )); } else { break; } @@ -153,6 +154,76 @@ where Ok(None) } +#[derive(Debug, Clone)] +pub struct ProstCodec { + _pd: PhantomData<(T, U)>, +} + +impl ProstCodec { + pub fn new() -> Self { + Self { _pd: PhantomData } + } +} + +impl Codec for ProstCodec +where + T: Message, + U: Message + Default, +{ + type Encode = T; + type Decode = U; + + type Encoder = ProstEncoder; + type Decoder = ProstDecoder; + + const CONTENT_TYPE: &'static str = "application/groc+proto"; + + fn encoder(&mut self) -> Self::Encoder { + ProstEncoder(PhantomData) + } + + fn decoder(&mut self) -> Self::Decoder { + ProstDecoder(PhantomData) + } +} + +pub struct ProstEncoder(PhantomData); + +impl Encoder for ProstEncoder { + type Item = T; + type Error = Status; + + fn encode(&mut self, item: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> { + let len = item.encoded_len(); + + if buf.remaining_mut() < len { + buf.reserve(len); + } + + item.encode(buf) + .map_err(|_| unreachable!("Message only errors if not enough space")) + } +} + +pub struct ProstDecoder(PhantomData); + +impl Decoder for ProstDecoder { + type Item = U; + type Error = Status; + + fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { + Message::decode(buf.take()) + .map(Option::Some) + .map_err(from_decode_error) + } +} + +fn from_decode_error(error: prost::DecodeError) -> crate::Status { + // Map Protobuf parse errors to an INTERNAL status code, as per + // https://github.com/grpc/grpc/blob/master/doc/statuscodes.md + crate::Status::new(crate::Code::Internal, error.to_string()) +} + #[derive(Default)] pub struct UnitCodec; @@ -163,6 +234,8 @@ impl Codec for UnitCodec { type Encoder = UnitEncoder; type Decoder = UnitDecoder; + const CONTENT_TYPE: &'static str = "()"; + fn encoder(&mut self) -> Self::Encoder { UnitEncoder } diff --git a/tonic/src/server/mod.rs b/tonic/src/server/mod.rs index dc36d1a..e15d455 100644 --- a/tonic/src/server/mod.rs +++ b/tonic/src/server/mod.rs @@ -1,12 +1,15 @@ use crate::{ + body::{BoxAsyncBody, BytesBuf}, codec::{self, Codec}, - Request, Status, + Request, Response, Status, }; -use futures_core::{Future, TryStream}; +use futures_core::{Future, Stream, TryStream}; use futures_util::{future, stream, TryStreamExt}; use http_body::Body; use std::pin::Pin; +type BoxStream = Pin> + Send + 'static>>; + pub struct Grpc { codec: T, } @@ -64,13 +67,12 @@ pub trait StreamingService { fn call(&mut self, request: Request) -> Self::Future; } -type BoxStream = Pin + Send + 'static>>; - impl Grpc where T: Codec, T::Decode: Unpin + 'static, - T::Encode: Unpin + 'static, + T::Encoder: Send + 'static, + T::Encode: Send + Unpin + 'static, { pub fn new(codec: T) -> Self { Self { codec } @@ -80,7 +82,7 @@ where &mut self, mut service: S, req: http::Request, - ) -> http::Response> + ) -> http::Response where S: UnaryService, B: Body, @@ -91,21 +93,22 @@ where futures_util::pin_mut!(stream); let message = stream.try_next().await.unwrap().unwrap(); let request = Request::new(message); - let response = service.call(request).await.unwrap(); - let message = response.into_inner(); - let source = stream::once(future::ok(message)); - let body = codec::encode(self.codec.encoder(), source).await; + let response = service + .call(request) + .await + .map(|r| r.map(|m| stream::once(future::ok(m)))); - http::Response::new(body) + self.map_response(response).map(BoxAsyncBody::new_try) } pub async fn server_streaming( &mut self, mut service: S, req: http::Request, - ) -> http::Response> + ) -> http::Response where S: ServerStreamingService, + S::ResponseStream: Send + 'static, B: Body, B::Error: Into, { @@ -114,11 +117,9 @@ where futures_util::pin_mut!(stream); let message = stream.try_next().await.unwrap().unwrap(); let request = Request::new(message); - let response = service.call(request).await.unwrap(); - let source = response.into_inner(); - let body = codec::encode(self.codec.encoder(), source).await; + let response = service.call(request).await; - http::Response::new(body) + self.map_response(response).map(BoxAsyncBody::new_try) } pub async fn client_streaming( @@ -135,13 +136,13 @@ where B::Error: Into + Send, { let (_parts, body) = req.into_parts(); - let stream = codec::decode(self.codec.decoder(), body); + let stream = codec::decode(self.codec.decoder(), body).into_stream(); let stream = Box::pin(stream) as BoxStream; let request = Request::new(stream); let response = service.call(request).await.unwrap(); let message = response.into_inner(); let source = stream::once(future::ok(message)); - let body = codec::encode(self.codec.encoder(), source).await; + let body = codec::encode(self.codec.encoder(), source); http::Response::new(body) } @@ -160,13 +161,55 @@ where B::Error: Into + Send, { let (_parts, body) = req.into_parts(); - let stream = codec::decode(self.codec.decoder(), body); + let stream = codec::decode(self.codec.decoder(), body).into_stream(); let stream = Box::pin(stream) as BoxStream; let request = Request::new(stream); let response = service.call(request).await.unwrap(); let source = response.into_inner(); - let body = codec::encode(self.codec.encoder(), source).await; + let body = codec::encode(self.codec.encoder(), source); http::Response::new(body) } + + // fn map_request(&mut self, request: http::Request) -> Request { + // Request::from_http(request.map(|b| codec::decode(self.codec.decoder(), b))) + // } + + fn map_response( + &mut self, + response: Result, Status>, + ) -> http::Response> + where + B: TryStream + Send + Unpin + 'static, + { + match response { + Ok(r) => { + let (mut parts, body) = r.into_http().into_parts(); + + // Set the content type + parts.headers.insert( + http::header::CONTENT_TYPE, + http::header::HeaderValue::from_static(T::CONTENT_TYPE), + ); + + let body = codec::encode(self.codec.encoder(), body).into_stream(); + + let body = Box::pin(body) as BoxStream; + http::Response::from_parts(parts, body) + } + Err(status) => { + let status = stream::once(future::err(status)); + let body = codec::encode(self.codec.encoder(), status).into_stream(); + let (mut parts, _body) = Response::new(()).into_http().into_parts(); + + parts.headers.insert( + http::header::CONTENT_TYPE, + http::header::HeaderValue::from_static(T::CONTENT_TYPE), + ); + + let body = Box::pin(body) as BoxStream; + http::Response::from_parts(parts, body) + } + } + } } diff --git a/tonic/tests/server.rs b/tonic/tests/server.rs new file mode 100644 index 0000000..ddd6750 --- /dev/null +++ b/tonic/tests/server.rs @@ -0,0 +1,55 @@ +#![feature(async_await, type_alias_impl_trait)] + +use std::future::Future; +use std::task::{Context, Poll}; +use tokio_buf::BufStream; +use tonic::codec::UnitCodec; +use tonic::server::*; +use tonic::{Request, Response, Status}; + +struct SayHello; + +impl UnaryService<()> for SayHello { + type Response = (); + type Future = impl Future, Status>>; + + fn call(&mut self, _request: Request<()>) -> Self::Future { + async move { Ok(Response::new(())) } + } +} + +#[tokio::test] +async fn say_hello() { + let codec = UnitCodec::default(); + let mut grpc = Grpc::new(codec); + + let request = http::Request::new(Body(Vec::new())); + grpc.unary(SayHello, request).await; +} + +#[derive(Debug, Default, Clone)] +struct Body(Vec); + +impl From> for Body { + fn from(t: Vec) -> Self { + Body(t) + } +} + +impl BufStream for Body { + type Item = std::io::Cursor>; + type Error = std::io::Error; + + fn poll_buf(&mut self, _cx: &mut Context<'_>) -> Poll>> { + if self.0.is_empty() { + return None.into(); + } + + use std::{io, mem}; + + let bytes = mem::replace(&mut self.0, Default::default()); + let buf = io::Cursor::new(bytes); + + Some(Ok(buf)).into() + } +}