rpc/client: subscribe to notifications impl added, script/research/rpc_cleint_notifications: example usage of rpc/client.subscribe()

bin/lilith/src/main.rs

@@ -141,7 +141,7 @@ impl Lilith {
             return JsonError::new(InvalidParams, None, id).into()
         }
 
-        JsonSubscriber::new(id, self.subscriber.clone()).into()
+        JsonSubscriber::new(self.subscriber.clone()).into()
     }
 }
 

script/research/rpc_client_notifications/.gitignore

@@ -0,0 +1,2 @@
+/target
+Cargo.lock

script/research/rpc_client_notifications/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "rpc_client_notifications"
+version = "0.1.0"
+authors = ["Dyne.org foundation <foundation@dyne.org>"]
+license = "AGPL-3.0-only"
+edition = "2021"
+
+[workspace]
+
+[dependencies]
+async-std = "1.12.0"
+darkfi = {path = "../../../", features = ["rpc"]}
+futures = "0.3.25"
+serde_json = "1.0.87"
+url = {version = "2.3.1", features = ["serde"]}

script/research/rpc_client_notifications/src/main.rs

@@ -0,0 +1,79 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2022 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use darkfi::{
+    rpc::{
+        client::RpcClient,
+        jsonrpc::{JsonRequest, JsonResult},
+    },
+    system::{Subscriber, SubscriberPtr},
+    Result,
+};
+use futures::join;
+use serde_json::json;
+use url::Url;
+
+async fn listen(subscriber: SubscriberPtr<JsonResult>) -> Result<()> {
+    let subscription = subscriber.subscribe().await;
+    loop {
+        // Listen subscription for notifications
+        let notification = subscription.receive().await;
+        match notification {
+            JsonResult::Notification(n) => {
+                println!("Got notification: {:?}", n);
+            }
+            JsonResult::Error(e) => {
+                println!("Client returned an error: {}", serde_json::to_string(&e)?);
+                break
+            }
+            _ => {
+                println!("Client returned an unexpected reply.");
+                break
+            }
+        }
+    }
+    subscription.unsubscribe().await;
+
+    Ok(())
+}
+
+#[async_std::main]
+async fn main() -> Result<()> {
+    let endpoint = Url::parse("tcp://127.0.0.1:18927")?;
+    let notif_channel = "blockchain.notify_blocks";
+    println!("Creating subscriber for channel: {}", notif_channel);
+    let subscriber: SubscriberPtr<JsonResult> = Subscriber::new();
+
+    println!("Creating client for endpoint: {}", endpoint);
+    let rpc_client = RpcClient::new(endpoint).await?;
+    println!("Subscribing client");
+    let req = JsonRequest::new("blockchain.notify_blocks", json!([]));
+
+    println!("Starting listening");
+    let result = join!(listen(subscriber.clone()), rpc_client.subscribe(req, subscriber));
+    match result.0 {
+        Ok(_) => {}
+        Err(e) => println!("Listener failed: {}", e),
+    }
+    match result.1 {
+        Ok(_) => {}
+        Err(e) => println!("Subscriber failed: {}", e),
+    }
+
+    Ok(())
+}

src/rpc/client.rs

@@ -30,6 +30,7 @@ use crate::{
     net::transport::{
         TcpTransport, TorTransport, Transport, TransportName, TransportStream, UnixTransport,
     },
+    system::SubscriberPtr,
     Error, Result,
 };
 
@@ -54,6 +55,52 @@ impl RpcClient {
         Ok(())
     }
 
+    /// Listen instantiated client for notifications.
+    /// NOTE: Subscriber listeners must perform response handling.
+    pub async fn subscribe(
+        &self,
+        req: JsonRequest,
+        subscriber: SubscriberPtr<JsonResult>,
+    ) -> Result<()> {
+        // Perform initial request.
+        debug!(target: "jsonrpc-client", "--> {}", serde_json::to_string(&req)?);
+        // If the connection is closed, the sender will get an error for sending to a closed channel.
+        if let Err(e) = self.send.send(json!(req)).await {
+            error!(target: "jsonrpc-client", "JSON-RPC client unable to send to {} (channels closed): {}", self.url, e);
+            return Err(Error::NetworkOperationFailed)
+        }
+
+        loop {
+            // If the connection is closed, the receiver will get an error for waiting on a closed channel.
+            let notification = self.recv.recv().await;
+            if notification.is_err() {
+                error!(target: "jsonrpc-client", "JSON-RPC client unable to recv from {} (channels closed)", self.url);
+                break
+            }
+
+            // Notify subscribed channels
+            let notification = notification?;
+            debug!(target: "jsonrpc-client", "<-- {}", serde_json::to_string(&notification)?);
+
+            subscriber.notify(notification.clone()).await;
+
+            // Stop listenning on error
+            match notification {
+                JsonResult::Notification(_) => {}
+                _ => break,
+            }
+
+            // Triggering next consume
+            if let Err(e) = self.send.send(json!(req)).await {
+                error!(target: "jsonrpc-client", "JSON-RPC client unable to send to {} (channels closed): {}", self.url, e);
+                break
+            }
+        }
+
+        subscriber.notify(JsonError::new(ErrorCode::InternalError, None, req.id).into()).await;
+        Err(Error::NetworkOperationFailed)
+    }
+
     /// Send a given JSON-RPC request over the instantiated client.
     pub async fn request(&self, value: JsonRequest) -> Result<Value> {
         let req_id = value.id.clone().as_u64().unwrap();

src/rpc/jsonrpc.rs

@@ -154,15 +154,13 @@ impl JsonNotification {
 pub struct JsonSubscriber {
     /// JSON-RPC version
     pub jsonrpc: Value,
-    /// Request ID
-    pub id: Value,
     /// Notification subscriber
     pub subscriber: SubscriberPtr<JsonNotification>,
 }
 
 impl JsonSubscriber {
-    pub fn new(id: Value, subscriber: SubscriberPtr<JsonNotification>) -> Self {
-        Self { jsonrpc: json!("2.0"), id, subscriber }
+    pub fn new(subscriber: SubscriberPtr<JsonNotification>) -> Self {
+        Self { jsonrpc: json!("2.0"), subscriber }
     }
 }
 
@@ -170,7 +168,6 @@ impl fmt::Debug for JsonSubscriber {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("JsonSubscriber")
             .field("jsonrpc", &self.jsonrpc)
-            .field("id", &self.id)
             .field("pointer", &Arc::as_ptr(&self.subscriber))
             .finish()
     }