rpc/jsonrpc: simplyfied subscibers usage

Cargo.toml

@@ -147,7 +147,6 @@ async-runtime = [
 
 blockchain = [
     "blake3",
-    "bs58", # <-- TODO: remove after we get rid of json for notifications
     "crypto_api_chachapoly",
     "dashu",
     "halo2_proofs",
@@ -213,6 +212,7 @@ net = [
 ]
 
 rpc = [
+    "bs58",
     "rand",
     "serde",
     "serde_json",

src/consensus/task/consensus_sync.rs

@@ -96,7 +96,7 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
     // Listen for next finalization
     info!(target: "consensus::consensus_sync", "Waiting for next finalization...");
     let subscriber = state.read().await.subscribers.get("blocks").unwrap().clone();
-    let subscription = subscriber.subscribe().await;
+    let subscription = subscriber.subscriber.subscribe().await;
     subscription.receive().await;
     subscription.unsubscribe().await;
 

src/consensus/validator.rs

@@ -32,13 +32,11 @@ use darkfi_serial::{serialize, Decodable, Encodable, WriteExt};
 use halo2_proofs::arithmetic::Field;
 use log::{debug, error, info, warn};
 use rand::rngs::OsRng;
-use serde_json::json;
 
 use crate::{
     blockchain::{BlockInfo, Blockchain, BlockchainOverlay, BlockchainOverlayPtr},
-    rpc::jsonrpc::JsonNotification,
+    rpc::jsonrpc::MethodSubscriber,
     runtime::vm_runtime::Runtime,
-    system::{Subscriber, SubscriberPtr},
     tx::Transaction,
     util::time::{TimeKeeper, Timestamp},
     wallet::WalletPtr,
@@ -72,10 +70,7 @@ pub struct ValidatorState {
     /// Canonical (finalized) blockchain
     pub blockchain: Blockchain,
     /// A map of various subscribers exporting live info from the blockchain
-    /// TODO: Instead of JsonNotification, it can be an enum of internal objects,
-    ///       and then we don't have to deal with json in this module but only
-    //        externally.
-    pub subscribers: HashMap<&'static str, SubscriberPtr<JsonNotification>>,
+    pub subscribers: HashMap<&'static str, MethodSubscriber>,
     /// Wallet interface
     pub wallet: WalletPtr,
     /// Flag signalling node has finished initial sync
@@ -195,8 +190,8 @@ impl ValidatorState {
 
         // Here we initialize various subscribers that can export live consensus/blockchain data.
         let mut subscribers = HashMap::new();
-        let block_subscriber = Subscriber::new();
-        let err_txs_subscriber = Subscriber::new();
+        let block_subscriber = MethodSubscriber::new("blockchain.subscribe_err_txs".into());
+        let err_txs_subscriber = MethodSubscriber::new("blockchain.subscribe_blocks".into());
         subscribers.insert("blocks", block_subscriber);
         subscribers.insert("err_txs", err_txs_subscriber);
 
@@ -340,14 +335,11 @@ impl ValidatorState {
         info!(target: "consensus::validator", "purge_pending_txs(): Removing {} erroneous transactions...", erroneous_txs.len());
         self.blockchain.remove_pending_txs(&erroneous_txs)?;
 
-        // TODO: Don't hardcode this:
         let err_txs_subscriber = self.subscribers.get("err_txs").unwrap();
         for err_tx in erroneous_txs {
             let tx_hash = blake3::hash(&serialize(&err_tx)).to_hex().as_str().to_string();
-            let params = json!([bs58::encode(&serialize(&tx_hash)).into_string()]);
-            let notif = JsonNotification::new("blockchain.subscribe_err_txs", params);
             info!(target: "consensus::validator", "purge_pending_txs(): Sending notification about erroneous transaction");
-            err_txs_subscriber.notify(notif).await;
+            err_txs_subscriber.notify(&tx_hash).await;
         }
 
         Ok(())
@@ -816,11 +808,8 @@ impl ValidatorState {
                 return Err(e)
             }
 
-            // TODO: Don't hardcode this:
-            let params = json!([bs58::encode(&serialize(proposal)).into_string()]);
-            let notif = JsonNotification::new("blockchain.subscribe_blocks", params);
             info!(target: "consensus::validator", "consensus: Sending notification about finalized block");
-            blocks_subscriber.notify(notif).await;
+            blocks_subscriber.notify(proposal).await;
         }
 
         // Setting leaders history to last proposal leaders count
@@ -925,12 +914,9 @@ impl ValidatorState {
         info!(target: "consensus::validator", "receive_finalized_block(): Executing state transitions");
         self.receive_blocks(&[block.clone()]).await?;
 
-        // TODO: Don't hardcode this:
         let blocks_subscriber = self.subscribers.get("blocks").unwrap();
-        let params = json!([bs58::encode(&serialize(&block)).into_string()]);
-        let notif = JsonNotification::new("blockchain.subscribe_blocks", params);
         info!(target: "consensus::validator", "consensus: Sending notification about finalized block");
-        blocks_subscriber.notify(notif).await;
+        blocks_subscriber.notify(&block).await;
 
         info!(target: "consensus::validator", "receive_finalized_block(): Removing block transactions from pending txs store");
         self.blockchain.remove_pending_txs(&block.txs)?;
@@ -975,13 +961,10 @@ impl ValidatorState {
         info!(target: "consensus::validator", "receive_sync_blocks(): Executing state transitions");
         self.receive_blocks(&new_blocks[..]).await?;
 
-        // TODO: Don't hardcode this:
         let blocks_subscriber = self.subscribers.get("blocks").unwrap();
         for block in new_blocks {
-            let params = json!([bs58::encode(&serialize(&block)).into_string()]);
-            let notif = JsonNotification::new("blockchain.subscribe_blocks", params);
             info!(target: "consensus::validator", "consensus: Sending notification about finalized block");
-            blocks_subscriber.notify(notif).await;
+            blocks_subscriber.notify(&block).await;
         }
 
         Ok(())

src/rpc/jsonrpc.rs

@@ -20,11 +20,12 @@
 use std::fmt;
 
 use async_std::sync::Arc;
+use darkfi_serial::{serialize, Encodable};
 use rand::{rngs::OsRng, Rng};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_json::{json, Value};
 
-use crate::system::SubscriberPtr;
+use crate::system::{Subscriber, SubscriberPtr};
 
 /// JSON-RPC error codes.
 /// The error codes from and including -32768 to -32000 are reserved for pre-defined errors.
@@ -142,35 +143,58 @@ pub struct JsonNotification {
 }
 
 impl JsonNotification {
-    pub fn new(method: &str, parameters: Value) -> Self {
-        Self { jsonrpc: json!("2.0"), method: json!(method), params: parameters }
+    pub fn new(method: Value, params: Value) -> Self {
+        Self { jsonrpc: json!("2.0"), method, params }
     }
 }
 
-/// A JSON-RPC subscriber for notifications
+/// A method specific JSON-RPC subscriber for notifications
 #[derive(Clone)]
-pub struct JsonSubscriber {
-    /// JSON-RPC version
-    pub jsonrpc: Value,
+pub struct MethodSubscriber {
+    /// Notification method
+    pub method: Value,
     /// Notification subscriber
     pub subscriber: SubscriberPtr<JsonNotification>,
 }
 
-impl JsonSubscriber {
-    pub fn new(subscriber: SubscriberPtr<JsonNotification>) -> Self {
-        Self { jsonrpc: json!("2.0"), subscriber }
+impl MethodSubscriber {
+    pub fn new(method: Value) -> Self {
+        let subscriber = Subscriber::new();
+        Self { method, subscriber }
+    }
+
+    /// Auxiliary function to format provided message and notify the subscriber.
+    pub async fn notify<T: Encodable>(&self, message: &T) {
+        let params = json!([bs58::encode(&serialize(message)).into_string()]);
+        let notif = JsonNotification::new(self.method.clone(), params);
+        self.subscriber.notify(notif).await;
     }
 }
 
-impl fmt::Debug for JsonSubscriber {
+impl fmt::Debug for MethodSubscriber {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("JsonSubscriber")
-            .field("jsonrpc", &self.jsonrpc)
+        f.debug_struct("MethodSubscriber")
+            .field("method", &self.method)
             .field("pointer", &Arc::as_ptr(&self.subscriber))
             .finish()
     }
 }
 
+/// A JSON-RPC subscriber for notifications
+#[derive(Clone, Debug)]
+pub struct JsonSubscriber {
+    /// JSON-RPC version
+    pub jsonrpc: Value,
+    /// Method subscriber
+    pub subscriber: MethodSubscriber,
+}
+
+impl JsonSubscriber {
+    pub fn new(subscriber: MethodSubscriber) -> Self {
+        Self { jsonrpc: json!("2.0"), subscriber }
+    }
+}
+
 impl Serialize for JsonSubscriber {
     fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
     where

src/rpc/server.rs

@@ -76,7 +76,8 @@ async fn accept(
         let reply = rh.handle_request(r).await;
         match reply {
             JsonResult::Subscriber(sub) => {
-                let subscription = sub.subscriber.subscribe().await;
+                // Subscribe to the inner method subscriber
+                let subscription = sub.subscriber.subscriber.subscribe().await;
                 loop {
                     // Listen subscription for notifications
                     let notification = subscription.receive().await;