pollable-map 0.1.7

Utilites for storing futures and streams in as a pollable map
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use futures::{Stream, StreamExt};
use std::pin::Pin;

use super::StreamMap;
use futures::stream::FusedStream;
use std::task::{Context, Poll};

pub struct StreamSet<S> {
    id: i64,
    map: StreamMap<i64, S>,
}

impl<S> Default for StreamSet<S>
where
    S: Stream + Send + Unpin + 'static,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<S> StreamSet<S>
where
    S: Stream + Send + Unpin + 'static,
{
    /// Creates an empty ['StreamSet`]
    pub fn new() -> Self {
        Self {
            id: 0,
            map: StreamMap::default(),
        }
    }

    /// Insert a stream into the set of streams.
    pub fn insert(&mut self, stream: S) -> bool {
        self.id = self.id.wrapping_add(1);
        self.map.insert(self.id, stream)
    }

    /// An iterator visiting all streams in arbitrary order.
    pub fn iter(&self) -> impl Iterator<Item = &S> {
        self.map.iter().map(|(_, st)| st)
    }

    /// An iterator visiting all streams mutably in arbitrary order.
    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut S> {
        self.map.iter_mut().map(|(_, st)| st)
    }

    /// An iterator visiting all streams pinned valued in arbitrary order
    pub fn iter_pin(&mut self) -> impl Iterator<Item = Pin<&mut S>> {
        self.map.iter_pin().map(|(_, st)| st)
    }

    /// Clears the set.
    pub fn clear(&mut self) {
        self.map.clear();
    }

    /// Returns the number of streams in the set.
    pub fn len(&self) -> usize {
        self.map.len()
    }

    /// Return `true` map contains no elements.
    pub fn is_empty(&self) -> bool {
        self.map.is_empty()
    }
}

impl<S> FromIterator<S> for StreamSet<S>
where
    S: Stream + Send + Unpin + 'static,
{
    fn from_iter<I: IntoIterator<Item = S>>(iter: I) -> Self {
        let mut maps = Self::new();
        for st in iter {
            maps.insert(st);
        }
        maps
    }
}

impl<S> Stream for StreamSet<S>
where
    S: Stream + Send + Unpin + 'static,
{
    type Item = S::Item;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.map
            .poll_next_unpin(cx)
            .map(|output| output.map(|(_, item)| item))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.map.size_hint()
    }
}

impl<S> FusedStream for StreamSet<S>
where
    S: Stream + Send + Unpin + 'static,
{
    fn is_terminated(&self) -> bool {
        self.map.is_terminated()
    }
}

#[cfg(test)]
mod test {
    use crate::stream::set::StreamSet;
    use futures::StreamExt;

    #[test]
    fn valid_stream_set() {
        let mut list = StreamSet::new();
        assert!(list.insert(futures::stream::once(async { 0 }).boxed()));
        assert!(list.insert(futures::stream::once(async { 1 }).boxed()));

        futures::executor::block_on(async move {
            let val = list.next().await;
            assert_eq!(val, Some(0));
            let val = list.next().await;
            assert_eq!(val, Some(1));
        });
    }
}