liminal-server 0.23.1

//! Cancellation follows the currently selected carrier, never another link's device context.
use super::Stop;
use crate::link_open::OpenError;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicBool, Ordering},
};
#[derive(Clone)]
pub struct StopSlot(Arc<State>);
struct State {
    current: Mutex<Option<Stop>>,
    stopped: AtomicBool,
}
impl std::fmt::Debug for StopSlot {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StopSlot").finish_non_exhaustive()
    }
}
impl StopSlot {
    pub(crate) fn new(stop: Option<Stop>) -> Self {
        Self(Arc::new(State {
            current: Mutex::new(stop),
            stopped: AtomicBool::new(false),
        }))
    }
    pub(crate) fn set(&self, stop: Option<Stop>) -> Result<(), OpenError> {
        if self.stopped() {
            if let Some(stop) = stop {
                stop()?;
            }
            return Err(OpenError("link_stopped".to_owned()));
        }
        let mut current = self
            .0
            .current
            .lock()
            .map_err(|_| OpenError("carrier_stop_state_poisoned".to_owned()))?;
        if self.stopped() {
            drop(current);
            if let Some(stop) = stop {
                stop()?;
            }
            return Err(OpenError("link_stopped".to_owned()));
        }
        *current = stop;
        drop(current);
        Ok(())
    }
    pub(crate) fn stopped(&self) -> bool {
        self.0.stopped.load(Ordering::Acquire)
    }
    pub(crate) fn handle(&self) -> Result<Stop, OpenError> {
        if self
            .0
            .current
            .lock()
            .map_err(|_| OpenError("carrier_stop_state_poisoned".to_owned()))?
            .is_none()
        {
            return Err(OpenError("carrier_has_no_shutdown_interrupt".to_owned()));
        }
        let state = self.0.clone();
        Ok(Arc::new(move || {
            let current = state
                .current
                .lock()
                .map_err(|_| OpenError("carrier_stop_state_poisoned".to_owned()))?;
            if state.stopped.swap(true, Ordering::AcqRel) {
                return Ok(());
            }
            let stop = current
                .clone()
                .ok_or_else(|| OpenError("carrier_has_no_shutdown_interrupt".to_owned()))?;
            drop(current);
            stop()
        }))
    }
}