diff options
author | hozan23 <hozan23@karyontech.net> | 2024-05-21 02:20:45 +0200 |
---|---|---|
committer | hozan23 <hozan23@karyontech.net> | 2024-05-22 15:02:06 +0200 |
commit | 028940fe3e0a87cdc421a6d07f1ecfb6c208b9d0 (patch) | |
tree | 3272d5c71cafb098e548cb9811e8f9ddc260ef2f /jsonrpc/src/server/channel.rs | |
parent | 0f0cefb62ee8b641dcabcc0a2a1cf019c1de4843 (diff) |
jsonrpc: support pubsub
Diffstat (limited to 'jsonrpc/src/server/channel.rs')
-rw-r--r-- | jsonrpc/src/server/channel.rs | 69 |
1 files changed, 69 insertions, 0 deletions
diff --git a/jsonrpc/src/server/channel.rs b/jsonrpc/src/server/channel.rs new file mode 100644 index 0000000..1498825 --- /dev/null +++ b/jsonrpc/src/server/channel.rs @@ -0,0 +1,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); + } +} |