nosy 0.3.0

Change notification / observation / broadcast channels, with filtering and coalescing. no_std compatible.
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

use core::future::Future as _;
use core::pin::pin;
use core::sync::atomic::{AtomicBool, Ordering::Relaxed};
use std::sync::Arc;
use std::task::{self, Poll};

use futures::StreamExt as _;

use nosy::Listener as _;

use crate::tools::{run_task_without_waiting, yield_now};

// -------------------------------------------------------------------------------------------------

struct WakeDetector {
    woken: AtomicBool,
}
impl std::task::Wake for WakeDetector {
    fn wake(self: Arc<Self>) {
        self.woken.store(true, Relaxed);
    }
}
impl WakeDetector {
    fn take(&self) -> bool {
        self.woken.swap(false, Relaxed)
    }
}

/// Basic functionality test using explicit `poll()` and checking the waker behavior too.
#[test]
fn wake_flag_polling() {
    let wake_detector = Arc::new(WakeDetector {
        woken: AtomicBool::new(false),
    });
    let waker = task::Waker::from(wake_detector.clone());
    let ctx = &mut task::Context::from_waker(&waker);
    let (mut flag, listener) = nosy::future::WakeFlag::new(true);

    // First poll succeeds immediately because we initialized with true.
    assert_eq!(pin!(flag.wait()).as_mut().poll(ctx), Poll::Ready(true));
    assert_eq!(wake_detector.take(), false);

    {
        // Second poll of a new future returns Pending.
        let mut future = pin!(flag.wait());
        assert_eq!(future.as_mut().poll(ctx), Poll::Pending);
        assert_eq!(wake_detector.take(), false);

        // When a message is received, then the waker will be called
        // and polling will return Ready(true).
        listener.receive(&[()]);
        assert_eq!(wake_detector.take(), true);
        assert_eq!(future.as_mut().poll(ctx), Poll::Ready(true));
        assert_eq!(wake_detector.take(), false);
    }

    // When the listener is dropped, then polling will return Ready(false),
    // and the waker (*if* one is registered again) will be woken;
    {
        let mut future = pin!(flag.wait());
        // Get the waker registered again
        assert_eq!(future.as_mut().poll(ctx), Poll::Pending);
        assert_eq!(wake_detector.take(), false);

        drop::<nosy::future::WakeFlagListener>(listener);

        assert_eq!(wake_detector.take(), true);
        assert_eq!(future.as_mut().poll(ctx), Poll::Ready(false));
        assert_eq!(wake_detector.take(), false);
    }
}

/// Test `impl WakeFlag for Stream`.
#[test]
fn wake_flag_stream() {
    // just using this as interior mutable Vec, not something under test
    let event_log = nosy::Log::new();
    let log_listener = event_log.listener();

    run_task_without_waiting(async move {
        let (mut flag, listener) = nosy::future::WakeFlag::new(false);

        futures::join!(
            async {
                while let Some(()) = flag.next().await {
                    log_listener.receive(&["received"]);
                }
            },
            async {
                for _ in 0..2 {
                    log_listener.receive(&["sending"]);
                    listener.receive(&[()]);
                    log_listener.receive(&["sent"]);
                    yield_now().await;
                }
                drop(listener);
            }
        );
    });

    // Check the expected events occurred
    assert_eq!(
        event_log.drain(),
        vec!["sending", "sent", "received", "sending", "sent", "received"]
    );
}

#[test]
fn wake_flag_debug() {
    let (mut flag, listener) = nosy::future::WakeFlag::new(false);
    assert_eq!(
        format!("{flag:?} {listener:?}"),
        "WakeFlag(false) WakeFlagListener { alive: true, value: false }"
    );
    listener.receive(&[123]);
    assert_eq!(
        format!("{flag:?} {listener:?}"),
        "WakeFlag(true) WakeFlagListener { alive: true, value: true }"
    );
    futures::executor::block_on(flag.wait());
    assert_eq!(
        format!("{flag:?} {listener:?}"),
        "WakeFlag(false) WakeFlagListener { alive: true, value: false }"
    );
    drop(flag);
    assert_eq!(format!("{listener:?}"), "WakeFlagListener { alive: false }");
}

#[test]
fn wake_flag_pointer_fmt_eq() {
    let (flag, listener) = nosy::future::WakeFlag::new(false);
    assert_eq!(
        format!("{flag:p}"),
        format!("{listener:p}"),
        "pointer formatting not equal as expected"
    )
}