rpc: simplyfied notify handler

bin/lilith/src/main.rs

@@ -32,11 +32,13 @@ use darkfi::{
     rpc::{
         jsonrpc::{
             ErrorCode::{InvalidParams, MethodNotFound},
-            JsonError, JsonRequest, JsonResponse, JsonResult,
+            JsonError, JsonNotification, JsonRequest, JsonResponse, JsonResult, JsonSubscriber,
         },
         server::{listen_and_serve, RequestHandler},
     },
+    system::{Subscriber, SubscriberPtr},
     util::{
+        async_util::sleep,
         file::{load_file, save_file},
         path::{expand_path, get_config_path},
     },
@@ -81,6 +83,8 @@ struct Lilith {
     urls: Vec<Url>,
     /// Spawned networks
     spawns: Vec<Spawn>,
+    // TODO: Subscriber should come from ValidatorState or something
+    subscriber: SubscriberPtr<JsonNotification>,
 }
 
 impl Lilith {
@@ -126,6 +130,30 @@ impl Lilith {
     async fn pong(&self, id: Value, _params: &[Value]) -> JsonResult {
         JsonResponse::new(json!("pong"), id).into()
     }
+
+    // RPCAPI:
+    // Create a new subscriber for new blocks to notify connected peer.
+    //
+    // --> {"jsonrpc": "2.0", "method": "blockchain.notify_blocks", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": {...}, "id": 1}
+    pub async fn blockchain_notify_blocks(&self, id: Value, params: &[Value]) -> JsonResult {
+        if !params.is_empty() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        JsonSubscriber::new(id, self.subscriber.clone()).into()
+    }
+}
+
+// TODO: remove this
+async fn simulate_blocks(subscriber: SubscriberPtr<JsonNotification>) {
+    // Notifications simulation
+    let message =
+        JsonNotification::new("blockchain.notify_blocks", Value::from("New Block created!"));
+    loop {
+        subscriber.notify(message.clone()).await;
+        sleep(10).await;
+    }
 }
 
 #[async_trait]
@@ -140,6 +168,9 @@ impl RequestHandler for Lilith {
         match req.method.as_str() {
             Some("spawns") => return self.spawns(req.id, params).await,
             Some("ping") => return self.pong(req.id, params).await,
+            Some("blockchain.notify_blocks") => {
+                return self.blockchain_notify_blocks(req.id, params).await
+            }
             Some(_) | None => return JsonError::new(MethodNotFound, None, req.id).into(),
         }
     }
@@ -303,13 +334,20 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
         }
     }
 
-    let lilith = Lilith { urls, spawns };
+    // TODO: Subscriber should come from ValidatorState or something
+    let subscriber: SubscriberPtr<JsonNotification> = Subscriber::new();
+
+    let lilith = Lilith { urls, spawns, subscriber: subscriber.clone() };
     let lilith = Arc::new(lilith);
 
     // JSON-RPC server
     info!("Starting JSON-RPC server");
     ex.spawn(listen_and_serve(args.rpc_listen, lilith.clone())).detach();
 
+    // JSON-RPC notifications simulation
+    let _ex = ex.clone();
+    ex.spawn(simulate_blocks(subscriber)).detach();
+
     // Wait for SIGINT
     shutdown.recv().await?;
     print!("\r");

src/rpc/client.rs

@@ -106,6 +106,7 @@ impl RpcClient {
                 //self.stop_signal.send(()).await?;
                 Err(Error::JsonRpcError("Unexpected reply".to_string()))
             }
+            JsonResult::Subscriber(_) => Err(Error::JsonRpcError("Unexpected reply".to_string())),
         }
     }
 

src/rpc/jsonrpc.rs

@@ -17,10 +17,15 @@
  */
 
 //! JSON-RPC 2.0 primitives
+use std::fmt;
+
+use async_std::sync::Arc;
 use rand::Rng;
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_json::{json, Value};
 
+use crate::system::SubscriberPtr;
+
 /// JSON-RPC error codes.
 /// The error codes from and including -32768 to -32000 are reserved for pre-defined errors.
 #[derive(Debug, Clone)]
@@ -71,6 +76,7 @@ pub enum JsonResult {
     Response(JsonResponse),
     Error(JsonError),
     Notification(JsonNotification),
+    Subscriber(JsonSubscriber),
 }
 // ANCHOR_END: jsonresult
 
@@ -92,6 +98,12 @@ impl From<JsonNotification> for JsonResult {
     }
 }
 
+impl From<JsonSubscriber> for JsonResult {
+    fn from(sub: JsonSubscriber) -> Self {
+        Self::Subscriber(sub)
+    }
+}
+
 /// A JSON-RPC request object.
 // ANCHOR: jsonrequest
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -137,6 +149,51 @@ impl JsonNotification {
     }
 }
 
+/// A JSON-RPC subscriber for notifications
+#[derive(Clone)]
+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 }
+    }
+}
+
+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()
+    }
+}
+
+impl Serialize for JsonSubscriber {
+    fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        unimplemented!();
+    }
+}
+
+impl<'de> Deserialize<'de> for JsonSubscriber {
+    fn deserialize<D>(_deserializer: D) -> Result<JsonSubscriber, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        unimplemented!();
+    }
+}
+
 /// A JSON-RPC response object.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct JsonResponse {

src/rpc/server.rs

@@ -77,13 +77,35 @@ async fn accept(
         };
 
         let reply = rh.handle_request(r).await;
-        let j = serde_json::to_string(&reply).unwrap();
-        debug!(target: "jsonrpc-server", "{} <-- {}", peer_addr, j);
-
-        if let Err(e) = stream.write_all(j.as_bytes()).await {
-            error!("JSON-RPC server failed writing to {} socket: {}", peer_addr, e);
-            debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
-            break
+        match reply {
+            JsonResult::Subscriber(sub) => {
+                let subscription = sub.subscriber.subscribe().await;
+                loop {
+                    // Listen subscription for notifications
+                    let notification = subscription.receive().await;
+
+                    // Push notification
+                    let j = serde_json::to_string(&notification).unwrap();
+                    debug!(target: "jsonrpc-server", "{} <-- {}", peer_addr, j);
+
+                    if let Err(e) = stream.write_all(j.as_bytes()).await {
+                        error!(target: "jsonrpc-server", "JSON-RPC server failed writing to {} socket: {}", peer_addr, e);
+                        debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
+                        break
+                    }
+                }
+                subscription.unsubscribe().await;
+            }
+            _ => {
+                let j = serde_json::to_string(&reply).unwrap();
+                debug!(target: "jsonrpc-server", "{} <-- {}", peer_addr, j);
+
+                if let Err(e) = stream.write_all(j.as_bytes()).await {
+                    error!(target: "jsonrpc-server", "JSON-RPC server failed writing to {} socket: {}", peer_addr, e);
+                    debug!(target: "jsonrpc-server", "Closed connection for {}", peer_addr);
+                    break
+                }
+            }
         }
     }