azums_core/backend/
stream.rs

1use crate::{
2    backend::NotificationStream,
3    model::{ConsumerGroupStatus, Event, NewEvent},
4};
5use async_trait::async_trait;
6
7/// Interface for append-only, replayable event streams with consumer groups and acknowledgments.
8#[async_trait]
9pub trait StreamBackend: Send + Sync {
10    /// Appends a new event to the specified stream log, returning its assigned sequence number.
11    async fn publish(&self, stream: &str, event: NewEvent) -> anyhow::Result<i64>;
12
13    /// Subscribes to notification events when new entries are appended to a stream.
14    async fn subscribe_stream(
15        &self,
16        stream: &str,
17        consumer_group: &str,
18        last_seq: Option<i64>,
19    ) -> anyhow::Result<NotificationStream>;
20
21    /// Acknowledges event processing up to `seq` for a consumer group on a stream log.
22    async fn ack(&self, stream: &str, consumer_group: &str, seq: i64) -> anyhow::Result<()>;
23
24    /// Reads events from a stream with sequence numbers strictly greater than `after_seq`.
25    async fn read_events(
26        &self,
27        stream: &str,
28        after_seq: i64,
29        limit: i64,
30    ) -> anyhow::Result<Vec<Event>>;
31
32    /// Reads the next events for a consumer group from its durable acknowledged offset.
33    async fn read_next(
34        &self,
35        stream: &str,
36        consumer_group: &str,
37        limit: i64,
38    ) -> anyhow::Result<Vec<Event>> {
39        let last_acked_seq = self
40            .consumer_group_info(stream)
41            .await?
42            .into_iter()
43            .find(|status| status.consumer_group == consumer_group)
44            .map(|status| status.last_acked_seq)
45            .unwrap_or(0);
46
47        self.read_events(stream, last_acked_seq, limit).await
48    }
49
50    /// Prunes retained events for `stream` with `sequence_no <= through_seq`.
51    ///
52    /// Implementations must not prune past the lowest known consumer-group offset. If no consumer
53    /// groups are known for the stream, `through_seq` is used as the retention cutoff.
54    async fn prune_events(&self, stream: &str, through_seq: i64) -> anyhow::Result<u64>;
55
56    /// Fetches consumer group offset status for a stream log.
57    async fn consumer_group_info(&self, stream: &str) -> anyhow::Result<Vec<ConsumerGroupStatus>>;
58}
azums_core/backend/stream.rs

azums_core/backend/
stream.rs
