Skip to content

Commit

Permalink
feat(body): put Stream impl for Body behind stream feature
Browse files Browse the repository at this point in the history
BREAKING CHANGE: Using a `Body` as a `Stream`, and constructing one via
  `Body::wrap_stream`, require enabling the unstable `stream` feature.
  • Loading branch information
seanmonstar committed Sep 5, 2019
1 parent b3e5506 commit 511ea38
Show file tree
Hide file tree
Showing 4 changed files with 43 additions and 25 deletions.
23 changes: 14 additions & 9 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,6 @@ tokio-fs = "=0.2.0-alpha.4"
tokio-test = "=0.2.0-alpha.4"
url = "1.0"


[features]
default = [
"__internal_flaky_tests",
Expand All @@ -78,6 +77,12 @@ nightly = []
__internal_flaky_tests = []
__internal_happy_eyeballs_tests = []

[package.metadata.docs.rs]
features = [
"runtime",
"stream",
]

[profile.release]
codegen-units = 1
incremental = false
Expand All @@ -94,12 +99,12 @@ required-features = ["runtime"]
[[example]]
name = "client_json"
path = "examples/client_json.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]

[[example]]
name = "echo"
path = "examples/echo.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]

[[example]]
name = "hello"
Expand All @@ -114,7 +119,7 @@ required-features = ["runtime"]
[[example]]
name = "params"
path = "examples/params.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]

[[example]]
name = "proxy"
Expand Down Expand Up @@ -155,7 +160,7 @@ required-features = ["runtime"]
[[example]]
name = "web_api"
path = "examples/web_api.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]


[[bench]]
Expand All @@ -171,20 +176,20 @@ required-features = ["runtime"]
[[bench]]
name = "server"
path = "benches/server.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]


[[test]]
name = "client"
path = "tests/client.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]

[[test]]
name = "integration"
path = "tests/integration.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]

[[test]]
name = "server"
path = "tests/server.rs"
required-features = ["runtime"]
required-features = ["runtime", "stream"]
25 changes: 20 additions & 5 deletions src/body/body.rs
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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,
Expand All @@ -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>>),
}

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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))),
Expand Down Expand Up @@ -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,
}
}
Expand All @@ -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();
Expand All @@ -361,19 +374,20 @@ 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),
};

builder.finish()
}
}

#[cfg(feature = "stream")]
impl Stream for Body {
type Item = crate::Result<Chunk>;

Expand All @@ -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
Expand Down
9 changes: 9 additions & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,15 @@
//!
//! If looking for just a convenient HTTP client, consider the
//! [reqwest](https://crates.io/crates/reqwest) crate.
//!
//! # Optional Features
//!
//! The following optional features are available:
//!
//! - `runtime` (*enabled by default*): Enables convenient integration with
//! `tokio`, providing connectors and acceptors for TCP, and a default
//! executor.
//! - `stream` (*unstable*): Provides `futures::Stream` capabilities.

#[doc(hidden)] pub use http;
#[macro_use] extern crate log;
Expand Down
11 changes: 0 additions & 11 deletions src/server/tcp.rs
Original file line number Diff line number Diff line change
Expand Up @@ -156,17 +156,6 @@ impl AddrIncoming {
}
}

/*
impl Stream for AddrIncoming {
type Item = io::Result<AddrStream>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
let result = ready!(self.poll_next_(cx));
Poll::Ready(Some(result))
}
}
*/

impl Accept for AddrIncoming {
type Conn = AddrStream;
type Error = io::Error;
Expand Down

0 comments on commit 511ea38

Please sign in to comment.