dutils 0.1.14

I am speed
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

use futures::prelude::Stream;
use futures::stream::FusedStream;
use std::fmt::Debug;
use std::pin::Pin;

#[pin_project::pin_project]
pub struct DropFuse<S> {
    #[pin]
    stream: Option<S>,
}
impl<S: Stream> Stream for DropFuse<S> {
    type Item = S::Item;
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let mut this = self.project();

        let Some(stream) = this.stream.as_mut().as_pin_mut() else {
            return std::task::Poll::Ready(None);
        };
        let item = futures::ready!(stream.poll_next(cx));
        if let Some(v) = item {
            return std::task::Poll::Ready(Some(v));
        }
        this.stream.set(None); // Will drop right after None returned
        std::task::Poll::Ready(None)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        if let Some(st) = self.stream.as_ref() {
            st.size_hint()
        } else {
            (0, Some(0))
        }
    }
}
impl<S: Stream> FusedStream for DropFuse<S> {
    fn is_terminated(&self) -> bool {
        self.stream.is_none()
    }
}

pub trait StreamUtils: Sized {
    fn drop_fuse(self) -> DropFuse<Self>;
}
impl<S: Stream> StreamUtils for S {
    fn drop_fuse(self) -> DropFuse<Self> {
        DropFuse { stream: Some(self) }
    }
}

#[pin_project::pin_project]
pub struct SelectOkStream<S>(#[pin] S);
impl<S: Stream<Item = Result<T, E>>, T, E: Debug> Stream for SelectOkStream<S> {
    type Item = T;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let mut this = self.project();
        loop {
            return match this.0.as_mut().poll_next(cx) {
                std::task::Poll::Ready(Some(Ok(v))) => std::task::Poll::Ready(Some(v)),
                std::task::Poll::Ready(Some(Err(_))) => continue,
                std::task::Poll::Ready(None) => std::task::Poll::Ready(None),
                std::task::Poll::Pending => std::task::Poll::Pending,
            };
        }
    }
    fn size_hint(&self) -> (usize, Option<usize>) {
        (0, self.0.size_hint().1)
    }
}
pub trait StreamErrors<T, E>: Sized {
    fn select_ok(self) -> SelectOkStream<Self>;
}
impl<S: Stream<Item = Result<T, E>>, T, E: Debug> StreamErrors<T, E> for S {
    fn select_ok(self) -> SelectOkStream<Self> {
        SelectOkStream(self)
    }
}