1 | use crate::Stream; |
2 | |
3 | use pin_project_lite::pin_project; |
4 | use std::pin::Pin; |
5 | use std::task::{Context, Poll}; |
6 | |
7 | pin_project! { |
8 | /// Stream returned by [`fuse()`][super::StreamExt::fuse]. |
9 | #[derive(Debug)] |
10 | pub struct Fuse<T> { |
11 | #[pin] |
12 | stream: Option<T>, |
13 | } |
14 | } |
15 | |
16 | impl<T> Fuse<T> |
17 | where |
18 | T: Stream, |
19 | { |
20 | pub(crate) fn new(stream: T) -> Fuse<T> { |
21 | Fuse { |
22 | stream: Some(stream), |
23 | } |
24 | } |
25 | } |
26 | |
27 | impl<T> Stream for Fuse<T> |
28 | where |
29 | T: Stream, |
30 | { |
31 | type Item = T::Item; |
32 | |
33 | fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T::Item>> { |
34 | let res = match Option::as_pin_mut(self.as_mut().project().stream) { |
35 | Some(stream) => ready!(stream.poll_next(cx)), |
36 | None => return Poll::Ready(None), |
37 | }; |
38 | |
39 | if res.is_none() { |
40 | // Do not poll the stream anymore |
41 | self.as_mut().project().stream.set(None); |
42 | } |
43 | |
44 | Poll::Ready(res) |
45 | } |
46 | |
47 | fn size_hint(&self) -> (usize, Option<usize>) { |
48 | match self.stream { |
49 | Some(ref stream) => stream.size_hint(), |
50 | None => (0, Some(0)), |
51 | } |
52 | } |
53 | } |
54 | |