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
45                        .process_message(
46                            &message.namespace,
47                            &message.payload,
48                            message.filter_value.as_deref(),
49                        )
50                        .await
51                    {
52                        error!(namespace = %message.namespace, %error, "failed to process queue message");
53                    }
54                }
55                Err(error) => {
56                    warn!(%error, "queue consumer error");
57                }
58            }
59        }
60        Ok(())
61    }
62
63    async fn process_message(
64        &self,
65        namespace: &str,
66        payload: &[u8],
67        filter_value: Option<&str>,
68    ) -> crate::Result<()> {
69        let namespace = crate::namespace::StreamNamespace::from(namespace);
70        StreamNamespace::validate(namespace.as_str())?;
71        self.store.ensure_namespace(&namespace).await?;
72
73        let event = self
74            .store
75            .persist_next(&namespace, payload, filter_value)
76            .await?;
77        debug!(
78            namespace = namespace.as_str(),
79            version = event.version,
80            "persisted event"
81        );
82
83        if self.handler.is_read_stream_ready(&namespace).await {
84            self.read_stream.publish_to_stream(&event).await?;
85        }
86
87        Ok(())
88    }
89}
90
91/// Raw queue message consumed by the worker.
92#[derive(Debug, Clone)]
93pub struct QueueMessage {
94    pub namespace: String,
95    pub payload: Vec<u8>,
96    pub filter_value: Option<String>,
97}
98
99/// Abstraction over the RabbitMQ queue consumer loop.
100#[async_trait]
101pub trait QueueConsumer: Send {
102    async fn next(&mut self) -> Option<std::result::Result<QueueMessage, crate::Error>>;
103}
104
nstreams_core/worker.rs

nstreams_core/
worker.rs
