From 815173f8e5e7c38bcaa85c433a0e8074432c5daf Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Thu, 19 Sep 2019 12:11:21 -0700 Subject: [PATCH] chore: rm tokio-buf (#1574) The crate has not been updated and it does not seem like it is a good path forward. --- Cargo.toml | 1 - azure-pipelines.yml | 1 - ci/patch.toml | 1 - tokio-buf/CHANGELOG.md | 19 ----- tokio-buf/Cargo.toml | 34 -------- tokio-buf/LICENSE | 25 ------ tokio-buf/README.md | 13 --- tokio-buf/src/lib.rs | 102 ---------------------- tokio-buf/src/never.rs | 18 ---- tokio-buf/src/size_hint.rs | 56 ------------- tokio-buf/src/str.rs | 55 ------------ tokio-buf/src/u8.rs | 66 --------------- tokio-buf/src/util/chain.rs | 46 ---------- tokio-buf/src/util/collect.rs | 101 ---------------------- tokio-buf/src/util/from.rs | 154 ---------------------------------- tokio-buf/src/util/iter.rs | 50 ----------- tokio-buf/src/util/limit.rs | 76 ----------------- tokio-buf/src/util/mod.rs | 87 ------------------- tokio-buf/src/util/stream.rs | 76 ----------------- tokio-buf/tests/buf_stream.rs | 8 -- tokio-buf/tests/chain.rs | 38 --------- tokio-buf/tests/collect.rs | 63 -------------- tokio-buf/tests/iter.rs | 29 ------- tokio-buf/tests/limit.rs | 60 ------------- tokio-buf/tests/size_hint.rs | 40 --------- tokio-buf/tests/stream.rs | 44 ---------- tokio-buf/tests/string.rs | 34 -------- tokio-buf/tests/support.rs | 133 ----------------------------- 28 files changed, 1430 deletions(-) delete mode 100644 tokio-buf/CHANGELOG.md delete mode 100644 tokio-buf/Cargo.toml delete mode 100644 tokio-buf/LICENSE delete mode 100644 tokio-buf/README.md delete mode 100644 tokio-buf/src/lib.rs delete mode 100644 tokio-buf/src/never.rs delete mode 100644 tokio-buf/src/size_hint.rs delete mode 100644 tokio-buf/src/str.rs delete mode 100644 tokio-buf/src/u8.rs delete mode 100644 tokio-buf/src/util/chain.rs delete mode 100644 tokio-buf/src/util/collect.rs delete mode 100644 tokio-buf/src/util/from.rs delete mode 100644 tokio-buf/src/util/iter.rs delete mode 100644 tokio-buf/src/util/limit.rs delete mode 100644 tokio-buf/src/util/mod.rs delete mode 100644 tokio-buf/src/util/stream.rs delete mode 100644 tokio-buf/tests/buf_stream.rs delete mode 100644 tokio-buf/tests/chain.rs delete mode 100644 tokio-buf/tests/collect.rs delete mode 100644 tokio-buf/tests/iter.rs delete mode 100644 tokio-buf/tests/limit.rs delete mode 100644 tokio-buf/tests/size_hint.rs delete mode 100644 tokio-buf/tests/stream.rs delete mode 100644 tokio-buf/tests/string.rs delete mode 100644 tokio-buf/tests/support.rs diff --git a/Cargo.toml b/Cargo.toml index 156b481c6..05e3de22c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,6 @@ members = [ "tokio", - "tokio-buf", "tokio-codec", "tokio-executor", "tokio-fs", diff --git a/azure-pipelines.yml b/azure-pipelines.yml index 8c0b01a73..e8cbc84f3 100644 --- a/azure-pipelines.yml +++ b/azure-pipelines.yml @@ -61,7 +61,6 @@ jobs: displayName: Test sub crates - rust: $(nightly) crates: - tokio-buf: [] tokio-codec: [] tokio-executor: - current-thread diff --git a/ci/patch.toml b/ci/patch.toml index 710db68ba..e615fab9b 100644 --- a/ci/patch.toml +++ b/ci/patch.toml @@ -2,7 +2,6 @@ # repository. [patch.crates-io] tokio = { path = "tokio" } -tokio-buf = { path = "tokio-buf" } tokio-codec = { path = "tokio-codec" } tokio-executor = { path = "tokio-executor" } tokio-fs = { path = "tokio-fs" } diff --git a/tokio-buf/CHANGELOG.md b/tokio-buf/CHANGELOG.md deleted file mode 100644 index 7f8fe5a2b..000000000 --- a/tokio-buf/CHANGELOG.md +++ /dev/null @@ -1,19 +0,0 @@ -# 0.2.0-alpha.1 (August 8, 2019) - -### Changed -- Switch to `async`, `await`, and `std::future`. - -# 0.1.1 (April 22, 2019) - -### Added -- Utilities for creating a `BufStream` from iterators and streams (#1011). -- Add `BufStream::into_stream` (#1048). -- Implement `FromBufStream` for `Bytes` (#1009). -- Implement `Error` for `CollectVecError` (#1010). - -### Fixed -- Implement `size_hint` for string types (#1012). - -# 0.1.0 (February 23, 2019) - -* Initial release diff --git a/tokio-buf/Cargo.toml b/tokio-buf/Cargo.toml deleted file mode 100644 index 1327e1d8a..000000000 --- a/tokio-buf/Cargo.toml +++ /dev/null @@ -1,34 +0,0 @@ -[package] -name = "tokio-buf" -# When releasing to crates.io: -# - Remove path dependencies -# - Update html_root_url. -# - Update doc url -# - Cargo.toml -# - Update CHANGELOG.md. -# - Create "v0.2.x" git tag. -version = "0.2.0-alpha.1" -edition = "2018" -authors = ["Tokio Contributors "] -license = "MIT" -repository = "https://github.com/tokio-rs/tokio" -homepage = "https://tokio.rs" -documentation = "https://docs.rs/tokio-buf/0.2.0-alpha.1/tokio_buf" -description = """ -Asynchronous stream of byte buffers -""" -categories = ["asynchronous"] - -[dependencies] -bytes = "0.4.10" -either = { version = "1.5", optional = true} - -[features] -# default = ["util"] -# util = ["bytes/either", "either"] - -[dev-dependencies] -tokio-mock-task = "0.1.1" - -[package.metadata.docs.rs] -all-features = true diff --git a/tokio-buf/LICENSE b/tokio-buf/LICENSE deleted file mode 100644 index cdb28b4b5..000000000 --- a/tokio-buf/LICENSE +++ /dev/null @@ -1,25 +0,0 @@ -Copyright (c) 2019 Tokio Contributors - -Permission is hereby granted, free of charge, to any -person obtaining a copy of this software and associated -documentation files (the "Software"), to deal in the -Software without restriction, including without -limitation the rights to use, copy, modify, merge, -publish, distribute, sublicense, and/or sell copies of -the Software, and to permit persons to whom the Software -is furnished to do so, subject to the following -conditions: - -The above copyright notice and this permission notice -shall be included in all copies or substantial portions -of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF -ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED -TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A -PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT -SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY -CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION -OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR -IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -DEALINGS IN THE SOFTWARE. diff --git a/tokio-buf/README.md b/tokio-buf/README.md deleted file mode 100644 index 787033a9d..000000000 --- a/tokio-buf/README.md +++ /dev/null @@ -1,13 +0,0 @@ -# tokio-buf - -Asynchronous stream of byte buffers - -## 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/lib.rs b/tokio-buf/src/lib.rs deleted file mode 100644 index f9adf8a6b..000000000 --- a/tokio-buf/src/lib.rs +++ /dev/null @@ -1,102 +0,0 @@ -#![doc(html_root_url = "https://docs.rs/tokio-buf/0.2.0-alpha.1")] -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - unreachable_pub -)] -#![deny(intra_doc_link_resolution_failure)] -#![doc(test( - no_crate_inject, - attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables)) -))] - -//! Asynchronous stream of bytes. -//! -//! This crate contains the `BufStream` trait and a number of combinators for -//! this trait. The trait is similar to `Stream` in the `futures` library, but -//! instead of yielding arbitrary values, it only yields types that implement -//! `Buf` (i.e, byte collections). - -// mod never; -mod size_hint; -// mod str; -// mod u8; -// #[cfg(feature = "util")] -// pub mod util; - -pub use self::size_hint::SizeHint; -// #[doc(inline)] -// #[cfg(feature = "util")] -// pub use crate::util::BufStreamExt; - -use bytes::Buf; -use std::task::{Context, 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: - /// - /// - `Poll::Pending` 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. - /// - /// - `Poll::Ready(Some(Ok(buf)))` means that the stream has successfully - /// produced a value, `buf`, and may produce further values on subsequent - /// `poll_buf` calls. - /// - /// - `Poll::Ready(None)` means that the stream has terminated, and - /// `poll_buf` should not be invoked again. - /// - /// # Panics - /// - /// Once a stream is finished, i.e. `Poll::Ready(None)` has been returned, - /// further calls to `poll_buf` may result in a panic or other "bad - /// behavior". - fn poll_buf(&mut self, cx: &mut Context<'_>) -> Poll>>; - - /// 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() - } -} diff --git a/tokio-buf/src/never.rs b/tokio-buf/src/never.rs deleted file mode 100644 index 370701be8..000000000 --- a/tokio-buf/src/never.rs +++ /dev/null @@ -1,18 +0,0 @@ -use std::{error, fmt}; - -/// An error that can never occur -pub enum Never {} - -impl fmt::Debug for Never { - fn fmt(&self, _f: &mut fmt::Formatter<'_>) -> fmt::Result { - match *self {} - } -} - -impl fmt::Display for Never { - fn fmt(&self, _f: &mut fmt::Formatter<'_>) -> fmt::Result { - match *self {} - } -} - -impl error::Error for Never {} diff --git a/tokio-buf/src/size_hint.rs b/tokio-buf/src/size_hint.rs deleted file mode 100644 index 8518f427e..000000000 --- a/tokio-buf/src/size_hint.rs +++ /dev/null @@ -1,56 +0,0 @@ -use std::u64; - -/// A `BufStream` size hint -/// -/// The default implementation returns: -/// -/// * 0 for `available` -/// * 0 for `lower` -/// * `None` for `upper`. -#[derive(Debug, Default, Clone)] -pub struct SizeHint { - lower: u64, - upper: Option, -} - -impl SizeHint { - /// Returns a new `SizeHint` with default values - pub fn new() -> SizeHint { - SizeHint::default() - } - - /// Returns the lower bound of data that the `BufStream` will yield before - /// completing. - pub fn lower(&self) -> u64 { - self.lower - } - - /// Set the value of the `lower` hint. - /// - /// # Panics - /// - /// The function panics if `value` is greater than `upper`. - pub fn set_lower(&mut self, value: u64) { - assert!(value <= self.upper.unwrap_or(u64::MAX)); - self.lower = value; - } - - /// Returns the upper bound of data the `BufStream` will yield before - /// completing, or `None` if the value is unknown. - pub fn upper(&self) -> Option { - self.upper - } - - /// Set the value of the `upper` hint value. - /// - /// # Panics - /// - /// This function panics if `value` is less than `lower`. - pub fn set_upper(&mut self, value: u64) { - // There is no need to check `available` as that is guaranteed to be - // less than or equal to `lower`. - assert!(value >= self.lower, "`value` is less than than `lower`"); - - self.upper = Some(value); - } -} diff --git a/tokio-buf/src/str.rs b/tokio-buf/src/str.rs deleted file mode 100644 index 05327be87..000000000 --- a/tokio-buf/src/str.rs +++ /dev/null @@ -1,55 +0,0 @@ -use crate::never::Never; -use crate::BufStream; -use crate::SizeHint; - -use futures::Poll; - -use std::io; -use std::mem; - -impl BufStream for String { - type Item = io::Cursor>; - type Error = Never; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - if self.is_empty() { - return Ok(None.into()); - } - - let bytes = mem::replace(self, Default::default()).into_bytes(); - let buf = io::Cursor::new(bytes); - - Ok(Some(buf).into()) - } - - fn size_hint(&self) -> SizeHint { - size_hint(&self[..]) - } -} - -impl BufStream for &'static str { - 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()); - } - - let bytes = mem::replace(self, Default::default()).as_bytes(); - let buf = io::Cursor::new(bytes); - - Ok(Some(buf).into()) - } - - fn size_hint(&self) -> SizeHint { - size_hint(&self[..]) - } -} - -fn size_hint(s: &str) -> SizeHint { - let mut hint = SizeHint::new(); - hint.set_lower(s.len() as u64); - hint.set_upper(s.len() as u64); - hint -} diff --git a/tokio-buf/src/u8.rs b/tokio-buf/src/u8.rs deleted file mode 100644 index 5baf9d8ef..000000000 --- a/tokio-buf/src/u8.rs +++ /dev/null @@ -1,66 +0,0 @@ -use crate::never::Never; -use crate::BufStream; -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/util/chain.rs b/tokio-buf/src/util/chain.rs deleted file mode 100644 index 58e17ca50..000000000 --- a/tokio-buf/src/util/chain.rs +++ /dev/null @@ -1,46 +0,0 @@ -use crate::BufStream; - -use either::Either; -use futures::{try_ready, Poll}; - -/// A buf stream that sequences two buf streams together. -/// -/// `Chain` values are produced by the `chain` function on `BufStream`. -#[derive(Debug)] -pub struct Chain { - left: Option, - right: U, -} - -impl Chain { - pub(crate) fn new(left: T, right: U) -> Chain { - Chain { - left: Some(left), - right, - } - } -} - -impl BufStream for Chain -where - T: BufStream, - U: BufStream, -{ - type Item = Either; - type Error = T::Error; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - if let Some(ref mut stream) = self.left { - let res = try_ready!(stream.poll_buf()); - - if res.is_some() { - return Ok(res.map(Either::Left).into()); - } - } - - self.left = None; - - let res = try_ready!(self.right.poll_buf()); - Ok(res.map(Either::Right).into()) - } -} diff --git a/tokio-buf/src/util/collect.rs b/tokio-buf/src/util/collect.rs deleted file mode 100644 index 5ed0462c0..000000000 --- a/tokio-buf/src/util/collect.rs +++ /dev/null @@ -1,101 +0,0 @@ -use super::FromBufStream; -use crate::BufStream; - -use futures::{try_ready, Future, Poll}; - -/// Consumes a buf stream, collecting the data into a single byte container. -/// -/// `Collect` values are produced by `BufStream::collect`. -#[derive(Debug)] -pub struct Collect -where - T: BufStream, - U: FromBufStream, -{ - stream: T, - builder: Option, -} - -/// Errors returned from `Collect` future. -#[derive(Debug)] -pub struct CollectError { - inner: Error, -} - -#[derive(Debug)] -enum Error { - Stream(T), - Collect(U), -} - -impl Collect -where - T: BufStream, - U: FromBufStream, -{ - pub(crate) fn new(stream: T) -> Collect { - let builder = U::builder(&stream.size_hint()); - - Collect { - stream, - builder: Some(builder), - } - } -} - -impl Future for Collect -where - T: BufStream, - U: FromBufStream, -{ - type Item = U; - type Error = CollectError; - - fn poll(&mut self) -> Poll { - loop { - let res = self.stream.poll_buf().map_err(|err| { - let inner = Error::Stream(err); - CollectError { inner } - }); - - match try_ready!(res) { - Some(mut buf) => { - let builder = self.builder.as_mut().expect("cannot poll after done"); - - U::extend(builder, &mut buf, &self.stream.size_hint()).map_err(|err| { - let inner = Error::Collect(err); - CollectError { inner } - })?; - } - None => { - let builder = self.builder.take().expect("cannot poll after done"); - let value = U::build(builder).map_err(|err| { - let inner = Error::Collect(err); - CollectError { inner } - })?; - return Ok(value.into()); - } - } - } - } -} - -// ===== impl CollectError ===== - -impl CollectError { - /// Returns `true` if the error was caused by polling the stream. - pub fn is_stream_err(&self) -> bool { - match self.inner { - Error::Stream(_) => true, - _ => false, - } - } - - /// Returns `true` if the error happened while collecting the data. - pub fn is_collect_err(&self) -> bool { - match self.inner { - Error::Collect(_) => true, - _ => false, - } - } -} diff --git a/tokio-buf/src/util/from.rs b/tokio-buf/src/util/from.rs deleted file mode 100644 index 5b2aa1420..000000000 --- a/tokio-buf/src/util/from.rs +++ /dev/null @@ -1,154 +0,0 @@ -use crate::SizeHint; - -use bytes::{Buf, BufMut, Bytes}; - -use std::error::Error; -use std::fmt; -use std::usize; - -/// Conversion from a `BufStream`. -/// -/// By implementing `FromBufStream` for a type, you define how it will be -/// created from a buf stream. This is common for types which describe byte -/// storage of some kind. -/// -/// `FromBufStream` is rarely called explicitly, and it is instead used through -/// `BufStream`'s `collect` method. -pub trait FromBufStream: Sized { - /// Type that is used to build `Self` while the `BufStream` is being - /// consumed. - type Builder; - - /// Error that might happen on conversion. - type Error; - - /// Create a new, empty, builder. The provided `hint` can be used to inform - /// reserving capacity. - fn builder(hint: &SizeHint) -> Self::Builder; - - /// Extend the builder with the `Buf`. - /// - /// This method is called whenever a new `Buf` value is obtained from the - /// buf stream. - /// - /// The provided size hint represents the state of the stream **after** - /// `buf` has been yielded. The lower bound represents the minimum amount of - /// data that will be provided after this call to `extend` returns. - fn extend(builder: &mut Self::Builder, buf: &mut T, hint: &SizeHint) - -> Result<(), Self::Error>; - - /// Finalize the building of `Self`. - /// - /// Called once the buf stream is fully consumed. - fn build(builder: Self::Builder) -> Result; -} - -/// Error returned from collecting into a `Vec` -#[derive(Debug)] -pub struct CollectVecError { - _p: (), -} - -/// Error returned from collecting into a `Bytes` -#[derive(Debug)] -pub struct CollectBytesError { - _p: (), -} - -impl FromBufStream for Vec { - type Builder = Vec; - type Error = CollectVecError; - - fn builder(hint: &SizeHint) -> Vec { - Vec::with_capacity(hint.lower() as usize) - } - - fn extend(builder: &mut Self, buf: &mut T, hint: &SizeHint) -> Result<(), Self::Error> { - let lower = hint.lower(); - - // If the lower bound is greater than `usize::MAX` then we have a - // problem - if lower > usize::MAX as u64 { - return Err(CollectVecError { _p: () }); - } - - let mut reserve = lower as usize; - - // If `upper` is set, use this value if it is less than or equal to 64. - // This only really impacts the first iteration. - match hint.upper() { - Some(upper) if upper <= 64 => { - reserve = upper as usize; - } - _ => {} - } - - // hint.lower() represents the minimum amount of data that will be - // received *after* this function call. We reserve this amount on top of - // the amount of data in `buf`. - reserve = match reserve.checked_add(buf.remaining()) { - Some(n) => n, - None => return Err(CollectVecError { _p: () }), - }; - - // Always reserve 64 bytes the first time, unless `upper` is set and is - // less than 64. - if builder.is_empty() { - reserve = reserve.max(match hint.upper() { - Some(upper) if upper < 64 => upper as usize, - _ => 64, - }); - } - - // Make sure overflow won't happen when reserving - if reserve.checked_add(builder.len()).is_none() { - return Err(CollectVecError { _p: () }); - } - - // Reserve space - builder.reserve(reserve); - - // Copy the data - builder.put(buf); - - Ok(()) - } - - fn build(builder: Self) -> Result { - Ok(builder) - } -} - -impl FromBufStream for Bytes { - type Builder = Vec; - type Error = CollectBytesError; - - fn builder(hint: &SizeHint) -> Vec { - as FromBufStream>::builder(hint) - } - - fn extend(builder: &mut Vec, buf: &mut T, hint: &SizeHint) -> Result<(), Self::Error> { - as FromBufStream>::extend(builder, buf, hint) - .map_err(|_| CollectBytesError { _p: () }) - } - - fn build(builder: Vec) -> Result { - Ok(builder.into()) - } -} - -impl fmt::Display for CollectVecError { - fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(fmt, "BufStream is too big") - } -} - -impl Error for CollectVecError {} - -impl fmt::Display for CollectBytesError { - fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(fmt, "BufStream too big") - } -} - -impl Error for CollectBytesError {} diff --git a/tokio-buf/src/util/iter.rs b/tokio-buf/src/util/iter.rs deleted file mode 100644 index 3df9be369..000000000 --- a/tokio-buf/src/util/iter.rs +++ /dev/null @@ -1,50 +0,0 @@ -use crate::BufStream; -use bytes::Buf; -use futures::Poll; -use std::error::Error; -use std::fmt; - -/// Converts an `Iterator` into a `BufStream` which is always ready to yield the -/// next value. -/// -/// Iterators in Rust don't express the ability to block, so this adapter -/// simply always calls `iter.next()` and returns that. -pub fn iter(i: I) -> Iter -where - I: IntoIterator, - I::Item: Buf, -{ - Iter { - iter: i.into_iter(), - } -} - -/// `BufStream` returned by the [`iter`] function. -#[derive(Debug)] -pub struct Iter { - iter: I, -} - -#[derive(Debug)] -pub enum Never {} - -impl BufStream for Iter -where - I: Iterator, - I::Item: Buf, -{ - type Item = I::Item; - type Error = Never; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - Ok(self.iter.next().into()) - } -} - -impl fmt::Display for Never { - fn fmt(&self, _: &mut fmt::Formatter<'_>) -> fmt::Result { - unreachable!(); - } -} - -impl Error for Never {} diff --git a/tokio-buf/src/util/limit.rs b/tokio-buf/src/util/limit.rs deleted file mode 100644 index 03c0d9138..000000000 --- a/tokio-buf/src/util/limit.rs +++ /dev/null @@ -1,76 +0,0 @@ -use crate::BufStream; - -use bytes::Buf; -use futures::Poll; - -/// Limits the stream to a maximum amount of data. -#[derive(Debug)] -pub struct Limit { - stream: T, - remaining: u64, -} - -/// Errors returned from `Limit`. -#[derive(Debug)] -pub struct LimitError { - /// When `None`, limit was reached - inner: Option, -} - -impl Limit { - pub(crate) fn new(stream: T, amount: u64) -> Limit { - Limit { - stream, - remaining: amount, - } - } -} - -impl BufStream for Limit -where - T: BufStream, -{ - type Item = T::Item; - type Error = LimitError; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - use futures::Async::Ready; - - if self.stream.size_hint().lower() > self.remaining { - return Err(LimitError { inner: None }); - } - - let res = self - .stream - .poll_buf() - .map_err(|err| LimitError { inner: Some(err) }); - - match res { - Ok(Ready(Some(ref buf))) => { - if buf.remaining() as u64 > self.remaining { - self.remaining = 0; - return Err(LimitError { inner: None }); - } - - self.remaining -= buf.remaining() as u64; - } - _ => {} - } - - res - } -} - -// ===== impl LimitError ===== - -impl LimitError { - /// Returns `true` if the error was caused by polling the stream. - pub fn is_stream_err(&self) -> bool { - self.inner.is_some() - } - - /// Returns `true` if the stream reached its limit. - pub fn is_limit_err(&self) -> bool { - self.inner.is_none() - } -} diff --git a/tokio-buf/src/util/mod.rs b/tokio-buf/src/util/mod.rs deleted file mode 100644 index d6848e885..000000000 --- a/tokio-buf/src/util/mod.rs +++ /dev/null @@ -1,87 +0,0 @@ -//! Types and utilities for working with `BufStream`. - -mod chain; -mod collect; -mod from; -mod iter; -mod limit; -mod stream; - -pub use self::chain::Chain; -pub use self::collect::Collect; -pub use self::from::FromBufStream; -pub use self::iter::iter; -pub use self::limit::Limit; -pub use self::stream::{stream, IntoStream}; - -pub mod error { - //! Error types - - pub use super::collect::CollectError; - pub use super::from::{CollectBytesError, CollectVecError}; - pub use super::limit::LimitError; -} - -use crate::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) - } - - /// Creates a `Stream` from a `BufStream`. - /// - /// This produces a `Stream` of `BufStream::Items`. - fn into_stream(self) -> IntoStream - where - Self: Sized, - { - IntoStream::new(self) - } -} diff --git a/tokio-buf/src/util/stream.rs b/tokio-buf/src/util/stream.rs deleted file mode 100644 index 25af75b38..000000000 --- a/tokio-buf/src/util/stream.rs +++ /dev/null @@ -1,76 +0,0 @@ -use crate::BufStream; -use bytes::Buf; -use futures::{Async, Poll, Stream}; - -/// Converts a `Stream` of `Buf` types into a `BufStream`. -/// -/// While `Stream` and `BufStream` are very similar, they are not identical. The -/// `stream` function returns a `BufStream` that is backed by the provided -/// `Stream` type. -pub fn stream(stream: T) -> FromStream -where - T: Stream, - T::Item: Buf, -{ - FromStream { stream } -} - -/// `BufStream` returned by the [`stream`] function. -#[derive(Debug)] -pub struct FromStream { - stream: T, -} - -impl BufStream for FromStream -where - T: Stream, - T::Item: Buf, -{ - type Item = T::Item; - type Error = T::Error; - - fn poll_buf(&mut self) -> Poll, Self::Error> { - self.stream.poll() - } -} - -/// Converts a `BufStream` into a `Stream`. -#[derive(Debug)] -pub struct IntoStream { - buf: T, -} - -impl IntoStream { - /// Create a new `Stream` from the provided `BufStream`. - pub fn new(buf: T) -> Self { - IntoStream { buf } - } - - /// Get a reference to the inner `BufStream`. - pub fn get_ref(&self) -> &T { - &self.buf - } - - /// Get a mutable reference to the inner `BufStream` - pub fn get_mut(&mut self) -> &mut T { - &mut self.buf - } - - /// Get the inner `BufStream`. - pub fn into_inner(self) -> T { - self.buf - } -} - -impl Stream for IntoStream { - type Item = T::Item; - type Error = T::Error; - - fn poll(&mut self) -> Poll, Self::Error> { - match self.buf.poll_buf()? { - Async::Ready(Some(buf)) => Ok(Async::Ready(Some(buf))), - Async::Ready(None) => Ok(Async::Ready(None)), - Async::NotReady => Ok(Async::NotReady), - } - } -} diff --git a/tokio-buf/tests/buf_stream.rs b/tokio-buf/tests/buf_stream.rs deleted file mode 100644 index 9f3679b2e..000000000 --- a/tokio-buf/tests/buf_stream.rs +++ /dev/null @@ -1,8 +0,0 @@ -use tokio_buf::BufStream; - -#[test] -fn object_safe() { - // Ensures that `BufStream` can be a trait object - #[allow(dead_code)] - fn obj(_: &mut dyn BufStream) {} -} diff --git a/tokio-buf/tests/chain.rs b/tokio-buf/tests/chain.rs deleted file mode 100644 index 153061b06..000000000 --- a/tokio-buf/tests/chain.rs +++ /dev/null @@ -1,38 +0,0 @@ -#![cfg(feature = "broken")] -#![cfg(feature = "util")] - -use futures::Async::*; -use tokio_buf::{BufStream, BufStreamExt}; - -mod support; -use support::*; - -#[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()); -} diff --git a/tokio-buf/tests/collect.rs b/tokio-buf/tests/collect.rs deleted file mode 100644 index c57bfde2f..000000000 --- a/tokio-buf/tests/collect.rs +++ /dev/null @@ -1,63 +0,0 @@ -#![cfg(feature = "broken")] -#![cfg(feature = "util")] - -use bytes::Bytes; -use futures::Future; -use tokio_buf::BufStreamExt; - -mod support; -use support::*; - -macro_rules! test_collect_impl { - ($t:ty $(, $capacity:ident)*) => { - // While unfortunate, this test makes some assumptions on vec's resizing - // behavior. - // - // Collect one - // - let bs = one("hello world"); - - let vec: $t = 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: $t = 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: $t = 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: $t = bs.collect().wait().unwrap(); - - assert_eq!(vec, &b"hello world, one two three"[..]); - } -} - -#[test] -fn collect_vec() { - test_collect_impl!(Vec, capacity); -} - -#[test] -fn collect_bytes() { - test_collect_impl!(Bytes); -} diff --git a/tokio-buf/tests/iter.rs b/tokio-buf/tests/iter.rs deleted file mode 100644 index d237a71b1..000000000 --- a/tokio-buf/tests/iter.rs +++ /dev/null @@ -1,29 +0,0 @@ -#![cfg(feature = "broken")] -use futures::Async::*; -use std::io::Cursor; -use tokio_buf::{util, BufStream}; - -mod support; - -type Buf = Cursor<&'static [u8]>; - -#[test] -fn empty_iter() { - let mut bs = util::iter(Vec::::new()); - assert_none!(bs.poll_buf()); -} - -#[test] -fn full_iter() { - let bufs = vec![buf(b"one"), buf(b"two"), buf(b"three")]; - - let mut bs = util::iter(bufs); - assert_buf_eq!(bs.poll_buf(), "one"); - assert_buf_eq!(bs.poll_buf(), "two"); - assert_buf_eq!(bs.poll_buf(), "three"); - assert_none!(bs.poll_buf()); -} - -fn buf(data: &'static [u8]) -> Buf { - Cursor::new(data) -} diff --git a/tokio-buf/tests/limit.rs b/tokio-buf/tests/limit.rs deleted file mode 100644 index 2bce76b60..000000000 --- a/tokio-buf/tests/limit.rs +++ /dev/null @@ -1,60 +0,0 @@ -#![cfg(feature = "broken")] -#![cfg(feature = "util")] - -use futures::Async::*; -use futures::Future; -use tokio_buf::{BufStream, BufStreamExt}; - -mod support; -use support::*; - -#[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/size_hint.rs b/tokio-buf/tests/size_hint.rs deleted file mode 100644 index 467a4a2d1..000000000 --- a/tokio-buf/tests/size_hint.rs +++ /dev/null @@ -1,40 +0,0 @@ -use tokio_buf::SizeHint; - -#[test] -fn size_hint() { - let hint = SizeHint::new(); - assert_eq!(hint.lower(), 0); - assert!(hint.upper().is_none()); - - let mut hint = SizeHint::new(); - hint.set_lower(100); - assert_eq!(hint.lower(), 100); - assert!(hint.upper().is_none()); - - let mut hint = SizeHint::new(); - hint.set_upper(200); - assert_eq!(hint.lower(), 0); - assert_eq!(hint.upper(), Some(200)); - - let mut hint = SizeHint::new(); - hint.set_lower(100); - hint.set_upper(100); - assert_eq!(hint.lower(), 100); - assert_eq!(hint.upper(), Some(100)); -} - -#[test] -#[should_panic] -fn size_hint_lower_bigger_than_upper() { - let mut hint = SizeHint::new(); - hint.set_upper(100); - hint.set_lower(200); -} - -#[test] -#[should_panic] -fn size_hint_upper_less_than_lower() { - let mut hint = SizeHint::new(); - hint.set_lower(200); - hint.set_upper(100); -} diff --git a/tokio-buf/tests/stream.rs b/tokio-buf/tests/stream.rs deleted file mode 100644 index 06f2ee275..000000000 --- a/tokio-buf/tests/stream.rs +++ /dev/null @@ -1,44 +0,0 @@ -#![cfg(feature = "broken")] -use futures::sync::mpsc; -use futures::Async::*; -use std::io::Cursor; -use tokio_buf::{util, BufStream}; -use tokio_mock_task::MockTask; - -mod support; - -type Buf = Cursor<&'static [u8]>; - -#[test] -fn empty_stream() { - let (_, rx) = mpsc::unbounded::(); - let mut bs = util::stream(rx); - assert_none!(bs.poll_buf()); -} - -#[test] -fn full_stream() { - let (tx, rx) = mpsc::unbounded(); - let mut bs = util::stream(rx); - let mut task = MockTask::new(); - - tx.unbounded_send(buf(b"one")).unwrap(); - - assert_buf_eq!(bs.poll_buf(), "one"); - task.enter(|| assert_not_ready!(bs.poll_buf())); - - tx.unbounded_send(buf(b"two")).unwrap(); - - assert!(task.is_notified()); - assert_buf_eq!(bs.poll_buf(), "two"); - task.enter(|| assert_not_ready!(bs.poll_buf())); - - drop(tx); - - assert!(task.is_notified()); - assert_none!(bs.poll_buf()); -} - -fn buf(data: &'static [u8]) -> Buf { - Cursor::new(data) -} diff --git a/tokio-buf/tests/string.rs b/tokio-buf/tests/string.rs deleted file mode 100644 index 995f6fe8a..000000000 --- a/tokio-buf/tests/string.rs +++ /dev/null @@ -1,34 +0,0 @@ -#![cfg(feature = "broken")] -use futures::Async::*; -use std::fmt; -use tokio_buf::BufStream; - -mod support; - -fn test_hello_world(mut bs: B) -where - B: BufStream + fmt::Debug, - B::Item: fmt::Debug, - B::Error: fmt::Debug, -{ - let hint = bs.size_hint(); - assert_eq!(hint.lower(), 11); - assert_eq!(hint.upper(), Some(11)); - - assert_buf_eq!(bs.poll_buf(), "hello world"); - - let hint = bs.size_hint(); - assert_eq!(hint.lower(), 0); - assert_eq!(hint.upper(), Some(0)); - assert_none!(bs.poll_buf()); -} - -#[test] -fn string() { - test_hello_world("hello world".to_string()); -} - -#[test] -fn str() { - test_hello_world("hello world"); -} diff --git a/tokio-buf/tests/support.rs b/tokio-buf/tests/support.rs deleted file mode 100644 index 9a79489f0..000000000 --- a/tokio-buf/tests/support.rs +++ /dev/null @@ -1,133 +0,0 @@ -#![cfg(feature = "broken")] -#![allow(unused)] - -use bytes::Buf; -use futures::Async::*; -use futures::Poll; -use tokio_buf::{BufStream, SizeHint}; - -use std::collections::VecDeque; -use std::io::Cursor; - -#[macro_export] -macro_rules! assert_buf_eq { - ($actual:expr, $expect:expr) => {{ - use bytes::Buf; - 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_export] -macro_rules! assert_none { - ($actual:expr) => { - match $actual { - Ok(Ready(None)) => {} - actual => panic!("expected None; actual = {:?}", actual), - } - }; -} - -#[macro_export] -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) - } -}