futures-rx 0.3.3

Rx implementations for the futures crate
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

use std::task::Poll;

use futures::{stream, Stream};

/// A stream that yields `Poll::Pending` once before every event, so that
/// adapters are driven through their pending paths instead of only their
/// always-ready ones.
pub(crate) fn stuttering<T>(items: impl IntoIterator<Item = T>) -> impl Stream<Item = T> + Unpin {
    let mut items = items.into_iter();
    let mut is_pending = false;

    Box::pin(stream::poll_fn(move |cx| {
        is_pending = !is_pending;

        if is_pending {
            cx.waker().wake_by_ref();

            Poll::Pending
        } else {
            Poll::Ready(items.next())
        }
    }))
}

/// A stream that never yields an event and never completes.
pub(crate) fn never<T>() -> impl Stream<Item = T> + Unpin {
    Box::pin(stream::poll_fn(|_| Poll::Pending))
}