feat: [#1358] add the and run the event listener when the tracker starts

This creates independent services that are not used yet in
the tracker-core, meaning the `Swarms` object created in the `TorrentRepositoryContainer` will not store any torrent yet. The tracker core is still creating its own fresh instance.

Author: josecelano

Cargo.lock

@@ -55,6 +55,7 @@ torrust-rest-tracker-api-core = { version = "3.0.0-develop", path = "packages/re
 torrust-server-lib = { version = "3.0.0-develop", path = "packages/server-lib" }
 torrust-tracker-clock = { version = "3.0.0-develop", path = "packages/clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "packages/configuration" }
+torrust-tracker-torrent-repository = { version = "3.0.0-develop", path = "packages/torrent-repository" }
 torrust-udp-tracker-server = { version = "3.0.0-develop", path = "packages/udp-tracker-server" }
 tracing = "0"
 tracing-subscriber = { version = "0", features = ["json"] }

Cargo.toml

@@ -0,0 +1,37 @@
+use std::sync::Arc;
+
+use crate::event::bus::EventBus;
+use crate::event::sender::Broadcaster;
+use crate::event::{self};
+use crate::statistics::repository::Repository;
+use crate::{statistics, Swarms};
+
+pub struct TorrentRepositoryContainer {
+    pub swarms: Arc<Swarms>,
+    pub event_bus: Arc<event::bus::EventBus>,
+    pub stats_event_sender: event::sender::Sender,
+    pub stats_repository: Arc<statistics::repository::Repository>,
+}
+
+impl TorrentRepositoryContainer {
+    #[must_use]
+    pub fn initialize() -> Self {
+        let swarms = Arc::new(Swarms::default());
+
+        // Torrent repository stats
+        let broadcaster = Broadcaster::default();
+        let stats_repository = Arc::new(Repository::new());
+
+        // todo: add a config option to enable/disable stats for this package
+        let event_bus = Arc::new(EventBus::new(true, broadcaster.clone()));
+
+        let stats_event_sender = event_bus.sender();
+
+        Self {
+            swarms,
+            event_bus,
+            stats_event_sender,
+            stats_repository,
+        }
+    }
+}

packages/torrent-repository/src/container.rs

@@ -1,3 +1,4 @@
+pub mod container;
 pub mod event;
 pub mod statistics;
 pub mod swarm;

packages/torrent-repository/src/lib.rs

@@ -72,9 +72,11 @@ async fn load_data_from_database(config: &Configuration, app_container: &Arc<App
 async fn start_jobs(config: &Configuration, app_container: &Arc<AppContainer>) -> JobManager {
     let mut job_manager = JobManager::new();
 
+    start_torrent_repository_event_listener(config, app_container, &mut job_manager);
     start_http_core_event_listener(config, app_container, &mut job_manager);
     start_udp_core_event_listener(config, app_container, &mut job_manager);
     start_udp_server_event_listener(config, app_container, &mut job_manager);
+
     start_the_udp_instances(config, app_container, &mut job_manager).await;
     start_the_http_instances(config, app_container, &mut job_manager).await;
     start_the_http_api(config, app_container, &mut job_manager).await;
@@ -126,6 +128,18 @@ fn load_torrents_from_database(config: &Configuration, app_container: &Arc<AppCo
     }
 }
 
+fn start_torrent_repository_event_listener(
+    config: &Configuration,
+    app_container: &Arc<AppContainer>,
+    job_manager: &mut JobManager,
+) {
+    let opt_handle = jobs::torrent_repository::start_event_listener(config, app_container);
+
+    if let Some(handle) = opt_handle {
+        job_manager.push("torrent_repository_event_listener", handle);
+    }
+}
+
 fn start_http_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
     let opt_handle = jobs::http_tracker_core::start_event_listener(config, app_container);
 

src/app.rs

@@ -11,6 +11,7 @@ pub mod http_tracker;
 pub mod http_tracker_core;
 pub mod manager;
 pub mod torrent_cleanup;
+pub mod torrent_repository;
 pub mod tracker_apis;
 pub mod udp_tracker;
 pub mod udp_tracker_core;

src/bootstrap/jobs/mod.rs

@@ -0,0 +1,20 @@
+use std::sync::Arc;
+
+use tokio::task::JoinHandle;
+use torrust_tracker_configuration::Configuration;
+
+use crate::container::AppContainer;
+
+pub fn start_event_listener(config: &Configuration, app_container: &Arc<AppContainer>) -> Option<JoinHandle<()>> {
+    if config.core.tracker_usage_statistics {
+        let job = torrust_tracker_torrent_repository::statistics::event::listener::run_event_listener(
+            app_container.torrent_repository_container.event_bus.receiver(),
+            &app_container.torrent_repository_container.stats_repository,
+        );
+
+        Some(job)
+    } else {
+        tracing::info!("HTTP tracker core event listener job is disabled.");
+        None
+    }
+}

src/bootstrap/jobs/torrent_repository.rs

@@ -9,6 +9,7 @@ use bittorrent_udp_tracker_core::{self};
 use torrust_rest_tracker_api_core::container::TrackerHttpApiCoreContainer;
 use torrust_server_lib::registar::Registar;
 use torrust_tracker_configuration::{Configuration, HttpApi};
+use torrust_tracker_torrent_repository::container::TorrentRepositoryContainer;
 use torrust_udp_tracker_server::container::UdpTrackerServerContainer;
 use tracing::instrument;
 
@@ -28,6 +29,9 @@ pub struct AppContainer {
     // Registar
     pub registar: Arc<Registar>,
 
+    // Torrent Repository
+    pub torrent_repository_container: Arc<TorrentRepositoryContainer>,
+
     // Core
     pub tracker_core_container: Arc<TrackerCoreContainer>,
 
@@ -54,6 +58,10 @@ impl AppContainer {
 
         let registar = Arc::new(Registar::default());
 
+        // Torrent Repository
+
+        let torrent_repository_container = Arc::new(TorrentRepositoryContainer::initialize());
+
         // Core
 
         let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
@@ -84,6 +92,9 @@ impl AppContainer {
             // Registar
             registar,
 
+            // Torrent Repository
+            torrent_repository_container,
+
             // Core
             tracker_core_container,