#futures #reactive #notify #change #watch

async-change-tracker

reactive change notifications using futures

4 releases (2 breaking)

0.3.1 Oct 4, 2020
0.3.0 Dec 15, 2019
0.2.0 Nov 8, 2019
0.1.0 Oct 11, 2018

#106 in Asynchronous

Download history 126/week @ 2020-11-07 83/week @ 2020-11-14 41/week @ 2020-11-21 102/week @ 2020-11-28 74/week @ 2020-12-05 36/week @ 2020-12-12 58/week @ 2020-12-19 109/week @ 2020-12-26 245/week @ 2021-01-02 105/week @ 2021-01-09 111/week @ 2021-01-16 66/week @ 2021-01-23 66/week @ 2021-01-30 71/week @ 2021-02-06 25/week @ 2021-02-13 129/week @ 2021-02-20

78 downloads per month
Used in bui-backend

MIT/Apache

10KB
52 lines

async-change-tracker

Reactive change notifications using futures.

The ChangeTracker<T> type wraps an owned value T. Changes to T are done within a function or closure implementing FnOnce(&mut T). When this returns, any changes are sent to listeners using a futures::Stream.

In slightly more detail, create a ChangeTracker<T> with ChangeTracker::new(value: T). This will take ownership of the value of type T. You can then create a futures::Stream (with get_changes()) that emits a tuple (old_value, new_value) of type (T, T) upon every change to the owned value. The value can be changed with the modify() method of ChangeTracker and read using the as_ref() method from the AsRef trait.

Example

In this example, the functionality of ChangeTracker is shown.

use futures::stream::StreamExt;

// Wrap an integer `with ChangeTracker`
let mut change_tracker = async_change_tracker::ChangeTracker::new( 123 );

// Create an receiver that fires when the value changes. The channel size
// is 1, meaning at most one change can be buffered before backpressure
// is applied.
let rx = change_tracker.get_changes(1);

// In this example take a single change and check that the old and new value
// are correct.
let rx_printer = rx.take(1).for_each(|(old_value, new_value)| {
    assert_eq!( old_value, 123);
    assert_eq!( new_value, 124);
    futures::future::ready(())
});

// Now, check and then change the value.
change_tracker.modify(|mut_ref_value| {
    assert_eq!(*mut_ref_value, 123);
    *mut_ref_value += 1;
});

// Wait until the stream is done. In this example, the stream ends due to
// the use of `.take(1)` prior to `for_each` above. In normal usage,
// typically the stream would finish for a different reason.
futures::executor::block_on(rx_printer);

// Finally, check that the final value is as expected.
assert!(*change_tracker.as_ref() == 124);

Testing

To test, you need the thread-pool feature for the futures create:

cargo test --features "futures/thread-pool"

License: MIT/Apache-2.0

Dependencies

~1.5MB
~28K SLoC