liminal-rs 0.16.1

//! What a recording publish saw (SRV-012, the tap's stations and hops): whom
//! its fan-out queued it for and when, and when a durable channel's append
//! completed. Filled by [`ChannelHandle::publish_recording`] into a record
//! its caller keeps and reuses.
//!
//! [`ChannelHandle::publish_recording`]: crate::channel::ChannelHandle::publish_recording

use std::time::Instant;

/// One recipient: the subscriber pid, and the instant its inbox queued the
/// envelope.
pub type Recipient = (u64, Instant);

/// What a recording publish saw. Cleared at the start of each recording
/// publish and its allocation reused, so a caller that keeps one record
/// allocates only while it grows to the widest fan-out.
#[derive(Debug, Default)]
pub struct PublishRecord {
    pub(crate) recipients: Vec<Recipient>,
    pub(crate) appended_at: Option<Instant>,
}

impl PublishRecord {
    /// An empty record with room for `recipients` before it grows.
    #[must_use]
    pub fn with_capacity(recipients: usize) -> Self {
        Self {
            recipients: Vec::with_capacity(recipients),
            appended_at: None,
        }
    }

    /// How many recipients it holds before it grows.
    #[must_use]
    pub fn capacity(&self) -> usize {
        self.recipients.capacity()
    }

    /// Each subscriber whose inbox queued the envelope, with the instant it
    /// did, in fan-out order: exactly
    /// [`ChannelDelivery::delivered_count`](crate::channel::ChannelDelivery::delivered_count)
    /// of them, never one that matched and refused.
    #[must_use]
    pub fn recipients(&self) -> &[Recipient] {
        &self.recipients
    }

    /// When a durable channel's append of the envelope completed, before its
    /// fan-out; `None` for a channel that is not durable.
    #[must_use]
    pub const fn appended_at(&self) -> Option<Instant> {
        self.appended_at
    }

    /// Empties the record, keeping its allocation.
    pub fn clear(&mut self) {
        self.recipients.clear();
        self.appended_at = None;
    }
}

/// The instant a recording publish reads: only ever called while a record
/// is lent, so a publish with none reads no clock. A test build counts the
/// reads, so a pin can show it.
pub fn recorded_now() -> Instant {
    #[cfg(test)]
    CLOCK_READS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    Instant::now()
}

/// Clock reads [`recorded_now`] has made in this process. Test-only.
#[cfg(test)]
pub static CLOCK_READS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);