darkfid: enable dnet

Add dnet_subscribe_events and p2p.get_info RPC calls. Fix a typo in
dnet.switch RPC call. Activate the dnet subscription in main.rs.

bin/darkfid/src/main.rs

@@ -21,7 +21,7 @@ use std::{
     sync::Arc,
 };
 
-use log::{error, info};
+use log::{debug, error, info};
 use smol::{lock::Mutex, stream::StreamExt};
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use url::Url;
@@ -200,6 +200,8 @@ pub struct Darkfid {
     rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
     /// JSON-RPC client to execute requests to the miner daemon
     rpc_client: Option<Mutex<MinerRpcCLient>>,
+    /// dnet JSON-RPC subscriber
+    dnet_sub: JsonSubscriber,
 }
 
 impl Darkfid {
@@ -210,6 +212,7 @@ impl Darkfid {
         txs_batch_size: usize,
         subscribers: HashMap<&'static str, JsonSubscriber>,
         rpc_client: Option<Mutex<MinerRpcCLient>>,
+        dnet_sub: JsonSubscriber,
     ) -> Self {
         Self {
             p2p,
@@ -219,6 +222,7 @@ impl Darkfid {
             subscribers,
             rpc_connections: Mutex::new(HashSet::new()),
             rpc_client,
+            dnet_sub,
         }
     }
 }
@@ -311,6 +315,29 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
         None => 50,
     };
 
+    info!("Starting dnet subs task");
+    let dnet_sub = JsonSubscriber::new("dnet.subscribe_events");
+    let dnet_sub_ = dnet_sub.clone();
+    let p2p_ = p2p.clone();
+    let dnet_task = StoppableTask::new();
+    dnet_task.clone().start(
+        async move {
+            let dnet_sub = p2p_.dnet_subscribe().await;
+            loop {
+                let event = dnet_sub.receive().await;
+                debug!("Got dnet event: {:?}", event);
+                dnet_sub_.notify(vec![event.into()].into()).await;
+            }
+        },
+        |res| async {
+            match res {
+                Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                Err(e) => panic!("{}", e),
+            }
+        },
+        Error::DetachedTaskStopped,
+        ex.clone(),
+    );
     // Initialize node
     let darkfid = Darkfid::new(
         p2p.clone(),
@@ -319,6 +346,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
         txs_batch_size,
         subscribers,
         rpc_client,
+        dnet_sub,
     )
     .await;
     let darkfid = Arc::new(darkfid);

bin/darkfid/src/rpc.rs

@@ -24,9 +24,11 @@ use smol::lock::MutexGuard;
 use tinyjson::JsonValue;
 
 use darkfi::{
+    net::P2pPtr,
     rpc::{
         client::RpcChadClient,
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
+        p2p_method::HandlerP2p,
         server::RequestHandler,
     },
     system::{sleep, StoppableTaskPtr},
@@ -51,8 +53,11 @@ impl RequestHandler for Darkfid {
             // =====================
             "ping" => self.pong(req.id, req.params).await,
             "clock" => self.clock(req.id, req.params).await,
-            "dnet_switch" => self.dnet_switch(req.id, req.params).await,
             "ping_miner" => self.ping_miner(req.id, req.params).await,
+            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
+            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
+            // TODO: Make this optional
+            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
 
             // ==================
             // Blockchain methods
@@ -124,6 +129,22 @@ impl Darkfid {
         JsonResponse::new(JsonValue::Boolean(true), id).into()
     }
 
+    // RPCAPI:
+    // Initializes a subscription to p2p dnet events.
+    // Once a subscription is established, `darkirc` will send JSON-RPC notifications of
+    // new network events to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
+    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        self.dnet_sub.clone().into()
+    }
+
     // RPCAPI:
     // Pings configured miner daemon for liveness.
     // Returns `true` on success.
@@ -200,3 +221,9 @@ impl Darkfid {
         }
     }
 }
+
+impl HandlerP2p for Darkfid {
+    fn p2p(&self) -> P2pPtr {
+        self.p2p.clone()
+    }
+}

bin/darkfid/src/tests/harness.rs

@@ -232,6 +232,8 @@ pub async fn generate_node(
     vks::inject(&sled_db, vks)?;
 
     let validator = Validator::new(&sled_db, config.clone()).await?;
+    // We initialize a dnet subscriber but do not activate it.
+    let dnet_sub = JsonSubscriber::new("dnet.subscribe_events");
 
     let mut subscribers = HashMap::new();
     subscribers.insert("blocks", JsonSubscriber::new("blockchain.subscribe_blocks"));
@@ -239,7 +241,7 @@ pub async fn generate_node(
     subscribers.insert("proposals", JsonSubscriber::new("blockchain.subscribe_proposals"));
 
     let p2p = spawn_p2p(settings, &validator, &subscribers, ex.clone()).await;
-    let node = Darkfid::new(p2p.clone(), validator, miner, 50, subscribers, None).await;
+    let node = Darkfid::new(p2p.clone(), validator, miner, 50, subscribers, None, dnet_sub).await;
 
     p2p.start().await?;