flo_stream 0.7.1

Pubsub and related streams for Rust futures
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

extern crate futures;
extern crate flo_stream;

use flo_stream::*;

use futures::*;
use futures::executor;
use futures::task::{ArcWake, Poll};

use std::thread;
use std::sync::*;
use std::time::Duration;

#[derive(Clone)]
struct NotifyNothing;

impl ArcWake for NotifyNothing {
    fn wake_by_ref(_arc_self: &Arc<Self>) { }
}

#[test]
pub fn blocks_if_there_are_no_subscribers() {
    let waker           = task::waker(Arc::new(NotifyNothing));
    let mut ctxt        = task::Context::from_waker(&waker);

    let mut publisher   = BlockingPublisher::new(2, 10);

    assert!(publisher.publish(1).poll_unpin(&mut ctxt) == Poll::Pending);
}

#[test]
pub fn blocks_if_there_are_insufficient_subscribers() {
    let waker           = task::waker(Arc::new(NotifyNothing));
    let mut ctxt        = task::Context::from_waker(&waker);

    let mut publisher   = BlockingPublisher::new(2, 10);
    let _subscriber     = publisher.subscribe();

    assert!(publisher.publish(1).poll_unpin(&mut ctxt) == Poll::Pending);
}

#[test]
pub fn unblocks_if_there_are_sufficient_subscribers() {
    let waker           = task::waker(Arc::new(NotifyNothing));
    let mut ctxt        = task::Context::from_waker(&waker);

    let mut publisher   = BlockingPublisher::new(2, 10);
    let _subscriber1    = publisher.subscribe();
    let _subscriber2    = publisher.subscribe();

    assert!(publisher.publish(1).poll_unpin(&mut ctxt) == Poll::Ready(()));
}

#[test]
pub fn read_from_subscribers() {
    let mut publisher   = BlockingPublisher::new(2, 10);
    let mut subscriber1 = publisher.subscribe();
    let mut subscriber2 = publisher.subscribe();

    executor::block_on(async {
        publisher.publish(1).await;
        publisher.publish(2).await;
        publisher.publish(3).await;

        assert!(subscriber1.next().await == Some(1));
        assert!(subscriber1.next().await == Some(2));
        assert!(subscriber1.next().await == Some(3));

        assert!(subscriber2.next().await == Some(1));
        assert!(subscriber2.next().await == Some(2));
        assert!(subscriber2.next().await == Some(3));
    });
}

#[test]
pub fn read_from_thread() {
    let mut subscriber = {
        // Create a shared publisher
        let publisher = BlockingPublisher::new(1, 1);
        let publisher = Arc::new(Mutex::new(publisher));

        // Create a thread to publish some values
        let thread_publisher = publisher.clone();
        thread::spawn(move || {
            let wait_for_subscribers = thread_publisher.lock().unwrap().when_ready();

            executor::block_on(async {
                wait_for_subscribers.await;

                thread_publisher.lock().unwrap().publish(1).await;
                thread_publisher.lock().unwrap().publish(2).await;
                thread_publisher.lock().unwrap().publish(3).await;
            });
        });

        // Pause for a bit to let the thread get ahead of us
        thread::sleep(Duration::from_millis(20));

        // Subscribe to the thread (which should now wake up)
        let subscriber = publisher.lock().unwrap().subscribe();
        subscriber
    };

    // Should receive the values from the thread
    executor::block_on(async {
        assert!(subscriber.next().await == Some(1));
        assert!(subscriber.next().await == Some(2));
        assert!(subscriber.next().await == Some(3));

        // As we don't retain the publisher, the thread is its only owner. When it finishes, the stream should close.
        assert!(subscriber.next().await == None);
    });
}

#[test]
pub fn read_from_thread_late_start() {
    let mut subscriber = {
        // Create a shared publisher
        let publisher = BlockingPublisher::new(1, 1);
        let publisher = Arc::new(Mutex::new(publisher));

        // Create a thread to publish some values
        let thread_publisher = publisher.clone();
        thread::spawn(move || {
            // Wait for the subscriber to be created
            thread::sleep(Duration::from_millis(20));

            let wait_for_subscribers = thread_publisher.lock().unwrap().when_ready();
            executor::block_on(async {
                wait_for_subscribers.await;

                thread_publisher.lock().unwrap().publish(1).await;
                thread_publisher.lock().unwrap().publish(2).await;
                thread_publisher.lock().unwrap().publish(3).await;
            });
        });

        // Subscribe to the thread (which should now wake up)
        let subscriber = publisher.lock().unwrap().subscribe();
        subscriber
    };

    // Should receive the values from the thread
    executor::block_on(async {
        assert!(subscriber.next().await == Some(1));
        assert!(subscriber.next().await == Some(2));
        assert!(subscriber.next().await == Some(3));

        // As we don't retain the publisher, the thread is its only owner. When it finishes, the stream should close.
        assert!(subscriber.next().await == None);
    });
}