nstreams_core/
publisher.rs

1use async_trait::async_trait;
2
3use crate::namespace::Namespace;
4use crate::stream::WriteQueueBackend;
5
6/// Publishes events to the write-side queue for durable processing.
7#[derive(Clone)]
8pub struct EventPublisher<Q: WriteQueueBackend> {
9    queue: Q,
10}
11
12impl<Q: WriteQueueBackend> EventPublisher<Q> {
13    pub fn new(queue: Q) -> Self {
14        Self { queue }
15    }
16
17    pub async fn publish<N: Namespace>(
18        &self,
19        namespace: &N,
20        payload: &[u8],
21        filter_value: Option<&str>,
22    ) -> crate::Result<()> {
23        self.queue.publish(namespace, payload, filter_value).await
24    }
25}
26
27/// Convenience trait for typed publishing.
28#[async_trait]
29pub trait TypedPublisher: Send + Sync {
30    async fn publish_typed<N, C>(&self, namespace: &N, codec: &C, event: &C::Event) -> crate::Result<()>
31    where
32        N: Namespace,
33        C: crate::event::EventCodec + ?Sized;
34}
35
36#[async_trait]
37impl<Q: WriteQueueBackend> TypedPublisher for EventPublisher<Q> {
38    async fn publish_typed<N, C>(&self, namespace: &N, codec: &C, event: &C::Event) -> crate::Result<()>
39    where
40        N: Namespace,
41        C: crate::event::EventCodec + ?Sized,
42    {
43        let payload = codec.encode(event)?;
44        self.publish(namespace, &payload, None).await
45    }
46}
nstreams_core/publisher.rs

nstreams_core/
publisher.rs
