typed_eventbus/
lib.rs

1use async_trait::async_trait;
2use serde::{Serialize, de::DeserializeOwned};
3use serde_json::to_string;
4use std::pin::Pin;
5use std::{future::Future, marker::PhantomData};
6mod envelop;
7pub use envelop::{Event, EventMetaData, Identifier};
8use std::sync::Arc;
9pub type EventError = Box<dyn std::error::Error + Send + Sync>;
10pub type EventHandler =
11    Box<dyn Fn(Vec<u8>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
12
13#[async_trait]
14pub trait Handler: Send + Sync + 'static {
15    async fn handle(&self, subject: String, message: Vec<u8>);
16}
17
18pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
19
20pub trait EventStream: Send + Sync {
21    fn publish<'a>(
22        &'a self,
23        subject: String,
24        payload: Vec<u8>,
25    ) -> BoxFuture<'a, Result<(), EventError>>;
26
27    fn subscribe<'a>(
28        &'a self,
29        subject: String,
30        handler: Arc<dyn Handler>,
31    ) -> BoxFuture<'a, Result<(), EventError>>;
32}
33mod local;
34pub use local::LocalEventStream;
35mod nats;
36pub use nats::NatsEventStream;
37
38#[async_trait]
39pub trait Publishable: Serialize {
40    const SUBJECT: &'static str;
41
42    async fn publish(&self, bus: Arc<dyn EventStream>) -> Result<(), EventError> {
43        bus.publish(
44            Self::SUBJECT.to_string(),
45            to_string(self).unwrap().into_bytes(),
46        )
47        .await
48    }
49
50    async fn subscribe(
51        bus: Arc<dyn EventStream>,
52        handler: Arc<dyn Handler>,
53    ) -> Result<(), EventError> {
54        bus.subscribe(Self::SUBJECT.to_string(), handler).await
55    }
56}
57
58pub trait Subscribable: DeserializeOwned + Send + Sync + 'static {
59    const SUBJECT: &'static str;
60}
61
62#[async_trait]
63pub trait Subscriber<T: Subscribable>: Send + Sync + Sized + 'static {
64    // Now receives the full Event<T> with metadata
65    async fn on_message(&self, event: Event<T>, subject: &str);
66
67    async fn subscribe(self, es: Arc<dyn EventStream>) -> Result<(), EventError> {
68        struct MessageHandler<C: Subscriber<T> + Send + Sync + 'static, T: Subscribable> {
69            subscriber: C,
70            _marker: PhantomData<T>,
71        }
72
73        #[async_trait]
74        impl<C: Subscriber<T> + Send + Sync + 'static, T: Subscribable> Handler for MessageHandler<C, T> {
75            async fn handle(&self, subject: String, message: Vec<u8>) {
76                // Deserialize the full Event<T>
77                match serde_json::from_slice::<Event<T>>(&message) {
78                    Ok(event) => {
79                        // Pass both payload and metadata
80                        self.subscriber.on_message(event, &subject).await;
81                    }
82                    Err(e) => {
83                        eprintln!("Failed to deserialize event on {}: {}", subject, e);
84                    }
85                }
86            }
87        }
88
89        let handler = Arc::new(MessageHandler::<Self, T> {
90            subscriber: self,
91            _marker: PhantomData,
92        });
93
94        es.subscribe(T::SUBJECT.to_string(), handler).await
95    }
96}
typed_eventbus/lib.rs

typed_eventbus/
lib.rs
