karyon_jsonrpc/server/
channel.rs

1use std::collections::HashSet;
2use std::sync::{Arc, Weak};
3
4use karyon_core::{async_runtime::lock::RwLock, util::random_32};
5use karyon_net::Endpoint;
6
7use crate::{
8    error::{Error, Result},
9    message::SubscriptionID,
10};
11
12#[derive(Debug)]
13pub struct NewNotification {
14    pub sub_id: SubscriptionID,
15    pub result: serde_json::Value,
16    pub method: String,
17}
18
19/// Represents a new subscription
20#[derive(Clone)]
21pub struct Subscription {
22    id: SubscriptionID,
23    parent: Weak<Channel>,
24    chan: async_channel::Sender<NewNotification>,
25    method: String,
26}
27
28impl Subscription {
29    /// Creates a new [`Subscription`]
30    fn new(
31        parent: Weak<Channel>,
32        id: SubscriptionID,
33        chan: async_channel::Sender<NewNotification>,
34        method: &str,
35    ) -> Self {
36        Self {
37            parent,
38            id,
39            chan,
40            method: method.to_string(),
41        }
42    }
43
44    /// Returns the subscription id.
45    pub fn id(&self) -> SubscriptionID {
46        self.id
47    }
48
49    /// Sends a notification to the subscriber
50    pub async fn notify(&self, res: serde_json::Value) -> Result<()> {
51        if self.still_subscribed().await {
52            let nt = NewNotification {
53                sub_id: self.id,
54                result: res,
55                method: self.method.clone(),
56            };
57            self.chan.send(nt).await?;
58            Ok(())
59        } else {
60            Err(Error::SubscriptionNotFound(self.id.to_string()))
61        }
62    }
63
64    /// Checks from the partent if this subscription is still subscribed
65    async fn still_subscribed(&self) -> bool {
66        match self.parent.upgrade() {
67            Some(parent) => parent.subs.read().await.contains(&self.id),
68            None => false,
69        }
70    }
71}
72
73/// Represents a connection channel for creating/removing subscriptions
74pub struct Channel {
75    chan: async_channel::Sender<NewNotification>,
76    subs: RwLock<HashSet<SubscriptionID>>,
77    peer: Option<Endpoint>,
78}
79
80impl Channel {
81    /// Creates a new [`Channel`]
82    pub(crate) fn new(
83        chan: async_channel::Sender<NewNotification>,
84        peer: Option<Endpoint>,
85    ) -> Arc<Channel> {
86        Arc::new(Self {
87            chan,
88            subs: RwLock::new(HashSet::new()),
89            peer,
90        })
91    }
92
93    /// Returns the remote endpoint of the connection, if known.
94    pub fn peer_endpoint(&self) -> Option<Endpoint> {
95        self.peer.clone()
96    }
97
98    /// Creates a new [`Subscription`]
99    pub async fn new_subscription(
100        self: &Arc<Self>,
101        method: &str,
102        sub_id: Option<SubscriptionID>,
103    ) -> Result<Subscription> {
104        let sub_id = sub_id.unwrap_or_else(random_32);
105        if !self.subs.write().await.insert(sub_id) {
106            return Err(Error::SubscriptionDuplicated(sub_id.to_string()));
107        }
108
109        let sub = Subscription::new(Arc::downgrade(self), sub_id, self.chan.clone(), method);
110        Ok(sub)
111    }
112
113    /// Removes a [`Subscription`]
114    pub async fn remove_subscription(&self, id: &SubscriptionID) -> Result<()> {
115        let mut subs = self.subs.write().await;
116        if !subs.remove(id) {
117            return Err(Error::SubscriptionNotFound(id.to_string()));
118        }
119        Ok(())
120    }
121
122    /// Closes the [`Channel`]
123    pub(crate) fn close(&self) {
124        self.chan.close();
125    }
126}
karyon_jsonrpc/server/channel.rs

karyon_jsonrpc/server/
channel.rs
