Merge #1476: Refactor event listeners initialization

6d50a78 refactor: normalize container field names (Jose Celano)
74e174d refactor: [#1444] udp server event listener start in app start (Jose Celano)
2fa4e15 refactor: [#1444] udp core event listener start in app start (Jose Celano)
07c5858 refactor: [#1444] http core event listener start in app start. Step 4 (Jose Celano)
2d9af45 chore: [#1444] event listener has to be run manually on tests (Jose Celano)
19bb37d refactor: [#1444] renaem variables (Jose Celano)
6d49a13 refactor: [#1444] rename fields (Jose Celano)
5906037 refactor: [#1444] http core event listener start in app start. Step 3 (Jose Celano)
b2cf5d9 refactor: [#1444] rename variable (Jose Celano)
07d1314 refactor: [#1444] http core event listener start in app start. Step 2 (Jose Celano)
17fb909 refactor: [#1444] http core event listener start in app start. Step 1 (Jose Celano)

Pull request description:

  Move the spawning of new tasks for event listeners from app container instantiation to job execution.

  This will enable us to easily add more event listeners in the future. Additionally, it enhances the way the tracker handles spawned tasks. We should have control over all JoinHandles for all tasks.

  This promotes a cleaner task lifecycle. Some good practices:

  - There should always be an owner, responsible for the spawned tasks. No orphan tasks.
  - It should be possible to gracefully shut down the task.
  - The state of the task should be known.
  - Sometimes the owner needs to retrieve data (for example, status) from the task (when it starts or while it's active) or send the task additional commands.

  ### Subtasks

  - [x] HTTP Tracker Core event listener
    - [x] Step 1: Separate factory from spawning task in the current `setup::factory` function.
    - [x] Step 2: Split factory function into factory and running listener.
    - [x] Step 3: Add `Keeper` to `HttpTrackerCoreContainer` (so we can run the listener later).
    - [x] Step 4: Move task creation to app start.
    - [x] Step 5. Start event listener in test environment.
  - [x] UDP Tracker Core event listener (Step 1 to 4)
  - [x] UDP Tracker Server event listener (Step 1 to 4)

  NOTE: This PR will not include the refactor to allow multiple listeners. See #1385.

ACKs for top commit:
  josecelano:
    ACK 6d50a78

Tree-SHA512: 7cc637227caf3fc2e86041a25b2e2b0cdf52819dc587b972494c54f3e8337c99472bd38fc0dc76ce4b9811161cdfdf4fb4d1c65fcedea0a90f77584a88f06994

Author: josecelano

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 {