talksik/tonic
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Clean up main crate and more work on macro

Browse Source 
Signed-off-by: Lucio Franco <[email protected]>
This commit is contained in:
Lucio Franco
2019-08-15 15:09:54 -04:00
parent 55941c0dd1
commit 0a8aaa8071
13 changed files with 438 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tonic/src/codec.rs
+8 -13
View File
@@ -1,10 +1,8 @@
#![allow(dead_code)]
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 futures_core::{Stream, TryStream};
use futures_util::{future, StreamExt};
use http_body::Body;
use prost::Message;
use std::marker::PhantomData;
@@ -27,16 +25,14 @@ pub trait Codec {
pub fn encode<T, U>(mut encoder: T, mut source: U) -> impl TryStream<Ok = BytesBuf, Error = Status>
where
T: Encoder<Error = Status>,
U: TryStream<Ok = T::Item, Error = Status> + Unpin,
U: Stream<Item = Result<T::Item, Status>> + Unpin,
{
stream! {
let mut buf = BytesMut::with_capacity(1024);
loop {
match source.try_next().await {
Ok(Some(item)) => {
match source.next().await {
Some(Ok(item)) => {
buf.reserve(5);
unsafe {
buf.advance_mut(5);
@@ -54,8 +50,8 @@ where
yield Ok(buf.split_to(len + 5).freeze().into_buf());
},
Ok(None) => break,
Err(status) => yield Err(status),
Some(Err(status)) => yield Err(status),
None => break,
}
}
}
@@ -241,7 +237,7 @@ impl<U: Message + Default> Decoder for ProstDecoder<U> {
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())
Status::new(Code::Internal, error.to_string())
}
#[derive(Default)]
@@ -291,5 +287,4 @@ impl Decoder for UnitDecoder {
enum State {
ReadHeader,
ReadBody { compression: bool, len: usize },
Done,
}
tonic/src/request.rs
+8 -4
View File
@@ -40,15 +40,19 @@ impl<T> Request<T> {
self.message
}
/// Convert an HTTP request to a gRPC request
pub fn from_http(http: http::Request<T>) -> Self {
let (head, message) = http.into_parts();
pub(crate) fn from_http_parts(parts: http::request::Parts, message: T) -> Self {
Request {
metadata: MetadataMap::from_headers(head.headers),
metadata: MetadataMap::from_headers(parts.headers),
message,
}
}
/// Convert an HTTP request to a gRPC request
pub fn from_http(http: http::Request<T>) -> Self {
let (parts, message) = http.into_parts();
Request::from_http_parts(parts, message)
}
pub fn into_http(self, uri: http::Uri) -> http::Request<T> {
let mut request = http::Request::new(self.message);
tonic/src/response.rs
-1
View File
@@ -16,7 +16,6 @@ impl<T> Response<T> {
}
}
/// Get a reference to the message
pub fn get_ref(&self) -> &T {
&self.message
}
tonic/src/server/grpc.rs
+201
View File
@@ -0,0 +1,201 @@
use crate::{
body::{BoxAsyncBody, BytesBuf},
codec::{decode, encode, Codec},
server::{ClientStreamingService, ServerStreamingService, StreamingService, UnaryService},
Code, Request, Response, Status,
};
use futures_core::{Stream, TryStream};
use futures_util::{future, stream, StreamExt, TryStreamExt};
use http_body::Body;
use std::pin::Pin;
type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>;
pub struct Grpc<T> {
codec: T,
}
impl<T> Grpc<T>
where
T: Codec,
T::Decoder: Send + 'static,
T::Decode: Send + Unpin + 'static,
T::Encoder: Send + 'static,
T::Encode: Send + Unpin + 'static,
{
pub fn new(codec: T) -> Self {
Self { codec }
}
pub async fn unary<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<BoxAsyncBody>
where
S: UnaryService<T::Decode, Response = T::Encode>,
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let request = match self.map_request_unary(req).await {
Ok(r) => r,
Err(status) => {
return self
.map_response::<stream::Once<future::Ready<Result<T::Encode, Status>>>>(Err(
status,
))
.map(BoxAsyncBody::new_try);
}
};
let response = service
.call(request)
.await
.map(|r| r.map(|m| stream::once(future::ok(m))));
self.map_response(response).map(BoxAsyncBody::new_try)
}
pub async fn server_streaming<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<BoxAsyncBody>
where
S: ServerStreamingService<T::Decode, Response = T::Encode>,
S::ResponseStream: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let request = match self.map_request_unary(req).await {
Ok(r) => r,
Err(status) => {
return self
.map_response::<S::ResponseStream>(Err(status))
.map(BoxAsyncBody::new_try);
}
};
let response = service.call(request).await;
self.map_response(response).map(BoxAsyncBody::new_try)
}
pub async fn client_streaming<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<BoxAsyncBody>
where
S: ClientStreamingService<BoxStream<T::Decode>, Response = T::Encode>,
T::Decode: Send,
T::Decoder: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let request = self.map_request_streaming(req);
let response = service
.call(request)
.await
.map(|r| r.map(|m| stream::once(future::ok(m))));
self.map_response(response).map(BoxAsyncBody::new_try)
}
pub async fn streaming<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<BoxAsyncBody>
where
S: StreamingService<BoxStream<T::Decode>, Response = T::Encode> + Send,
S::ResponseStream: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let request = self.map_request_streaming(req);
let response = service.call(request).await;
self.map_response(response).map(BoxAsyncBody::new_try)
}
async fn map_request_unary<B>(
&mut self,
request: http::Request<B>,
) -> Result<Request<T::Decode>, Status>
where
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let (parts, body) = request.into_parts();
let stream = decode(self.codec.decoder(), body).into_stream();
futures_util::pin_mut!(stream);
let message = stream
.try_next()
.await?
.ok_or(Status::new(Code::Internal, "Missing request message."))?;
Ok(Request::from_http_parts(parts, message))
}
fn map_request_streaming<B>(
&mut self,
request: http::Request<B>,
) -> Request<BoxStream<T::Decode>>
where
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
Request::from_http(
request.map(|b| {
decode(self.codec.decoder(), b).into_stream().boxed() as BoxStream<T::Decode>
}),
)
}
fn map_response<B>(
&mut self,
response: Result<crate::Response<B>, Status>,
) -> http::Response<BoxStream<BytesBuf>>
where
B: TryStream<Ok = T::Encode, Error = Status> + Send + '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),
);
// 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 = Box::pin(body) as BoxStream<BytesBuf>;
http::Response::from_parts(parts, body)
}
Err(status) => {
let status = stream::once(future::err(status));
let body = 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<BytesBuf>;
http::Response::from_parts(parts, body)
}
}
}
}
tonic/src/server/mod.rs
+6 -214
View File
@@ -1,215 +1,7 @@
use crate::{
body::{BoxAsyncBody, BytesBuf},
codec::{self, Codec},
Request, Response, Status,
mod grpc;
mod service;
pub use self::grpc::Grpc;
pub use self::service::{
ClientStreamingService, ServerStreamingService, StreamingService, UnaryService,
};
use futures_core::{Future, Stream, TryStream};
use futures_util::{future, stream, TryStreamExt};
use http_body::Body;
use std::pin::Pin;
type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>;
pub struct Grpc<T> {
codec: T,
}
// type UnaryFuture<B> = Once<Ready<Result<B, Status>>>;
// type ResponseBody = impl Stream<Item = Result<crate::body::BytesBuf, Status>>;
pub trait UnaryService<R> {
/// Protobuf response message type
type Response;
/// Response future
type Future: Future<Output = Result<crate::Response<Self::Response>, Status>>;
/// Call the service
fn call(&mut self, request: Request<R>) -> Self::Future;
}
pub trait ServerStreamingService<R> {
/// Protobuf response message type
type Response;
/// Stream of outbound response messages
type ResponseStream: TryStream<Ok = Self::Response, Error = crate::Status> + Unpin;
/// Response future
type Future: Future<Output = Result<crate::Response<Self::ResponseStream>, crate::Status>>;
/// Call the service
fn call(&mut self, request: Request<R>) -> Self::Future;
}
pub trait ClientStreamingService<RequestStream> {
/// Protobuf response message type
type Response;
/// Response future
type Future: Future<Output = Result<crate::Response<Self::Response>, Status>>;
/// Call the service
fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
}
pub trait StreamingService<RequestStream> {
/// Protobuf response message type
type Response;
/// Stream of outbound response messages
type ResponseStream: TryStream<Ok = Self::Response, Error = crate::Status> + Unpin;
/// Response future
type Future: Future<Output = Result<crate::Response<Self::ResponseStream>, crate::Status>>;
/// Call the service
fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
}
impl<T> Grpc<T>
where
T: Codec,
T::Decode: Unpin + 'static,
T::Encoder: Send + 'static,
T::Encode: Send + Unpin + 'static,
{
pub fn new(codec: T) -> Self {
Self { codec }
}
pub async fn unary<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<BoxAsyncBody>
where
S: UnaryService<T::Decode, Response = T::Encode>,
B: Body,
B::Error: Into<crate::Error>,
{
let (_parts, body) = req.into_parts();
let stream = codec::decode(self.codec.decoder(), body).into_stream();
futures_util::pin_mut!(stream);
let message = stream.try_next().await.unwrap().unwrap();
let request = Request::new(message);
let response = service
.call(request)
.await
.map(|r| r.map(|m| stream::once(future::ok(m))));
self.map_response(response).map(BoxAsyncBody::new_try)
}
pub async fn server_streaming<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<BoxAsyncBody>
where
S: ServerStreamingService<T::Decode, Response = T::Encode>,
S::ResponseStream: Send + 'static,
B: Body,
B::Error: Into<crate::Error>,
{
let (_parts, body) = req.into_parts();
let stream = codec::decode(self.codec.decoder(), body).into_stream();
futures_util::pin_mut!(stream);
let message = stream.try_next().await.unwrap().unwrap();
let request = Request::new(message);
let response = service.call(request).await;
self.map_response(response).map(BoxAsyncBody::new_try)
}
pub async fn client_streaming<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<impl TryStream<Ok = crate::body::BytesBuf, Error = Status>>
where
S: ClientStreamingService<BoxStream<T::Decode>, Response = T::Encode>,
T::Decode: Send,
T::Decoder: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let (_parts, body) = req.into_parts();
let stream = codec::decode(self.codec.decoder(), body).into_stream();
let stream = Box::pin(stream) as BoxStream<T::Decode>;
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);
http::Response::new(body)
}
pub async fn streaming<S, B>(
&mut self,
mut service: S,
req: http::Request<B>,
) -> http::Response<impl TryStream<Ok = crate::body::BytesBuf, Error = Status>>
where
S: StreamingService<BoxStream<T::Decode>, Response = T::Encode>,
T::Decode: Send,
T::Decoder: Send + 'static,
B: Body + Send + 'static,
B::Data: Send,
B::Error: Into<crate::Error> + Send,
{
let (_parts, body) = req.into_parts();
let stream = codec::decode(self.codec.decoder(), body).into_stream();
let stream = Box::pin(stream) as BoxStream<T::Decode>;
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);
http::Response::new(body)
}
// fn map_request<B>(&mut self, request: http::Request<B>) -> Request<B> {
// Request::from_http(request.map(|b| codec::decode(self.codec.decoder(), b)))
// }
fn map_response<B>(
&mut self,
response: Result<crate::Response<B>, Status>,
) -> http::Response<BoxStream<BytesBuf>>
where
B: TryStream<Ok = T::Encode, Error = Status> + 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<BytesBuf>;
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<BytesBuf>;
http::Response::from_parts(parts, body)
}
}
}
}
tonic/src/server/service.rs
+53
View File
@@ -0,0 +1,53 @@
use crate::{Request, Response, Status};
use futures_core::Stream;
use std::future::Future;
pub trait UnaryService<R> {
/// Protobuf response message type
type Response;
/// Response future
type Future: Future<Output = Result<Response<Self::Response>, Status>>;
/// Call the service
fn call(&mut self, request: Request<R>) -> Self::Future;
}
pub trait ServerStreamingService<R> {
/// Protobuf response message type
type Response;
/// Stream of outbound response messages
type ResponseStream: Stream<Item = Result<Self::Response, Status>> + Unpin;
/// Response future
type Future: Future<Output = Result<Response<Self::ResponseStream>, Status>>;
/// Call the service
fn call(&mut self, request: Request<R>) -> Self::Future;
}
pub trait ClientStreamingService<RequestStream> {
/// Protobuf response message type
type Response;
/// Response future
type Future: Future<Output = Result<Response<Self::Response>, Status>>;
/// Call the service
fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
}
pub trait StreamingService<RequestStream> {
/// Protobuf response message type
type Response;
/// Stream of outbound response messages
type ResponseStream: Stream<Item = Result<Self::Response, Status>> + Unpin;
/// Response future
type Future: Future<Output = Result<Response<Self::ResponseStream>, Status>>;
/// Call the service
fn call(&mut self, request: Request<RequestStream>) -> Self::Future;
}