usecrate::fns::FnOnce1; usecrate::stream::{Fuse, StreamExt}; use core::fmt; use core::marker::PhantomData; use core::pin::Pin; use futures_core::future::{FusedFuture, Future}; use futures_core::ready; use futures_core::stream::{FusedStream, Stream}; use futures_core::task::{Context, Poll}; #[cfg(feature = "sink")] use futures_sink::Sink; use pin_project_lite::pin_project;
pin_project! { /// A `Stream` that implements a `peek` method. /// /// The `peek` method can be used to retrieve a reference /// to the next `Stream::Item` if available. A subsequent /// call to `poll` will return the owned item. #[derive(Debug)] #[must_use = "streams do nothing unless polled"] pubstruct Peekable<St: Stream> { #[pin]
stream: Fuse<St>,
peeked: Option<St::Item>,
}
}
/// Produces a future which retrieves a reference to the next item /// in the stream, or `None` if the underlying stream terminates. pubfn peek(self: Pin<&mutSelf>) -> Peek<'_, St> {
Peek { inner: Some(self) }
}
/// Peek retrieves a reference to the next item in the stream. /// /// This method polls the underlying stream and return either a reference /// to the next item if the stream is ready or passes through any errors. pubfn poll_peek(self: Pin<&mutSelf>, cx: &mut Context<'_>) -> Poll<Option<&St::Item>> { letmut this = self.project();
/// Produces a future which retrieves a mutable reference to the next item /// in the stream, or `None` if the underlying stream terminates. /// /// # Examples /// /// ``` /// # futures::executor::block_on(async { /// use futures::stream::{self, StreamExt}; /// use futures::pin_mut; /// /// let stream = stream::iter(vec![1, 2, 3]).peekable(); /// pin_mut!(stream); /// /// assert_eq!(stream.as_mut().peek_mut().await, Some(&mut 1)); /// assert_eq!(stream.as_mut().next().await, Some(1)); /// /// // Peek into the stream and modify the value which will be returned next /// if let Some(p) = stream.as_mut().peek_mut().await { /// if *p == 2 { /// *p = 5; /// } /// } /// /// assert_eq!(stream.collect::<Vec<_>>().await, vec![5, 3]); /// # }); /// ``` pubfn peek_mut(self: Pin<&mutSelf>) -> PeekMut<'_, St> {
PeekMut { inner: Some(self) }
}
/// Peek retrieves a mutable reference to the next item in the stream. pubfn poll_peek_mut( self: Pin<&mutSelf>,
cx: &mut Context<'_>,
) -> Poll<Option<&mut St::Item>> { letmut this = self.project();
/// Creates a future which will consume and return the next value of this /// stream if a condition is true. /// /// If `func` returns `true` for the next value of this stream, consume and /// return it. Otherwise, return `None`. /// /// # Examples /// /// Consume a number if it's equal to 0. /// /// ``` /// # futures::executor::block_on(async { /// use futures::stream::{self, StreamExt}; /// use futures::pin_mut; /// /// let stream = stream::iter(0..5).peekable(); /// pin_mut!(stream); /// // The first item of the stream is 0; consume it. /// assert_eq!(stream.as_mut().next_if(|&x| x == 0).await, Some(0)); /// // The next item returned is now 1, so `consume` will return `false`. /// assert_eq!(stream.as_mut().next_if(|&x| x == 0).await, None); /// // `next_if` saves the value of the next item if it was not equal to `expected`. /// assert_eq!(stream.next().await, Some(1)); /// # }); /// ``` /// /// Consume any number less than 10. /// /// ``` /// # futures::executor::block_on(async { /// use futures::stream::{self, StreamExt}; /// use futures::pin_mut; /// /// let stream = stream::iter(1..20).peekable(); /// pin_mut!(stream); /// // Consume all numbers less than 10 /// while stream.as_mut().next_if(|&x| x < 10).await.is_some() {} /// // The next value returned will be 10 /// assert_eq!(stream.next().await, Some(10)); /// # }); /// ``` pubfn next_if<F>(self: Pin<&mutSelf>, func: F) -> NextIf<'_, St, F> where
F: FnOnce(&St::Item) -> bool,
{
NextIf { inner: Some((self, func)) }
}
/// Creates a future which will consume and return the next item if it is /// equal to `expected`. /// /// # Example /// /// Consume a number if it's equal to 0. /// /// ``` /// # futures::executor::block_on(async { /// use futures::stream::{self, StreamExt}; /// use futures::pin_mut; /// /// let stream = stream::iter(0..5).peekable(); /// pin_mut!(stream); /// // The first item of the stream is 0; consume it. /// assert_eq!(stream.as_mut().next_if_eq(&0).await, Some(0)); /// // The next item returned is now 1, so `consume` will return `false`. /// assert_eq!(stream.as_mut().next_if_eq(&0).await, None); /// // `next_if_eq` saves the value of the next item if it was not equal to `expected`. /// assert_eq!(stream.next().await, Some(1)); /// # }); /// ``` pubfn next_if_eq<'a, T>(self: Pin<&'a mutSelf>, expected: &'a T) -> NextIfEq<'a, St, T> where
T: ?Sized,
St::Item: PartialEq<T>,
{
NextIfEq {
inner: NextIf { inner: Some((self, NextIfEqFn { expected, _next: PhantomData })) },
}
}
}
fn size_hint(&self) -> (usize, Option<usize>) { let peek_len = usize::from(self.peeked.is_some()); let (lower, upper) = self.stream.size_hint(); let lower = lower.saturating_add(peek_len); let upper = match upper {
Some(x) => x.checked_add(peek_len),
None => None,
};
(lower, upper)
}
}
// Forwarding impl of Sink from the underlying stream #[cfg(feature = "sink")] impl<S, Item> Sink<Item> for Peekable<S> where
S: Sink<Item> + Stream,
{ type Error = S::Error;
delegate_sink!(stream, Item);
}
pin_project! { /// Future for the [`Peekable::peek`](self::Peekable::peek) method. #[must_use = "futures do nothing unless polled"] pubstruct Peek<'a, St: Stream> {
inner: Option<Pin<&'a mut Peekable<St>>>,
}
}
#[allow(single_use_lifetimes)] // https://github.com/rust-lang/rust/issues/55058 impl<St, F> Future for NextIf<'_, St, F> where
St: Stream,
F: for<'a> FnOnce1<&'a St::Item, Output = bool>,
{ type Output = Option<St::Item>;
fn poll(self: Pin<&mutSelf>, cx: &mut Context<'_>) -> Poll<Self::Output> { let inner = self.project().inner; iflet Some((peekable, _)) = inner { let res = ready!(peekable.as_mut().poll_next(cx));
let (peekable, func) = inner.take().unwrap(); match res {
Some(ref matched) if func.call_once(matched) => Poll::Ready(res),
other => { let peekable = peekable.project(); // Since we called `self.next()`, we consumed `self.peeked`.
assert!(peekable.peeked.is_none());
*peekable.peeked = other;
Poll::Ready(None)
}
}
} else {
panic!("NextIf polled after completion")
}
}
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.