forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(body): put Stream impl for Body behind stream feature

Browse Source 
BREAKING CHANGE: Using a `Body` as a `Stream`, and constructing one via
  `Body::wrap_stream`, require enabling the unstable `stream` feature.
This commit is contained in:
Sean McArthur
2019-09-05 14:46:12 -07:00
parent b3e5506261
commit 511ea3889b
4 changed files with 43 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/body/body.rs
View File

@@ -1,10 +1,14 @@
use std::borrow::Cow;
#[cfg(feature = "stream")]
use std::error::Error as StdError;
use std::fmt;
use bytes::Bytes;
use futures_core::{Stream, TryStream};
use futures_core::Stream; // for mpsc::Receiver
#[cfg(feature = "stream")]
use futures_core::TryStream;
use futures_channel::{mpsc, oneshot};
#[cfg(feature = "stream")]
use futures_util::TryStreamExt;
use http_body::{SizeHint, Body as HttpBody};
use http::HeaderMap;
@@ -18,8 +22,6 @@ type BodySender = mpsc::Sender<Result<Chunk, crate::Error>>;
/// A stream of `Chunk`s, used when receiving bodies.
///
/// A good default `Payload` to use in many applications.
///
/// Also implements `futures::Stream`, so stream combinators may be used.
#[must_use = "streams do nothing unless polled"]
pub struct Body {
kind: Kind,
@@ -43,6 +45,7 @@ enum Kind {
// while a borrow of a `Request<Body>` exists.
//
// See https://github.com/rust-lang/rust/issues/57017
#[cfg(feature = "stream")]
Wrapped(Pin<Box<dyn Stream<Item = Result<Chunk, Box<dyn StdError + Send + Sync>>> + Send + Sync>>),
}
@@ -140,6 +143,12 @@ impl Body {
/// let body = Body::wrap_stream(stream);
/// # }
/// ```
///
/// # Unstable
///
/// This function requires enabling the unstable `stream` feature in your
/// `Cargo.toml`.
#[cfg(feature = "stream")]
pub fn wrap_stream<S>(stream: S) -> Body
where
S: TryStream + Send + Sync + 'static,
@@ -277,6 +286,8 @@ impl Body {
Some(Err(e)) => Poll::Ready(Some(Err(crate::Error::new_body(e)))),
None => Poll::Ready(None),
},
#[cfg(feature = "stream")]
Kind::Wrapped(ref mut s) => {
match ready!(s.as_mut().poll_next(cx)) {
Some(res) => Poll::Ready(Some(res.map_err(crate::Error::new_body))),
@@ -326,6 +337,7 @@ impl HttpBody for Body {
Kind::Once(ref val) => val.is_none(),
Kind::Chan { content_length, .. } => content_length == Some(0),
Kind::H2 { recv: ref h2, .. } => h2.is_end_stream(),
#[cfg(feature = "stream")]
Kind::Wrapped(..) => false,
}
}
@@ -340,6 +352,7 @@ impl HttpBody for Body {
Kind::Once(None) => {
SizeHint::default()
},
#[cfg(feature = "stream")]
Kind::Wrapped(..) => SizeHint::default(),
Kind::Chan { content_length, .. } | Kind::H2 { content_length, .. } => {
let mut hint = SizeHint::default();
@@ -361,12 +374,12 @@ impl fmt::Debug for Body {
#[derive(Debug)]
struct Empty;
#[derive(Debug)]
struct Once<'a>(&'a Chunk);
struct Full<'a>(&'a Chunk);
let mut builder = f.debug_tuple("Body");
match self.kind {
Kind::Once(None) => builder.field(&Empty),
Kind::Once(Some(ref chunk)) => builder.field(&Once(chunk)),
Kind::Once(Some(ref chunk)) => builder.field(&Full(chunk)),
_ => builder.field(&Streaming),
};
@@ -374,6 +387,7 @@ impl fmt::Debug for Body {
}
}
#[cfg(feature = "stream")]
impl Stream for Body {
type Item = crate::Result<Chunk>;
@@ -383,6 +397,7 @@ impl Stream for Body {
}
#[cfg(feature = "stream")]
impl
From<Box<dyn Stream<Item = Result<Chunk, Box<dyn StdError + Send + Sync>>> + Send + Sync>>
for Body