liminal-server 0.24.2

//! Observers wake after a durable append and never participate in its admission.

use super::SelfStreams;
use crate::ServerError;
use crate::link_bridge::subscriptions::ExportSubscriptions;
use std::sync::{Arc, OnceLock, Weak};

#[derive(Default)]
pub(super) struct ExportObservers {
    first: OnceLock<Box<Observer>>,
}

struct Observer {
    feeds: Weak<ExportSubscriptions>,
    next: OnceLock<Box<Self>>,
}

impl ExportObservers {
    pub(super) fn install(&self, feeds: &Arc<ExportSubscriptions>) -> Result<(), ServerError> {
        let mut next = &self.first;
        let mut observer = Box::new(Observer {
            feeds: Arc::downgrade(feeds),
            next: OnceLock::new(),
        });
        loop {
            match next.set(observer) {
                Ok(()) => return Ok(()),
                Err(uninstalled) => {
                    observer = uninstalled;
                    next = &next
                        .get()
                        .ok_or_else(|| {
                            SelfStreams::failure("bridge observer registration refused")
                        })?
                        .next;
                }
            }
        }
    }

    pub(super) fn published(&self, subject: &str, head: u64) {
        let mut next = self.first.get();
        while let Some(observer) = next {
            if let Some(feeds) = observer.feeds.upgrade() {
                feeds.stored_published(subject, head);
            }
            next = observer.next.get();
        }
    }
}