cargo clippy a little clean up

bin/tau/taud/src/main.rs

@@ -182,8 +182,8 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         .spawn(start_sync_loop(
             commits_received.clone(),
             broadcast_rcv,
-            raft.get_msgs_channel(),
-            raft.get_commits_channel(),
+            raft.sender(),
+            raft.receiver(),
             datastore_path.clone(),
             configured_ws,
             rng,
@@ -200,7 +200,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     let registry = p2p.protocol_registry();
 
-    let raft_node_id = raft.get_id();
+    let raft_node_id = raft.id();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let raft_node_id = raft_node_id.clone();
@@ -229,7 +229,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     })
     .unwrap();
 
-    raft.start(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
+    raft.run(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
 
     Ok(())
 }

script/research/raft-diag/src/main.rs

@@ -141,7 +141,7 @@ async fn start(args: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     let registry = p2p.protocol_registry();
 
-    let raft_node_id = raft.get_id();
+    let raft_node_id = raft.id();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let raft_node_id = raft_node_id.clone();
@@ -171,12 +171,12 @@ async fn start(args: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     .unwrap();
 
     if args.broadcast != 0 {
-        executor.spawn(start_broadcasting(args.broadcast, raft.get_msgs_channel())).detach();
+        executor.spawn(start_broadcasting(args.broadcast, raft.sender())).detach();
     }
 
-    executor.spawn(receive_loop(raft.get_commits_channel())).detach();
+    executor.spawn(receive_loop(raft.receiver())).detach();
 
-    raft.start(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
+    raft.run(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
     Ok(())
 }
 

src/net/p2p.rs

@@ -9,7 +9,7 @@ use url::Url;
 
 use crate::{
     system::{Subscriber, SubscriberPtr, Subscription},
-    Error, Result,
+    Result,
 };
 
 use super::{

src/net/session/inbound_session.rs

@@ -147,15 +147,12 @@ impl InboundSession {
 impl Session for InboundSession {
     async fn get_info(&self) -> serde_json::Value {
         let mut infos = FxHashMap::default();
-        match self.p2p().settings().inbound.as_ref() {
-            Some(accept_addr) => {
-                for (addr, info) in self.connect_infos.lock().await.iter() {
-                    let json_addr = json!({ "accept_addr": accept_addr });
-                    let info = vec![json_addr, info.get_info().await];
-                    infos.insert(addr.to_string(), info);
-                }
+        if let Some(accept_addr) = self.p2p().settings().inbound.as_ref() {
+            for (addr, info) in self.connect_infos.lock().await.iter() {
+                let json_addr = json!({ "accept_addr": accept_addr });
+                let info = vec![json_addr, info.get_info().await];
+                infos.insert(addr.to_string(), info);
             }
-            None => {}
         }
         json!({
             "connected": infos,

src/raft/consensus.rs

@@ -37,7 +37,6 @@ async fn send_node_id_loop(sender: async_channel::Sender<()>, timeout: i64) -> R
 }
 
 pub struct Raft<T> {
-    // this will be derived from the ip
     id: NodeId,
 
     pub(super) role: Role,
@@ -53,7 +52,7 @@ pub struct Raft<T> {
 
     pub(super) last_term: u64,
 
-    sender: Sender,
+    p2p_sender: Sender,
 
     msgs_channel: Channel<T>,
     commits_channel: Channel<T>,
@@ -81,7 +80,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         let msgs_channel = async_channel::unbounded::<T>();
         let commits_channel = async_channel::unbounded::<T>();
 
-        let sender = async_channel::unbounded::<NetMsg>();
+        let p2p_sender = async_channel::unbounded::<NetMsg>();
 
         let id = match datastore.id.get_last()? {
             Some(_id) => _id,
@@ -103,7 +102,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             acked_length: MapLength(FxHashMap::default()),
             nodes: Arc::new(Mutex::new(FxHashMap::default())),
             last_term: 0,
-            sender,
+            p2p_sender,
             msgs_channel,
             commits_channel,
             datastore,
@@ -112,14 +111,17 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         })
     }
 
-    pub async fn start(
+    ///  
+    ///  Run raft consensus and wait stop_signal channel to terminate
+    ///
+    pub async fn run(
         &mut self,
         p2p: net::P2pPtr,
         p2p_recv_channel: async_channel::Receiver<NetMsg>,
         executor: Arc<Executor<'_>>,
         stop_signal: async_channel::Receiver<()>,
     ) -> Result<()> {
-        let p2p_send_task = executor.spawn(p2p_send_loop(self.sender.1.clone(), p2p.clone()));
+        let p2p_send_task = executor.spawn(p2p_send_loop(self.p2p_sender.1.clone(), p2p.clone()));
 
         let prune_seen_messages_task = executor.spawn(prune_map::<String>(
             self.seen_msgs.clone(),
@@ -176,6 +178,29 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         Ok(())
     }
 
+    ///  
+    /// Return async receiver channel which can be used to receive T Messages
+    /// from raft consensus
+    ///
+    pub fn receiver(&self) -> async_channel::Receiver<T> {
+        self.commits_channel.1.clone()
+    }
+
+    ///  
+    /// Return async sender channel which can be used to broadcast T Messages
+    /// to raft consensus
+    ///
+    pub fn sender(&self) -> async_channel::Sender<T> {
+        self.msgs_channel.0.clone()
+    }
+
+    ///  
+    /// Return the raft node id
+    ///
+    pub fn id(&self) -> NodeId {
+        self.id.clone()
+    }
+
     async fn send_node_id_msg(&self) -> Result<()> {
         let node_id_msg = serialize(&NodeIdMsg { id: self.id.clone() });
         self.send(None, &node_id_msg, NetMsgMethod::NodeIdMsg, None).await?;
@@ -255,7 +280,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
         let net_msg = NetMsg { id: random_id, recipient_id, payload: payload.to_vec(), method };
         self.seen_msgs.lock().await.insert(random_id.to_string(), Utc::now().timestamp());
-        self.sender.0.send(net_msg).await?;
+        self.p2p_sender.0.send(net_msg).await?;
 
         Ok(())
     }
@@ -300,18 +325,6 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         Ok(self.datastore.voted_for.get_last()?.flatten())
     }
 
-    pub fn get_commits_channel(&self) -> async_channel::Receiver<T> {
-        self.commits_channel.1.clone()
-    }
-
-    pub fn get_msgs_channel(&self) -> async_channel::Sender<T> {
-        self.msgs_channel.0.clone()
-    }
-
-    pub fn get_id(&self) -> NodeId {
-        self.id.clone()
-    }
-
     pub(super) fn commits_len(&self) -> u64 {
         self.datastore.commits.len()
     }

src/raft/consensus_candidate.rs

@@ -10,7 +10,7 @@ use super::{
 
 impl<T: Decodable + Encodable + Clone> Raft<T> {
     pub(super) async fn send_vote_request(&mut self) -> Result<()> {
-        let self_id = self.get_id();
+        let self_id = self.id();
 
         self.set_current_term(&(self.current_term()? + 1))?;
         self.role = Role::Candidate;
@@ -41,7 +41,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
             if self.votes_received.len() >= ((nodes_cloned.len() + 1) / 2) {
                 self.role = Role::Leader;
-                self.current_leader = self.get_id();
+                self.current_leader = self.id();
                 for node in nodes_cloned.iter() {
                     self.sent_length.insert(node.0, self.logs_len());
                     self.acked_length.insert(node.0, 0);

src/raft/consensus_follower.rs

@@ -31,7 +31,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             if let Some(voted_for) = self.voted_for()? { voted_for == vr.node_id } else { true };
 
         let mut response =
-            VoteResponse { node_id: self.get_id(), current_term: self.current_term()?, ok: false };
+            VoteResponse { node_id: self.id(), current_term: self.current_term()?, ok: false };
 
         if vr.current_term == self.current_term()? && vote_ok && vote {
             self.set_voted_for(&Some(vr.node_id.clone()))?;
@@ -71,7 +71,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         }
 
         let response =
-            LogResponse { node_id: self.get_id(), current_term: self.current_term()?, ack, ok };
+            LogResponse { node_id: self.id(), current_term: self.current_term()?, ack, ok };
 
         debug!(target: "raft",
          "Send LogResponse current_term: {} ack: {} ok: {}",

src/raft/consensus_leader.rs

@@ -43,7 +43,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         }
 
         let request = LogRequest {
-            leader_id: self.get_id(),
+            leader_id: self.id(),
             current_term: self.current_term()?,
             prefix_len,
             prefix_term,

src/raft/datastore.rs

@@ -67,7 +67,7 @@ impl<T: Decodable + Encodable> DataTree<T> {
         Ok(())
     }
 
-    pub fn wipe_insert_all(&self, data: &Vec<T>) -> Result<()> {
+    pub fn wipe_insert_all(&self, data: &[T]) -> Result<()> {
         self.tree.clear()?;
 
         let mut batch = Batch::default();