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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
use std::sync::{Arc, Weak};
use karyon_core::{async_runtime::lock::Mutex, util::random_32};
use crate::{message::SubscriptionID, Error, Result};
#[derive(Debug)]
pub(crate) struct NewNotification {
pub sub_id: SubscriptionID,
pub result: serde_json::Value,
pub method: String,
}
/// Represents a new subscription
#[derive(Clone)]
pub struct Subscription {
pub id: SubscriptionID,
parent: Weak<Channel>,
chan: async_channel::Sender<NewNotification>,
method: String,
}
impl Subscription {
/// Creates a new [`Subscription`]
fn new(
parent: Weak<Channel>,
id: SubscriptionID,
chan: async_channel::Sender<NewNotification>,
method: &str,
) -> Self {
Self {
parent,
id,
chan,
method: method.to_string(),
}
}
/// Sends a notification to the subscriber
pub async fn notify(&self, res: serde_json::Value) -> Result<()> {
if self.still_subscribed().await {
let nt = NewNotification {
sub_id: self.id,
result: res,
method: self.method.clone(),
};
self.chan.send(nt).await?;
Ok(())
} else {
Err(Error::SubscriptionNotFound(self.id.to_string()))
}
}
/// Checks from the partent if this subscription is still subscribed
async fn still_subscribed(&self) -> bool {
match self.parent.upgrade() {
Some(parent) => parent.subs.lock().await.contains(&self.id),
None => false,
}
}
}
/// Represents a connection channel for creating/removing subscriptions
pub struct Channel {
chan: async_channel::Sender<NewNotification>,
subs: Mutex<Vec<SubscriptionID>>,
}
impl Channel {
/// Creates a new [`Channel`]
pub(crate) fn new(chan: async_channel::Sender<NewNotification>) -> Arc<Channel> {
Arc::new(Self {
chan,
subs: Mutex::new(Vec::new()),
})
}
/// Creates a new [`Subscription`]
pub async fn new_subscription(self: &Arc<Self>, method: &str) -> Subscription {
let sub_id = random_32();
let sub = Subscription::new(Arc::downgrade(self), sub_id, self.chan.clone(), method);
self.subs.lock().await.push(sub_id);
sub
}
/// Removes a [`Subscription`]
pub async fn remove_subscription(&self, id: &SubscriptionID) {
let mut subs = self.subs.lock().await;
let i = match subs.iter().position(|i| i == id) {
Some(i) => i,
None => return,
};
subs.remove(i);
}
/// Closes the [`Channel`]
pub(crate) fn close(&self) {
self.chan.close();
}
}
|
