async-change-tracker 0.3.5

reactive change notifications using 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

use std::{
    cell::RefCell,
    rc::Rc,
    sync::{Arc, RwLock},
};

use futures::stream::StreamExt;

use async_change_tracker::ChangeTracker;

#[test]
fn test_change_tracker() {
    #[derive(Clone, PartialEq, Debug)]
    struct StoreType {
        val: i32,
    }

    let mut change_tracker = ChangeTracker::new(StoreType { val: 123 });
    let rx = change_tracker.get_changes(1);

    // Create a future to cause a change
    let cause_change = async move {
        change_tracker.modify(|scoped_store| {
            assert!((*scoped_store).val == 123);
            (*scoped_store).val += 1;
        });
    };

    futures::executor::block_on(cause_change);

    let check_change = rx.take(1).for_each(|(old_value, new_value)| {
        assert!(old_value.val == 123);
        assert!(new_value.val == 124);
        futures::future::ready(())
    });

    futures::executor::block_on(check_change);
}

#[test]
fn test_dropped_rx() {
    #[derive(Clone, PartialEq, Debug)]
    struct StoreType {
        val: i32,
    }

    let data_store = ChangeTracker::new(StoreType { val: 123 });
    let data_store_rc = Rc::new(RefCell::new(data_store));

    {
        let _rx = data_store_rc.borrow_mut().get_changes(1);
        // drop rx at end of this scope
    }

    let dsclone = data_store_rc.clone();
    // Create a future to cause a change
    let cause_change = async move {
        dsclone.borrow_mut().modify(|scoped_store| {
            assert!((*scoped_store).val == 123);
            (*scoped_store).val += 1;
        });
    };

    futures::executor::block_on(cause_change);

    assert!(data_store_rc.borrow().as_ref().val == 124);
}

#[test]
fn test_multiple_changes_no_rx() {
    #[derive(Clone, PartialEq, Debug)]
    struct StoreType {
        val: i32,
    }

    let data_store = ChangeTracker::new(StoreType { val: 123 });
    let data_store_rc = Rc::new(RefCell::new(data_store));

    let dsclone1 = data_store_rc.clone();
    let dsclone2 = data_store_rc.clone();

    // Create a future to cause a change
    let cause_change1 = async move {
        let mut data_store = dsclone1.borrow_mut();
        data_store.modify(|scoped_store| {
            (*scoped_store).val += 1;
        });
    };
    futures::executor::block_on(cause_change1);

    // Create a future to cause a change
    let cause_change2 = async move {
        let mut data_store = dsclone2.borrow_mut();
        data_store.modify(|scoped_store| {
            (*scoped_store).val += 1;
        });
    };
    futures::executor::block_on(cause_change2);
    assert!(data_store_rc.borrow().as_ref().val == 125);
}

#[test]
fn test_multithreaded_change_tracker() {
    use futures::task::SpawnExt;

    #[derive(Clone, PartialEq, Debug)]
    struct StoreType {
        val: i32,
    }

    let data_store = ChangeTracker::new(StoreType { val: 123 });
    let rx = data_store.get_changes(1);
    let data_store_arc = Arc::new(RwLock::new(data_store));
    let rx_printer = rx.take(1).for_each(|(old_value, new_value)| {
        assert!(old_value.val == 123);
        assert!(new_value.val == 124);
        futures::future::ready(())
    });

    // use multi-threaded runtime
    use futures::executor::ThreadPool;
    let rt = ThreadPool::new().unwrap();

    let dsclone = data_store_arc.clone();
    // Create a future to cause a change
    let cause_change = async move {
        dsclone.write().unwrap().modify(|scoped_store| {
            assert!((*scoped_store).val == 123);
            (*scoped_store).val += 1;
        });
    };

    let do_all_fut = async move {
        cause_change.await;
        rx_printer.await;
    };

    let join_handle_fut = rt.spawn_with_handle(do_all_fut).unwrap();
    futures::executor::block_on(join_handle_fut);

    assert!(data_store_arc.read().unwrap().as_ref().val == 124);
}