clippy

Author: rklaehn

examples/multiprovider.rs

@@ -236,7 +236,7 @@ async fn download(args: DownloadArgs) -> anyhow::Result<()> {
             let bitmap = bitmap(current, requested, rows as usize);
             print!("\r{bitmap}");
             if progress.is_done() {
-                println!("");
+                println!();
                 break;
             }
         }

src/downloader2.rs

@@ -51,6 +51,9 @@ use state::*;
 mod actor;
 use actor::*;
 
+mod content_discovery;
+pub use content_discovery::*;
+
 #[derive(
     Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, derive_more::From,
 )]
@@ -76,35 +79,6 @@ struct PeerDownloadId(u64);
 )]
 struct BitfieldSubscriptionId(u64);
 
-/// Announce kind
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Default)]
-pub enum AnnounceKind {
-    /// The peer supposedly has some of the data.
-    Partial = 0,
-    /// The peer supposedly has the complete data.
-    #[default]
-    Complete,
-}
-
-/// Options for finding peers
-#[derive(Debug, Default)]
-pub struct FindPeersOpts {
-    /// Kind of announce
-    pub kind: AnnounceKind,
-}
-
-/// A pluggable content discovery mechanism
-pub trait ContentDiscovery: std::fmt::Debug + Send + 'static {
-    /// Find peers that have the given blob.
-    ///
-    /// The returned stream is a handle for the discovery task. It should be an
-    /// infinite stream that only stops when it is dropped.
-    fn find_peers(&mut self, hash: Hash, opts: FindPeersOpts) -> BoxStream<'static, NodeId>;
-}
-
-/// A boxed content discovery
-pub type BoxedContentDiscovery = Box<dyn ContentDiscovery>;
-
 /// A pluggable bitfield subscription mechanism
 pub trait BitfieldSubscription: std::fmt::Debug + Send + 'static {
     /// Subscribe to a bitfield
@@ -265,7 +239,7 @@ impl<S> DownloaderBuilder<S> {
     {
         let store = self.store;
         let discovery = self.discovery.expect("discovery not set");
-        let local_pool = self.local_pool.unwrap_or_else(|| LocalPool::single());
+        let local_pool = self.local_pool.unwrap_or_else(LocalPool::single);
         let planner = self
             .planner
             .unwrap_or_else(|| Box::new(StripePlanner2::new(0, 10)));
@@ -350,33 +324,6 @@ impl Downloader {
     }
 }
 
-/// A simple static content discovery mechanism
-#[derive(Debug)]
-pub struct StaticContentDiscovery {
-    info: BTreeMap<Hash, Vec<NodeId>>,
-    default: Vec<NodeId>,
-}
-
-impl StaticContentDiscovery {
-    /// Create a new static content discovery mechanism
-    pub fn new(mut info: BTreeMap<Hash, Vec<NodeId>>, mut default: Vec<NodeId>) -> Self {
-        default.sort();
-        default.dedup();
-        for (_, peers) in info.iter_mut() {
-            peers.sort();
-            peers.dedup();
-        }
-        Self { info, default }
-    }
-}
-
-impl ContentDiscovery for StaticContentDiscovery {
-    fn find_peers(&mut self, hash: Hash, _opts: FindPeersOpts) -> BoxStream<'static, NodeId> {
-        let peers = self.info.get(&hash).unwrap_or(&self.default).clone();
-        Box::pin(futures_lite::stream::iter(peers).chain(futures_lite::stream::pending()))
-    }
-}
-
 /// A bitfield subscription that just returns nothing for local and everything(*) for remote
 ///
 /// * Still need to figure out how to deal with open ended chunk ranges.
@@ -423,7 +370,7 @@ impl<S> SimpleBitfieldSubscription<S> {
 }
 
 async fn get_valid_ranges_local<S: Store>(hash: &Hash, store: S) -> anyhow::Result<ChunkRanges> {
-    if let Some(entry) = store.get_mut(&hash).await? {
+    if let Some(entry) = store.get_mut(hash).await? {
         crate::get::db::valid_ranges::<S>(&entry).await
     } else {
         Ok(ChunkRanges::empty())
@@ -436,7 +383,7 @@ async fn get_valid_ranges_remote(
     hash: &Hash,
 ) -> anyhow::Result<ChunkRanges> {
     let conn = endpoint.connect(id, crate::ALPN).await?;
-    let (size, _) = crate::get::request::get_verified_size(&conn, &hash).await?;
+    let (size, _) = crate::get::request::get_verified_size(&conn, hash).await?;
     let chunks = (size + 1023) / 1024;
     Ok(ChunkRanges::from(ChunkNum(0)..ChunkNum(chunks)))
 }
@@ -491,6 +438,7 @@ impl<S: Store> BitfieldSubscription for SimpleBitfieldSubscription<S> {
 
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::single_range_in_vec_init)]
     use std::ops::Range;
 
     use crate::net_protocol::Blobs;
@@ -534,7 +482,7 @@ mod tests {
         let mut planner = StripePlanner2::new(0, 4);
         let hash = Hash::new(b"test");
         let mut ranges = make_range_map(&[chunk_ranges([0..50]), chunk_ranges([50..100])]);
-        println!("");
+        println!();
         print_range_map(&ranges);
         println!("planning");
         planner.plan(hash, &mut ranges);
@@ -550,7 +498,7 @@ mod tests {
             chunk_ranges([0..100]),
             chunk_ranges([0..100]),
         ]);
-        println!("");
+        println!();
         print_range_map(&ranges);
         println!("planning");
         planner.plan(hash, &mut ranges);
@@ -567,7 +515,7 @@ mod tests {
             chunk_ranges([0..120]),
             chunk_ranges([0..50]),
         ]);
-        println!("");
+        println!();
         print_range_map(&ranges);
         println!("planning");
         planner.plan(hash, &mut ranges);
@@ -656,10 +604,7 @@ mod tests {
             .discovery_n0()
             .bind()
             .await?;
-        let discovery = StaticContentDiscovery {
-            info: BTreeMap::new(),
-            default: vec![peer],
-        };
+        let discovery = StaticContentDiscovery::new(BTreeMap::new(), vec![peer]);
         let bitfield_subscription = TestBitfieldSubscription;
         let downloader = Downloader::builder(endpoint, store)
             .discovery(discovery)
@@ -697,10 +642,7 @@ mod tests {
             .discovery_n0()
             .bind()
             .await?;
-        let discovery = StaticContentDiscovery {
-            info: BTreeMap::new(),
-            default: peers,
-        };
+        let discovery = StaticContentDiscovery::new(BTreeMap::new(), peers);
         let downloader = Downloader::builder(endpoint, store)
             .discovery(discovery)
             .planner(StripePlanner2::new(0, 8))

src/downloader2/actor.rs

@@ -15,17 +15,23 @@ pub(super) enum UserCommand {
 }
 
 pub(super) struct DownloaderActor<S> {
+    state: DownloaderState,
     local_pool: LocalPool,
     endpoint: Endpoint,
     command_rx: mpsc::Receiver<Command>,
     command_tx: mpsc::Sender<Command>,
-    state: DownloaderState,
     store: S,
+    /// Content discovery
     discovery: BoxedContentDiscovery,
+    /// Bitfield subscription
     subscribe_bitfield: BoxedBitfieldSubscription,
+    /// Futures for downloads
     download_futs: BTreeMap<DownloadId, tokio::sync::oneshot::Sender<()>>,
+    /// Tasks for peer downloads
     peer_download_tasks: BTreeMap<PeerDownloadId, local_pool::Run<()>>,
+    /// Tasks for discovery
     discovery_tasks: BTreeMap<DiscoveryId, AbortOnDropHandle<()>>,
+    /// Tasks for bitfield subscriptions
     bitfield_subscription_tasks: BTreeMap<BitfieldSubscriptionId, AbortOnDropHandle<()>>,
     /// Id generator for download ids
     download_id_gen: IdGenerator<DownloadId>,
@@ -67,13 +73,13 @@ impl<S: Store> DownloaderActor<S> {
         }
     }
 
-    pub(super) async fn run(mut self, mut channel: mpsc::Receiver<UserCommand>) {
+    pub(super) async fn run(mut self, mut user_commands: mpsc::Receiver<UserCommand>) {
         let mut ticks = tokio::time::interval(Duration::from_millis(100));
         loop {
             trace!("downloader actor tick");
             tokio::select! {
                 biased;
-                Some(cmd) = channel.recv() => {
+                Some(cmd) = user_commands.recv() => {
                     debug!("user command {cmd:?}");
                     match cmd {
                         UserCommand::Download {
@@ -228,6 +234,7 @@ impl<S: Store> DownloaderActor<S> {
     }
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn peer_download_task<S: Store>(
     id: PeerDownloadId,
     endpoint: Endpoint,
@@ -327,6 +334,7 @@ async fn peer_download<S: Store>(
     Ok(stats)
 }
 
+/// Spawn a future and wrap it in a [`AbortOnDropHandle`]
 pub(super) fn spawn<F, T>(f: F) -> AbortOnDropHandle<T>
 where
     F: Future<Output = T> + Send + 'static,

src/downloader2/content_discovery.rs

@@ -0,0 +1,63 @@
+use std::collections::BTreeMap;
+
+use crate::Hash;
+use futures_lite::stream::StreamExt;
+use futures_util::stream::BoxStream;
+use iroh::NodeId;
+use serde::{Deserialize, Serialize};
+
+/// Announce kind
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Default)]
+pub enum AnnounceKind {
+    /// The peer supposedly has some of the data.
+    Partial = 0,
+    /// The peer supposedly has the complete data.
+    #[default]
+    Complete,
+}
+
+/// Options for finding peers
+#[derive(Debug, Default)]
+pub struct FindPeersOpts {
+    /// Kind of announce
+    pub kind: AnnounceKind,
+}
+
+/// A pluggable content discovery mechanism
+pub trait ContentDiscovery: std::fmt::Debug + Send + 'static {
+    /// Find peers that have the given blob.
+    ///
+    /// The returned stream is a handle for the discovery task. It should be an
+    /// infinite stream that only stops when it is dropped.
+    fn find_peers(&mut self, hash: Hash, opts: FindPeersOpts) -> BoxStream<'static, NodeId>;
+}
+
+/// A boxed content discovery
+pub type BoxedContentDiscovery = Box<dyn ContentDiscovery>;
+
+/// A simple static content discovery mechanism
+#[derive(Debug)]
+pub struct StaticContentDiscovery {
+    info: BTreeMap<Hash, Vec<NodeId>>,
+    default: Vec<NodeId>,
+}
+
+impl StaticContentDiscovery {
+    /// Create a new static content discovery mechanism
+    pub fn new(mut info: BTreeMap<Hash, Vec<NodeId>>, mut default: Vec<NodeId>) -> Self {
+        default.sort();
+        default.dedup();
+        for (_, peers) in info.iter_mut() {
+            peers.sort();
+            peers.dedup();
+        }
+        Self { info, default }
+    }
+}
+
+impl ContentDiscovery for StaticContentDiscovery {
+    fn find_peers(&mut self, hash: Hash, _opts: FindPeersOpts) -> BoxStream<'static, NodeId> {
+        let peers = self.info.get(&hash).unwrap_or(&self.default).clone();
+        Box::pin(futures_lite::stream::iter(peers).chain(futures_lite::stream::pending()))
+    }
+}

src/downloader2/state.rs

@@ -64,13 +64,13 @@ pub(super) enum Command {
     DropPeer { peer: NodeId },
     /// A peer has been discovered
     PeerDiscovered { peer: NodeId, hash: Hash },
-    ///
+    /// Start observing a local bitfield
     ObserveLocal {
         id: ObserveId,
         hash: Hash,
         ranges: ChunkRanges,
     },
-    ///
+    /// Stop observing a local bitfield
     StopObserveLocal { id: ObserveId },
     /// A tick from the driver, for rebalancing
     Tick { time: Duration },
@@ -319,9 +319,9 @@ impl DownloaderState {
                         evs.push(Event::UnsubscribeBitfield {
                             id: state.subscription_id,
                         });
-                        return false;
+                        false
                     } else {
-                        return true;
+                        true
                     }
                 });
                 self.peers.remove(&peer);
@@ -666,7 +666,7 @@ impl DownloaderState {
             "Stopping {} old peer downloads",
             download.peer_downloads.len()
         );
-        for (_, state) in &download.peer_downloads {
+        for state in download.peer_downloads.values() {
             // stop all downloads
             evs.push(Event::StopPeerDownload { id: state.id });
         }
@@ -906,6 +906,7 @@ fn total_chunks(chunks: &ChunkRanges) -> Option<u64> {
 
 #[cfg(test)]
 mod tests {
+    #![allow(clippy::single_range_in_vec_init)]
 
     use super::super::tests::{
         chunk_ranges, has_all_events, has_one_event, has_one_event_matching, noop_planner,