tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-07 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore: rm tokio-buf (#1574)

Browse Source 
The crate has not been updated and it does not seem like it is a good
path forward.
This commit is contained in:
Carl Lerche
2019-09-19 12:11:21 -07:00
committed by GitHub
parent 34e388619f
commit 815173f8e5
28 changed files with 0 additions and 1430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
-1
View File
@@ -2,7 +2,6 @@
members = [
"tokio",
"tokio-buf",
"tokio-codec",
"tokio-executor",
"tokio-fs",
azure-pipelines.yml
-1
View File
@@ -61,7 +61,6 @@ jobs:
displayName: Test sub crates -
rust: $(nightly)
crates:
tokio-buf: []
tokio-codec: []
tokio-executor:
- current-thread
ci/patch.toml
-1
View File
@@ -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" }
tokio-buf/CHANGELOG.md
-19
View File
@@ -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
tokio-buf/Cargo.toml
-34
View File
@@ -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 <[email protected]>"]
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
tokio-buf/LICENSE
-25
View File
@@ -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.
tokio-buf/README.md
-13
View File
@@ -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.
tokio-buf/src/lib.rs
-102
View File
@@ -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<Option<Result<Self::Item, 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()
}
}
tokio-buf/src/never.rs
-18
View File
@@ -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 {}
tokio-buf/src/size_hint.rs
-56
View File
@@ -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<u64>,
}
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<u64> {
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);
}
}
tokio-buf/src/str.rs
-55
View File
@@ -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<Vec<u8>>;
type Error = Never;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, 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<Option<Self::Item>, 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
}
tokio-buf/src/u8.rs
-66
View File
@@ -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<u8> {
type Item = io::Cursor<Vec<u8>>;
type Error = Never;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, 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<Option<Self::Item>, Self::Error> {
if self.is_empty() {
return Ok(None.into());
}
poll_bytes(self)
}
}
impl BufStream for Bytes {
type Item = io::Cursor<Bytes>;
type Error = Never;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
if self.is_empty() {
return Ok(None.into());
}
poll_bytes(self)
}
}
impl BufStream for BytesMut {
type Item = io::Cursor<BytesMut>;
type Error = Never;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
if self.is_empty() {
return Ok(None.into());
}
poll_bytes(self)
}
}
fn poll_bytes<T: Default>(buf: &mut T) -> Poll<Option<io::Cursor<T>>, Never> {
use std::mem;
let bytes = mem::replace(buf, Default::default());
let buf = io::Cursor::new(bytes);
Ok(Some(buf).into())
}
tokio-buf/src/util/chain.rs
-46
View File
@@ -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<T, U> {
left: Option<T>,
right: U,
}
impl<T, U> Chain<T, U> {
pub(crate) fn new(left: T, right: U) -> Chain<T, U> {
Chain {
left: Some(left),
right,
}
}
}
impl<T, U> BufStream for Chain<T, U>
where
T: BufStream,
U: BufStream<Error = T::Error>,
{
type Item = Either<T::Item, U::Item>;
type Error = T::Error;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, 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())
}
}
tokio-buf/src/util/collect.rs
-101
View File
@@ -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<T, U>
where
T: BufStream,
U: FromBufStream<T::Item>,
{
stream: T,
builder: Option<U::Builder>,
}
/// Errors returned from `Collect` future.
#[derive(Debug)]
pub struct CollectError<T, U> {
inner: Error<T, U>,
}
#[derive(Debug)]
enum Error<T, U> {
Stream(T),
Collect(U),
}
impl<T, U> Collect<T, U>
where
T: BufStream,
U: FromBufStream<T::Item>,
{
pub(crate) fn new(stream: T) -> Collect<T, U> {
let builder = U::builder(&stream.size_hint());
Collect {
stream,
builder: Some(builder),
}
}
}
impl<T, U> Future for Collect<T, U>
where
T: BufStream,
U: FromBufStream<T::Item>,
{
type Item = U;
type Error = CollectError<T::Error, U::Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
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<T, U> CollectError<T, U> {
/// 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,
}
}
}
tokio-buf/src/util/from.rs
-154
View File
@@ -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<T: Buf>: 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<Self, Self::Error>;
}
/// Error returned from collecting into a `Vec<u8>`
#[derive(Debug)]
pub struct CollectVecError {
_p: (),
}
/// Error returned from collecting into a `Bytes`
#[derive(Debug)]
pub struct CollectBytesError {
_p: (),
}
impl<T: Buf> FromBufStream<T> for Vec<u8> {
type Builder = Vec<u8>;
type Error = CollectVecError;
fn builder(hint: &SizeHint) -> Vec<u8> {
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<Self, Self::Error> {
Ok(builder)
}
}
impl<T: Buf> FromBufStream<T> for Bytes {
type Builder = Vec<u8>;
type Error = CollectBytesError;
fn builder(hint: &SizeHint) -> Vec<u8> {
<Vec<u8> as FromBufStream<T>>::builder(hint)
}
fn extend(builder: &mut Vec<u8>, buf: &mut T, hint: &SizeHint) -> Result<(), Self::Error> {
<Vec<u8> as FromBufStream<T>>::extend(builder, buf, hint)
.map_err(|_| CollectBytesError { _p: () })
}
fn build(builder: Vec<u8>) -> Result<Self, Self::Error> {
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 {}
tokio-buf/src/util/iter.rs
-50
View File
@@ -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: I) -> Iter<I::IntoIter>
where
I: IntoIterator,
I::Item: Buf,
{
Iter {
iter: i.into_iter(),
}
}
/// `BufStream` returned by the [`iter`] function.
#[derive(Debug)]
pub struct Iter<I> {
iter: I,
}
#[derive(Debug)]
pub enum Never {}
impl<I> BufStream for Iter<I>
where
I: Iterator,
I::Item: Buf,
{
type Item = I::Item;
type Error = Never;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, 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 {}
tokio-buf/src/util/limit.rs
-76
View File
@@ -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<T> {
stream: T,
remaining: u64,
}
/// Errors returned from `Limit`.
#[derive(Debug)]
pub struct LimitError<T> {
/// When `None`, limit was reached
inner: Option<T>,
}
impl<T> Limit<T> {
pub(crate) fn new(stream: T, amount: u64) -> Limit<T> {
Limit {
stream,
remaining: amount,
}
}
}
impl<T> BufStream for Limit<T>
where
T: BufStream,
{
type Item = T::Item;
type Error = LimitError<T::Error>;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, 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<T> LimitError<T> {
/// 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()
}
}
tokio-buf/src/util/mod.rs
-87
View File
@@ -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<T> 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<T>(self, other: T) -> Chain<Self, T>
where
Self: Sized,
T: BufStream<Error = Self::Error>,
{
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<T>(self) -> Collect<Self, T>
where
Self: Sized,
T: FromBufStream<Self::Item>,
{
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<Self>
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<Self>
where
Self: Sized,
{
IntoStream::new(self)
}
}
tokio-buf/src/util/stream.rs
-76
View File
@@ -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<T>(stream: T) -> FromStream<T>
where
T: Stream,
T::Item: Buf,
{
FromStream { stream }
}
/// `BufStream` returned by the [`stream`] function.
#[derive(Debug)]
pub struct FromStream<T> {
stream: T,
}
impl<T> BufStream for FromStream<T>
where
T: Stream,
T::Item: Buf,
{
type Item = T::Item;
type Error = T::Error;
fn poll_buf(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
self.stream.poll()
}
}
/// Converts a `BufStream` into a `Stream`.
#[derive(Debug)]
pub struct IntoStream<T> {
buf: T,
}
impl<T> IntoStream<T> {
/// 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<T: BufStream> Stream for IntoStream<T> {
type Item = T::Item;
type Error = T::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, 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),
}
}
}
tokio-buf/tests/buf_stream.rs
-8
View File
@@ -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<Item = u32, Error = ()>) {}
}
tokio-buf/tests/chain.rs
-38
View File
@@ -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());
}
tokio-buf/tests/collect.rs
-63
View File
@@ -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<u8>, capacity);
}
#[test]
fn collect_bytes() {
test_collect_impl!(Bytes);
}
tokio-buf/tests/iter.rs
-29
View File
@@ -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::<Buf>::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)
}
tokio-buf/tests/limit.rs
-60
View File
@@ -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::<Vec<_>>()
.wait()
.unwrap();
assert_eq!(res, b"hello world");
let res = list(&["hello", " ", "world"])
.limit(100)
.collect::<Vec<_>>()
.wait()
.unwrap();
assert_eq!(res, b"hello world");
let res = list(&["hello", " ", "world"])
.limit(11)
.collect::<Vec<_>>()
.wait()
.unwrap();
assert_eq!(res, b"hello world");
// Limited
let res = one("hello world").limit(5).collect::<Vec<_>>().wait();
assert!(res.is_err());
let res = one("hello world").limit(10).collect::<Vec<_>>().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());
}
tokio-buf/tests/size_hint.rs
-40
View File
@@ -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);
}
tokio-buf/tests/stream.rs
-44
View File
@@ -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::<Buf>();
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)
}
tokio-buf/tests/string.rs
-34
View File
@@ -1,34 +0,0 @@
#![cfg(feature = "broken")]
use futures::Async::*;
use std::fmt;
use tokio_buf::BufStream;
mod support;
fn test_hello_world<B>(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");
}
tokio-buf/tests/support.rs
-133
View File
@@ -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<Poll<&'static [u8], ()>>,
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<Option<Self::Item>, 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)
}
}