tor_async_utils/
watch.rs

1//! Extension trait for more efficient use of [`postage::watch`].
2
3use extend::ext;
4use std::ops::{Deref, DerefMut};
5use void::{ResultVoidExt as _, Void};
6
7/// Extension trait for some `postage::watch::Sender` to provide `maybe_send`
8///
9/// Ideally these, or something like them, would be upstream:
10/// See <https://github.com/austinjones/postage-rs/issues/56>.
11///
12/// We provide this as an extension trait became the implementation is a bit fiddly.
13/// This lets us concentrate on the actual logic, when we use it.
14#[ext(name = PostageWatchSenderExt)]
15pub impl<T> postage::watch::Sender<T> {
16    /// Update, by calling a fallible function, sending only if necessary
17    ///
18    /// Calls `update` on the current value in the watch, to obtain a new value.
19    /// If the new value doesn't compare equal, updates the watch, notifying receivers.
20    fn try_maybe_send<F, E>(&mut self, update: F) -> Result<(), E>
21    where
22        T: PartialEq,
23        F: FnOnce(&T) -> Result<T, E>,
24    {
25        let lock = self.borrow();
26        let new = update(&*lock)?;
27        if new != *lock {
28            // We must drop the lock guard, because otherwise borrow_mut will deadlock.
29            // There is no race, because we hold &mut self, so no-one else can get a look in.
30            // (postage::watch::Sender is not one of those facilities which is mereely a
31            // handle, and Clone.)
32            drop(lock);
33            *self.borrow_mut() = new;
34        }
35        Ok(())
36    }
37
38    /// Update, by calling a function, sending only if necessary
39    ///
40    /// Calls `update` on the current value in the watch, to obtain a new value.
41    /// If the new value doesn't compare equal, updates the watch, notifying receivers.
42    fn maybe_send<F>(&mut self, update: F)
43    where
44        T: PartialEq,
45        F: FnOnce(&T) -> T,
46    {
47        self.try_maybe_send(|t| Ok::<_, Void>(update(t)))
48            .void_unwrap();
49    }
50}
51
52#[derive(Debug)]
53/// Wrapper for `postage::watch::Sender` that sends `DropNotifyEof::eof()` when dropped
54///
55/// Derefs to the inner `Sender`.
56///
57/// Ideally this would be behaviour promised by upstream, or something
58/// See <https://github.com/austinjones/postage-rs/issues/57>.
59pub struct DropNotifyWatchSender<T: DropNotifyEofSignallable>(Option<postage::watch::Sender<T>>);
60
61/// Values that can signal EOF
62///
63/// Implemented for `Option`, which is usually what you want to use.
64pub trait DropNotifyEofSignallable {
65    /// Generate the EOF value
66    fn eof() -> Self;
67
68    /// Does this value indicate EOF?
69    ///
70    /// ### Deprecated
71    ///
72    /// This method is deprecated.
73    /// It should not be called, or defined, in new programs.
74    /// It is not required by [`DropNotifyWatchSender`].
75    /// The provided implementation always returns `false`.
76    #[deprecated]
77    fn is_eof(&self) -> bool {
78        false
79    }
80}
81
82impl<T> DropNotifyEofSignallable for Option<T> {
83    fn eof() -> Self {
84        None
85    }
86
87    fn is_eof(&self) -> bool {
88        self.is_none()
89    }
90}
91
92impl<T: DropNotifyEofSignallable> DropNotifyWatchSender<T> {
93    /// Arrange to send `T::Default` when `inner` is dropped
94    pub fn new(inner: postage::watch::Sender<T>) -> Self {
95        DropNotifyWatchSender(Some(inner))
96    }
97
98    /// Unwrap the inner sender, defusing the drop notification
99    pub fn into_inner(mut self) -> postage::watch::Sender<T> {
100        self.0.take().expect("inner was None")
101    }
102}
103
104impl<T: DropNotifyEofSignallable> Deref for DropNotifyWatchSender<T> {
105    type Target = postage::watch::Sender<T>;
106    fn deref(&self) -> &Self::Target {
107        self.0.as_ref().expect("inner was None")
108    }
109}
110
111impl<T: DropNotifyEofSignallable> DerefMut for DropNotifyWatchSender<T> {
112    fn deref_mut(&mut self) -> &mut Self::Target {
113        self.0.as_mut().expect("inner was None")
114    }
115}
116
117impl<T: DropNotifyEofSignallable> Drop for DropNotifyWatchSender<T> {
118    fn drop(&mut self) {
119        if let Some(mut inner) = self.0.take() {
120            // None means into_inner() was called
121            *inner.borrow_mut() = DropNotifyEofSignallable::eof();
122        }
123    }
124}
125
126#[cfg(test)]
127mod test {
128    // @@ begin test lint list maintained by maint/add_warning @@
129    #![allow(clippy::bool_assert_comparison)]
130    #![allow(clippy::clone_on_copy)]
131    #![allow(clippy::dbg_macro)]
132    #![allow(clippy::mixed_attributes_style)]
133    #![allow(clippy::print_stderr)]
134    #![allow(clippy::print_stdout)]
135    #![allow(clippy::single_char_pattern)]
136    #![allow(clippy::unwrap_used)]
137    #![allow(clippy::unchecked_time_subtraction)]
138    #![allow(clippy::useless_vec)]
139    #![allow(clippy::needless_pass_by_value)]
140    #![allow(clippy::string_slice)] // See arti#2571
141    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
142
143    use super::*;
144    use futures::select_biased;
145    use futures_await_test::async_test;
146
147    #[async_test]
148    async fn postage_sender_ext() {
149        use futures::FutureExt;
150        use futures::stream::StreamExt;
151
152        let (mut s, mut r) = postage::watch::channel_with(20);
153        // Receiver of a fresh watch wakes once, but let's not rely on this
154        select_biased! {
155            i = r.next().fuse() => assert_eq!(i, Some(20)),
156            _ = futures::future::ready(()) => { }, // tolerate nothing
157        };
158        // Now, not ready
159        select_biased! {
160            _ = r.next().fuse() => panic!(),
161            _ = futures::future::ready(()) => { },
162        };
163
164        s.maybe_send(|i| *i);
165        // Still not ready
166        select_biased! {
167            _ = r.next().fuse() => panic!(),
168            _ = futures::future::ready(()) => { },
169        };
170
171        s.maybe_send(|i| *i + 1);
172        // Ready, with 21
173        select_biased! {
174            i = r.next().fuse() => assert_eq!(i, Some(21)),
175            _ = futures::future::ready(()) => panic!(),
176        };
177
178        let () = s.try_maybe_send(|_i| Err(())).unwrap_err();
179        // Not ready
180        select_biased! {
181            _ = r.next().fuse() => panic!(),
182            _ = futures::future::ready(()) => { },
183        };
184    }
185
186    #[test]
187    fn postage_drop() {
188        #[derive(Clone, Copy, Debug, Eq, PartialEq)]
189        struct I(i32);
190
191        impl DropNotifyEofSignallable for I {
192            fn eof() -> I {
193                I(0)
194            }
195            fn is_eof(&self) -> bool {
196                self.0 == 0
197            }
198        }
199
200        let (s, r) = postage::watch::channel_with(I(20));
201        let s = DropNotifyWatchSender::new(s);
202
203        assert_eq!(*r.borrow(), I(20));
204        drop(s);
205        assert_eq!(*r.borrow(), I(0));
206
207        let (s, r) = postage::watch::channel_with(I(44));
208        let s = DropNotifyWatchSender::new(s);
209
210        assert_eq!(*r.borrow(), I(44));
211        drop(s.into_inner());
212        assert_eq!(*r.borrow(), I(44));
213    }
214}
tor_async_utils/watch.rs

tor_async_utils/
watch.rs
