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 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189
//! This crate provides a synchronous message passing channel that only
//! retains the most recent value.
//!
//! This crate provides a `parking_lot` feature. When enabled, the crate will
//! use the mutex from the `parking_lot` crate rather than the one from std.
use std::{
sync::Arc,
time::{Duration, Instant},
};
#[cfg(not(feature = "parking_lot"))]
mod sync_std;
#[cfg(not(feature = "parking_lot"))]
use sync_std::{Condvar, Mutex};
#[cfg(feature = "parking_lot")]
mod sync_parking_lot;
#[cfg(feature = "parking_lot")]
use sync_parking_lot::{Condvar, Mutex};
/// The sender for the watch channel.
///
/// The sender can be cloned to obtain multiple senders for the same channel.
pub struct WatchSender<T> {
shared: Arc<Shared<T>>,
}
/// The receiver for the watch channel.
///
/// The receiver can be cloned. Each clone will yield a new receiver that
/// receives the same messages.
pub struct WatchReceiver<T> {
shared: Arc<Shared<T>>,
last_seen_version: u64,
}
impl<T> Clone for WatchSender<T> {
fn clone(&self) -> WatchSender<T> {
WatchSender {
shared: self.shared.clone(),
}
}
}
impl<T> Clone for WatchReceiver<T> {
fn clone(&self) -> WatchReceiver<T> {
WatchReceiver {
shared: self.shared.clone(),
last_seen_version: self.last_seen_version,
}
}
}
struct Shared<T> {
lock: Mutex<SharedValue<T>>,
on_update: Condvar,
}
struct SharedValue<T> {
value: T,
version: u64,
}
/// Creates a new watch channel.
///
/// The starting value in the channel is not initially considered seen by the receiver.
pub fn channel<T: Clone>(value: T) -> (WatchSender<T>, WatchReceiver<T>) {
let shared = Arc::new(Shared {
lock: Mutex::new(SharedValue { value, version: 1 }),
on_update: Condvar::new(),
});
(
WatchSender {
shared: shared.clone(),
},
WatchReceiver {
shared,
last_seen_version: 0,
},
)
}
impl<T> WatchSender<T> {
/// Send a new message and notify all receivers currently waiting for a
/// message.
pub fn send(&self, mut value: T) {
{
let mut lock = self.shared.lock.lock();
std::mem::swap(&mut lock.value, &mut value);
lock.version = lock.version.wrapping_add(1);
}
self.shared.on_update.notify_all();
// Destroy old value after releasing lock.
drop(value);
}
/// Update the message by a closure and notify all receivers currently waiting for a message.
pub fn update<F>(&self, f: F)
where
F: FnOnce(&mut T),
{
{
let mut lock = self.shared.lock.lock();
f(&mut lock.value);
lock.version = lock.version.wrapping_add(1);
}
self.shared.on_update.notify_all();
}
/// Create a new receiver for the channel.
///
/// Any messages sent before this method was called are considered seen by
/// the new receiver.
pub fn subscribe(&self) -> WatchReceiver<T> {
let version = {
let lock = self.shared.lock.lock();
lock.version
};
WatchReceiver {
shared: self.shared.clone(),
last_seen_version: version,
}
}
}
impl<T: Clone> WatchReceiver<T> {
/// Get a clone of the latest value sent on the channel.
pub fn get(&mut self) -> T {
let lock = self.shared.lock.lock();
self.last_seen_version = lock.version;
lock.value.clone()
}
/// Get a clone of the latest value if that value has not previously been
/// seen by this receiver.
pub fn get_if_new(&mut self) -> Option<T> {
let lock = self.shared.lock.lock();
if self.last_seen_version == lock.version {
return None;
}
self.last_seen_version = lock.version;
Some(lock.value.clone())
}
/// This method waits until a new value becomes available and return a clone
/// of it.
pub fn wait(&mut self) -> T {
let mut lock = self.shared.lock.lock();
while lock.version == self.last_seen_version {
lock = self.shared.on_update.wait(lock);
}
self.last_seen_version = lock.version;
lock.value.clone()
}
/// This method waits until a new value becomes available and return a clone
/// of it, timing out after specified duration.
pub fn wait_timeout(&mut self, duration: Duration) -> Option<T> {
let mut lock = self.shared.lock.lock();
let deadline = Instant::now() + duration;
while lock.version == self.last_seen_version {
let timeout = deadline.saturating_duration_since(Instant::now());
lock = self.shared.on_update.wait_timeout(lock, timeout)?;
// Note: checking after `on_update.wait_timeout` to call it at least once,
// event when `duration` was zero.
if timeout.is_zero() && lock.version == self.last_seen_version {
return None;
}
}
self.last_seen_version = lock.version;
Some(lock.value.clone())
}
}
impl<T> WatchReceiver<T> {
/// Create a new sender for this channel.
pub fn new_sender(&self) -> WatchSender<T> {
WatchSender {
shared: self.shared.clone(),
}
}
}