radiate_engines/events/
subscriber.rs

1use crate::events::{EventStream, Subscription, SubscriptionId};
2use radiate_core::{Executor, error::RadiateResult};
3use std::sync::{Arc, Mutex};
4
5pub trait Event: Send + Sync + 'static {}
6impl<T: Send + Sync + 'static> Event for T {}
7
8pub trait Handler<E: Event>: Send + 'static {
9    fn handle(&mut self, event: &E, ctx: &EventContext<'_, Self>)
10    where
11        Self: Sized;
12}
13
14pub trait EventHandler: Send + 'static {
15    fn start(&mut self, _ctx: &EventContext<'_, Self>) -> RadiateResult<()>
16    where
17        Self: Sized,
18    {
19        Ok(())
20    }
21}
22
23impl<E, F> Handler<E> for F
24where
25    E: Event,
26    F: FnMut(&E) + Send + 'static,
27{
28    fn handle(&mut self, event: &E, _ctx: &EventContext<'_, Self>) {
29        self(event)
30    }
31}
32
33pub struct EventContext<'a, H>(&'a Subscriber<H>);
34
35impl<H> EventContext<'_, H> {
36    pub fn publish<E: Event>(&self, event: E) {
37        self.0.stream.publish(event);
38    }
39
40    pub fn subscribe<E>(&self) -> Subscription
41    where
42        E: Event,
43        H: Handler<E>,
44    {
45        self.0.subscribe::<E>()
46    }
47}
48
49pub struct Subscriber<H> {
50    handler: Arc<Mutex<H>>,
51    executor: Arc<Executor>,
52    stream: EventStream,
53}
54
55impl<H: Send + 'static> Subscriber<H> {
56    pub(super) fn new(handler: H, executor: Arc<Executor>, stream: EventStream) -> Self {
57        Subscriber {
58            handler: Arc::new(Mutex::new(handler)),
59            executor,
60            stream,
61        }
62    }
63
64    pub fn subscribe<E>(&self) -> Subscription
65    where
66        E: Event,
67        H: Handler<E>,
68    {
69        self.stream.subscribe_existing::<E, H>(self)
70    }
71
72    pub fn unsubscribe(&self, id: SubscriptionId) {
73        self.stream.unsubscribe(id);
74    }
75
76    pub(super) fn start(&self) -> RadiateResult<()>
77    where
78        H: EventHandler,
79    {
80        let ctx = EventContext(self);
81        self.handler.lock().unwrap().start(&ctx)
82    }
83
84    pub(super) fn send_shared<E>(&self, event: Arc<E>)
85    where
86        E: Event,
87        H: Handler<E>,
88    {
89        match self.executor.as_ref() {
90            Executor::Serial => {
91                let ctx = EventContext(self);
92                self.handler.lock().unwrap().handle(event.as_ref(), &ctx);
93            }
94            _ => {
95                let owned = self.clone();
96                self.executor.submit(move || {
97                    let ctx = EventContext(&owned);
98                    owned.handler.lock().unwrap().handle(event.as_ref(), &ctx);
99                });
100            }
101        }
102    }
103}
104
105impl<H> Clone for Subscriber<H> {
106    fn clone(&self) -> Self {
107        Subscriber {
108            handler: Arc::clone(&self.handler),
109            executor: Arc::clone(&self.executor),
110            stream: self.stream.clone(),
111        }
112    }
113}
radiate_engines/events/subscriber.rs

radiate_engines/events/
subscriber.rs
