tor_async_utils/
sink_close_channel.rs

1//! [`SinkCloseChannel`]
2
3use std::pin::Pin;
4
5use futures::Sink;
6use futures::channel::mpsc;
7
8//---------- principal API ----------
9
10/// A [`Sink`] with a `close_channel` method like [`futures::channel::mpsc::Sender`]'s.
11pub trait SinkCloseChannel<T>: Sink<T> {
12    /// Close the channel from the sending end, giving EOF at the receiver
13    ///
14    /// Future attempts to send will get a disconnected error.
15    ///
16    /// This closes *all* equivalent senders for the underlying data sink.
17    /// For example, if `Self` is `Clone`, all clones are affected.
18    ///
19    /// If the Sink is for a channel,
20    /// the receiver will see EOF after reading the messages that were successfully sent so far.
21    fn close_channel(self: Pin<&mut Self>);
22}
23
24//---------- impl for futures::channel::mpsc ----------
25
26impl<T> SinkCloseChannel<T> for mpsc::Sender<T> {
27    fn close_channel(self: Pin<&mut Self>) {
28        let self_: &mut Self = Pin::into_inner(self);
29        self_.close_channel();
30    }
31}
32
33#[cfg(test)]
34mod test {
35    // @@ begin test lint list maintained by maint/add_warning @@
36    #![allow(clippy::bool_assert_comparison)]
37    #![allow(clippy::clone_on_copy)]
38    #![allow(clippy::dbg_macro)]
39    #![allow(clippy::mixed_attributes_style)]
40    #![allow(clippy::print_stderr)]
41    #![allow(clippy::print_stdout)]
42    #![allow(clippy::single_char_pattern)]
43    #![allow(clippy::unwrap_used)]
44    #![allow(clippy::unchecked_time_subtraction)]
45    #![allow(clippy::useless_vec)]
46    #![allow(clippy::needless_pass_by_value)]
47    #![allow(clippy::string_slice)] // See arti#2571
48    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
49    #![allow(clippy::arithmetic_side_effects)] // don't mind potential panicking ops in tests
50    #![allow(clippy::useless_format)] // sorely
51
52    use super::*;
53    use futures::{SinkExt as _, StreamExt as _};
54
55    #[test]
56    fn close_channel() {
57        tor_rtmock::MockRuntime::test_with_various(|_rt| async move {
58            let (mut tx, mut rx) = mpsc::channel::<i32>(20);
59            tx.send(0).await.unwrap();
60            let mut tx2 = tx.clone();
61            tx2.send(1).await.unwrap();
62            tx2.close_channel();
63            let _: mpsc::SendError = tx.send(66).await.unwrap_err();
64            for i in 0..=1 {
65                assert_eq!(rx.next().await.unwrap(), i);
66            }
67            assert_eq!(rx.next().await, None);
68        });
69    }
70}
tor_async_utils/sink_close_channel.rs

tor_async_utils/
sink_close_channel.rs
