chore: fix warns

CHANGELOG.md

@@ -16,7 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - added config seeds for avoiding manual process with seeds ([#6](https://github.com/8xFF/atm0s-small-p2p/pull/6))
 - switch to use peer_id for send and recv ([#5](https://github.com/8xFF/atm0s-small-p2p/pull/5))
 - visualization service ([#3](https://github.com/8xFF/atm0s-small-p2p/pull/3))
-- mininum working version with service pkt and stream
+- minimum working version with service pkt and stream
 
 ### Fixed
 

src/discovery.rs

@@ -58,6 +58,7 @@ impl PeerDiscovery {
     pub fn apply_sync(&mut self, now_ms: u64, sync: PeerDiscoverySync) {
         log::debug!("[PeerDiscovery] apply sync with addrs: {:?}", sync.0);
         for (peer, last_updated, address) in sync.0.into_iter() {
+            #[allow(clippy::collapsible_if)]
             if last_updated + TIMEOUT_AFTER > now_ms {
                 if self.remotes.insert(peer, (last_updated, address)).is_none() {
                     log::info!("[PeerDiscovery] added new peer {peer}");
@@ -66,7 +67,7 @@ impl PeerDiscovery {
         }
     }
     pub fn remotes(&self) -> impl Iterator<Item = PeerAddress> + '_ {
-        self.remotes.iter().map(|(p, (_, a))| PeerAddress(*p, a.clone())).chain(self.seeds.iter().map(|s| s.clone()))
+        self.remotes.iter().map(|(p, (_, a))| PeerAddress(*p, a.clone())).chain(self.seeds.iter().cloned())
     }
 }
 

src/lib.rs

@@ -108,6 +108,7 @@ enum PeerMainData {
     Sync { route: RouterTableSync, advertise: PeerDiscoverySync },
 }
 
+#[allow(clippy::enum_variant_names)]
 enum InternalEvent {
     PeerConnected(ConnectionId, PeerId, u16),
     PeerConnectError(ConnectionId, Option<PeerId>, anyhow::Error),
@@ -216,7 +217,7 @@ impl<SECURE: HandshakeProtocol> P2pNetwork<SECURE> {
     fn process_tick(&mut self, now_ms: u64) -> anyhow::Result<P2pNetworkEvent> {
         self.discovery.clear_timeout(now_ms);
         for conn in self.neighbours.connected_conns() {
-            let peer_id = conn.peer_id().expect("conected neighbours should have peer_id");
+            let peer_id = conn.peer_id().expect("accepted neighbours should have peer_id");
             let conn_id = conn.conn_id();
             let route: router::RouterTableSync = self.router.create_sync(&peer_id);
             let advertise = self.discovery.create_sync_for(now_ms, &peer_id);

src/neighbours.rs

@@ -13,7 +13,7 @@ impl NetworkNeighbours {
     }
 
     pub fn has_peer(&self, peer: &PeerId) -> bool {
-        self.conns.values().find(|c| c.peer_id().eq(&Some(*peer))).is_some()
+        self.conns.values().any(|c| c.peer_id().eq(&Some(*peer)))
     }
 
     pub fn mark_connected(&mut self, conn_id: &ConnectionId, peer: PeerId) -> Option<()> {
@@ -22,11 +22,11 @@ impl NetworkNeighbours {
     }
 
     pub fn remove(&mut self, conn_id: &ConnectionId) -> Option<()> {
-        self.conns.remove(&conn_id)?;
+        self.conns.remove(conn_id)?;
         Some(())
     }
 
     pub fn connected_conns(&self) -> impl Iterator<Item = &PeerConnection> {
-        self.conns.values().into_iter().filter(|c| c.is_connected())
+        self.conns.values().filter(|c| c.is_connected())
     }
 }

src/peer.rs

@@ -124,6 +124,7 @@ struct ConnectRes {
     result: Result<Vec<u8>, String>,
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn run_connection<SECURE: HandshakeProtocol>(
     secure: Arc<SECURE>,
     ctx: SharedCtx,

src/peer/peer_alias.rs

@@ -1,5 +1,5 @@
 //! PeerAlias allow control a peer-connection from othert task
-//! This is done by using control_tx to send control to running task over chanel
+//! This is done by using control_tx to send control to running task over channel
 
 use tokio::sync::{mpsc::Sender, oneshot};
 
@@ -49,6 +49,6 @@ impl PeerConnectionAlias {
     pub(crate) async fn open_stream(&self, service: P2pServiceId, source: PeerId, dest: PeerId, meta: Vec<u8>) -> anyhow::Result<P2pQuicStream> {
         let (tx, rx) = oneshot::channel();
         self.control_tx.send(PeerConnectionControl::OpenStream(service, source, dest, meta, tx)).await?;
-        Ok(rx.await??)
+        rx.await?
     }
 }

src/peer/peer_internal.rs

@@ -3,7 +3,7 @@
 //! We have some strict rules
 //!
 //! - Only use async with current connection stream
-//! - For other communication shoud use try_send for avoding blocking
+//! - For other communication should use try_send for avoiding blocking
 
 use std::{net::SocketAddr, time::Duration};
 
@@ -42,6 +42,7 @@ pub struct PeerConnectionInternal {
 }
 
 impl PeerConnectionInternal {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         ctx: SharedCtx,
         conn_id: ConnectionId,
@@ -186,7 +187,7 @@ async fn accept_bi(to_peer: PeerId, mut stream: P2pQuicStream, ctx: SharedCtx) -
                 service_tx
                     .send(P2pServiceEvent::Stream(source, meta, stream))
                     .await
-                    .print_on_err("[PeerConnectionInternal] send accpeted stream to service");
+                    .print_on_err("[PeerConnectionInternal] send accepted stream to service");
                 Ok(())
             } else {
                 log::warn!("[PeerConnectionInternal {to_peer}] stream service {service} source {source} to dest {dest} => service not found");

src/router.rs

@@ -79,7 +79,7 @@ impl RouterTable {
             self.peers.entry(*peer).or_default();
         }
 
-        let mut new_paths = BTreeMap::<PeerId, PathMetric>::from_iter(sync.0.into_iter());
+        let mut new_paths = BTreeMap::<PeerId, PathMetric>::from_iter(sync.0);
         // only loop over peer which don't equal source, because it is direct connection
         for (peer, memory) in self.peers.iter_mut().filter(|(p, _)| !from_peer.eq(p)) {
             let previous = memory.paths.contains_key(&conn);
@@ -121,7 +121,7 @@ impl RouterTable {
     fn del_direct(&mut self, conn: &ConnectionId) {
         if let Some((to, _)) = self.directs.remove(conn) {
             if let Some(memory) = self.peers.get_mut(&to) {
-                memory.paths.remove(&conn);
+                memory.paths.remove(conn);
                 Self::select_best_for(&to, memory);
                 if memory.best().is_none() {
                     self.peers.remove(&to);
@@ -225,7 +225,7 @@ impl SharedRouterTable {
     }
 
     pub fn create_sync(&self, dest: &PeerId) -> RouterTableSync {
-        self.table.read().create_sync(&dest)
+        self.table.read().create_sync(dest)
     }
 
     pub fn apply_sync(&self, conn: ConnectionId, sync: RouterTableSync) {

src/service/alias_service.rs

@@ -253,7 +253,7 @@ impl AliasServiceInternal {
     fn on_msg(&mut self, now: u64, from: PeerId, msg: AliasMessage) {
         match msg {
             AliasMessage::NotifySet(alias_id) => {
-                let slot = self.cache.get_or_insert_mut(alias_id, || HashSet::new());
+                let slot = self.cache.get_or_insert_mut(alias_id, HashSet::new);
                 slot.insert(from);
             }
             AliasMessage::NotifyDel(alias_id) => {
@@ -277,7 +277,7 @@ impl AliasServiceInternal {
                 }
             }
             AliasMessage::Found(alias_id) => {
-                let slot = self.cache.get_or_insert_mut(alias_id, || HashSet::new());
+                let slot = self.cache.get_or_insert_mut(alias_id, HashSet::new);
                 slot.insert(from);
 
                 if let Some(req) = self.find_reqs.remove(&alias_id) {

src/service/pubsub_service.rs

@@ -228,7 +228,7 @@ impl PubsubService {
                                     for (_, sub_tx) in state.local_subscribers.iter() {
                                         let _ = sub_tx.send(SubscriberEvent::PeerJoined(PeerSrc::Remote(from_peer)));
                                     }
-                                    // we also send subscribe state it remote, as publisher it only care about whereever this node is a subscriber
+                                    // we also send subscribe state it remote, as publisher it only care about wherever this node is a subscriber
                                     if !state.local_subscribers.is_empty() {
                                         self.send_to(from_peer, &PubsubMessage::SubscriberJoined(channel)).await;
                                     }
@@ -254,9 +254,9 @@ impl PubsubService {
                                     for (_, pub_tx) in state.local_publishers.iter() {
                                         let _ = pub_tx.send(PublisherEvent::PeerJoined(PeerSrc::Remote(from_peer)));
                                     }
-                                    // we also send publisher state it remote, as subscriber it only care about whereever this node is a publisher
+                                    // we also send publisher state it remote, as subscriber it only care about wherever this node is a publisher
                                     if !state.local_publishers.is_empty() {
-                                        self.send_to(from_peer, &&PubsubMessage::PublisherJoined(channel)).await;
+                                        self.send_to(from_peer, &PubsubMessage::PublisherJoined(channel)).await;
                                     }
                                 }
                             }
@@ -395,7 +395,7 @@ impl PubsubService {
                 let state = self.channels.entry(channel).or_default();
                 state.local_publishers.remove(&local_id);
                 if state.local_publishers.is_empty() {
-                    // if this is last local_publisher => notify all subscibers
+                    // if this is last local_publisher => notify all subscribers
                     for (_, sub_tx) in state.local_subscribers.iter() {
                         let _ = sub_tx.send(SubscriberEvent::PeerLeaved(PeerSrc::Local));
                     }
@@ -410,7 +410,7 @@ impl PubsubService {
                     let _ = tx.send(SubscriberEvent::PeerJoined(PeerSrc::Local));
                 }
                 if state.local_subscribers.is_empty() {
-                    // if this is first local_subsciber => notify to all local_publishers
+                    // if this is first local_subscriber => notify to all local_publishers
                     for (_, pub_tx) in state.local_publishers.iter() {
                         let _ = pub_tx.send(PublisherEvent::PeerJoined(PeerSrc::Local));
                     }
@@ -574,6 +574,7 @@ impl PubsubService {
                 if let PeerSrc::Remote(peer) = peer_src {
                     let _ = self.send_to(peer, &PubsubMessage::PublishRpcAnswer(data, rpc_id)).await;
                 } else {
+                    #[allow(clippy::collapsible_if)]
                     if let Some(mut req) = self.publish_rpc_reqs.remove(&rpc_id) {
                         let _ = req.tx.take().expect("should have req_tx").send(Ok(data));
                     } else {
@@ -585,6 +586,7 @@ impl PubsubService {
                 if let PeerSrc::Remote(peer) = peer_src {
                     let _ = self.send_to(peer, &PubsubMessage::FeedbackRpcAnswer(data, rpc_id)).await;
                 } else {
+                    #[allow(clippy::collapsible_if)]
                     if let Some(mut req) = self.feedback_rpc_reqs.remove(&rpc_id) {
                         let _ = req.tx.take().expect("should have req_tx").send(Ok(data));
                     } else {