aboutsummaryrefslogtreecommitdiff
path: root/jsonrpc/src/server/channel.rs
blob: a9d10025af79823b00d82097e1c745168859ea95 (plain)
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
use std::sync::Arc;

use karyon_core::{async_runtime::lock::Mutex, util::random_32};

use crate::{message::SubscriptionID, Error, Result};

pub type ArcChannel = Arc<Channel>;

pub(crate) struct NewNotification {
    pub sub_id: SubscriptionID,
    pub result: serde_json::Value,
    pub method: String,
}

/// Represents a new subscription
pub struct Subscription {
    pub id: SubscriptionID,
    parent: Arc<Channel>,
    chan: async_channel::Sender<NewNotification>,
    method: String,
}

impl Subscription {
    /// Creates a new `Subscription`
    fn new(
        parent: Arc<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.parent.subs.lock().await.contains(&self.id) {
            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()))
        }
    }
}

/// Represents a 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>) -> ArcChannel {
        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(self.clone(), 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);
    }

    pub fn close(&self) {
        self.chan.close();
    }
}