liminal-server 0.25.4

//! Existing test signals expose the hub's actual transitions.

use super::{ConsoleHub, TapError, TapSubscription};

impl ConsoleHub {
    /// How many topology changes reached [`Self::emit`]. Test-only.
    #[cfg(test)]
    pub(crate) fn emits(&self) -> u64 {
        self.emits.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Each connection's participant as the model holds it. Test-only.
    #[cfg(test)]
    pub(crate) fn participants(&self) -> Vec<Option<String>> {
        self.topology
            .lock()
            .map(|state| state.model.participants())
            .unwrap_or_default()
    }

    /// How many times the topology lock was taken. Test-only.
    #[cfg(test)]
    pub(crate) fn topology_locks(&self) -> u64 {
        self.topology_locks
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Waits until the model holds `count` connections and answers whether
    /// it does; the guard is a minute, never a deadline. Test-only.
    #[cfg(test)]
    pub(crate) fn wait_for_connections(&self, count: usize) -> bool {
        self.topology
            .lock()
            .ok()
            .and_then(|state| {
                self.applied
                    .wait_timeout_while(state, std::time::Duration::from_secs(60), |state| {
                        state.model.connection_count() != count
                    })
                    .ok()
            })
            .is_some_and(|(_, guard)| !guard.timed_out())
    }

    /// Makes the next take of `subject`'s tap fail once. Test-only.
    #[cfg(test)]
    pub(crate) fn fail_next_take(&self, subject: &str) {
        if let Ok(mut failing) = self.failing_take.lock() {
            *failing = Some(subject.to_owned());
        }
    }

    /// Takes `subscription`'s batches, unless [`Self::fail_next_take`] named
    /// `subject`: that take fails once. Test-only.
    #[cfg(test)]
    pub(super) fn take_batches(
        &self,
        subject: &str,
        subscription: &TapSubscription,
    ) -> Result<Vec<crate::tap::TapBatch>, TapError> {
        if let Ok(mut failing) = self.failing_take.lock()
            && failing.as_deref() == Some(subject)
        {
            *failing = None;
            return Err(TapError::Poisoned {
                what: "tap subscriber",
            });
        }
        subscription.take_batches()
    }
}