Merge branch 'master' into add_future_delay
k-nasa
5 years ago
commit
1545d24e50
@ -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,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