refactor: [torrust#1444] http core event listener start in app start. Step 1

This is the first step in a bigger refactor to move the start of event listeners from app container instantiation to app start (jobs creation).

Author: josecelano

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

@@ -274,7 +274,6 @@ mod tests {
         let (http_stats_event_sender, http_stats_repository) =
             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 tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
 

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

@@ -65,7 +65,6 @@ impl HttpTrackerCoreServices {
         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);
         let http_announce_service = Arc::new(AnnounceService::new(
             tracker_core_container.core_config.clone(),
             tracker_core_container.announce_handler.clone(),

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/statistics/event/handler.rs

@@ -1,4 +1,5 @@
 use std::net::IpAddr;
+use std::sync::Arc;
 
 use torrust_tracker_metrics::label::{LabelSet, LabelValue};
 use torrust_tracker_metrics::{label_name, metric_name};
@@ -12,7 +13,7 @@ use crate::statistics::HTTP_TRACKER_CORE_REQUESTS_RECEIVED_TOTAL;
 ///
 /// This function panics if the client IP address is not the same as the IP
 /// version of the event.
-pub async fn handle_event(event: Event, stats_repository: &Repository, now: DurationSinceUnixEpoch) {
+pub async fn handle_event(event: Event, stats_repository: &Arc<Repository>, now: DurationSinceUnixEpoch) {
     match event {
         Event::TcpAnnounce { connection, .. } => {
             // Global fixed metrics
@@ -72,6 +73,7 @@ pub async fn handle_event(event: Event, stats_repository: &Repository, now: Dura
 #[cfg(test)]
 mod tests {
     use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+    use std::sync::Arc;
 
     use bittorrent_http_tracker_protocol::v1::services::peer_ip_resolver::{RemoteClientAddr, ResolvedIp};
     use torrust_tracker_clock::clock::Time;
@@ -85,7 +87,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_increase_the_tcp4_announces_counter_when_it_receives_a_tcp4_announce_event() {
-        let stats_repository = Repository::new();
+        let stats_repository = Arc::new(Repository::new());
         let peer = sample_peer_using_ipv4();
         let remote_client_ip = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2));
 
@@ -110,7 +112,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_increase_the_tcp4_scrapes_counter_when_it_receives_a_tcp4_scrape_event() {
-        let stats_repository = Repository::new();
+        let stats_repository = Arc::new(Repository::new());
 
         handle_event(
             Event::TcpScrape {
@@ -134,7 +136,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_increase_the_tcp6_announces_counter_when_it_receives_a_tcp6_announce_event() {
-        let stats_repository = Repository::new();
+        let stats_repository = Arc::new(Repository::new());
         let peer = sample_peer_using_ipv6();
         let remote_client_ip = IpAddr::V6(Ipv6Addr::new(0x6969, 0x6969, 0x6969, 0x6969, 0x6969, 0x6969, 0x6969, 0x6969));
 
@@ -159,7 +161,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_increase_the_tcp6_scrapes_counter_when_it_receives_a_tcp6_scrape_event() {
-        let stats_repository = Repository::new();
+        let stats_repository = Arc::new(Repository::new());
 
         handle_event(
             Event::TcpScrape {

packages/http-tracker-core/src/statistics/event/listener.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use tokio::sync::broadcast;
 use torrust_tracker_clock::clock::Time;
 
@@ -6,7 +8,7 @@ use crate::event::Event;
 use crate::statistics::repository::Repository;
 use crate::{CurrentClock, HTTP_TRACKER_LOG_TARGET};
 
-pub async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_repository: Repository) {
+pub async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_repository: Arc<Repository>) {
     loop {
         match receiver.recv().await {
             Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,

packages/http-tracker-core/src/statistics/keeper.rs

@@ -1,42 +1,69 @@
-use tokio::sync::broadcast::Receiver;
+use std::sync::Arc;
+
+use tokio::task::JoinHandle;
 
 use super::event::listener::dispatch_events;
 use super::repository::Repository;
-use crate::event::Event;
+use crate::event::sender::{self, Broadcaster};
 use crate::HTTP_TRACKER_LOG_TARGET;
 
 /// The service responsible for keeping tracker metrics (listening to statistics events and handle them).
 ///
 /// It actively listen to new statistics events. When it receives a new event
 /// it accordingly increases the counters.
 pub struct Keeper {
-    pub repository: Repository,
+    pub enable_sender: bool,
+    pub broadcaster: Broadcaster,
+    pub repository: Arc<Repository>,
 }
 
 impl Default for Keeper {
     fn default() -> Self {
-        Self::new()
+        let enable_sender = true;
+        let broadcaster = Broadcaster::default();
+        let repository = Arc::new(Repository::new());
+
+        Self::new(enable_sender, broadcaster, repository)
     }
 }
 
 impl Keeper {
+    /// Creates a new instance of [`Keeper`].
     #[must_use]
-    pub fn new() -> Self {
+    pub fn new(enable_sender: bool, broadcaster: Broadcaster, repository: Arc<Repository>) -> Self {
         Self {
-            repository: Repository::new(),
+            enable_sender,
+            broadcaster,
+            repository,
+        }
+    }
+
+    #[must_use]
+    pub fn sender(&self) -> Option<Box<dyn sender::Sender>> {
+        if self.enable_sender {
+            Some(Box::new(self.broadcaster.clone()))
+        } else {
+            None
         }
     }
 
-    pub fn run_event_listener(&mut self, receiver: Receiver<Event>) {
+    #[must_use]
+    pub fn repository(&self) -> Arc<Repository> {
+        self.repository.clone()
+    }
+
+    #[must_use]
+    pub fn run_event_listener(&self) -> JoinHandle<()> {
         let stats_repository = self.repository.clone();
+        let receiver = self.broadcaster.subscribe();
 
         tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Starting HTTP tracker core event listener");
 
         tokio::spawn(async move {
             dispatch_events(receiver, stats_repository).await;
 
             tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "HTTP tracker core event listener finished");
-        });
+        })
     }
 }
 
@@ -48,7 +75,7 @@ mod tests {
 
     #[tokio::test]
     async fn should_contain_the_tracker_statistics() {
-        let stats_tracker = Keeper::new();
+        let stats_tracker = Keeper::default();
 
         let stats = stats_tracker.repository.get_stats().await;
 

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

@@ -89,9 +89,8 @@ mod tests {
         let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
 
         let (_http_stats_event_sender, http_stats_repository) = statistics::setup::factory(config.core.tracker_usage_statistics);
-        let http_stats_repository = Arc::new(http_stats_repository);
 
-        let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), http_stats_repository.clone()).await;
+        let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), http_stats_repository).await;
 
         assert_eq!(
             tracker_metrics,

packages/http-tracker-core/src/statistics/setup.rs

@@ -1,8 +1,12 @@
 //! Setup for the tracker statistics.
 //!
 //! The [`factory`] function builds the structs needed for handling the tracker metrics.
+use std::sync::Arc;
+
+use super::keeper::Keeper;
+use super::repository::Repository;
+use crate::event;
 use crate::event::sender::Broadcaster;
-use crate::{event, statistics};
 
 /// It builds the structs needed for handling the tracker metrics.
 ///
@@ -17,20 +21,23 @@ use crate::{event, statistics};
 /// not run the event listeners, consequently the statistics events are sent are
 /// received but not dispatched to the handler.
 #[must_use]
-pub fn factory(tracker_usage_statistics: bool) -> (Option<Box<dyn event::sender::Sender>>, statistics::repository::Repository) {
-    let mut keeper = statistics::keeper::Keeper::new();
-
-    let opt_event_sender: Option<Box<dyn event::sender::Sender>> = if tracker_usage_statistics {
-        let broadcaster = Broadcaster::default();
+pub fn factory(tracker_usage_statistics: bool) -> (Option<Box<dyn event::sender::Sender>>, Arc<Repository>) {
+    let keeper = keeper_factory(tracker_usage_statistics);
 
-        keeper.run_event_listener(broadcaster.subscribe());
+    if tracker_usage_statistics {
+        // todo: this should be started like the other jobs during `app::start`
+        // and keep the join handle in a list of jobs.
+        let _unused = keeper.run_event_listener();
+    }
 
-        Some(Box::new(broadcaster))
-    } else {
-        None
-    };
+    (keeper.sender(), keeper.repository())
+}
 
-    (opt_event_sender, keeper.repository)
+#[must_use]
+pub fn keeper_factory(tracker_usage_statistics: bool) -> Arc<Keeper> {
+    let broadcaster = Broadcaster::default();
+    let repository = Arc::new(Repository::new());
+    Arc::new(Keeper::new(tracker_usage_statistics, broadcaster.clone(), repository.clone()))
 }
 
 #[cfg(test)]

packages/rest-tracker-api-core/src/statistics/services.rs

@@ -149,7 +149,6 @@ mod tests {
         // HTTP core stats
         let (_http_stats_event_sender, http_stats_repository) =
             bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics);
-        let http_stats_repository = Arc::new(http_stats_repository);
 
         // UDP core stats
         let (_udp_stats_event_sender, _udp_stats_repository) =