bin: Update respective binaries to new RPC server API.

bin/darkfid/src/main.rs

@@ -16,12 +16,15 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{path::Path, str::FromStr, sync::Arc};
+use std::{collections::HashSet, path::Path, str::FromStr, sync::Arc};
 
 use async_trait::async_trait;
 use darkfi_sdk::crypto::PublicKey;
 use log::{error, info};
-use smol::{lock::Mutex, stream::StreamExt};
+use smol::{
+    lock::{Mutex, MutexGuard},
+    stream::StreamExt,
+};
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use url::Url;
 
@@ -45,7 +48,7 @@ use darkfi::{
         jsonrpc::{ErrorCode::MethodNotFound, JsonError, JsonRequest, JsonResult},
         server::{listen_and_serve, RequestHandler},
     },
-    system::StoppableTask,
+    system::{StoppableTask, StoppableTaskPtr},
     util::path::expand_path,
     wallet::{WalletDb, WalletPtr},
     Error, Result,
@@ -184,6 +187,7 @@ pub struct Darkfid {
     sync_p2p: Option<P2pPtr>,
     _wallet: WalletPtr,
     validator_state: ValidatorStatePtr,
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 // JSON-RPC methods
@@ -250,6 +254,10 @@ impl RequestHandler for Darkfid {
             _ => return JsonError::new(MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl Darkfid {
@@ -259,7 +267,14 @@ impl Darkfid {
         sync_p2p: Option<P2pPtr>,
         _wallet: WalletPtr,
     ) -> Self {
-        Self { synced: Mutex::new(false), consensus_p2p, sync_p2p, _wallet, validator_state }
+        Self {
+            synced: Mutex::new(false),
+            consensus_p2p,
+            sync_p2p,
+            _wallet,
+            validator_state,
+            rpc_connections: Mutex::new(HashSet::new()),
+        }
     }
 }
 
@@ -430,14 +445,14 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     info!("Starting JSON-RPC server");
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, darkfid.clone(), ex.clone()),
+        listen_and_serve(args.rpc_listen, darkfid.clone(), None, ex.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "darkfid", "Failed starting sync JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         ex.clone(),
     );
 

bin/darkfid2/src/main.rs

@@ -16,10 +16,13 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 
 use log::{error, info};
-use smol::stream::StreamExt;
+use smol::{lock::Mutex, stream::StreamExt};
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use url::Url;
 
@@ -29,7 +32,7 @@ use darkfi::{
     cli_desc,
     net::{settings::SettingsOpt, P2pPtr},
     rpc::{jsonrpc::JsonSubscriber, server::listen_and_serve},
-    system::StoppableTask,
+    system::{StoppableTask, StoppableTaskPtr},
     util::time::TimeKeeper,
     validator::{Validator, ValidatorConfig, ValidatorPtr},
     Error, Result,
@@ -112,6 +115,8 @@ pub struct Darkfid {
     validator: ValidatorPtr,
     /// A map of various subscribers exporting live info from the blockchain
     subscribers: HashMap<&'static str, JsonSubscriber>,
+    /// JSON-RPC connection tracker
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 impl Darkfid {
@@ -121,7 +126,13 @@ impl Darkfid {
         validator: ValidatorPtr,
         subscribers: HashMap<&'static str, JsonSubscriber>,
     ) -> Self {
-        Self { sync_p2p, consensus_p2p, validator, subscribers }
+        Self {
+            sync_p2p,
+            consensus_p2p,
+            validator,
+            subscribers,
+            rpc_connections: Mutex::new(HashSet::new()),
+        }
     }
 }
 
@@ -192,14 +203,14 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     // created for it.
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, darkfid.clone(), ex.clone()),
+        listen_and_serve(args.rpc_listen, darkfid.clone(), None, ex.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "darkfid", "Failed starting sync JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         ex.clone(),
     );
 

bin/darkfid2/src/rpc.rs

@@ -16,8 +16,11 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::collections::HashSet;
+
 use async_trait::async_trait;
 use log::debug;
+use smol::lock::MutexGuard;
 use tinyjson::JsonValue;
 
 use darkfi::{
@@ -25,6 +28,7 @@ use darkfi::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::RequestHandler,
     },
+    system::StoppableTaskPtr,
     util::time::Timestamp,
 };
 
@@ -79,6 +83,10 @@ impl RequestHandler for Darkfid {
             _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl Darkfid {

bin/darkirc/src/main.rs

@@ -16,7 +16,10 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 
 use chrono::{Duration, Utc};
 use irc::ClientSubMsg;
@@ -240,17 +243,18 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
         addr: rpc_listen_addr.clone(),
         p2p: p2p.clone(),
         dnet_sub: json_sub,
+        rpc_connections: Mutex::new(HashSet::new()),
     });
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(rpc_listen_addr, rpc_interface, executor.clone()),
+        listen_and_serve(rpc_listen_addr, rpc_interface, None, executor.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "darkirc", "Failed starting JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         executor.clone(),
     );
 

bin/darkirc/src/rpc.rs

@@ -16,8 +16,11 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::collections::HashSet;
+
 use async_trait::async_trait;
 use log::debug;
+use smol::lock::{Mutex, MutexGuard};
 use tinyjson::JsonValue;
 use url::Url;
 
@@ -28,12 +31,15 @@ use darkfi::{
         p2p_method::HandlerP2p,
         server::RequestHandler,
     },
+    system::StoppableTaskPtr,
 };
 
 pub struct JsonRpcInterface {
     pub addr: Url,
     pub p2p: net::P2pPtr,
     pub dnet_sub: JsonSubscriber,
+    /// JSON-RPC connection tracker
+    pub rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 #[async_trait]
@@ -50,6 +56,10 @@ impl RequestHandler for JsonRpcInterface {
             _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl JsonRpcInterface {

bin/faucetd/src/main.rs

@@ -16,7 +16,12 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashMap, path::Path, str::FromStr, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    path::Path,
+    str::FromStr,
+    sync::Arc,
+};
 
 use async_trait::async_trait;
 use chrono::Utc;
@@ -42,7 +47,7 @@ use darkfi_serial::{deserialize, serialize, Encodable};
 use log::{debug, error, info};
 use rand::rngs::OsRng;
 use smol::{
-    lock::{Mutex, RwLock},
+    lock::{Mutex, MutexGuard, RwLock},
     stream::StreamExt,
 };
 use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
@@ -72,7 +77,7 @@ use darkfi::{
         },
         server::{listen_and_serve, RequestHandler},
     },
-    system::{sleep, StoppableTask},
+    system::{sleep, StoppableTask, StoppableTaskPtr},
     tx::Transaction,
     util::{parse::decode_base10, path::expand_path},
     wallet::{WalletDb, WalletPtr},
@@ -184,6 +189,7 @@ pub struct Faucetd {
     airdrop_map: AirdropMap,
     challenge_map: ChallengeMap,
     proving_keys: ProvingKeyMap,
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 #[async_trait]
@@ -195,6 +201,10 @@ impl RequestHandler for Faucetd {
             _ => return JsonError::new(MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl Faucetd {
@@ -271,6 +281,7 @@ impl Faucetd {
             airdrop_map: Arc::new(Mutex::new(HashMap::new())),
             challenge_map: Arc::new(Mutex::new(HashMap::new())),
             proving_keys,
+            rpc_connections: Mutex::new(HashSet::new()),
         };
 
         Ok(faucetd)
@@ -753,14 +764,14 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
     info!(target: "faucetd", "Starting JSON-RPC server");
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, faucetd.clone(), ex.clone()),
+        listen_and_serve(args.rpc_listen, faucetd.clone(), None, ex.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "faucetd", "Failed starting JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         ex.clone(),
     );
 

bin/fud/fud/src/main.rs

@@ -23,7 +23,13 @@ use std::{
 
 use async_trait::async_trait;
 use log::{debug, error, info, warn};
-use smol::{channel, fs::File, lock::RwLock, stream::StreamExt, Executor};
+use smol::{
+    channel,
+    fs::File,
+    lock::{Mutex, MutexGuard, RwLock},
+    stream::StreamExt,
+    Executor,
+};
 use structopt_toml::{structopt::StructOpt, StructOptToml};
 use tinyjson::JsonValue;
 use url::Url;
@@ -39,7 +45,7 @@ use darkfi::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::{listen_and_serve, RequestHandler},
     },
-    system::StoppableTask,
+    system::{StoppableTask, StoppableTaskPtr},
     util::path::expand_path,
     Error, Result,
 };
@@ -97,6 +103,8 @@ pub struct Fud {
     file_fetch_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
     chunk_fetch_tx: channel::Sender<(blake3::Hash, Result<()>)>,
     chunk_fetch_rx: channel::Receiver<(blake3::Hash, Result<()>)>,
+
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 #[async_trait]
@@ -112,6 +120,10 @@ impl RequestHandler for Fud {
             _ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl Fud {
@@ -542,6 +554,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         file_fetch_rx,
         chunk_fetch_tx,
         chunk_fetch_rx,
+        rpc_connections: Mutex::new(HashSet::new()),
     });
 
     info!(target: "fud", "Starting fetch file task");
@@ -575,14 +588,14 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "fud", "Starting JSON-RPC server on {}", args.rpc_listen);
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, fud.clone(), ex.clone()),
+        listen_and_serve(args.rpc_listen, fud.clone(), None, ex.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "fud", "Failed starting sync JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         ex.clone(),
     );
 

bin/genev/genevd/src/main.rs

@@ -154,14 +154,14 @@ async fn realmain(args: Args, executor: Arc<smol::Executor<'static>>) -> Result<
     ));
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, rpc_interface, executor.clone()),
+        listen_and_serve(args.rpc_listen, rpc_interface, None, executor.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "genevd", "Failed starting JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         executor.clone(),
     );
 

bin/genev/genevd/src/rpc.rs

@@ -16,11 +16,11 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
+use std::{collections::HashSet, sync::Arc};
 
 use async_trait::async_trait;
 use log::debug;
-use smol::lock::Mutex;
+use smol::lock::{Mutex, MutexGuard};
 use tinyjson::JsonValue;
 
 use darkfi::{
@@ -33,6 +33,7 @@ use darkfi::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
         server::RequestHandler,
     },
+    system::StoppableTaskPtr,
     util::{encoding::base64, time::Timestamp},
 };
 use darkfi_serial::deserialize;
@@ -44,6 +45,7 @@ pub struct JsonRpcInterface {
     model: ModelPtr<GenEvent>,
     seen: SeenPtr<EventId>,
     p2p: net::P2pPtr,
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 #[async_trait]
@@ -58,6 +60,10 @@ impl RequestHandler for JsonRpcInterface {
             _ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl JsonRpcInterface {
@@ -68,7 +74,14 @@ impl JsonRpcInterface {
         seen: SeenPtr<EventId>,
         p2p: net::P2pPtr,
     ) -> Self {
-        Self { _nickname, missed_events, model, seen, p2p }
+        Self {
+            _nickname,
+            missed_events,
+            model,
+            seen,
+            p2p,
+            rpc_connections: Mutex::new(HashSet::new()),
+        }
     }
 
     // RPCAPI:

bin/lilith/src/main.rs

@@ -27,7 +27,11 @@ use async_trait::async_trait;
 use futures::future::join_all;
 use log::{debug, error, info, warn};
 use semver::Version;
-use smol::{stream::StreamExt, Executor};
+use smol::{
+    lock::{Mutex, MutexGuard},
+    stream::StreamExt,
+    Executor,
+};
 use structopt::StructOpt;
 use structopt_toml::StructOptToml;
 use tinyjson::JsonValue;
@@ -41,7 +45,7 @@ use darkfi::{
         jsonrpc::*,
         server::{listen_and_serve, RequestHandler},
     },
-    system::{sleep, StoppableTask},
+    system::{sleep, StoppableTask, StoppableTaskPtr},
     util::{
         file::{load_file, save_file},
         path::{expand_path, get_config_path},
@@ -138,6 +142,8 @@ struct NetInfo {
 struct Lilith {
     /// Spawned networks
     pub networks: Vec<Spawn>,
+    /// JSON-RPC connection tracker
+    pub rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 impl Lilith {
@@ -235,6 +241,10 @@ impl RequestHandler for Lilith {
             _ => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 /// Attempt to read existing hosts tsv
@@ -445,7 +455,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     }
 
     // Set up main daemon and background tasks
-    let lilith = Arc::new(Lilith { networks });
+    let lilith = Arc::new(Lilith { networks, rpc_connections: Mutex::new(HashSet::new()) });
     let mut periodic_tasks = HashMap::new();
     for network in &lilith.networks {
         let name = network.name.clone();
@@ -468,14 +478,14 @@ async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "lilith", "Starting JSON-RPC server on {}", args.rpc_listen);
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(args.rpc_listen, lilith.clone(), ex.clone()),
+        listen_and_serve(args.rpc_listen, lilith.clone(), None, ex.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "lilith", "Failed starting JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         ex.clone(),
     );
 

bin/tau/taud/src/jsonrpc.rs

@@ -16,12 +16,17 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashMap, fs::create_dir_all, path::PathBuf, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    fs::create_dir_all,
+    path::PathBuf,
+    sync::Arc,
+};
 
 use async_trait::async_trait;
 use crypto_box::ChaChaBox;
 use log::{debug, warn};
-use smol::lock::Mutex;
+use smol::lock::{Mutex, MutexGuard};
 use tinyjson::JsonValue;
 
 use darkfi::{
@@ -30,6 +35,7 @@ use darkfi::{
         jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult},
         server::RequestHandler,
     },
+    system::StoppableTaskPtr,
     util::{path::expand_path, time::Timestamp},
     Error,
 };
@@ -48,6 +54,7 @@ pub struct JsonRpcInterface {
     workspace: Mutex<String>,
     workspaces: Arc<HashMap<String, ChaChaBox>>,
     p2p: net::P2pPtr,
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
 }
 
 #[async_trait]
@@ -73,6 +80,10 @@ impl RequestHandler for JsonRpcInterface {
 
         to_json_result(rep, req.id)
     }
+
+    async fn get_connections(&self) -> MutexGuard<'_, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
 }
 
 impl JsonRpcInterface {
@@ -84,7 +95,15 @@ impl JsonRpcInterface {
         p2p: net::P2pPtr,
     ) -> Self {
         let workspace = Mutex::new(workspaces.iter().last().unwrap().0.clone());
-        Self { dataset_path, nickname, workspace, workspaces, notify_queue_sender, p2p }
+        Self {
+            dataset_path,
+            nickname,
+            workspace,
+            workspaces,
+            notify_queue_sender,
+            p2p,
+            rpc_connections: Mutex::new(HashSet::new()),
+        }
     }
 
     // RPCAPI:

bin/tau/taud/src/main.rs

@@ -403,14 +403,14 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'static>>) -> Res
     ));
     let rpc_task = StoppableTask::new();
     rpc_task.clone().start(
-        listen_and_serve(settings.rpc_listen, rpc_interface, executor.clone()),
+        listen_and_serve(settings.rpc_listen, rpc_interface, None, executor.clone()),
         |res| async {
             match res {
-                Ok(()) | Err(Error::RPCServerStopped) => { /* Do nothing */ }
+                Ok(()) | Err(Error::RpcServerStopped) => { /* Do nothing */ }
                 Err(e) => error!(target: "taud", "Failed starting JSON-RPC server: {}", e),
             }
         },
-        Error::RPCServerStopped,
+        Error::RpcServerStopped,
         executor.clone(),
     );