xstd 0.2.0

eXtended STandarD library
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

use std::pin::Pin;
use pin_utils::{unsafe_pinned, unsafe_unpinned};
use futures_core::{
    ready,
    stream::{
        Stream
    },
    task::{
        Context,
        Poll,
    }
};


pub trait UniqueBy: Stream {
    fn unique_by<F, E>(self, f: F) -> Unique<Self, F, E>
    where Self: Sized,
          F: FnMut(&Self::Item) -> E,
    {
        Unique::new(self, f)
    }
}

impl<T> UniqueBy for T where T: Stream {}

/// Creates a Stream which skips already emitted elements.
///
///
/// # Examples
///
/// ```
/// # futures::executor::block_on(async {
/// use xstd::prelude::*;
/// use futures::{stream::{self, StreamExt}, executor};
///
/// let mut stream = stream::iter(vec![3, 1, 3, 2, 1, 4])
///                     .unique_by(|x| x.clone());
///
/// assert_eq!(Some(3), stream.next().await);
/// assert_eq!(Some(1), stream.next().await);
/// assert_eq!(Some(2), stream.next().await);
/// assert_eq!(Some(4), stream.next().await);
/// # });
/// ```
#[must_use = "streams do nothing unless polled"]
pub struct Unique<St, F, E> {
    stream: St,
    f: F,
    seen: Vec<E>,
}

impl<St, F, E> Unique<St, F, E>
where St: Stream,
      F: FnMut(&St::Item) -> E,
{
    unsafe_pinned!(stream: St);
    unsafe_unpinned!(f: F);
    unsafe_unpinned!(seen: Vec<E>);

    pub fn new(stream: St, f: F) -> Unique<St, F, E>
    {
        Self { stream, f, seen: Vec::new() }
    }
}

impl<St, F, E> Stream for Unique<St, F, E>
where St: Stream,
      F: FnMut(&St::Item) -> E,
      E: PartialEq,
{
    type Item = St::Item;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<St::Item>> {
        match ready!(self.as_mut().stream().poll_next(cx)) {
            Some(item) => {
                let e = self.as_mut().f()(&item);
                if self.as_mut().seen().contains(&e) {
                    self.poll_next(cx)
                } else {
                    self.as_mut().seen().push(e);
                    Poll::Ready(Some(item))
                }
            },
            x => Poll::Ready(x),
        }
    }
}


#[cfg(test)]
mod tests {
    use crate::prelude::*;
    use futures::{stream::{self, StreamExt}, executor};

    #[test]
    fn test_unique() {
        executor::block_on(async {
            let stream = stream::iter(vec![3, 1, 2, 3, 4, 1, 6, 3, 8])
                .unique_by(|x| x.clone());
            let res = stream.collect::<Vec<i32>>().await;
            assert_eq!(vec![3, 1, 2, 4, 6, 8], res);
        });
    }
}