dev: more refactoring torrent/repository

Author: da2ce7

src/core/torrent/entry.rs

@@ -23,7 +23,7 @@ pub struct Entry {
     /// The number of peers that have ever completed downloading the torrent associated to this entry
     pub(crate) completed: u32,
 }
-
+pub type Single = Entry;
 pub type MutexStd = Arc<std::sync::Mutex<Entry>>;
 pub type MutexTokio = Arc<tokio::sync::Mutex<Entry>>;
 
@@ -40,6 +40,7 @@ pub trait ReadInfo {
     fn peers_is_empty(&self) -> bool;
 }
 
+/// Same as [`ReadInfo`], but async.
 pub trait ReadInfoAsync {
     /// It returns the swarm metadata (statistics) as a struct:
     ///
@@ -65,15 +66,10 @@ pub trait ReadPeers {
     fn get_peers_for_peer(&self, client: &peer::Peer, limit: Option<usize>) -> Vec<Arc<peer::Peer>>;
 }
 
+/// Same as [`ReadPeers`], but async.
 pub trait ReadPeersAsync {
-    /// Get all swarm peers, optionally limiting the result.
     fn get_peers(&self, limit: Option<usize>) -> impl std::future::Future<Output = Vec<Arc<peer::Peer>>> + Send;
 
-    /// It returns the list of peers for a given peer client, optionally limiting the
-    /// result.
-    ///
-    /// It filters out the input peer, typically because we want to return this
-    /// list of peers to that client peer.
     fn get_peers_for_peer(
         &self,
         client: &peer::Peer,
@@ -95,12 +91,14 @@ pub trait Update {
     fn remove_inactive_peers(&mut self, max_peer_timeout: u32);
 }
 
+/// Same as [`Update`], except not `mut`.
 pub trait UpdateSync {
     fn insert_or_update_peer(&self, peer: &peer::Peer) -> bool;
     fn insert_or_update_peer_and_get_stats(&self, peer: &peer::Peer) -> (bool, SwarmMetadata);
     fn remove_inactive_peers(&self, max_peer_timeout: u32);
 }
 
+/// Same as [`Update`], except not `mut` and async.
 pub trait UpdateAsync {
     fn insert_or_update_peer(&self, peer: &peer::Peer) -> impl std::future::Future<Output = bool> + Send;
 
@@ -112,7 +110,7 @@ pub trait UpdateAsync {
     fn remove_inactive_peers(&self, max_peer_timeout: u32) -> impl std::future::Future<Output = ()> + Send;
 }
 
-impl ReadInfo for Entry {
+impl ReadInfo for Single {
     #[allow(clippy::cast_possible_truncation)]
     fn get_stats(&self) -> SwarmMetadata {
         let complete: u32 = self.peers.values().filter(|peer| peer.is_seeder()).count() as u32;
@@ -170,7 +168,7 @@ impl ReadInfoAsync for MutexTokio {
     }
 }
 
-impl ReadPeers for Entry {
+impl ReadPeers for Single {
     fn get_peers(&self, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
         match limit {
             Some(limit) => self.peers.values().take(limit).cloned().collect(),
@@ -220,7 +218,7 @@ impl ReadPeersAsync for MutexTokio {
     }
 }
 
-impl Update for Entry {
+impl Update for Single {
     fn insert_or_update_peer(&mut self, peer: &peer::Peer) -> bool {
         let mut did_torrent_stats_change: bool = false;
 

src/core/torrent/mod.rs

@@ -11,8 +11,6 @@
 //! That's the most valuable information the peer want to get from the tracker, because it allows them to
 //! start downloading torrent from those peers.
 //!
-//! > **NOTICE**: that both swarm data (torrent entries) and swarm metadata (aggregate counters) are related to only one torrent.
-//!
 //! The "swarm metadata" contains aggregate data derived from the torrent entries. There two types of data:
 //!
 //! - For **active peers**: metrics related to the current active peers in the swarm.
@@ -35,10 +33,10 @@ use derive_more::Constructor;
 
 pub type Torrents = TorrentsRwLockStd; // Currently Used
 
-pub type TorrentsRwLockStd = repository::RwLockStd<entry::Entry>;
+pub type TorrentsRwLockStd = repository::RwLockStd<entry::Single>;
 pub type TorrentsRwLockStdMutexStd = repository::RwLockStd<entry::MutexStd>;
 pub type TorrentsRwLockStdMutexTokio = repository::RwLockStd<entry::MutexTokio>;
-pub type TorrentsRwLockTokio = repository::RwLockTokio<entry::Entry>;
+pub type TorrentsRwLockTokio = repository::RwLockTokio<entry::Single>;
 pub type TorrentsRwLockTokioMutexStd = repository::RwLockTokio<entry::MutexStd>;
 pub type TorrentsRwLockTokioMutexTokio = repository::RwLockTokio<entry::MutexTokio>;
 
@@ -147,14 +145,14 @@ mod tests {
 
         #[test]
         fn the_default_torrent_entry_should_contain_an_empty_list_of_peers() {
-            let torrent_entry = entry::Entry::default();
+            let torrent_entry = entry::Single::default();
 
             assert_eq!(torrent_entry.get_peers(None).len(), 0);
         }
 
         #[test]
         fn a_new_peer_can_be_added_to_a_torrent_entry() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let torrent_peer = TorrentPeerBuilder::default().into();
 
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add the peer
@@ -165,7 +163,7 @@ mod tests {
 
         #[test]
         fn a_torrent_entry_should_contain_the_list_of_peers_that_were_added_to_the_torrent() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let torrent_peer = TorrentPeerBuilder::default().into();
 
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add the peer
@@ -175,7 +173,7 @@ mod tests {
 
         #[test]
         fn a_peer_can_be_updated_in_a_torrent_entry() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let mut torrent_peer = TorrentPeerBuilder::default().into();
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add the peer
 
@@ -187,7 +185,7 @@ mod tests {
 
         #[test]
         fn a_peer_should_be_removed_from_a_torrent_entry_when_the_peer_announces_it_has_stopped() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let mut torrent_peer = TorrentPeerBuilder::default().into();
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add the peer
 
@@ -199,7 +197,7 @@ mod tests {
 
         #[test]
         fn torrent_stats_change_when_a_previously_known_peer_announces_it_has_completed_the_torrent() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let mut torrent_peer = TorrentPeerBuilder::default().into();
 
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add the peer
@@ -213,7 +211,7 @@ mod tests {
         #[test]
         fn torrent_stats_should_not_change_when_a_peer_announces_it_has_completed_the_torrent_if_it_is_the_first_announce_from_the_peer(
         ) {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let torrent_peer_announcing_complete_event = TorrentPeerBuilder::default().with_event_completed().into();
 
             // Add a peer that did not exist before in the entry
@@ -225,7 +223,7 @@ mod tests {
         #[test]
         fn a_torrent_entry_should_return_the_list_of_peers_for_a_given_peer_filtering_out_the_client_that_is_making_the_request()
         {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let peer_socket_address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080);
             let torrent_peer = TorrentPeerBuilder::default().with_peer_address(peer_socket_address).into();
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add peer
@@ -238,7 +236,7 @@ mod tests {
 
         #[test]
         fn two_peers_with_the_same_ip_but_different_port_should_be_considered_different_peers() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
 
             let peer_ip = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
 
@@ -272,7 +270,7 @@ mod tests {
 
         #[test]
         fn the_tracker_should_limit_the_list_of_peers_to_74_when_clients_scrape_torrents() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
 
             // We add one more peer than the scrape limit
             for peer_number in 1..=74 + 1 {
@@ -289,7 +287,7 @@ mod tests {
 
         #[test]
         fn torrent_stats_should_have_the_number_of_seeders_for_a_torrent() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let torrent_seeder = a_torrent_seeder();
 
             torrent_entry.insert_or_update_peer(&torrent_seeder); // Add seeder
@@ -299,7 +297,7 @@ mod tests {
 
         #[test]
         fn torrent_stats_should_have_the_number_of_leechers_for_a_torrent() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let torrent_leecher = a_torrent_leecher();
 
             torrent_entry.insert_or_update_peer(&torrent_leecher); // Add leecher
@@ -310,7 +308,7 @@ mod tests {
         #[test]
         fn torrent_stats_should_have_the_number_of_peers_that_having_announced_at_least_two_events_the_latest_one_is_the_completed_event(
         ) {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let mut torrent_peer = TorrentPeerBuilder::default().into();
             torrent_entry.insert_or_update_peer(&torrent_peer); // Add the peer
 
@@ -325,7 +323,7 @@ mod tests {
 
         #[test]
         fn torrent_stats_should_not_include_a_peer_in_the_completed_counter_if_the_peer_has_announced_only_one_event() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
             let torrent_peer_announcing_complete_event = TorrentPeerBuilder::default().with_event_completed().into();
 
             // Announce "Completed" torrent download event.
@@ -339,7 +337,7 @@ mod tests {
 
         #[test]
         fn a_torrent_entry_should_remove_a_peer_not_updated_after_a_timeout_in_seconds() {
-            let mut torrent_entry = entry::Entry::default();
+            let mut torrent_entry = entry::Single::default();
 
             let timeout = 120u32;
 

src/core/torrent/repository/mod.rs

@@ -4,14 +4,14 @@ use crate::core::services::torrent::Pagination;
 use crate::core::{peer, TorrentsMetrics, TrackerPolicy};
 use crate::shared::bit_torrent::info_hash::InfoHash;
 
-pub mod std_single;
-pub mod std_std;
-pub mod std_tokio;
-pub mod tokio_single;
-pub mod tokio_std;
-pub mod tokio_tokio;
+pub mod rw_lock_std;
+pub mod rw_lock_std_mutex_std;
+pub mod rw_lock_std_mutex_tokio;
+pub mod rw_lock_tokio;
+pub mod rw_lock_tokio_mutex_std;
+pub mod rw_lock_tokio_mutex_tokio;
 
-pub trait Repository<T>: Default {
+pub trait Repository<T>: Default + 'static {
     fn get(&self, key: &InfoHash) -> impl std::future::Future<Output = Option<T>> + Send;
     fn get_metrics(&self) -> impl std::future::Future<Output = TorrentsMetrics> + Send;
     fn get_paginated(&self, pagination: &Pagination) -> impl std::future::Future<Output = Vec<(InfoHash, T)>> + Send;
@@ -21,16 +21,6 @@ pub trait Repository<T>: Default {
     fn remove_peerless_torrents(&self, policy: &TrackerPolicy) -> impl std::future::Future<Output = ()> + Send;
 }
 
-#[derive(Default)]
-pub struct RwLockTokio<T> {
-    torrents: tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, T>>,
-}
-
-#[derive(Default)]
-pub struct RwLockStd<T> {
-    torrents: std::sync::RwLock<std::collections::BTreeMap<InfoHash, T>>,
-}
-
 pub trait UpdateTorrentSync {
     fn update_torrent_with_peer_and_get_stats(&self, info_hash: &InfoHash, peer: &peer::Peer) -> (bool, SwarmMetadata);
 }
@@ -42,3 +32,13 @@ pub trait UpdateTorrentAsync {
         peer: &peer::Peer,
     ) -> impl std::future::Future<Output = (bool, SwarmMetadata)> + Send;
 }
+
+#[derive(Default)]
+pub struct RwLockTokio<T> {
+    torrents: tokio::sync::RwLock<std::collections::BTreeMap<InfoHash, T>>,
+}
+
+#[derive(Default)]
+pub struct RwLockStd<T> {
+    torrents: std::sync::RwLock<std::collections::BTreeMap<InfoHash, T>>,
+}

src/core/torrent/repository/std_single.rs src/core/torrent/repository/rw_lock_std.rs

@@ -3,48 +3,48 @@ use std::sync::Arc;
 
 use futures::future::join_all;
 
-use super::{Repository, RwLockStd, UpdateTorrentSync};
+use super::{Repository, UpdateTorrentSync};
 use crate::core::databases::PersistentTorrents;
 use crate::core::services::torrent::Pagination;
-use crate::core::torrent::entry::{Entry, ReadInfo, Update};
-use crate::core::torrent::SwarmMetadata;
+use crate::core::torrent::entry::{self, ReadInfo, Update};
+use crate::core::torrent::{SwarmMetadata, TorrentsRwLockStd};
 use crate::core::{peer, TorrentsMetrics};
 use crate::shared::bit_torrent::info_hash::InfoHash;
 
-impl RwLockStd<Entry> {
-    fn get_torrents<'a>(&'a self) -> std::sync::RwLockReadGuard<'a, std::collections::BTreeMap<InfoHash, Entry>>
+impl TorrentsRwLockStd {
+    fn get_torrents<'a>(&'a self) -> std::sync::RwLockReadGuard<'a, std::collections::BTreeMap<InfoHash, entry::Single>>
     where
-        std::collections::BTreeMap<InfoHash, Entry>: 'a,
+        std::collections::BTreeMap<InfoHash, entry::Single>: 'a,
     {
         self.torrents.read().expect("it should get the read lock")
     }
 
-    fn get_torrents_mut<'a>(&'a self) -> std::sync::RwLockWriteGuard<'a, std::collections::BTreeMap<InfoHash, Entry>>
+    fn get_torrents_mut<'a>(&'a self) -> std::sync::RwLockWriteGuard<'a, std::collections::BTreeMap<InfoHash, entry::Single>>
     where
-        std::collections::BTreeMap<InfoHash, Entry>: 'a,
+        std::collections::BTreeMap<InfoHash, entry::Single>: 'a,
     {
         self.torrents.write().expect("it should get the write lock")
     }
 }
 
-impl UpdateTorrentSync for RwLockStd<Entry> {
+impl UpdateTorrentSync for TorrentsRwLockStd {
     fn update_torrent_with_peer_and_get_stats(&self, info_hash: &InfoHash, peer: &peer::Peer) -> (bool, SwarmMetadata) {
         let mut db = self.get_torrents_mut();
 
-        let entry = db.entry(*info_hash).or_insert(Entry::default());
+        let entry = db.entry(*info_hash).or_insert(entry::Single::default());
 
         entry.insert_or_update_peer_and_get_stats(peer)
     }
 }
 
-impl UpdateTorrentSync for Arc<RwLockStd<Entry>> {
+impl UpdateTorrentSync for Arc<TorrentsRwLockStd> {
     fn update_torrent_with_peer_and_get_stats(&self, info_hash: &InfoHash, peer: &peer::Peer) -> (bool, SwarmMetadata) {
         self.as_ref().update_torrent_with_peer_and_get_stats(info_hash, peer)
     }
 }
 
-impl Repository<Entry> for RwLockStd<Entry> {
-    async fn get(&self, key: &InfoHash) -> Option<Entry> {
+impl Repository<entry::Single> for TorrentsRwLockStd {
+    async fn get(&self, key: &InfoHash) -> Option<entry::Single> {
         let db = self.get_torrents();
         db.get(key).cloned()
     }
@@ -71,7 +71,7 @@ impl Repository<Entry> for RwLockStd<Entry> {
         *metrics.lock_owned().await
     }
 
-    async fn get_paginated(&self, pagination: &Pagination) -> Vec<(InfoHash, Entry)> {
+    async fn get_paginated(&self, pagination: &Pagination) -> Vec<(InfoHash, entry::Single)> {
         let db = self.get_torrents();
         db.iter()
             .skip(pagination.offset as usize)
@@ -89,7 +89,7 @@ impl Repository<Entry> for RwLockStd<Entry> {
                 continue;
             }
 
-            let entry = Entry {
+            let entry = entry::Single {
                 peers: BTreeMap::default(),
                 completed: *completed,
             };
@@ -98,7 +98,7 @@ impl Repository<Entry> for RwLockStd<Entry> {
         }
     }
 
-    async fn remove(&self, key: &InfoHash) -> Option<Entry> {
+    async fn remove(&self, key: &InfoHash) -> Option<entry::Single> {
         let mut db = self.get_torrents_mut();
         db.remove(key)
     }