Merge #1479: Overhaul stats: Extract events package (part 1)

b0951aa refactor: [#1478] move EventBus to event mod (Jose Celano)
b32072c refactor: [#1478] rename struct fields for Keeper type (Jose Celano)
c103b60 refactor: [#1478] rename Keeper variables (Jose Celano)
8a9314b refactor: [#1478] rename Keeper to EventBus (Jose Celano)
a660be8 refactor: [#1478] inline factory fn in udp server stats (Jose Celano)
8e8b1dd refactor: [#1478] inline factory fn in udp core stats (Jose Celano)
cc7ead8 refactor: [#1478] inline factory fn in http core stats (Jose Celano)
a055ab9 refactor: [#1478] inline keeper_factory fn (Jose Celano)
5f38357 refactor: [#1478] decouple events from stats in udp server keeper (Jose Celano)
f9f13a4 refactor: [#1478] decouple events from stats in udp core keeper (Jose Celano)
f25438a refactor: [#1478] decouple events from stats in http core keeper (Jose Celano)

Pull request description:

  Overhaul stats: Extract events package.

  ### Subtasks

  - [x] Decouple events from stats in the http core keeper.
  - [x] Decouple events from stats in the udp core keeper.
  - [x] Decouple events from stats in the udp server keeper.
  - [x] Inline function `statistics::setup::keeper_factory` and `statistics::setup::factory`. No need for them anymore.
  - [x] Rename all `Keeper` types to `EventBus`.
  - [x] Move `EventBus` to `event` module.

  ### Implement in part 2 of the PR

  - Create new package `events`.
  - Create a generic `EventBus` in the `events` package and replace concrete versions in packages.

  More info in #1480.

ACKs for top commit:
  josecelano:
    ACK b0951aa

Tree-SHA512: 48337ed46d234f2d6c77df514ce55b8f2abbc0f3191ce3a818d88917eba95f3bd03291f4690501e5adc969bb260167fc968eea0a3790d43350083ece5cbe0d12

Author: josecelano

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

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use bittorrent_http_tracker_core::container::HttpTrackerCoreContainer;
+use bittorrent_http_tracker_core::statistics::event::listener::run_event_listener;
 use bittorrent_primitives::info_hash::InfoHash;
 use bittorrent_tracker_core::container::TrackerCoreContainer;
 use futures::executor::block_on;
@@ -68,7 +69,10 @@ impl Environment<Stopped> {
     #[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();
+        let event_listener_job = run_event_listener(
+            self.container.http_tracker_core_container.event_bus.receiver(),
+            &self.container.http_tracker_core_container.stats_repository,
+        );
 
         // Start the server
         let server = self

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

@@ -248,8 +248,12 @@ mod tests {
     use std::sync::Arc;
 
     use bittorrent_http_tracker_core::container::HttpTrackerCoreContainer;
+    use bittorrent_http_tracker_core::event::bus::EventBus;
+    use bittorrent_http_tracker_core::event::sender::Broadcaster;
     use bittorrent_http_tracker_core::services::announce::AnnounceService;
     use bittorrent_http_tracker_core::services::scrape::ScrapeService;
+    use bittorrent_http_tracker_core::statistics::event::listener::run_event_listener;
+    use bittorrent_http_tracker_core::statistics::repository::Repository;
     use bittorrent_tracker_core::container::TrackerCoreContainer;
     use torrust_axum_server::tsl::make_rust_tls;
     use torrust_server_lib::registar::Registar;
@@ -271,13 +275,17 @@ mod tests {
         let http_tracker_config = Arc::new(http_tracker_config.clone());
 
         // HTTP core stats
-        let http_stats_keeper =
-            bittorrent_http_tracker_core::statistics::setup::factory(configuration.core.tracker_usage_statistics);
-        let http_stats_event_sender = http_stats_keeper.sender();
-        let http_stats_repository = http_stats_keeper.repository();
+        let http_core_broadcaster = Broadcaster::default();
+        let http_stats_repository = Arc::new(Repository::new());
+        let http_stats_event_bus = Arc::new(EventBus::new(
+            configuration.core.tracker_usage_statistics,
+            http_core_broadcaster.clone(),
+        ));
+
+        let http_stats_event_sender = http_stats_event_bus.sender();
 
         if configuration.core.tracker_usage_statistics {
-            let _unused = http_stats_keeper.run_event_listener();
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
         }
 
         let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config));
@@ -300,7 +308,7 @@ mod tests {
         HttpTrackerCoreContainer {
             tracker_core_container,
             http_tracker_config,
-            stats_keeper: http_stats_keeper,
+            event_bus: http_stats_event_bus,
             stats_event_sender: http_stats_event_sender,
             stats_repository: http_stats_repository,
             announce_service,

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

@@ -107,7 +107,11 @@ mod tests {
     use std::sync::Arc;
 
     use aquatic_udp_protocol::PeerId;
+    use bittorrent_http_tracker_core::event::bus::EventBus;
+    use bittorrent_http_tracker_core::event::sender::Broadcaster;
     use bittorrent_http_tracker_core::services::announce::AnnounceService;
+    use bittorrent_http_tracker_core::statistics::event::listener::run_event_listener;
+    use bittorrent_http_tracker_core::statistics::repository::Repository;
     use bittorrent_http_tracker_protocol::v1::requests::announce::Announce;
     use bittorrent_http_tracker_protocol::v1::responses;
     use bittorrent_http_tracker_protocol::v1::services::peer_ip_resolver::ClientIpSources;
@@ -161,12 +165,17 @@ mod tests {
         ));
 
         // 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();
+        let http_core_broadcaster = Broadcaster::default();
+        let http_stats_repository = Arc::new(Repository::new());
+        let http_stats_event_bus = Arc::new(EventBus::new(
+            config.core.tracker_usage_statistics,
+            http_core_broadcaster.clone(),
+        ));
+
+        let http_stats_event_sender = http_stats_event_bus.sender();
 
         if config.core.tracker_usage_statistics {
-            let _unused = http_stats_keeper.run_event_listener();
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
         }
 
         let announce_service = Arc::new(AnnounceService::new(

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

@@ -83,6 +83,10 @@ mod tests {
     use std::str::FromStr;
     use std::sync::Arc;
 
+    use bittorrent_http_tracker_core::event::bus::EventBus;
+    use bittorrent_http_tracker_core::event::sender::Broadcaster;
+    use bittorrent_http_tracker_core::statistics::event::listener::run_event_listener;
+    use bittorrent_http_tracker_core::statistics::repository::Repository;
     use bittorrent_http_tracker_protocol::v1::requests::scrape::Scrape;
     use bittorrent_http_tracker_protocol::v1::responses;
     use bittorrent_http_tracker_protocol::v1::services::peer_ip_resolver::ClientIpSources;
@@ -132,12 +136,17 @@ mod tests {
         let scrape_handler = Arc::new(ScrapeHandler::new(&whitelist_authorization, &in_memory_torrent_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();
+        let http_core_broadcaster = Broadcaster::default();
+        let http_stats_repository = Arc::new(Repository::new());
+        let http_stats_event_bus = Arc::new(EventBus::new(
+            config.core.tracker_usage_statistics,
+            http_core_broadcaster.clone(),
+        ));
+
+        let http_stats_event_sender = http_stats_event_bus.sender();
 
         if config.core.tracker_usage_statistics {
-            let _unused = http_stats_keeper.run_event_listener();
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
         }
 
         (

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

@@ -2,8 +2,12 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::sync::Arc;
 
 use aquatic_udp_protocol::{AnnounceEvent, NumberOfBytes, PeerId};
+use bittorrent_http_tracker_core::event;
+use bittorrent_http_tracker_core::event::bus::EventBus;
+use bittorrent_http_tracker_core::event::sender::Broadcaster;
 use bittorrent_http_tracker_core::event::Event;
-use bittorrent_http_tracker_core::{event, statistics};
+use bittorrent_http_tracker_core::statistics::event::listener::run_event_listener;
+use bittorrent_http_tracker_core::statistics::repository::Repository;
 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;
@@ -56,12 +60,17 @@ pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> (
     ));
 
     // 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();
+    let http_core_broadcaster = Broadcaster::default();
+    let http_stats_repository = Arc::new(Repository::new());
+    let http_stats_event_bus = Arc::new(EventBus::new(
+        config.core.tracker_usage_statistics,
+        http_core_broadcaster.clone(),
+    ));
+
+    let http_stats_event_sender = http_stats_event_bus.sender();
 
     if config.core.tracker_usage_statistics {
-        let _unused = http_stats_keeper.run_event_listener();
+        let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
     }
 
     (

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

@@ -3,8 +3,11 @@ use std::sync::Arc;
 use bittorrent_tracker_core::container::TrackerCoreContainer;
 use torrust_tracker_configuration::{Core, HttpTracker};
 
+use crate::event::bus::EventBus;
+use crate::event::sender::Broadcaster;
 use crate::services::announce::AnnounceService;
 use crate::services::scrape::ScrapeService;
+use crate::statistics::repository::Repository;
 use crate::{event, services, statistics};
 
 pub struct HttpTrackerCoreContainer {
@@ -13,7 +16,7 @@ pub struct HttpTrackerCoreContainer {
     pub tracker_core_container: Arc<TrackerCoreContainer>,
 
     // `HttpTrackerCoreServices`
-    pub stats_keeper: Arc<statistics::keeper::Keeper>,
+    pub event_bus: Arc<event::bus::EventBus>,
     pub stats_event_sender: Arc<Option<Box<dyn event::sender::Sender>>>,
     pub stats_repository: Arc<statistics::repository::Repository>,
     pub announce_service: Arc<AnnounceService>,
@@ -45,7 +48,7 @@ impl HttpTrackerCoreContainer {
         Arc::new(Self {
             tracker_core_container: tracker_core_container.clone(),
             http_tracker_config: http_tracker_config.clone(),
-            stats_keeper: http_tracker_core_services.stats_keeper.clone(),
+            event_bus: http_tracker_core_services.event_bus.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(),
@@ -55,7 +58,7 @@ impl HttpTrackerCoreContainer {
 }
 
 pub struct HttpTrackerCoreServices {
-    pub stats_keeper: Arc<statistics::keeper::Keeper>,
+    pub event_bus: Arc<event::bus::EventBus>,
     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>,
@@ -66,9 +69,14 @@ impl HttpTrackerCoreServices {
     #[must_use]
     pub fn initialize_from(tracker_core_container: &Arc<TrackerCoreContainer>) -> Arc<Self> {
         // 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_core_broadcaster = Broadcaster::default();
+        let http_stats_repository = Arc::new(Repository::new());
+        let http_stats_event_bus = Arc::new(EventBus::new(
+            tracker_core_container.core_config.tracker_usage_statistics,
+            http_core_broadcaster.clone(),
+        ));
+
+        let http_stats_event_sender = http_stats_event_bus.sender();
 
         let http_announce_service = Arc::new(AnnounceService::new(
             tracker_core_container.core_config.clone(),
@@ -86,7 +94,7 @@ impl HttpTrackerCoreServices {
         ));
 
         Arc::new(Self {
-            stats_keeper: http_stats_keeper,
+            event_bus: http_stats_event_bus,
             stats_event_sender: http_stats_event_sender,
             stats_repository: http_stats_repository,
             announce_service: http_announce_service,

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

@@ -0,0 +1,44 @@
+use std::sync::Arc;
+
+use tokio::sync::broadcast::Receiver;
+
+use crate::event::sender::{self, Broadcaster};
+use crate::event::Event;
+
+pub struct EventBus {
+    pub enable_sender: bool,
+    pub broadcaster: Broadcaster,
+}
+
+impl Default for EventBus {
+    fn default() -> Self {
+        let enable_sender = true;
+        let broadcaster = Broadcaster::default();
+
+        Self::new(enable_sender, broadcaster)
+    }
+}
+
+impl EventBus {
+    #[must_use]
+    pub fn new(enable_sender: bool, broadcaster: Broadcaster) -> Self {
+        Self {
+            enable_sender,
+            broadcaster,
+        }
+    }
+
+    #[must_use]
+    pub fn sender(&self) -> Arc<Option<Box<dyn sender::Sender>>> {
+        if self.enable_sender {
+            Arc::new(Some(Box::new(self.broadcaster.clone())))
+        } else {
+            Arc::new(None)
+        }
+    }
+
+    #[must_use]
+    pub fn receiver(&self) -> Receiver<Event> {
+        self.broadcaster.subscribe()
+    }
+}

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

@@ -1,3 +1,6 @@
+pub mod bus;
+pub mod sender;
+
 use std::net::{IpAddr, SocketAddr};
 
 use bittorrent_http_tracker_protocol::v1::services::peer_ip_resolver::RemoteClientAddr;
@@ -7,8 +10,6 @@ use torrust_tracker_metrics::label_name;
 use torrust_tracker_primitives::peer::PeerAnnouncement;
 use torrust_tracker_primitives::service_binding::ServiceBinding;
 
-pub mod sender;
-
 /// A HTTP core event.
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum Event {

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

@@ -253,12 +253,17 @@ mod tests {
         ));
 
         // 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();
+        let http_core_broadcaster = Broadcaster::default();
+        let http_stats_repository = Arc::new(Repository::new());
+        let http_stats_event_bus = Arc::new(EventBus::new(
+            config.core.tracker_usage_statistics,
+            http_core_broadcaster.clone(),
+        ));
+
+        let http_stats_event_sender = http_stats_event_bus.sender();
 
         if config.core.tracker_usage_statistics {
-            let _unused = http_stats_keeper.run_event_listener();
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
         }
 
         (
@@ -297,9 +302,13 @@ mod tests {
     use mockall::mock;
     use tokio::sync::broadcast::error::SendError;
 
+    use crate::event;
+    use crate::event::bus::EventBus;
+    use crate::event::sender::Broadcaster;
     use crate::event::Event;
+    use crate::statistics::event::listener::run_event_listener;
+    use crate::statistics::repository::Repository;
     use crate::tests::sample_info_hash;
-    use crate::{event, statistics};
 
     mock! {
         HttpStatsEventSender {}

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

@@ -259,23 +259,26 @@ mod tests {
         use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
         use torrust_tracker_test_helpers::configuration;
 
+        use crate::event;
+        use crate::event::bus::EventBus;
+        use crate::event::sender::Broadcaster;
         use crate::event::{ConnectionContext, Event};
         use crate::services::scrape::tests::{
             initialize_services_with_configuration, sample_info_hashes, sample_peer, MockHttpStatsEventSender,
         };
         use crate::services::scrape::ScrapeService;
         use crate::tests::sample_info_hash;
-        use crate::{event, statistics};
 
         #[tokio::test]
         async fn it_should_return_the_scrape_data_for_a_torrent() {
             let configuration = configuration::ephemeral_public();
             let core_config = Arc::new(configuration.core.clone());
 
             // HTTP core stats
-            let http_stats_keeper = statistics::setup::factory(false);
-            let http_stats_event_sender = http_stats_keeper.sender();
-            let _http_stats_repository = http_stats_keeper.repository();
+            let http_core_broadcaster = Broadcaster::default();
+            let http_stats_event_bus = Arc::new(EventBus::new(false, http_core_broadcaster.clone()));
+
+            let http_stats_event_sender = http_stats_event_bus.sender();
 
             let container = initialize_services_with_configuration(&configuration);
 
@@ -449,13 +452,15 @@ mod tests {
         use torrust_tracker_primitives::service_binding::{Protocol, ServiceBinding};
         use torrust_tracker_test_helpers::configuration;
 
+        use crate::event;
+        use crate::event::bus::EventBus;
+        use crate::event::sender::Broadcaster;
         use crate::event::{ConnectionContext, Event};
         use crate::services::scrape::tests::{
             initialize_services_with_configuration, sample_info_hashes, sample_peer, MockHttpStatsEventSender,
         };
         use crate::services::scrape::ScrapeService;
         use crate::tests::sample_info_hash;
-        use crate::{event, statistics};
 
         #[tokio::test]
         async fn it_should_return_the_zeroed_scrape_data_when_the_tracker_is_running_in_private_mode_and_the_peer_is_not_authenticated(
@@ -465,9 +470,10 @@ mod tests {
             let container = initialize_services_with_configuration(&config);
 
             // HTTP core stats
-            let http_stats_keeper = statistics::setup::factory(false);
-            let http_stats_event_sender = http_stats_keeper.sender();
-            let _http_stats_repository = http_stats_keeper.repository();
+            let http_core_broadcaster = Broadcaster::default();
+            let http_stats_event_bus = Arc::new(EventBus::new(false, http_core_broadcaster.clone()));
+
+            let http_stats_event_sender = http_stats_event_bus.sender();
 
             let info_hash = sample_info_hash();
             let info_hashes = vec![info_hash];