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

//! 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;

#[cfg(not(feature = "parking_lot"))]
mod sync_std;
#[cfg(not(feature = "parking_lot"))]
use sync_std::{Mutex, Condvar};

#[cfg(feature = "parking_lot")]
mod sync_parking_lot;
#[cfg(feature = "parking_lot")]
use sync_parking_lot::{Mutex, Condvar};

/// 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);
    }

    /// 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()
    }
}

impl<T> WatchReceiver<T> {
    /// Create a new sender for this channel.
    pub fn new_sender(&self) -> WatchSender<T> {
        WatchSender {
            shared: self.shared.clone(),
        }
    }
}