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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
use futures::task::{Context, Poll};
use futures::{Stream, StreamExt};
use once_cell::sync::Lazy;
use serde::export::PhantomData;
use slab::Slab;
use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Mutex;
static SUBSCRIBERS: Lazy<Mutex<HashMap<TypeId, Box<dyn Any + Send>>>> = Lazy::new(Default::default);
struct Senders<T>(Slab<UnboundedSender<T>>);
struct BrokerStream<T: Sync + Send + Clone + 'static>(usize, UnboundedReceiver<T>);
fn with_senders<T, F, R>(f: F) -> R
where
T: Sync + Send + Clone + 'static,
F: FnOnce(&mut Senders<T>) -> R,
{
let mut map = SUBSCRIBERS.lock().unwrap();
let senders = map
.entry(TypeId::of::<Senders<T>>())
.or_insert_with(|| Box::new(Senders::<T>(Default::default())));
f(senders.downcast_mut::<Senders<T>>().unwrap())
}
impl<T: Sync + Send + Clone + 'static> Drop for BrokerStream<T> {
fn drop(&mut self) {
with_senders::<T, _, _>(|senders| senders.0.remove(self.0));
}
}
impl<T: Sync + Send + Clone + 'static> Stream for BrokerStream<T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.1.poll_next_unpin(cx)
}
}
pub struct SimpleBroker<T>(PhantomData<T>);
impl<T: Sync + Send + Clone + 'static> SimpleBroker<T> {
pub fn publish(msg: T) {
with_senders::<T, _, _>(|senders| {
for (_, sender) in senders.0.iter_mut() {
sender.start_send(msg.clone()).ok();
}
});
}
pub fn subscribe() -> impl Stream<Item = T> {
with_senders::<T, _, _>(|senders| {
let (tx, rx) = mpsc::unbounded();
let id = senders.0.insert(tx);
BrokerStream(id, rx)
})
}
}