From 028940fe3e0a87cdc421a6d07f1ecfb6c208b9d0 Mon Sep 17 00:00:00 2001
From: hozan23 <hozan23@karyontech.net>
Date: Tue, 21 May 2024 02:20:45 +0200
Subject: jsonrpc: support pubsub

---
 jsonrpc/src/server/channel.rs | 69 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 69 insertions(+)
 create mode 100644 jsonrpc/src/server/channel.rs

(limited to 'jsonrpc/src/server/channel.rs')

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);
+    }
+}
-- 
cgit v1.2.3