From dd66096ea0ab388c7163ae1928fc29d3f20e02ea Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Tue, 19 Feb 2019 16:30:37 -0500 Subject: [PATCH] buf: Add `BufStreamExt` trait and add a core feature (#897) This change adds an extension trait to `BufStream` and puts the core trait behind a feature flag for optional use. This mainly adds the additional functions in an extension trait to allow the user to select if they want just the core trait or the fully featured version. Now the user can add the core feature to _not_ include the extension trait. By deafult, this feature is disabled. --- .travis.yml | 1 + tokio-buf/Cargo.toml | 8 +- tokio-buf/README.md | 35 +++ tokio-buf/src/buf_stream/bytes.rs | 70 ----- tokio-buf/src/buf_stream/mod.rs | 163 ----------- tokio-buf/src/{buf_stream => }/errors.rs | 9 +- tokio-buf/src/{buf_stream => ext}/chain.rs | 7 +- tokio-buf/src/{buf_stream => ext}/collect.rs | 3 +- tokio-buf/src/{buf_stream => ext}/from.rs | 2 +- tokio-buf/src/{buf_stream => ext}/limit.rs | 18 +- tokio-buf/src/ext/mod.rs | 69 +++++ tokio-buf/src/lib.rs | 147 +++++++++- tokio-buf/src/{buf_stream => }/size_hint.rs | 0 tokio-buf/src/{buf_stream => }/str.rs | 2 +- tokio-buf/tests/buf_stream.rs | 268 +------------------ tokio-buf/tests/buf_stream_ext.rs | 158 +++++++++++ tokio-buf/tests/support.rs | 133 +++++++++ 17 files changed, 562 insertions(+), 531 deletions(-) delete mode 100644 tokio-buf/src/buf_stream/bytes.rs delete mode 100644 tokio-buf/src/buf_stream/mod.rs rename tokio-buf/src/{buf_stream => }/errors.rs (80%) rename tokio-buf/src/{buf_stream => ext}/chain.rs (88%) rename tokio-buf/src/{buf_stream => ext}/collect.rs (98%) rename tokio-buf/src/{buf_stream => ext}/from.rs (99%) rename tokio-buf/src/{buf_stream => ext}/limit.rs (78%) create mode 100644 tokio-buf/src/ext/mod.rs rename tokio-buf/src/{buf_stream => }/size_hint.rs (100%) rename tokio-buf/src/{buf_stream => }/str.rs (95%) create mode 100644 tokio-buf/tests/buf_stream_ext.rs create mode 100644 tokio-buf/tests/support.rs diff --git a/.travis.yml b/.travis.yml index 122f22c56..22e2f7e7e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -102,6 +102,7 @@ script: | # Limit the execution time of loom tests. export LOOM_MAX_DURATION=10 cargo test --all --no-fail-fast + cargo test -p tokio-buf --no-default-features cargo doc --all fi diff --git a/tokio-buf/Cargo.toml b/tokio-buf/Cargo.toml index d139a688a..2eb938a50 100644 --- a/tokio-buf/Cargo.toml +++ b/tokio-buf/Cargo.toml @@ -17,6 +17,10 @@ Asynchronous stream of byte buffers categories = ["asynchronous"] [dependencies] -bytes = { version = "0.4.10", features = [ "either" ] } -either = "1.5" +bytes = { version = "0.4.10" } +either = { version = "1.5", optional = true} futures = "0.1.23" + +[features] +default = ["ext"] +ext = ["bytes/either", "either"] diff --git a/tokio-buf/README.md b/tokio-buf/README.md index e69de29bb..3f275fb56 100644 --- a/tokio-buf/README.md +++ b/tokio-buf/README.md @@ -0,0 +1,35 @@ +# tokio-buf + +Asynchronous stream of byte buffers + +[Documenation](https://docs.rs/tokio-buf) + +## Usage + +First, add this to your `Cargo.toml`: + +```toml +[dependencies] +tokio-buf = "0.1" +``` + +Next, add this to your crate: + +```rust +extern crate tokio_buf; +``` + +You can find extensive documentation and examples about how to use this crate +online at [https://tokio.rs](https://tokio.rs). The [API +documentation](https://docs.rs/tokio-buf) is also a great place to get started +for the nitty-gritty. + +## License + +This project is licensed under the [MIT license](LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/tokio-buf/src/buf_stream/bytes.rs b/tokio-buf/src/buf_stream/bytes.rs deleted file mode 100644 index 627e9b114..000000000 --- a/tokio-buf/src/buf_stream/bytes.rs +++ /dev/null @@ -1,70 +0,0 @@ -use BufStream; -use buf_stream::errors::internal::Never; - -use bytes::{Bytes, BytesMut}; -use futures::Poll; - -use std::io; - -impl BufStream for Vec { - type Item = io::Cursor>; - type Error = Never; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - if self.is_empty() { - return Ok(None.into()); - } - - poll_bytes(self) - } -} - -impl BufStream for &'static [u8] { - type Item = io::Cursor<&'static [u8]>; - type Error = Never; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - if self.is_empty() { - return Ok(None.into()); - } - - poll_bytes(self) - } -} - -impl BufStream for Bytes { - type Item = io::Cursor; - type Error = Never; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - if self.is_empty() { - return Ok(None.into()); - } - - poll_bytes(self) - } -} - -impl BufStream for BytesMut { - type Item = io::Cursor; - type Error = Never; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - if self.is_empty() { - return Ok(None.into()); - } - - poll_bytes(self) - } -} - -fn poll_bytes(buf: &mut T) - -> Poll>, Never> -{ - use std::mem; - - let bytes = mem::replace(buf, Default::default()); - let buf = io::Cursor::new(bytes); - - Ok(Some(buf).into()) -} diff --git a/tokio-buf/src/buf_stream/mod.rs b/tokio-buf/src/buf_stream/mod.rs deleted file mode 100644 index bde1199b9..000000000 --- a/tokio-buf/src/buf_stream/mod.rs +++ /dev/null @@ -1,163 +0,0 @@ -//! Types and utilities for working with `BufStream`. - -mod bytes; -mod chain; -mod collect; -pub mod errors; -mod from; -mod limit; -mod size_hint; -mod str; - -pub use self::chain::Chain; -pub use self::collect::Collect; -pub use self::from::FromBufStream; -pub use self::limit::Limit; -pub use self::size_hint::SizeHint; - -use bytes::Buf; -use futures::Poll; - -/// An asynchronous stream of bytes. -/// -/// `BufStream` asynchronously yields values implementing `Buf`, i.e. byte -/// buffers. -pub trait BufStream { - /// Values yielded by the `BufStream`. - /// - /// Each item is a sequence of bytes representing a chunk of the total - /// `ByteStream`. - type Item: Buf; - - /// The error type this `BufStream` might generate. - type Error; - - /// Attempt to pull out the next buffer of this stream, registering the - /// current task for wakeup if the value is not yet available, and returning - /// `None` if the stream is exhausted. - /// - /// # Return value - /// - /// There are several possible return values, each indicating a distinct - /// stream state: - /// - /// - `Ok(Async::NotReady)` means that this stream's next value is not ready - /// yet. Implementations will ensure that the current task will be notified - /// when the next value may be ready. - /// - /// - `Ok(Async::Ready(Some(buf)))` means that the stream has successfully - /// produced a value, `buf`, and may produce further values on subsequent - /// `poll_buf` calls. - /// - /// - `Ok(Async::Ready(None))` means that the stream has terminated, and - /// `poll_buf` should not be invoked again. - /// - /// # Panics - /// - /// Once a stream is finished, i.e. `Ready(None)` has been returned, further - /// calls to `poll_buf` may result in a panic or other "bad behavior". - fn poll_buf(&mut self) -> Poll, Self::Error>; - - /// Returns the bounds on the remaining length of the stream. - /// - /// The size hint allows the caller to perform certain optimizations that - /// are dependent on the byte stream size. For example, `collect` uses the - /// size hint to pre-allocate enough capacity to store the entirety of the - /// data received from the byte stream. - /// - /// When `SizeHint::upper()` returns `Some` with a value equal to - /// `SizeHint::lower()`, this represents the exact number of bytes that will - /// be yielded by the `BufStream`. - /// - /// # Implementation notes - /// - /// While not enforced, implementations are expected to respect the values - /// returned from `SizeHint`. Any deviation is considered an implementation - /// bug. Consumers may rely on correctness in order to use the value as part - /// of protocol impelmentations. For example, an HTTP library may use the - /// size hint to set the `content-length` header. - /// - /// However, `size_hint` must not be trusted to omit bounds checks in unsafe - /// code. An incorrect implementation of `size_hint()` must not lead to - /// memory safety violations. - fn size_hint(&self) -> SizeHint { - SizeHint::default() - } - - /// Indicates to the `BufStream` how much data the consumer is currently - /// able to process. - /// - /// The consume hint allows the stream to perform certain optimizations that - /// are dependent on the consumer's readiness. For example, the consume hint - /// may be used to request a remote peer to start sending up to `amount` - /// data. - /// - /// Calling `consume_hint` is not a requirement. If `consume_hint` is never - /// called, the stream should assume a default behavior. When `consume_hint` - /// is called, the stream should make a best effort to honor by the request. - /// - /// `amount` represents the number of bytes that the caller would like to - /// receive at the time the function is called. For example, if - /// `consume_hint` is called with 20, the consumer requests 20 bytes. The - /// stream may yield less than that. If the next call to `poll_buf` returns - /// 5 bytes, the consumer still has 15 bytes requested. At this point, - /// invoking `consume_hint` again with 20 resets the amount requested back - /// to 20 bytes. - /// - /// Calling `consume_hint` with 0 as the argument informs the stream that - /// the caller does not intend to call `poll_buf`. If `poll_buf` **is** - /// called, the stream may, but is not obligated to, return `NotReady` even - /// if it could produce data at that point. If it chooses to return - /// `NotReady`, when `consume_hint` is called with a non-zero argument, the - /// task must be notified in order to respect the `poll_buf` contract. - fn consume_hint(&mut self, amount: usize) { - // By default, this function does nothing - drop(amount); - } - - /// Takes two buf streams and creates a new buf stream over both in - /// sequence. - /// - /// `chain()` returns a new `BufStream` value which will first yield all - /// data from `self` then all data from `other`. - /// - /// In other words, it links two buf streams together, in a chain. - fn chain(self, other: T) -> Chain - where - Self: Sized, - T: BufStream, - { - Chain::new(self, other) - } - - /// Consumes all data from `self`, storing it in byte storage of type `T`. - /// - /// `collect()` returns a future that buffers all data yielded from `self` - /// into storage of type of `T`. The future completes once `self` yield - /// `None`, returning the buffered data. - /// - /// The collect future will yield an error if `self` yields an error or if - /// the collect operation errors. The collect error cases are dependent on - /// the target storage type. - fn collect(self) -> Collect - where - Self: Sized, - T: FromBufStream, - { - Collect::new(self) - } - - /// Limit the number of bytes that the stream can yield. - /// - /// `limit()` returns a new `BufStream` value which yields all the data from - /// `self` while ensuring that at most `amount` bytes are yielded. - /// - /// If `self` can yield greater than `amount` bytes, the returned stream - /// will yield an error. - fn limit(self, amount: u64) -> Limit - where - Self: Sized, - { - Limit::new(self, amount) - } -} diff --git a/tokio-buf/src/buf_stream/errors.rs b/tokio-buf/src/errors.rs similarity index 80% rename from tokio-buf/src/buf_stream/errors.rs rename to tokio-buf/src/errors.rs index 2a9760ff9..d87f1f8c2 100644 --- a/tokio-buf/src/buf_stream/errors.rs +++ b/tokio-buf/src/errors.rs @@ -1,8 +1,11 @@ //! Error types -pub use super::collect::CollectError; -pub use super::from::CollectVecError; -pub use super::limit::LimitError; +#[cfg(feature = "ext")] +pub use ext::CollectError; +#[cfg(feature = "ext")] +pub use ext::CollectVecError; +#[cfg(feature = "ext")] +pub use ext::LimitError; // Being crate-private, we should be able to swap the type out in a // backwards compatible way. diff --git a/tokio-buf/src/buf_stream/chain.rs b/tokio-buf/src/ext/chain.rs similarity index 88% rename from tokio-buf/src/buf_stream/chain.rs rename to tokio-buf/src/ext/chain.rs index 157b0e9e1..44fc40088 100644 --- a/tokio-buf/src/buf_stream/chain.rs +++ b/tokio-buf/src/ext/chain.rs @@ -1,4 +1,4 @@ -use super::{BufStream, SizeHint}; +use BufStream; use either::Either; use futures::Poll; @@ -43,9 +43,4 @@ where let res = try_ready!(self.right.poll_buf()); Ok(res.map(Either::Right).into()) } - - fn size_hint(&self) -> SizeHint { - // TODO: Implement - SizeHint::default() - } } diff --git a/tokio-buf/src/buf_stream/collect.rs b/tokio-buf/src/ext/collect.rs similarity index 98% rename from tokio-buf/src/buf_stream/collect.rs rename to tokio-buf/src/ext/collect.rs index c5731561a..95da54528 100644 --- a/tokio-buf/src/buf_stream/collect.rs +++ b/tokio-buf/src/ext/collect.rs @@ -1,4 +1,5 @@ -use super::{BufStream, FromBufStream}; +use BufStream; +use super::FromBufStream; use futures::{Future, Poll}; diff --git a/tokio-buf/src/buf_stream/from.rs b/tokio-buf/src/ext/from.rs similarity index 99% rename from tokio-buf/src/buf_stream/from.rs rename to tokio-buf/src/ext/from.rs index 6137bf829..c15057a4d 100644 --- a/tokio-buf/src/buf_stream/from.rs +++ b/tokio-buf/src/ext/from.rs @@ -1,4 +1,4 @@ -use super::SizeHint; +use SizeHint; use bytes::{Buf, BufMut}; diff --git a/tokio-buf/src/buf_stream/limit.rs b/tokio-buf/src/ext/limit.rs similarity index 78% rename from tokio-buf/src/buf_stream/limit.rs rename to tokio-buf/src/ext/limit.rs index 0c6baeb88..1d3e0ba3a 100644 --- a/tokio-buf/src/buf_stream/limit.rs +++ b/tokio-buf/src/ext/limit.rs @@ -1,4 +1,4 @@ -use super::{BufStream, SizeHint}; +use BufStream; use bytes::Buf; use futures::Poll; @@ -59,22 +59,6 @@ where res } - - fn size_hint(&self) -> SizeHint { - let mut hint = self.stream.size_hint(); - - let upper = hint.upper() - .map(|upper| upper.min(self.remaining)) - .unwrap_or(self.remaining); - - hint.set_upper(upper); - hint - } - - fn consume_hint(&mut self, amount: usize) { - // TODO: Should this be capped by `self.remaining`? - self.stream.consume_hint(amount) - } } // ===== impl LimitError ===== diff --git a/tokio-buf/src/ext/mod.rs b/tokio-buf/src/ext/mod.rs new file mode 100644 index 000000000..f0299129c --- /dev/null +++ b/tokio-buf/src/ext/mod.rs @@ -0,0 +1,69 @@ +//! Types and utilities for working with `BufStream`. + +mod chain; +mod collect; +mod from; +mod limit; + +pub use self::chain::Chain; +pub use self::collect::Collect; +pub use self::from::FromBufStream; +pub use self::limit::Limit; + +pub use self::collect::CollectError; +pub use self::from::CollectVecError; +pub use self::limit::LimitError; + +use BufStream; + +impl BufStreamExt for T where T: BufStream {} + +/// An extension trait for `BufStream`'s that provides a variety of convenient +/// adapters. +pub trait BufStreamExt: BufStream { + /// Takes two buf streams and creates a new buf stream over both in + /// sequence. + /// + /// `chain()` returns a new `BufStream` value which will first yield all + /// data from `self` then all data from `other`. + /// + /// In other words, it links two buf streams together, in a chain. + fn chain(self, other: T) -> Chain + where + Self: Sized, + T: BufStream, + { + Chain::new(self, other) + } + + /// Consumes all data from `self`, storing it in byte storage of type `T`. + /// + /// `collect()` returns a future that buffers all data yielded from `self` + /// into storage of type of `T`. The future completes once `self` yield + /// `None`, returning the buffered data. + /// + /// The collect future will yield an error if `self` yields an error or if + /// the collect operation errors. The collect error cases are dependent on + /// the target storage type. + fn collect(self) -> Collect + where + Self: Sized, + T: FromBufStream, + { + Collect::new(self) + } + + /// Limit the number of bytes that the stream can yield. + /// + /// `limit()` returns a new `BufStream` value which yields all the data from + /// `self` while ensuring that at most `amount` bytes are yielded. + /// + /// If `self` can yield greater than `amount` bytes, the returned stream + /// will yield an error. + fn limit(self, amount: u64) -> Limit + where + Self: Sized, + { + Limit::new(self, amount) + } +} diff --git a/tokio-buf/src/lib.rs b/tokio-buf/src/lib.rs index b706b932d..d816f5509 100644 --- a/tokio-buf/src/lib.rs +++ b/tokio-buf/src/lib.rs @@ -10,11 +10,154 @@ //! `Buf` (i.e, byte collections). extern crate bytes; +#[cfg(feature = "ext")] extern crate either; +#[allow(unused)] #[macro_use] extern crate futures; -pub mod buf_stream; +#[cfg(feature = "ext")] +pub mod ext; +pub mod errors; +mod size_hint; +mod str; #[doc(inline)] -pub use buf_stream::BufStream; +#[cfg(feature = "ext")] +pub use ext::BufStreamExt; +pub use self::size_hint::SizeHint; + +use futures::Poll; +use bytes::{Buf, Bytes, BytesMut}; +use std::io; +use errors::internal::Never; + +/// An asynchronous stream of bytes. +/// +/// `BufStream` asynchronously yields values implementing `Buf`, i.e. byte +/// buffers. +pub trait BufStream { + /// Values yielded by the `BufStream`. + /// + /// Each item is a sequence of bytes representing a chunk of the total + /// `ByteStream`. + type Item: Buf; + + /// The error type this `BufStream` might generate. + type Error; + + /// Attempt to pull out the next buffer of this stream, registering the + /// current task for wakeup if the value is not yet available, and returning + /// `None` if the stream is exhausted. + /// + /// # Return value + /// + /// There are several possible return values, each indicating a distinct + /// stream state: + /// + /// - `Ok(Async::NotReady)` means that this stream's next value is not ready + /// yet. Implementations will ensure that the current task will be notified + /// when the next value may be ready. + /// + /// - `Ok(Async::Ready(Some(buf)))` means that the stream has successfully + /// produced a value, `buf`, and may produce further values on subsequent + /// `poll_buf` calls. + /// + /// - `Ok(Async::Ready(None))` means that the stream has terminated, and + /// `poll_buf` should not be invoked again. + /// + /// # Panics + /// + /// Once a stream is finished, i.e. `Ready(None)` has been returned, further + /// calls to `poll_buf` may result in a panic or other "bad behavior". + fn poll_buf(&mut self) -> Poll, Self::Error>; + + /// Returns the bounds on the remaining length of the stream. + /// + /// The size hint allows the caller to perform certain optimizations that + /// are dependent on the byte stream size. For example, `collect` uses the + /// size hint to pre-allocate enough capacity to store the entirety of the + /// data received from the byte stream. + /// + /// When `SizeHint::upper()` returns `Some` with a value equal to + /// `SizeHint::lower()`, this represents the exact number of bytes that will + /// be yielded by the `BufStream`. + /// + /// # Implementation notes + /// + /// While not enforced, implementations are expected to respect the values + /// returned from `SizeHint`. Any deviation is considered an implementation + /// bug. Consumers may rely on correctness in order to use the value as part + /// of protocol impelmentations. For example, an HTTP library may use the + /// size hint to set the `content-length` header. + /// + /// However, `size_hint` must not be trusted to omit bounds checks in unsafe + /// code. An incorrect implementation of `size_hint()` must not lead to + /// memory safety violations. + fn size_hint(&self) -> SizeHint { + SizeHint::default() + } +} + +impl BufStream for Vec { + type Item = io::Cursor>; + type Error = Never; + + fn poll_buf(&mut self) -> Poll, Self::Error> { + if self.is_empty() { + return Ok(None.into()); + } + + poll_bytes(self) + } +} + +impl BufStream for &'static [u8] { + type Item = io::Cursor<&'static [u8]>; + type Error = Never; + + fn poll_buf(&mut self) -> Poll, Self::Error> { + if self.is_empty() { + return Ok(None.into()); + } + + poll_bytes(self) + } +} + +impl BufStream for Bytes { + type Item = io::Cursor; + type Error = Never; + + fn poll_buf(&mut self) -> Poll, Self::Error> { + if self.is_empty() { + return Ok(None.into()); + } + + poll_bytes(self) + } +} + +impl BufStream for BytesMut { + type Item = io::Cursor; + type Error = Never; + + fn poll_buf(&mut self) -> Poll, Self::Error> { + if self.is_empty() { + return Ok(None.into()); + } + + poll_bytes(self) + } +} + +fn poll_bytes(buf: &mut T) + -> Poll>, Never> +{ + use std::mem; + + let bytes = mem::replace(buf, Default::default()); + let buf = io::Cursor::new(bytes); + + Ok(Some(buf).into()) +} diff --git a/tokio-buf/src/buf_stream/size_hint.rs b/tokio-buf/src/size_hint.rs similarity index 100% rename from tokio-buf/src/buf_stream/size_hint.rs rename to tokio-buf/src/size_hint.rs diff --git a/tokio-buf/src/buf_stream/str.rs b/tokio-buf/src/str.rs similarity index 95% rename from tokio-buf/src/buf_stream/str.rs rename to tokio-buf/src/str.rs index fa12d8f04..b18080809 100644 --- a/tokio-buf/src/buf_stream/str.rs +++ b/tokio-buf/src/str.rs @@ -1,5 +1,5 @@ use BufStream; -use buf_stream::errors::internal::Never; +use errors::internal::Never; use futures::Poll; diff --git a/tokio-buf/tests/buf_stream.rs b/tokio-buf/tests/buf_stream.rs index e5b462b59..848a5de1e 100644 --- a/tokio-buf/tests/buf_stream.rs +++ b/tokio-buf/tests/buf_stream.rs @@ -2,45 +2,13 @@ extern crate tokio_buf; extern crate bytes; extern crate futures; -use tokio_buf::buf_stream::{BufStream, SizeHint}; +use tokio_buf::{BufStream, SizeHint}; use bytes::Buf; -use futures::{Future, Poll}; use futures::Async::*; -use std::collections::VecDeque; -use std::io::Cursor; -macro_rules! assert_buf_eq { - ($actual:expr, $expect:expr) => {{ - match $actual { - Ok(Ready(Some(val))) => { - assert_eq!(val.remaining(), val.bytes().len()); - assert_eq!(val.bytes(), $expect.as_bytes()); - } - Ok(Ready(None)) => panic!("expected value; BufStream yielded None"), - Ok(NotReady) => panic!("expected value; BufStream is not ready"), - Err(e) => panic!("expected value; got error = {:?}", e), - } - }}; -} - -macro_rules! assert_none { - ($actual:expr) => { - match $actual { - Ok(Ready(None)) => {} - actual => panic!("expected None; actual = {:?}", actual), - } - } -} - -macro_rules! assert_not_ready { - ($actual:expr) => { - match $actual { - Ok(NotReady) => {} - actual => panic!("expected NotReady; actual = {:?}", actual), - } - } -} +#[macro_use] +mod support; // ===== test `SizeHint` ===== @@ -83,149 +51,6 @@ fn size_hint_upper_less_than_lower() { hint.set_upper(100); } -// ===== test `chain()` ===== - -#[test] -fn chain() { - // Chain one with one - // - let mut bs = one("hello").chain(one("world")); - - assert_buf_eq!(bs.poll_buf(), "hello"); - assert_buf_eq!(bs.poll_buf(), "world"); - assert_none!(bs.poll_buf()); - - // Chain multi with multi - let mut bs = list(&["foo", "bar"]) - .chain(list(&["baz", "bok"])); - - assert_buf_eq!(bs.poll_buf(), "foo"); - assert_buf_eq!(bs.poll_buf(), "bar"); - assert_buf_eq!(bs.poll_buf(), "baz"); - assert_buf_eq!(bs.poll_buf(), "bok"); - assert_none!(bs.poll_buf()); - - // Chain includes a not ready call - // - let mut bs = new_mock(&[ - Ok(Ready("foo")), - Ok(NotReady), - Ok(Ready("bar")) - ]).chain(one("baz")); - - assert_buf_eq!(bs.poll_buf(), "foo"); - assert_not_ready!(bs.poll_buf()); - assert_buf_eq!(bs.poll_buf(), "bar"); - assert_buf_eq!(bs.poll_buf(), "baz"); - assert_none!(bs.poll_buf()); -} - -// ===== Test `collect()` ===== - -#[test] -fn collect_vec() { - // While unfortunate, this test makes some assumptions on vec's resizing - // behavior. - // - // Collect one - // - let bs = one("hello world"); - - let vec: Vec = bs.collect() - .wait().unwrap(); - - assert_eq!(vec, b"hello world"); - assert_eq!(vec.capacity(), 64); - - // Collect one, with size hint - // - let mut bs = one("hello world"); - bs.size_hint.set_lower(11); - - let vec: Vec = bs.collect() - .wait().unwrap(); - - assert_eq!(vec, b"hello world"); - assert_eq!(vec.capacity(), 64); - - // Collect one, with size hint - // - let mut bs = one("hello world"); - bs.size_hint.set_lower(10); - - let vec: Vec = bs.collect() - .wait().unwrap(); - - assert_eq!(vec, b"hello world"); - assert_eq!(vec.capacity(), 64); - - // Collect many - // - let bs = list(&["hello", " ", "world", ", one two three"]); - - let vec: Vec = bs.collect() - .wait().unwrap(); - - assert_eq!(vec, b"hello world, one two three"); -} - -// ===== Test limit() ===== - -#[test] -fn limit() { - // Not limited - - let res = one("hello world") - .limit(100) - .collect::>() - .wait().unwrap(); - - assert_eq!(res, b"hello world"); - - let res = list(&["hello", " ", "world"]) - .limit(100) - .collect::>() - .wait().unwrap(); - - assert_eq!(res, b"hello world"); - - let res = list(&["hello", " ", "world"]) - .limit(11) - .collect::>() - .wait().unwrap(); - - assert_eq!(res, b"hello world"); - - // Limited - - let res = one("hello world") - .limit(5) - .collect::>() - .wait(); - - assert!(res.is_err()); - - let res = one("hello world") - .limit(10) - .collect::>() - .wait(); - - assert!(res.is_err()); - - let mut bs = list(&["hello", " ", "world"]) - .limit(9); - - assert_buf_eq!(bs.poll_buf(), "hello"); - assert_buf_eq!(bs.poll_buf(), " "); - assert!(bs.poll_buf().is_err()); - - let mut bs = list(&["hello", " ", "world"]); - bs.size_hint.set_lower(11); - let mut bs = bs.limit(9); - - assert!(bs.poll_buf().is_err()); -} - // ===== BufStream impelmentations for misc types ===== #[test] @@ -240,90 +65,3 @@ fn str_buf_stream() { assert!(bs.is_empty()); assert_none!(bs.poll_buf()); } - -// ===== Test utils ===== - -fn one(buf: &'static str) -> Mock { - list(&[buf]) -} - -fn list(bufs: &[&'static str]) -> Mock { - let mut polls = VecDeque::new(); - - for &buf in bufs { - polls.push_back(Ok(Ready(buf.as_bytes()))); - } - - Mock { - polls, - size_hint: SizeHint::default(), - } -} - -fn new_mock(values: &[Poll<&'static str, ()>]) -> Mock { - let mut polls = VecDeque::new(); - - for &v in values { - polls.push_back(match v { - Ok(Ready(v)) => Ok(Ready(v.as_bytes())), - Ok(NotReady) => Ok(NotReady), - Err(e) => Err(e), - }); - } - - Mock { - polls, - size_hint: SizeHint::default(), - } -} - -#[derive(Debug)] -struct Mock { - polls: VecDeque>, - size_hint: SizeHint, -} - -#[derive(Debug)] -struct MockBuf { - data: Cursor<&'static [u8]>, -} - -impl BufStream for Mock { - type Item = MockBuf; - type Error = (); - - fn poll_buf(&mut self) -> Poll, Self::Error> { - match self.polls.pop_front() { - Some(Ok(Ready(value))) => Ok(Ready(Some(MockBuf::new(value)))), - Some(Ok(NotReady)) => Ok(NotReady), - Some(Err(e)) => Err(e), - None => Ok(Ready(None)), - } - } - - fn size_hint(&self) -> SizeHint { - self.size_hint.clone() - } -} - -impl MockBuf { - fn new(data: &'static [u8]) -> MockBuf { - MockBuf { - data: Cursor::new(data), - } - } -} - -impl Buf for MockBuf { - fn remaining(&self) -> usize { - self.data.remaining() - } - - fn bytes(&self) -> &[u8] { - self.data.bytes() - } - - fn advance(&mut self, cnt: usize) { - self.data.advance(cnt) - } -} diff --git a/tokio-buf/tests/buf_stream_ext.rs b/tokio-buf/tests/buf_stream_ext.rs new file mode 100644 index 000000000..2395d20e7 --- /dev/null +++ b/tokio-buf/tests/buf_stream_ext.rs @@ -0,0 +1,158 @@ +#![cfg(feature = "ext")] + +extern crate tokio_buf; +extern crate bytes; +extern crate futures; + +use tokio_buf::{BufStream, BufStreamExt}; +use futures::Future; +use futures::Async::*; +use bytes::Buf; + +#[macro_use] +mod support; + +use support::*; + +// ===== test `chain()` ===== + +#[test] +fn chain() { + // Chain one with one + // + let mut bs = one("hello").chain(one("world")); + + assert_buf_eq!(bs.poll_buf(), "hello"); + assert_buf_eq!(bs.poll_buf(), "world"); + assert_none!(bs.poll_buf()); + + // Chain multi with multi + let mut bs = list(&["foo", "bar"]) + .chain(list(&["baz", "bok"])); + + assert_buf_eq!(bs.poll_buf(), "foo"); + assert_buf_eq!(bs.poll_buf(), "bar"); + assert_buf_eq!(bs.poll_buf(), "baz"); + assert_buf_eq!(bs.poll_buf(), "bok"); + assert_none!(bs.poll_buf()); + + // Chain includes a not ready call + // + let mut bs = new_mock(&[ + Ok(Ready("foo")), + Ok(NotReady), + Ok(Ready("bar")) + ]).chain(one("baz")); + + assert_buf_eq!(bs.poll_buf(), "foo"); + assert_not_ready!(bs.poll_buf()); + assert_buf_eq!(bs.poll_buf(), "bar"); + assert_buf_eq!(bs.poll_buf(), "baz"); + assert_none!(bs.poll_buf()); +} + +// ===== Test `collect()` ===== + +#[test] +fn collect_vec() { + // While unfortunate, this test makes some assumptions on vec's resizing + // behavior. + // + // Collect one + // + let bs = one("hello world"); + + let vec: Vec = bs.collect() + .wait().unwrap(); + + assert_eq!(vec, b"hello world"); + assert_eq!(vec.capacity(), 64); + + // Collect one, with size hint + // + let mut bs = one("hello world"); + bs.size_hint.set_lower(11); + + let vec: Vec = bs.collect() + .wait().unwrap(); + + assert_eq!(vec, b"hello world"); + assert_eq!(vec.capacity(), 64); + + // Collect one, with size hint + // + let mut bs = one("hello world"); + bs.size_hint.set_lower(10); + + let vec: Vec = bs.collect() + .wait().unwrap(); + + assert_eq!(vec, b"hello world"); + assert_eq!(vec.capacity(), 64); + + // Collect many + // + let bs = list(&["hello", " ", "world", ", one two three"]); + + let vec: Vec = bs.collect() + .wait().unwrap(); + + assert_eq!(vec, b"hello world, one two three"); +} + +// ===== Test limit() ===== + +#[test] +fn limit() { + // Not limited + + let res = one("hello world") + .limit(100) + .collect::>() + .wait().unwrap(); + + assert_eq!(res, b"hello world"); + + let res = list(&["hello", " ", "world"]) + .limit(100) + .collect::>() + .wait().unwrap(); + + assert_eq!(res, b"hello world"); + + let res = list(&["hello", " ", "world"]) + .limit(11) + .collect::>() + .wait().unwrap(); + + assert_eq!(res, b"hello world"); + + // Limited + + let res = one("hello world") + .limit(5) + .collect::>() + .wait(); + + assert!(res.is_err()); + + let res = one("hello world") + .limit(10) + .collect::>() + .wait(); + + assert!(res.is_err()); + + let mut bs = list(&["hello", " ", "world"]) + .limit(9); + + assert_buf_eq!(bs.poll_buf(), "hello"); + assert_buf_eq!(bs.poll_buf(), " "); + assert!(bs.poll_buf().is_err()); + + let mut bs = list(&["hello", " ", "world"]); + bs.size_hint.set_lower(11); + let mut bs = bs.limit(9); + + assert!(bs.poll_buf().is_err()); +} diff --git a/tokio-buf/tests/support.rs b/tokio-buf/tests/support.rs new file mode 100644 index 000000000..7e7d1b4b1 --- /dev/null +++ b/tokio-buf/tests/support.rs @@ -0,0 +1,133 @@ +#![allow(unused)] + +extern crate tokio_buf; +extern crate bytes; +extern crate futures; + +use tokio_buf::{BufStream, SizeHint}; +use bytes::Buf; +use futures::Poll; +use futures::Async::*; + +use std::collections::VecDeque; +use std::io::Cursor; + +macro_rules! assert_buf_eq { + ($actual:expr, $expect:expr) => {{ + match $actual { + Ok(Ready(Some(val))) => { + assert_eq!(val.remaining(), val.bytes().len()); + assert_eq!(val.bytes(), $expect.as_bytes()); + } + Ok(Ready(None)) => panic!("expected value; BufStream yielded None"), + Ok(NotReady) => panic!("expected value; BufStream is not ready"), + Err(e) => panic!("expected value; got error = {:?}", e), + } + }}; +} + +macro_rules! assert_none { + ($actual:expr) => { + match $actual { + Ok(Ready(None)) => {} + actual => panic!("expected None; actual = {:?}", actual), + } + } +} + +macro_rules! assert_not_ready { + ($actual:expr) => { + match $actual { + Ok(NotReady) => {} + actual => panic!("expected NotReady; actual = {:?}", actual), + } + } +} + +// ===== Test utils ===== + +pub fn one(buf: &'static str) -> Mock { + list(&[buf]) +} + +pub fn list(bufs: &[&'static str]) -> Mock { + let mut polls = VecDeque::new(); + + for &buf in bufs { + polls.push_back(Ok(Ready(buf.as_bytes()))); + } + + Mock { + polls, + size_hint: SizeHint::default(), + } +} + +pub fn new_mock(values: &[Poll<&'static str, ()>]) -> Mock { + let mut polls = VecDeque::new(); + + for &v in values { + polls.push_back(match v { + Ok(Ready(v)) => Ok(Ready(v.as_bytes())), + Ok(NotReady) => Ok(NotReady), + Err(e) => Err(e), + }); + } + + Mock { + polls, + size_hint: SizeHint::default(), + } +} + +#[derive(Debug)] +pub struct Mock { + pub polls: VecDeque>, + pub size_hint: SizeHint, +} + +#[derive(Debug)] +pub struct MockBuf { + pub data: Cursor<&'static [u8]>, +} + +impl BufStream for Mock { + type Item = MockBuf; + type Error = (); + + fn poll_buf(&mut self) -> Poll, Self::Error> { + match self.polls.pop_front() { + Some(Ok(Ready(value))) => Ok(Ready(Some(MockBuf::new(value)))), + Some(Ok(NotReady)) => Ok(NotReady), + Some(Err(e)) => Err(e), + None => Ok(Ready(None)), + } + } + + fn size_hint(&self) -> SizeHint { + self.size_hint.clone() + } +} + +impl MockBuf { + fn new(data: &'static [u8]) -> MockBuf { + MockBuf { + data: Cursor::new(data), + } + } +} + +impl Buf for MockBuf { + fn remaining(&self) -> usize { + self.data.remaining() + } + + fn bytes(&self) -> &[u8] { + self.data.bytes() + } + + fn advance(&mut self, cnt: usize) { + self.data.advance(cnt) + } +} +