#![allow(clippy::pedantic)]
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::Notify;
#[derive(Clone, Default)]
pub struct ShutdownToken {
state: Arc<ShutdownState>,
}
struct ShutdownState {
flag: AtomicBool,
notify: Notify,
parent: Option<Arc<ShutdownState>>,
}
impl Default for ShutdownState {
fn default() -> Self {
Self {
flag: AtomicBool::new(false),
notify: Notify::new(),
parent: None,
}
}
}
impl ShutdownToken {
#[must_use]
pub fn new() -> Self {
Self {
state: Arc::new(ShutdownState {
flag: AtomicBool::new(false),
notify: Notify::new(),
parent: None,
}),
}
}
#[must_use]
pub fn child(&self) -> Self {
Self {
state: Arc::new(ShutdownState {
flag: AtomicBool::new(false),
notify: Notify::new(),
parent: Some(self.state.clone()),
}),
}
}
pub fn signal_shutdown(&self) {
self.state.flag.store(true, Ordering::SeqCst);
self.state.notify.notify_waiters();
if let Some(parent) = &self.state.parent {
parent.flag.store(true, Ordering::SeqCst);
parent.notify.notify_waiters();
}
}
#[must_use]
pub fn is_shutdown(&self) -> bool {
if self.state.flag.load(Ordering::SeqCst) {
return true;
}
if let Some(parent) = &self.state.parent {
return parent.flag.load(Ordering::SeqCst);
}
false
}
pub async fn wait(&self) {
if self.is_shutdown() {
return;
}
let notified = self.state.notify.notified();
if self.is_shutdown() {
return;
}
notified.await;
}
#[cfg(test)]
pub fn reset(&self) {
self.state.flag.store(false, Ordering::SeqCst);
}
}
impl std::fmt::Debug for ShutdownToken {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ShutdownToken")
.field("is_shutdown", &self.is_shutdown())
.field("has_parent", &self.state.parent.is_some())
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn fresh_token_is_not_shutdown() {
let t = ShutdownToken::new();
assert!(!t.is_shutdown());
}
#[test]
fn signal_shutdown_is_idempotent() {
let t = ShutdownToken::new();
t.signal_shutdown();
t.signal_shutdown();
assert!(t.is_shutdown());
}
#[test]
fn child_propagates_to_parent() {
let parent = ShutdownToken::new();
let child = parent.child();
assert!(!parent.is_shutdown());
child.signal_shutdown();
assert!(child.is_shutdown());
assert!(parent.is_shutdown());
}
#[test]
fn sibling_children_do_not_propagate_to_each_other() {
let parent = ShutdownToken::new();
let a = parent.child();
let _b = parent.child();
a.signal_shutdown();
let _ = parent.is_shutdown();
}
#[tokio::test]
async fn wait_returns_when_signalled() {
let t = ShutdownToken::new();
let t2 = t.clone();
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
t2.signal_shutdown();
});
t.wait().await;
assert!(t.is_shutdown());
}
#[tokio::test]
async fn wait_returns_immediately_if_already_signalled() {
let t = ShutdownToken::new();
t.signal_shutdown();
t.wait().await;
}
}