nstreams-core 0.3.1

Generic versioned event stream handler with history replay
Documentation 
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
39
40
41
42
43
44
45
46

use async_trait::async_trait;

use crate::namespace::Namespace;
use crate::stream::WriteQueueBackend;

/// Publishes events to the write-side queue for durable processing.
#[derive(Clone)]
pub struct EventPublisher<Q: WriteQueueBackend> {
    queue: Q,
}

impl<Q: WriteQueueBackend> EventPublisher<Q> {
    pub fn new(queue: Q) -> Self {
        Self { queue }
    }

    pub async fn publish<N: Namespace>(
        &self,
        namespace: &N,
        payload: &[u8],
        filter_value: Option<&str>,
    ) -> crate::Result<()> {
        self.queue.publish(namespace, payload, filter_value).await
    }
}

/// Convenience trait for typed publishing.
#[async_trait]
pub trait TypedPublisher: Send + Sync {
    async fn publish_typed<N, C>(&self, namespace: &N, codec: &C, event: &C::Event) -> crate::Result<()>
    where
        N: Namespace,
        C: crate::event::EventCodec + ?Sized;
}

#[async_trait]
impl<Q: WriteQueueBackend> TypedPublisher for EventPublisher<Q> {
    async fn publish_typed<N, C>(&self, namespace: &N, codec: &C, event: &C::Event) -> crate::Result<()>
    where
        N: Namespace,
        C: crate::event::EventCodec + ?Sized,
    {
        let payload = codec.encode(event)?;
        self.publish(namespace, &payload, None).await
    }
}