chore: Clippy lint

bin/darkfid/src/rpc_tx.rs

@@ -149,7 +149,7 @@ impl Darkfid {
 
         if let Some(sync_p2p) = &self.sync_p2p {
             sync_p2p.broadcast(&tx).await;
-            if sync_p2p.channels().lock().await.is_empty() {
+            if sync_p2p.channels().await.is_empty() {
                 error!("[RPC] tx.broadcast: Failed broadcasting tx, no connected channels");
                 return server_error(RpcError::TxBroadcastFail, id, None)
             }

bin/darkfid2/src/rpc_tx.rs

@@ -144,7 +144,7 @@ impl Darkfid {
         }
 
         self.sync_p2p.broadcast(&tx).await;
-        if self.sync_p2p.channels().lock().await.is_empty() {
+        if self.sync_p2p.channels().await.is_empty() {
             error!(target: "darkfid::rpc::tx_broadcast", "Failed broadcasting tx, no connected channels");
             return server_error(RpcError::TxBroadcastFail, id, None)
         }

bin/darkfid2/src/task/sync.rs

@@ -31,7 +31,7 @@ pub async fn sync_task(node: &Darkfid) -> Result<()> {
     info!(target: "darkfid::task::sync_task", "Starting blockchain sync...");
     // Block until at least node is connected to at least one peer
     loop {
-        if !node.sync_p2p.channels().lock().await.is_empty() {
+        if !node.sync_p2p.channels().await.is_empty() {
             break
         }
         warn!(target: "darkfid::task::sync_task", "Node is not connected to other nodes, waiting to retry...");

src/consensus/task/consensus_sync.rs

@@ -35,11 +35,8 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
     info!(target: "consensus::consensus_sync", "Starting consensus state sync...");
     let current_slot = state.read().await.consensus.time_keeper.current_slot();
     // Loop through connected channels
-    let channels_map = p2p.channels().lock().await;
-    let values = channels_map.values();
-    // Using len here because is_empty() uses unstable library feature
-    // called 'exact_size_is_empty'.
-    if values.len() == 0 {
+    let channels = p2p.channels().await;
+    if channels.is_empty() {
         warn!(target: "consensus::consensus_sync", "Node is not connected to other nodes");
         let mut lock = state.write().await;
         lock.consensus.bootstrap_slot = current_slot;
@@ -50,7 +47,7 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
 
     // Node iterates the channel peers to check if at least on peer has seen slots
     let mut peer = None;
-    for channel in values {
+    for channel in channels {
         // Communication setup
         let msg_subsystem = channel.message_subsystem();
         msg_subsystem.add_dispatch::<ConsensusSyncResponse>().await;
@@ -78,9 +75,6 @@ pub async fn consensus_sync_task(p2p: P2pPtr, state: ValidatorStatePtr) -> Resul
         break
     }
 
-    // Release channels lock
-    drop(channels_map);
-
     // If no peer knows about any slots, that means that the network was bootstrapped or restarted
     // and no node has started consensus.
     if peer.is_none() {

src/net/message_subscriber.rs

@@ -251,7 +251,8 @@ impl MessageSubsystem {
             return Err(Error::MissingDispatcher)
         };
 
-        Ok(dispatcher.trigger(payload).await)
+        dispatcher.trigger(payload).await;
+        Ok(())
     }
 
     /// Concurrently transmits an error message across dispatchers.
@@ -298,7 +299,7 @@ mod tests {
             // 2. Publish data there
             let msg = MyVersionMessage(110);
             let payload = serialize(&msg);
-            subsystem.notify("verver", &payload).await;
+            subsystem.notify("verver", &payload).await.unwrap();
 
             // Receive:
             // 1. Do a get easy