Merge branch 'master' into add_stdin_lock
k-nasa
5 years ago
commit
35cb11e398
@ -0,0 +1,43 @@
|
||||
use std::pin::Pin;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_timer::Delay;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::future::Future;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
#[doc(hidden)]
|
||||
#[derive(Debug)]
|
||||
pub struct DelayFuture<F> {
|
||||
#[pin]
|
||||
future: F,
|
||||
#[pin]
|
||||
delay: Delay,
|
||||
}
|
||||
}
|
||||
|
||||
impl<F> DelayFuture<F> {
|
||||
pub fn new(future: F, dur: Duration) -> DelayFuture<F> {
|
||||
let delay = Delay::new(dur);
|
||||
|
||||
DelayFuture { future, delay }
|
||||
}
|
||||
}
|
||||
|
||||
impl<F: Future> Future for DelayFuture<F> {
|
||||
type Output = F::Output;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let this = self.project();
|
||||
|
||||
match this.delay.poll(cx) {
|
||||
Poll::Pending => Poll::Pending,
|
||||
Poll::Ready(_) => match this.future.poll(cx) {
|
||||
Poll::Ready(v) => Poll::Ready(v),
|
||||
Poll::Pending => Poll::Pending,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,66 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::prelude::*;
|
||||
use crate::stream::{Stream, Product};
|
||||
|
||||
impl<T, U> Product<Option<U>> for Option<T>
|
||||
where
|
||||
T: Product<U>,
|
||||
{
|
||||
#[doc = r#"
|
||||
Takes each element in the `Stream`: if it is a `None`, no further
|
||||
elements are taken, and the `None` is returned. Should no `None` occur,
|
||||
the product of all elements is returned.
|
||||
|
||||
# Examples
|
||||
|
||||
This multiplies every integer in a vector, rejecting the product if a negative element is
|
||||
encountered:
|
||||
|
||||
```
|
||||
# fn main() { async_std::task::block_on(async {
|
||||
#
|
||||
use std::collections::VecDeque;
|
||||
use async_std::prelude::*;
|
||||
|
||||
let v: VecDeque<_> = vec![1, 2, 4].into_iter().collect();
|
||||
let prod: Option<i32> = v.map(|x|
|
||||
if x < 0 {
|
||||
None
|
||||
} else {
|
||||
Some(x)
|
||||
}).product().await;
|
||||
assert_eq!(prod, Some(8));
|
||||
#
|
||||
# }) }
|
||||
```
|
||||
"#]
|
||||
fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
|
||||
where S: Stream<Item = Option<U>> + 'a
|
||||
{
|
||||
Box::pin(async move {
|
||||
pin_utils::pin_mut!(stream);
|
||||
|
||||
// Using `scan` here because it is able to stop the stream early
|
||||
// if a failure occurs
|
||||
let mut found_none = false;
|
||||
let out = <T as Product<U>>::product(stream
|
||||
.scan((), |_, elem| {
|
||||
match elem {
|
||||
Some(elem) => Some(elem),
|
||||
None => {
|
||||
found_none = true;
|
||||
// Stop processing the stream on error
|
||||
None
|
||||
}
|
||||
}
|
||||
})).await;
|
||||
|
||||
if found_none {
|
||||
None
|
||||
} else {
|
||||
Some(out)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::prelude::*;
|
||||
use crate::stream::{Stream, Sum};
|
||||
|
||||
impl<T, U> Sum<Option<U>> for Option<T>
|
||||
where
|
||||
T: Sum<U>,
|
||||
{
|
||||
#[doc = r#"
|
||||
Takes each element in the `Iterator`: if it is a `None`, no further
|
||||
elements are taken, and the `None` is returned. Should no `None` occur,
|
||||
the sum of all elements is returned.
|
||||
|
||||
# Examples
|
||||
|
||||
This sums up the position of the character 'a' in a vector of strings,
|
||||
if a word did not have the character 'a' the operation returns `None`:
|
||||
|
||||
```
|
||||
# fn main() { async_std::task::block_on(async {
|
||||
#
|
||||
use std::collections::VecDeque;
|
||||
use async_std::prelude::*;
|
||||
|
||||
let words: VecDeque<_> = vec!["have", "a", "great", "day"]
|
||||
.into_iter()
|
||||
.collect();
|
||||
let total: Option<usize> = words.map(|w| w.find('a')).sum().await;
|
||||
assert_eq!(total, Some(5));
|
||||
#
|
||||
# }) }
|
||||
```
|
||||
"#]
|
||||
fn sum<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
|
||||
where S: Stream<Item = Option<U>> + 'a
|
||||
{
|
||||
Box::pin(async move {
|
||||
pin_utils::pin_mut!(stream);
|
||||
|
||||
// Using `scan` here because it is able to stop the stream early
|
||||
// if a failure occurs
|
||||
let mut found_none = false;
|
||||
let out = <T as Sum<U>>::sum(stream
|
||||
.scan((), |_, elem| {
|
||||
match elem {
|
||||
Some(elem) => Some(elem),
|
||||
None => {
|
||||
found_none = true;
|
||||
// Stop processing the stream on error
|
||||
None
|
||||
}
|
||||
}
|
||||
})).await;
|
||||
|
||||
if found_none {
|
||||
None
|
||||
} else {
|
||||
Some(out)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,64 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::prelude::*;
|
||||
use crate::stream::{Stream, Product};
|
||||
|
||||
impl<T, U, E> Product<Result<U, E>> for Result<T, E>
|
||||
where
|
||||
T: Product<U>,
|
||||
{
|
||||
#[doc = r#"
|
||||
Takes each element in the `Stream`: if it is an `Err`, no further
|
||||
elements are taken, and the `Err` is returned. Should no `Err` occur,
|
||||
the product of all elements is returned.
|
||||
|
||||
# Examples
|
||||
|
||||
This multiplies every integer in a vector, rejecting the product if a negative element is
|
||||
encountered:
|
||||
|
||||
```
|
||||
# fn main() { async_std::task::block_on(async {
|
||||
#
|
||||
use std::collections::VecDeque;
|
||||
use async_std::prelude::*;
|
||||
|
||||
let v: VecDeque<_> = vec![1, 2, 4].into_iter().collect();
|
||||
let res: Result<i32, &'static str> = v.map(|x|
|
||||
if x < 0 {
|
||||
Err("Negative element found")
|
||||
} else {
|
||||
Ok(x)
|
||||
}).product().await;
|
||||
assert_eq!(res, Ok(8));
|
||||
#
|
||||
# }) }
|
||||
```
|
||||
"#]
|
||||
fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>
|
||||
where S: Stream<Item = Result<U, E>> + 'a
|
||||
{
|
||||
Box::pin(async move {
|
||||
pin_utils::pin_mut!(stream);
|
||||
|
||||
// Using `scan` here because it is able to stop the stream early
|
||||
// if a failure occurs
|
||||
let mut found_error = None;
|
||||
let out = <T as Product<U>>::product(stream
|
||||
.scan((), |_, elem| {
|
||||
match elem {
|
||||
Ok(elem) => Some(elem),
|
||||
Err(err) => {
|
||||
found_error = Some(err);
|
||||
// Stop processing the stream on error
|
||||
None
|
||||
}
|
||||
}
|
||||
})).await;
|
||||
match found_error {
|
||||
Some(err) => Err(err),
|
||||
None => Ok(out)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,64 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::prelude::*;
|
||||
use crate::stream::{Stream, Sum};
|
||||
|
||||
impl<T, U, E> Sum<Result<U, E>> for Result<T, E>
|
||||
where
|
||||
T: Sum<U>,
|
||||
{
|
||||
#[doc = r#"
|
||||
Takes each element in the `Stream`: if it is an `Err`, no further
|
||||
elements are taken, and the `Err` is returned. Should no `Err` occur,
|
||||
the sum of all elements is returned.
|
||||
|
||||
# Examples
|
||||
|
||||
This sums up every integer in a vector, rejecting the sum if a negative
|
||||
element is encountered:
|
||||
|
||||
```
|
||||
# fn main() { async_std::task::block_on(async {
|
||||
#
|
||||
use std::collections::VecDeque;
|
||||
use async_std::prelude::*;
|
||||
|
||||
let v: VecDeque<_> = vec![1, 2].into_iter().collect();
|
||||
let res: Result<i32, &'static str> = v.map(|x|
|
||||
if x < 0 {
|
||||
Err("Negative element found")
|
||||
} else {
|
||||
Ok(x)
|
||||
}).sum().await;
|
||||
assert_eq!(res, Ok(3));
|
||||
#
|
||||
# }) }
|
||||
```
|
||||
"#]
|
||||
fn sum<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>
|
||||
where S: Stream<Item = Result<U, E>> + 'a
|
||||
{
|
||||
Box::pin(async move {
|
||||
pin_utils::pin_mut!(stream);
|
||||
|
||||
// Using `scan` here because it is able to stop the stream early
|
||||
// if a failure occurs
|
||||
let mut found_error = None;
|
||||
let out = <T as Sum<U>>::sum(stream
|
||||
.scan((), |_, elem| {
|
||||
match elem {
|
||||
Ok(elem) => Some(elem),
|
||||
Err(err) => {
|
||||
found_error = Some(err);
|
||||
// Stop processing the stream on error
|
||||
None
|
||||
}
|
||||
}
|
||||
})).await;
|
||||
match found_error {
|
||||
Some(err) => Err(err),
|
||||
None => Ok(out)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
use pin_project_lite::pin_project;
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::prelude::*;
|
||||
use crate::stream::stream::map::Map;
|
||||
use crate::stream::{IntoStream, Stream};
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
/// This `struct` is created by the [`flat_map`] method on [`Stream`]. See its
|
||||
/// documentation for more.
|
||||
///
|
||||
/// [`flat_map`]: trait.Stream.html#method.flat_map
|
||||
/// [`Stream`]: trait.Stream.html
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct FlatMap<S, U, T, F> {
|
||||
#[pin]
|
||||
stream: Map<S, F, T, U>,
|
||||
#[pin]
|
||||
inner_stream: Option<U>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, U, F> FlatMap<S, U, S::Item, F>
|
||||
where
|
||||
S: Stream,
|
||||
U: IntoStream,
|
||||
F: FnMut(S::Item) -> U,
|
||||
{
|
||||
pub(super) fn new(stream: S, f: F) -> FlatMap<S, U, S::Item, F> {
|
||||
FlatMap {
|
||||
stream: stream.map(f),
|
||||
inner_stream: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, U, F> Stream for FlatMap<S, U, S::Item, F>
|
||||
where
|
||||
S: Stream,
|
||||
S::Item: IntoStream<IntoStream = U, Item = U::Item>,
|
||||
U: Stream,
|
||||
F: FnMut(S::Item) -> U,
|
||||
{
|
||||
type Item = U::Item;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let mut this = self.project();
|
||||
loop {
|
||||
if let Some(inner) = this.inner_stream.as_mut().as_pin_mut() {
|
||||
if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
|
||||
return Poll::Ready(item);
|
||||
}
|
||||
}
|
||||
|
||||
match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
|
||||
None => return Poll::Ready(None),
|
||||
Some(inner) => this.inner_stream.set(Some(inner.into_stream())),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,58 @@
|
||||
use pin_project_lite::pin_project;
|
||||
use std::pin::Pin;
|
||||
|
||||
use crate::stream::{IntoStream, Stream};
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
/// This `struct` is created by the [`flatten`] method on [`Stream`]. See its
|
||||
/// documentation for more.
|
||||
///
|
||||
/// [`flatten`]: trait.Stream.html#method.flatten
|
||||
/// [`Stream`]: trait.Stream.html
|
||||
#[allow(missing_debug_implementations)]
|
||||
pub struct Flatten<S, U> {
|
||||
#[pin]
|
||||
stream: S,
|
||||
#[pin]
|
||||
inner_stream: Option<U>,
|
||||
}
|
||||
}
|
||||
|
||||
impl<S> Flatten<S, S::Item>
|
||||
where
|
||||
S: Stream,
|
||||
S::Item: IntoStream,
|
||||
{
|
||||
pub(super) fn new(stream: S) -> Flatten<S, S::Item> {
|
||||
Flatten {
|
||||
stream,
|
||||
inner_stream: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, U> Stream for Flatten<S, <S::Item as IntoStream>::IntoStream>
|
||||
where
|
||||
S: Stream,
|
||||
S::Item: IntoStream<IntoStream = U, Item = U::Item>,
|
||||
U: Stream,
|
||||
{
|
||||
type Item = U::Item;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let mut this = self.project();
|
||||
loop {
|
||||
if let Some(inner) = this.inner_stream.as_mut().as_pin_mut() {
|
||||
if let item @ Some(_) = futures_core::ready!(inner.poll_next(cx)) {
|
||||
return Poll::Ready(item);
|
||||
}
|
||||
}
|
||||
|
||||
match futures_core::ready!(this.stream.as_mut().poll_next(cx)) {
|
||||
None => return Poll::Ready(None),
|
||||
Some(inner) => this.inner_stream.set(Some(inner.into_stream())),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
use std::error::Error;
|
||||
use std::fmt;
|
||||
use std::pin::Pin;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_timer::Delay;
|
||||
use pin_project_lite::pin_project;
|
||||
|
||||
use crate::future::Future;
|
||||
use crate::stream::Stream;
|
||||
use crate::task::{Context, Poll};
|
||||
|
||||
pin_project! {
|
||||
/// A stream with timeout time set
|
||||
#[derive(Debug)]
|
||||
pub struct Timeout<S: Stream> {
|
||||
#[pin]
|
||||
stream: S,
|
||||
#[pin]
|
||||
delay: Delay,
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: Stream> Timeout<S> {
|
||||
pub fn new(stream: S, dur: Duration) -> Timeout<S> {
|
||||
let delay = Delay::new(dur);
|
||||
|
||||
Timeout { stream, delay }
|
||||
}
|
||||
}
|
||||
|
||||
impl<S: Stream> Stream for Timeout<S> {
|
||||
type Item = Result<S::Item, TimeoutError>;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let this = self.project();
|
||||
|
||||
match this.stream.poll_next(cx) {
|
||||
Poll::Ready(Some(v)) => Poll::Ready(Some(Ok(v))),
|
||||
Poll::Ready(None) => Poll::Ready(None),
|
||||
Poll::Pending => match this.delay.poll(cx) {
|
||||
Poll::Ready(_) => Poll::Ready(Some(Err(TimeoutError { _private: () }))),
|
||||
Poll::Pending => Poll::Pending,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An error returned when a stream times out.
|
||||
#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
|
||||
#[cfg(any(feature = "unstable", feature = "docs"))]
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
pub struct TimeoutError {
|
||||
_private: (),
|
||||
}
|
||||
|
||||
impl Error for TimeoutError {}
|
||||
|
||||
impl fmt::Display for TimeoutError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
"stream has timed out".fmt(f)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue