Refactor event listeners initialization

packages/axum-http-tracker-server/src/environment.rs

@@ -4,6 +4,7 @@ use bittorrent_http_tracker_core::container::HttpTrackerCoreContainer;
 use bittorrent_primitives::info_hash::InfoHash;
 use bittorrent_tracker_core::container::TrackerCoreContainer;
 use futures::executor::block_on;
+use tokio::task::JoinHandle;
 use torrust_axum_server::tsl::make_rust_tls;
 use torrust_server_lib::registar::Registar;
 use torrust_tracker_configuration::{logging, Configuration};
@@ -17,6 +18,7 @@ pub struct Environment<S> {
     pub container: Arc<EnvContainer>,
     pub registar: Registar,
     pub server: HttpServer<S>,
+    pub event_listener_job: Option<JoinHandle<()>>,
 }
 
 impl<S> Environment<S> {
@@ -54,22 +56,32 @@ impl Environment<Stopped> {
             container,
             registar: Registar::default(),
             server,
+            event_listener_job: None,
         }
     }
 
+    /// Starts the test environment and return a running environment.
+    ///
     /// # Panics
     ///
     /// Will panic if the server fails to start.    
     #[allow(dead_code)]
     pub async fn start(self) -> Environment<Running> {
+        // Start the event listener
+        let event_listener_job = self.container.http_tracker_core_container.stats_keeper.run_event_listener();
+
+        // Start the server
+        let server = self
+            .server
+            .start(self.container.http_tracker_core_container.clone(), self.registar.give_form())
+            .await
+            .expect("Failed to start the HTTP tracker server");
+
         Environment {
             container: self.container.clone(),
             registar: self.registar.clone(),
-            server: self
-                .server
-                .start(self.container.http_tracker_core_container.clone(), self.registar.give_form())
-                .await
-                .unwrap(),
+            server,
+            event_listener_job: Some(event_listener_job),
         }
     }
 }
@@ -79,14 +91,27 @@ impl Environment<Running> {
         Environment::<Stopped>::new(configuration).start().await
     }
 
+    /// Stops the test environment and return a stopped environment.
+    ///
     /// # Panics
     ///
     /// Will panic if the server fails to stop.
     pub async fn stop(self) -> Environment<Stopped> {
+        // Stop the event listener
+        if let Some(event_listener_job) = self.event_listener_job {
+            // todo: send a message to the event listener to stop and wait for
+            // it to finish
+            event_listener_job.abort();
+        }
+
+        // Stop the server
+        let server = self.server.stop().await.expect("Failed to stop the HTTP tracker server");
+
         Environment {
             container: self.container,
             registar: Registar::default(),
-            server: self.server.stop().await.unwrap(),
+            server,
+            event_listener_job: None,
         }
     }
 

packages/axum-http-tracker-server/src/server.rs

@@ -270,11 +270,15 @@ mod tests {
 
         let http_tracker_config = Arc::new(http_tracker_config.clone());
 
-        // HTTP stats
-        let (http_stats_event_sender, http_stats_repository) =
+        // HTTP core stats
+        let http_stats_keeper =
             bittorrent_http_tracker_core::statistics::setup::factory(configuration.core.tracker_usage_statistics);
-        let http_stats_event_sender = Arc::new(http_stats_event_sender);
-        let http_stats_repository = Arc::new(http_stats_repository);
+        let http_stats_event_sender = http_stats_keeper.sender();
+        let http_stats_repository = http_stats_keeper.repository();
+
+        if configuration.core.tracker_usage_statistics {
+            let _unused = http_stats_keeper.run_event_listener();
+        }
 
         let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
 
@@ -296,8 +300,9 @@ mod tests {
         HttpTrackerCoreContainer {
             tracker_core_container,
             http_tracker_config,
-            http_stats_event_sender,
-            http_stats_repository,
+            stats_keeper: http_stats_keeper,
+            stats_event_sender: http_stats_event_sender,
+            stats_repository: http_stats_repository,
             announce_service,
             scrape_service,
         }

packages/axum-http-tracker-server/src/v1/handlers/announce.rs

@@ -160,11 +160,14 @@ mod tests {
             &db_torrent_repository,
         ));
 
-        // HTTP stats
-        let (http_stats_event_sender, http_stats_repository) =
-            bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics);
-        let http_stats_event_sender = Arc::new(http_stats_event_sender);
-        let _http_stats_repository = Arc::new(http_stats_repository);
+        // HTTP core stats
+        let http_stats_keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics);
+        let http_stats_event_sender = http_stats_keeper.sender();
+        let _http_stats_repository = http_stats_keeper.repository();
+
+        if config.core.tracker_usage_statistics {
+            let _unused = http_stats_keeper.run_event_listener();
+        }
 
         let announce_service = Arc::new(AnnounceService::new(
             core_config.clone(),

packages/axum-http-tracker-server/src/v1/handlers/scrape.rs

@@ -131,10 +131,14 @@ mod tests {
         let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
         let scrape_handler = Arc::new(ScrapeHandler::new(&whitelist_authorization, &in_memory_torrent_repository));
 
-        // HTTP stats
-        let (http_stats_event_sender, _http_stats_repository) =
-            bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics);
-        let http_stats_event_sender = Arc::new(http_stats_event_sender);
+        // HTTP core stats
+        let http_stats_keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics);
+        let http_stats_event_sender = http_stats_keeper.sender();
+        let _http_stats_repository = http_stats_keeper.repository();
+
+        if config.core.tracker_usage_statistics {
+            let _unused = http_stats_keeper.run_event_listener();
+        }
 
         (
             CoreTrackerServices {

packages/axum-http-tracker-server/tests/server/v1/contract.rs

@@ -676,12 +676,7 @@ mod for_all_config_modes {
                 .announce(&QueryBuilder::default().query())
                 .await;
 
-            let stats = env
-                .container
-                .http_tracker_core_container
-                .http_stats_repository
-                .get_stats()
-                .await;
+            let stats = env.container.http_tracker_core_container.stats_repository.get_stats().await;
 
             assert_eq!(stats.tcp4_announces_handled, 1);
 
@@ -707,12 +702,7 @@ mod for_all_config_modes {
                 .announce(&QueryBuilder::default().query())
                 .await;
 
-            let stats = env
-                .container
-                .http_tracker_core_container
-                .http_stats_repository
-                .get_stats()
-                .await;
+            let stats = env.container.http_tracker_core_container.stats_repository.get_stats().await;
 
             assert_eq!(stats.tcp6_announces_handled, 1);
 
@@ -737,12 +727,7 @@ mod for_all_config_modes {
                 )
                 .await;
 
-            let stats = env
-                .container
-                .http_tracker_core_container
-                .http_stats_repository
-                .get_stats()
-                .await;
+            let stats = env.container.http_tracker_core_container.stats_repository.get_stats().await;
 
             assert_eq!(stats.tcp6_announces_handled, 0);
 
@@ -1130,12 +1115,7 @@ mod for_all_config_modes {
                 )
                 .await;
 
-            let stats = env
-                .container
-                .http_tracker_core_container
-                .http_stats_repository
-                .get_stats()
-                .await;
+            let stats = env.container.http_tracker_core_container.stats_repository.get_stats().await;
 
             assert_eq!(stats.tcp4_scrapes_handled, 1);
 
@@ -1167,12 +1147,7 @@ mod for_all_config_modes {
                 )
                 .await;
 
-            let stats = env
-                .container
-                .http_tracker_core_container
-                .http_stats_repository
-                .get_stats()
-                .await;
+            let stats = env.container.http_tracker_core_container.stats_repository.get_stats().await;
 
             assert_eq!(stats.tcp6_scrapes_handled, 1);
 

packages/http-tracker-core/benches/helpers/util.rs

@@ -2,6 +2,8 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::sync::Arc;
 
 use aquatic_udp_protocol::{AnnounceEvent, NumberOfBytes, PeerId};
+use bittorrent_http_tracker_core::event::Event;
+use bittorrent_http_tracker_core::{event, statistics};
 use bittorrent_http_tracker_protocol::v1::requests::announce::Announce;
 use bittorrent_http_tracker_protocol::v1::services::peer_ip_resolver::ClientIpSources;
 use bittorrent_primitives::info_hash::InfoHash;
@@ -13,6 +15,9 @@ use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepo
 use bittorrent_tracker_core::torrent::repository::persisted::DatabasePersistentTorrentRepository;
 use bittorrent_tracker_core::whitelist::authorization::WhitelistAuthorization;
 use bittorrent_tracker_core::whitelist::repository::in_memory::InMemoryWhitelist;
+use futures::future::BoxFuture;
+use mockall::mock;
+use tokio::sync::broadcast::error::SendError;
 use torrust_tracker_configuration::{Configuration, Core};
 use torrust_tracker_primitives::peer::Peer;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
@@ -50,10 +55,14 @@ pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> (
         &db_torrent_repository,
     ));
 
-    // HTTP stats
-    let (http_stats_event_sender, http_stats_repository) = statistics::setup::factory(config.core.tracker_usage_statistics);
-    let http_stats_event_sender = Arc::new(http_stats_event_sender);
-    let _http_stats_repository = Arc::new(http_stats_repository);
+    // HTTP core stats
+    let http_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics);
+    let http_stats_event_sender = http_stats_keeper.sender();
+    let _http_stats_repository = http_stats_keeper.repository();
+
+    if config.core.tracker_usage_statistics {
+        let _unused = http_stats_keeper.run_event_listener();
+    }
 
     (
         CoreTrackerServices {
@@ -105,12 +114,6 @@ pub fn sample_info_hash() -> InfoHash {
         .expect("String should be a valid info hash")
 }
 
-use bittorrent_http_tracker_core::event::Event;
-use bittorrent_http_tracker_core::{event, statistics};
-use futures::future::BoxFuture;
-use mockall::mock;
-use tokio::sync::broadcast::error::SendError;
-
 mock! {
     HttpStatsEventSender {}
     impl event::sender::Sender for HttpStatsEventSender {

packages/http-tracker-core/src/container.rs

@@ -13,8 +13,9 @@ pub struct HttpTrackerCoreContainer {
     pub tracker_core_container: Arc<TrackerCoreContainer>,
 
     // `HttpTrackerCoreServices`
-    pub http_stats_event_sender: Arc<Option<Box<dyn event::sender::Sender>>>,
-    pub http_stats_repository: Arc<statistics::repository::Repository>,
+    pub stats_keeper: Arc<statistics::keeper::Keeper>,
+    pub stats_event_sender: Arc<Option<Box<dyn event::sender::Sender>>>,
+    pub stats_repository: Arc<statistics::repository::Repository>,
     pub announce_service: Arc<AnnounceService>,
     pub scrape_service: Arc<ScrapeService>,
 }
@@ -44,35 +45,39 @@ impl HttpTrackerCoreContainer {
         Arc::new(Self {
             tracker_core_container: tracker_core_container.clone(),
             http_tracker_config: http_tracker_config.clone(),
-            http_stats_event_sender: http_tracker_core_services.http_stats_event_sender.clone(),
-            http_stats_repository: http_tracker_core_services.http_stats_repository.clone(),
-            announce_service: http_tracker_core_services.http_announce_service.clone(),
-            scrape_service: http_tracker_core_services.http_scrape_service.clone(),
+            stats_keeper: http_tracker_core_services.stats_keeper.clone(),
+            stats_event_sender: http_tracker_core_services.stats_event_sender.clone(),
+            stats_repository: http_tracker_core_services.stats_repository.clone(),
+            announce_service: http_tracker_core_services.announce_service.clone(),
+            scrape_service: http_tracker_core_services.scrape_service.clone(),
         })
     }
 }
 
 pub struct HttpTrackerCoreServices {
-    pub http_stats_event_sender: Arc<Option<Box<dyn event::sender::Sender>>>,
-    pub http_stats_repository: Arc<statistics::repository::Repository>,
-    pub http_announce_service: Arc<services::announce::AnnounceService>,
-    pub http_scrape_service: Arc<services::scrape::ScrapeService>,
+    pub stats_keeper: Arc<statistics::keeper::Keeper>,
+    pub stats_event_sender: Arc<Option<Box<dyn event::sender::Sender>>>,
+    pub stats_repository: Arc<statistics::repository::Repository>,
+    pub announce_service: Arc<services::announce::AnnounceService>,
+    pub scrape_service: Arc<services::scrape::ScrapeService>,
 }
 
 impl HttpTrackerCoreServices {
     #[must_use]
     pub fn initialize_from(tracker_core_container: &Arc<TrackerCoreContainer>) -> Arc<Self> {
-        let (http_stats_event_sender, http_stats_repository) =
-            statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics);
-        let http_stats_event_sender = Arc::new(http_stats_event_sender);
-        let http_stats_repository = Arc::new(http_stats_repository);
+        // HTTP core stats
+        let http_stats_keeper = statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics);
+        let http_stats_event_sender = http_stats_keeper.sender();
+        let http_stats_repository = http_stats_keeper.repository();
+
         let http_announce_service = Arc::new(AnnounceService::new(
             tracker_core_container.core_config.clone(),
             tracker_core_container.announce_handler.clone(),
             tracker_core_container.authentication_service.clone(),
             tracker_core_container.whitelist_authorization.clone(),
             http_stats_event_sender.clone(),
         ));
+
         let http_scrape_service = Arc::new(ScrapeService::new(
             tracker_core_container.core_config.clone(),
             tracker_core_container.scrape_handler.clone(),
@@ -81,10 +86,11 @@ impl HttpTrackerCoreServices {
         ));
 
         Arc::new(Self {
-            http_stats_event_sender,
-            http_stats_repository,
-            http_announce_service,
-            http_scrape_service,
+            stats_keeper: http_stats_keeper,
+            stats_event_sender: http_stats_event_sender,
+            stats_repository: http_stats_repository,
+            announce_service: http_announce_service,
+            scrape_service: http_scrape_service,
         })
     }
 }

packages/http-tracker-core/src/event/sender.rs

@@ -16,6 +16,7 @@ pub trait Sender: Sync + Send {
 }
 
 /// An event sender implementation using a broadcast channel.
+#[derive(Clone)]
 pub struct Broadcaster {
     pub(crate) sender: broadcast::Sender<Event>,
 }

packages/http-tracker-core/src/services/announce.rs

@@ -252,10 +252,14 @@ mod tests {
             &db_torrent_repository,
         ));
 
-        // HTTP stats
-        let (http_stats_event_sender, http_stats_repository) = statistics::setup::factory(config.core.tracker_usage_statistics);
-        let http_stats_event_sender = Arc::new(http_stats_event_sender);
-        let _http_stats_repository = Arc::new(http_stats_repository);
+        // HTTP core stats
+        let http_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics);
+        let http_stats_event_sender = http_stats_keeper.sender();
+        let _http_stats_repository = http_stats_keeper.repository();
+
+        if config.core.tracker_usage_statistics {
+            let _unused = http_stats_keeper.run_event_listener();
+        }
 
         (
             CoreTrackerServices {