1//! [`SinkCloseChannel`]
23use std::pin::Pin;
45use futures::channel::mpsc;
6use futures::Sink;
78//---------- principal API ----------
910/// 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.
21fn close_channel(self: Pin<&mut Self>);
22}
2324//---------- impl for futures::channel::mpsc ----------
2526impl<T> SinkCloseChannel<T> for mpsc::Sender<T> {
27fn close_channel(self: Pin<&mut Self>) {
28let self_: &mut Self = Pin::into_inner(self);
29 self_.close_channel();
30 }
31}
3233#[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_duration_subtraction)]
45 #![allow(clippy::useless_vec)]
46 #![allow(clippy::needless_pass_by_value)]
47//! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
48#![allow(clippy::arithmetic_side_effects)] // don't mind potential panicking ops in tests
49#![allow(clippy::useless_format)] // sorely
5051use super::*;
52use futures::{SinkExt as _, StreamExt as _};
5354#[test]
55fn close_channel() {
56 tor_rtmock::MockRuntime::test_with_various(|_rt| async move {
57let (mut tx, mut rx) = mpsc::channel::<i32>(20);
58 tx.send(0).await.unwrap();
59let mut tx2 = tx.clone();
60 tx2.send(1).await.unwrap();
61 tx2.close_channel();
62let _: mpsc::SendError = tx.send(66).await.unwrap_err();
63for i in 0..=1 {
64assert_eq!(rx.next().await.unwrap(), i);
65 }
66assert_eq!(rx.next().await, None);
67 });
68 }
69}