swansong/implementation/
interrupt.rs

1use crate::{Guard, Inner};
2use event_listener::EventListener;
3use std::{
4    ops::{Deref, DerefMut},
5    sync::{Arc, Weak},
6};
7
8#[cfg(any(feature = "tokio", feature = "futures-io"))]
9mod async_read;
10#[cfg(any(feature = "tokio", feature = "futures-io"))]
11mod async_write;
12
13mod future;
14mod iterator;
15mod stream;
16
17pin_project_lite::pin_project! {
18    /// A wrapper type that implements Stream when wrapping a [`Stream`] and [`Future`] when wrapping a
19    /// Future
20    ///
21    /// When the associated [`Swansong`][crate::Swansong] is stopped with
22    /// [`Swansong::shut_down`][crate::Swansong::shut_down] or all clones of the [`Swansong`] have dropped,
23    /// the Future or Stream within this Stop will wake and return `Poll::Ready(None)` on next poll,
24    /// regardless of where it is being polled.
25    #[derive(Debug)]
26    pub struct Interrupt<T> {
27        inner: WeakInner,
28        #[pin]
29        wrapped_type: T,
30        guard: Option<Guard>,
31        stop_listener: StopListener,
32    }
33}
34
35impl<T: Eq> Eq for Interrupt<T> {}
36
37impl<T, U> PartialEq<Interrupt<U>> for Interrupt<T>
38where
39    T: PartialEq<U>,
40{
41    fn eq(&self, other: &Interrupt<U>) -> bool {
42        self.inner.ptr_eq(&other.inner) && self.wrapped_type == other.wrapped_type
43    }
44}
45
46impl<T> Interrupt<T> {
47    pub(crate) fn new(inner: &Arc<Inner>, wrapped_type: T) -> Self {
48        Self {
49            inner: WeakInner(Arc::downgrade(inner)),
50            wrapped_type,
51            guard: None,
52            stop_listener: StopListener(None),
53        }
54    }
55
56    /// Chainable setter to delay shutdown until this wrapper type has dropped.
57    ///
58    /// The guard records the source location of this call for diagnostic
59    /// purposes; see [`Swansong::guard_report`][crate::Swansong::guard_report].
60    #[must_use]
61    #[track_caller]
62    pub fn guarded(mut self) -> Self {
63        if let Some(inner) = self.inner.upgrade() {
64            self.guard = Some(Guard::new(&inner));
65        }
66        self
67    }
68
69    /// Take the wrapped type out of this Interrupt.
70    ///
71    /// If the interrupt is guarded with [`Interrupt::guarded`], this will decrement the guard count.
72    pub fn into_inner(self) -> T {
73        self.wrapped_type
74    }
75
76    pub(crate) fn is_stopped(&self) -> bool {
77        self.inner.is_stopped()
78    }
79
80    #[cfg(any(feature = "futures-io", feature = "tokio"))]
81    pub(crate) fn is_stopped_relaxed(&self) -> bool {
82        self.inner.is_stopped_relaxed()
83    }
84}
85
86impl<T> Deref for Interrupt<T> {
87    type Target = T;
88
89    fn deref(&self) -> &Self::Target {
90        &self.wrapped_type
91    }
92}
93
94impl<T> DerefMut for Interrupt<T> {
95    fn deref_mut(&mut self) -> &mut Self::Target {
96        &mut self.wrapped_type
97    }
98}
99
100#[derive(Debug)]
101struct WeakInner(Weak<Inner>);
102impl Deref for WeakInner {
103    type Target = Weak<Inner>;
104
105    fn deref(&self) -> &Self::Target {
106        &self.0
107    }
108}
109impl WeakInner {
110    fn is_stopped(&self) -> bool {
111        self.upgrade().as_deref().is_none_or(Inner::is_stopped)
112    }
113    fn is_stopped_relaxed(&self) -> bool {
114        self.upgrade()
115            .as_deref()
116            .is_none_or(Inner::is_stopped_relaxed)
117    }
118}
119
120#[derive(Debug)]
121struct StopListener(Option<EventListener>);
122impl StopListener {
123    fn listen(&mut self, weak_inner: &WeakInner) -> Option<&mut EventListener> {
124        let Self(listener) = self;
125        if let Some(listener) = listener {
126            return Some(listener);
127        }
128        let inner = weak_inner.upgrade()?;
129        let listener = listener.insert(inner.listen_stop());
130        if inner.is_stopped() {
131            log::trace!("inner was stopped after registering new listener");
132            None
133        } else {
134            log::trace!("registered new listener");
135            Some(listener)
136        }
137    }
138}
139impl Deref for StopListener {
140    type Target = Option<EventListener>;
141
142    fn deref(&self) -> &Self::Target {
143        &self.0
144    }
145}
146impl DerefMut for StopListener {
147    fn deref_mut(&mut self) -> &mut Self::Target {
148        &mut self.0
149    }
150}
swansong/implementation/interrupt.rs

swansong/implementation/
interrupt.rs
