camber 0.1.6

Opinionated async Rust for IO-bound services on top of Tokio
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

use crate::RuntimeError;
use std::fmt;
use std::ops::Deref;
use tokio::sync::watch as tokio_watch;

/// Create a watch channel with an initial value.
///
/// A watch channel holds a single value. Receivers always see the latest
/// value — intermediate writes are skipped. Use this for configuration,
/// state snapshots, or shutdown signals where only the current value matters.
///
/// The sender can update the value; all receivers see the change.
/// Receivers can be cloned cheaply.
pub fn watch<T>(initial: T) -> (WatchSender<T>, WatchReceiver<T>) {
    let (tx, rx) = tokio_watch::channel(initial);
    (WatchSender { inner: tx }, WatchReceiver { inner: rx })
}

/// Sending half of a watch channel.
///
/// Holds the current value. Dropping the sender closes the channel —
/// receivers' `changed()` futures will resolve with `ChannelClosed`.
pub struct WatchSender<T> {
    inner: tokio_watch::Sender<T>,
}

impl<T> WatchSender<T> {
    /// Replace the current value. All receivers see the update.
    ///
    /// Returns `ChannelClosed` if all receivers have been dropped.
    pub fn send(&self, value: T) -> Result<(), RuntimeError> {
        self.inner
            .send(value)
            .map_err(|_| RuntimeError::ChannelClosed)
    }

    /// Modify the current value in place. Notifies receivers even if the
    /// value is unchanged.
    ///
    /// Unlike `send()`, this always succeeds — the sender owns the stored
    /// value regardless of whether receivers exist. The mutation is applied
    /// and visible via `borrow()` even if all receivers have been dropped.
    pub fn send_modify<F: FnOnce(&mut T)>(&self, modify: F) {
        self.inner.send_modify(modify);
    }

    /// Read the current value.
    pub fn borrow(&self) -> WatchRef<'_, T> {
        WatchRef {
            inner: self.inner.borrow(),
        }
    }
}

impl<T> Clone for WatchSender<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl<T: fmt::Debug> fmt::Debug for WatchSender<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WatchSender").finish()
    }
}

/// Receiving half of a watch channel. Cloneable — each clone tracks its own
/// "seen" state independently.
pub struct WatchReceiver<T> {
    inner: tokio_watch::Receiver<T>,
}

impl<T> WatchReceiver<T> {
    /// Read the current value without marking it as seen. Use
    /// `borrow_and_update()` to also mark the value as seen for
    /// `has_changed()` tracking.
    pub fn borrow(&self) -> WatchRef<'_, T> {
        WatchRef {
            inner: self.inner.borrow(),
        }
    }

    /// Read the current value and mark it as seen. After this call,
    /// `has_changed()` returns `false` until the sender writes again.
    pub fn borrow_and_update(&mut self) -> WatchRef<'_, T> {
        WatchRef {
            inner: self.inner.borrow_and_update(),
        }
    }

    /// Wait until the value changes. Marks the new value as seen.
    /// Returns `ChannelClosed` when the sender is dropped.
    pub async fn changed(&mut self) -> Result<(), RuntimeError> {
        self.inner
            .changed()
            .await
            .map_err(|_| RuntimeError::ChannelClosed)
    }

    /// Check if the value has changed since the receiver was created, or
    /// since the last call to `changed()` or `borrow_and_update()`.
    pub fn has_changed(&self) -> bool {
        self.inner.has_changed().unwrap_or(false)
    }
}

impl<T> Clone for WatchReceiver<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl<T: fmt::Debug> fmt::Debug for WatchReceiver<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WatchReceiver").finish()
    }
}

/// RAII borrow of the current watch value. Dereferences to `T`.
pub struct WatchRef<'a, T> {
    inner: tokio_watch::Ref<'a, T>,
}

impl<T> Deref for WatchRef<'_, T> {
    type Target = T;

    fn deref(&self) -> &T {
        &self.inner
    }
}

impl<T: fmt::Debug> fmt::Debug for WatchRef<'_, T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.inner.fmt(f)
    }
}