liminal-server 0.25.4

//! The acknowledgement, tap and live delivery share one allocated identity.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use liminal::channel::PublicationIdentity;
use liminal::error::LiminalError;
use liminal::protocol::DeliveredId;

use super::{LiminalConnectionServices, ServerError};

#[derive(Debug)]
pub(super) struct ServerPublicationIdentity {
    boot: u64,
    next: AtomicU64,
}

impl ServerPublicationIdentity {
    pub(super) fn new() -> Result<Arc<Self>, ServerError> {
        let elapsed = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_err(|_| refused("publication incarnation clock is before epoch"))?;
        let boot = u64::try_from(elapsed.as_millis())
            .map_err(|_| refused("publication incarnation clock overflow"))?;
        Ok(Arc::new(Self {
            boot,
            next: AtomicU64::new(1),
        }))
    }
}

impl PublicationIdentity for ServerPublicationIdentity {
    fn next(&self) -> Result<DeliveredId, LiminalError> {
        let id = self
            .next
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
                value.checked_add(1)
            })
            .map_err(|_| LiminalError::PublishFailed {
                message: "publication identity exhausted".into(),
            })?;
        Ok(DeliveredId {
            boot: self.boot,
            id,
        })
    }
}

impl LiminalConnectionServices {
    pub(super) fn identify_registered(
        &self,
        channel: &super::ConfiguredChannel,
    ) -> Result<(), super::ChannelRegistryError> {
        channel
            .handle
            .install_publication_identity(self.publication_identity.clone())
            .map_err(|error| super::ChannelRegistryError::RosterUnavailable {
                message: error.to_string(),
            })
    }

    pub(super) fn install_roster_identity(&self) -> Result<(), ServerError> {
        let channels = self.read_channels()?;
        for channel in channels.values() {
            channel
                .handle
                .install_publication_identity(self.publication_identity.clone())
                .map_err(|_| refused("channel publication identity already installed"))?;
        }
        drop(channels);
        Ok(())
    }

    pub(super) fn record_delivery_counts(delivery: &liminal::channel::ChannelDelivery) {
        crate::metrics::publish_accepted();
        crate::metrics::deliveries_recorded(
            u64::try_from(delivery.delivered_count()).unwrap_or(u64::MAX),
        );
    }
}

fn refused(message: &str) -> ServerError {
    ServerError::ConfigValidation {
        message: message.into(),
    }
}