dreamwell-runtime 1.0.0

//! Staging buffer — holds authority events until tick boundary application.
//!
//! Events from the authority adapter are staged here during PollAuthority,
//! then applied atomically during ApplyAuthorityEvents phase.

use crate::authority::AuthorityEvent;

/// Staging buffer for authority events.
/// Events are pushed during poll, consumed during apply.
pub struct StagingBuffer {
    events: Vec<AuthorityEvent>,
}

impl StagingBuffer {
    pub fn new() -> Self {
        Self {
            events: Vec::with_capacity(64),
        }
    }

    /// Stage an event for later application.
    pub fn push(&mut self, event: AuthorityEvent) {
        self.events.push(event);
    }

    /// Stage multiple events.
    pub fn extend(&mut self, events: impl IntoIterator<Item = AuthorityEvent>) {
        self.events.extend(events);
    }

    /// Drain all staged events for tick-boundary application.
    pub fn drain(&mut self) -> std::vec::Drain<'_, AuthorityEvent> {
        self.events.drain(..)
    }

    /// Number of staged events.
    pub fn len(&self) -> usize {
        self.events.len()
    }

    /// Whether the buffer is empty.
    pub fn is_empty(&self) -> bool {
        self.events.is_empty()
    }

    /// Discard all staged events.
    pub fn clear(&mut self) {
        self.events.clear();
    }
}

impl Default for StagingBuffer {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn push_and_drain() {
        let mut buf = StagingBuffer::new();
        buf.push(AuthorityEvent::Ack { seq: 1 });
        buf.push(AuthorityEvent::Ack { seq: 2 });
        assert_eq!(buf.len(), 2);

        let drained: Vec<_> = buf.drain().collect();
        assert_eq!(drained.len(), 2);
        assert!(buf.is_empty());
    }

    #[test]
    fn extend_from_iter() {
        let mut buf = StagingBuffer::new();
        let events = vec![
            AuthorityEvent::Ack { seq: 1 },
            AuthorityEvent::Reject {
                seq: 2,
                reason: "test".into(),
            },
        ];
        buf.extend(events);
        assert_eq!(buf.len(), 2);
    }

    #[test]
    fn clear_discards() {
        let mut buf = StagingBuffer::new();
        buf.push(AuthorityEvent::Ack { seq: 1 });
        buf.clear();
        assert!(buf.is_empty());
    }
}