This commit is contained in:
Lucio Franco
2019-09-04 18:17:39 -04:00
parent bbc261ae0c
commit 3c100c3a5e
2 changed files with 14 additions and 10 deletions
+5 -4
View File
@@ -63,7 +63,8 @@ impl<T> Grpc<T> {
<T::ResponseBody as HttpBody>::Data: Into<Bytes>,
C: Codec<Encode = M1, Decode = M2>,
M1: Send + 'static,
M2: Send + 'static {
M2: Send + 'static,
{
let request = request.map(|m| stream::once(future::ok(m)));
self.client_streaming(request, path, codec).await
}
@@ -83,7 +84,7 @@ impl<T> Grpc<T> {
S: Stream<Item = Result<M1, Status>> + Send + 'static,
C: Codec<Encode = M1, Decode = M2>,
M1: Send + 'static,
M2: Send + 'static
M2: Send + 'static,
{
let (mut parts, body) = self.streaming(request, path, codec).await?.into_parts();
@@ -115,7 +116,7 @@ impl<T> Grpc<T> {
<T::ResponseBody as HttpBody>::Data: Into<Bytes>,
C: Codec<Encode = M1, Decode = M2>,
M1: Send + 'static,
M2: Send + 'static
M2: Send + 'static,
{
let request = request.map(|m| stream::once(future::ok(m)));
self.streaming(request, path, codec).await
@@ -136,7 +137,7 @@ impl<T> Grpc<T> {
S: Stream<Item = Result<M1, Status>> + Send + 'static,
C: Codec<Encode = M1, Decode = M2>,
M1: Send + 'static,
M2: Send + 'static
M2: Send + 'static,
{
let mut parts = Parts::default();
parts.path_and_query = Some(path);
+9 -6
View File
@@ -1,11 +1,15 @@
use super::Decoder;
use crate::{Code, Status, BoxBody, metadata::MetadataMap};
use crate::{metadata::MetadataMap, BoxBody, Code, Status};
use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf};
use futures_core::Stream;
use futures_util::{future, ready};
use http::StatusCode;
use http_body::Body;
use std::{fmt, pin::Pin, task::{Context, Poll}};
use std::{
fmt,
pin::Pin,
task::{Context, Poll},
};
use tracing::{debug, trace};
/// Streaming requests and responses.
@@ -110,10 +114,9 @@ impl<T> Streaming<T> {
// Trailers were not caught during poll_next and thus lets poll for
// them manually.
let map =
future::poll_fn(|cx| Pin::new(&mut self.body).poll_trailers(cx))
.await
.map_err(|e| Status::from_error(&e))?;
let map = future::poll_fn(|cx| Pin::new(&mut self.body).poll_trailers(cx))
.await
.map_err(|e| Status::from_error(&e))?;
Ok(map.map(MetadataMap::from_headers))
}