nstreams_core/
worker.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use tracing::{debug, error, info, warn};
5
6use crate::handler::NStreamsHandler;
7use crate::namespace::{Namespace, StreamNamespace};
8use crate::store::EventStore;
9use crate::stream::{ReadStreamBackend, WriteQueueBackend};
10
11/// Consumes the write queue, persists events, and pipes to ready read streams.
12pub struct StreamWorker<S, Q, R> {
13    store: Arc<S>,
14    read_stream: Arc<R>,
15    handler: Arc<NStreamsHandler<S, R>>,
16    _queue: std::marker::PhantomData<Q>,
17}
18
19impl<S, Q, R> StreamWorker<S, Q, R>
20where
21    S: EventStore + 'static,
22    Q: WriteQueueBackend + 'static,
23    R: ReadStreamBackend + 'static,
24{
25    pub fn new(
26        store: S,
27        _queue: Q,
28        read_stream: R,
29        handler: NStreamsHandler<S, R>,
30    ) -> Self {
31        Self {
32            store: Arc::new(store),
33            read_stream: Arc::new(read_stream),
34            handler: Arc::new(handler),
35            _queue: std::marker::PhantomData,
36        }
37    }
38
39    pub async fn run<QC: QueueConsumer>(&self, mut consumer: QC) -> crate::Result<()> {
40        info!("stream worker started");
41        while let Some(delivery) = consumer.next().await {
42            match delivery {
43                Ok(message) => {
44                    if let Err(error) = self.process_message(&message.namespace, &message.payload).await
45                    {
46                        error!(namespace = %message.namespace, %error, "failed to process queue message");
47                    }
48                }
49                Err(error) => {
50                    warn!(%error, "queue consumer error");
51                }
52            }
53        }
54        Ok(())
55    }
56
57    async fn process_message(&self, namespace: &str, payload: &[u8]) -> crate::Result<()> {
58        let namespace = crate::namespace::StreamNamespace::from(namespace);
59        StreamNamespace::validate(namespace.as_str())?;
60        self.store.ensure_namespace(&namespace).await?;
61
62        let event = self.store.persist_next(&namespace, payload).await?;
63        debug!(
64            namespace = namespace.as_str(),
65            version = event.version,
66            "persisted event"
67        );
68
69        if self.handler.is_read_stream_ready(&namespace).await {
70            self.read_stream.publish_to_stream(&event).await?;
71        }
72
73        Ok(())
74    }
75}
76
77/// Raw queue message consumed by the worker.
78#[derive(Debug, Clone)]
79pub struct QueueMessage {
80    pub namespace: String,
81    pub payload: Vec<u8>,
82}
83
84/// Abstraction over the RabbitMQ queue consumer loop.
85#[async_trait]
86pub trait QueueConsumer: Send {
87    async fn next(&mut self) -> Option<std::result::Result<QueueMessage, crate::Error>>;
88}
89
nstreams_core/worker.rs

nstreams_core/
worker.rs
