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
66
67
68
69
|
use std::sync::Arc;
use karyon_core::{async_runtime::lock::Mutex, util::random_32};
use crate::{Error, Result};
pub type SubscriptionID = u32;
pub type ArcChannel = Arc<Channel>;
/// Represents a new subscription
pub struct Subscription {
pub id: SubscriptionID,
parent: Arc<Channel>,
chan: async_channel::Sender<(SubscriptionID, serde_json::Value)>,
}
impl Subscription {
/// Creates a new `Subscription`
fn new(
parent: Arc<Channel>,
id: SubscriptionID,
chan: async_channel::Sender<(SubscriptionID, serde_json::Value)>,
) -> Self {
Self { parent, id, chan }
}
/// Sends a notification to the subscriber
pub async fn notify(&self, res: serde_json::Value) -> Result<()> {
if self.parent.subs.lock().await.contains(&self.id) {
self.chan.send((self.id, res)).await?;
Ok(())
} else {
Err(Error::SubscriptionNotFound(self.id.to_string()))
}
}
}
/// Represents a channel for creating/removing subscriptions
pub struct Channel {
chan: async_channel::Sender<(SubscriptionID, serde_json::Value)>,
subs: Mutex<Vec<SubscriptionID>>,
}
impl Channel {
/// Creates a new `Channel`
pub fn new(chan: async_channel::Sender<(SubscriptionID, serde_json::Value)>) -> ArcChannel {
Arc::new(Self {
chan,
subs: Mutex::new(Vec::new()),
})
}
/// Creates a new subscription
pub async fn new_subscription(self: &Arc<Self>) -> Subscription {
let sub_id = random_32();
let sub = Subscription::new(self.clone(), sub_id, self.chan.clone());
self.subs.lock().await.push(sub_id);
sub
}
/// Removes a subscription
pub async fn remove_subscription(self: &Arc<Self>, id: &SubscriptionID) {
let i = match self.subs.lock().await.iter().position(|i| i == id) {
Some(i) => i,
None => return,
};
self.subs.lock().await.remove(i);
}
}
|
