nodata 0.1.0

nodata is a kafka like message broker that is simple and easy to use, while relying on either local or s3 like data storage for consistency
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

use crate::state::SharedState;

use super::{
    consumers::Consumers,
    staging::{Staging, StagingEvent},
};

#[derive(Clone)]
pub struct Ingest {
    staging: Staging,
    consumers: Consumers,
}

impl Ingest {
    pub fn new(staging: Staging, consumers: Consumers) -> Self {
        Self { staging, consumers }
    }

    pub async fn publish(&self, event: impl Into<StagingEvent>) -> anyhow::Result<()> {
        let event: StagingEvent = event.into();
        let topic = event.topic.clone();

        let offset = self.staging.publish(event).await?;
        self.consumers.notify_update(topic, offset).await?;

        Ok(())
    }
}

pub trait IngestState {
    fn ingest(&self) -> Ingest;
}

impl IngestState for SharedState {
    fn ingest(&self) -> Ingest {
        Ingest::new(self.staging.clone(), self.consumers.clone())
    }
}