From 17fb90943bdb95ab7ed1da72589f34e9a0a9d356 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Wed, 23 Apr 2025 19:56:49 +0100 Subject: [PATCH 01/11] refactor: [#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). --- .../axum-http-tracker-server/src/server.rs | 1 - packages/http-tracker-core/src/container.rs | 1 - .../http-tracker-core/src/event/sender.rs | 1 + .../src/statistics/event/handler.rs | 12 ++--- .../src/statistics/event/listener.rs | 4 +- .../src/statistics/keeper.rs | 45 +++++++++++++++---- .../src/statistics/services.rs | 3 +- .../http-tracker-core/src/statistics/setup.rs | 31 ++++++++----- .../src/statistics/services.rs | 1 - 9 files changed, 67 insertions(+), 32 deletions(-) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index eea00c142..896922751 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/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)); diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index 7fc2f48a6..913236483 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/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(), diff --git a/packages/http-tracker-core/src/event/sender.rs b/packages/http-tracker-core/src/event/sender.rs index 511a381d0..b720926bb 100644 --- a/packages/http-tracker-core/src/event/sender.rs +++ b/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, } diff --git a/packages/http-tracker-core/src/statistics/event/handler.rs b/packages/http-tracker-core/src/statistics/event/handler.rs index 7e8338edf..8d2ad1aa2 100644 --- a/packages/http-tracker-core/src/statistics/event/handler.rs +++ b/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, 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 { diff --git a/packages/http-tracker-core/src/statistics/event/listener.rs b/packages/http-tracker-core/src/statistics/event/listener.rs index 5e87b47df..00fce6b77 100644 --- a/packages/http-tracker-core/src/statistics/event/listener.rs +++ b/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, stats_repository: Repository) { +pub async fn dispatch_events(mut receiver: broadcast::Receiver, stats_repository: Arc) { loop { match receiver.recv().await { Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await, diff --git a/packages/http-tracker-core/src/statistics/keeper.rs b/packages/http-tracker-core/src/statistics/keeper.rs index 1b69f032d..fad9382d7 100644 --- a/packages/http-tracker-core/src/statistics/keeper.rs +++ b/packages/http-tracker-core/src/statistics/keeper.rs @@ -1,8 +1,10 @@ -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). @@ -10,25 +12,50 @@ use crate::HTTP_TRACKER_LOG_TARGET; /// 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, } 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) -> Self { Self { - repository: Repository::new(), + enable_sender, + broadcaster, + repository, + } + } + + #[must_use] + pub fn sender(&self) -> Option> { + if self.enable_sender { + Some(Box::new(self.broadcaster.clone())) + } else { + None } } - pub fn run_event_listener(&mut self, receiver: Receiver) { + #[must_use] + pub fn repository(&self) -> Arc { + 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"); @@ -36,7 +63,7 @@ impl Keeper { 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; diff --git a/packages/http-tracker-core/src/statistics/services.rs b/packages/http-tracker-core/src/statistics/services.rs index 418b0d082..2895d1b6d 100644 --- a/packages/http-tracker-core/src/statistics/services.rs +++ b/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, diff --git a/packages/http-tracker-core/src/statistics/setup.rs b/packages/http-tracker-core/src/statistics/setup.rs index e2974e4c0..e2b252c23 100644 --- a/packages/http-tracker-core/src/statistics/setup.rs +++ b/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>, statistics::repository::Repository) { - let mut keeper = statistics::keeper::Keeper::new(); - - let opt_event_sender: Option> = if tracker_usage_statistics { - let broadcaster = Broadcaster::default(); +pub fn factory(tracker_usage_statistics: bool) -> (Option>, Arc) { + 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 { + let broadcaster = Broadcaster::default(); + let repository = Arc::new(Repository::new()); + Arc::new(Keeper::new(tracker_usage_statistics, broadcaster.clone(), repository.clone())) } #[cfg(test)] diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index 9277df92b..744c8fd7c 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/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) = From 07d13146f5cd13055a37323e12fbf9f970f41379 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 15:54:47 +0100 Subject: [PATCH 02/11] refactor: [#1444] http core event listener start in app start. Step 2 --- .../axum-http-tracker-server/src/server.rs | 14 +++++--- .../src/v1/handlers/announce.rs | 15 ++++++--- .../src/v1/handlers/scrape.rs | 14 +++++--- .../http-tracker-core/benches/helpers/util.rs | 25 ++++++++------ packages/http-tracker-core/src/container.rs | 15 +++++++-- .../src/services/announce.rs | 14 +++++--- .../http-tracker-core/src/services/scrape.rs | 12 ++++--- .../src/statistics/keeper.rs | 6 ++-- .../src/statistics/services.rs | 11 ++++++- .../http-tracker-core/src/statistics/setup.rs | 33 ++++++++++--------- .../src/statistics/services.rs | 11 +++++-- 11 files changed, 115 insertions(+), 55 deletions(-) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index 896922751..40620674f 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/packages/axum-http-tracker-server/src/server.rs @@ -270,10 +270,16 @@ mod tests { let http_tracker_config = Arc::new(http_tracker_config.clone()); - // HTTP stats - 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); + // HTTP core stats + let keeper = bittorrent_http_tracker_core::statistics::setup::factory(configuration.core.tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let http_stats_repository = keeper.repository(); + + if configuration.core.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(); + } let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config)); diff --git a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs index 296cefcd5..3729f5bdc 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs @@ -160,11 +160,16 @@ 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 keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); + + if config.core.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(); + } let announce_service = Arc::new(AnnounceService::new( core_config.clone(), diff --git a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs index e5d94a072..9e5fafd46 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs @@ -131,10 +131,16 @@ 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 keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); + + if config.core.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(); + } ( CoreTrackerServices { diff --git a/packages/http-tracker-core/benches/helpers/util.rs b/packages/http-tracker-core/benches/helpers/util.rs index dff516063..957f70444 100644 --- a/packages/http-tracker-core/benches/helpers/util.rs +++ b/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,16 @@ 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 keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); + + if config.core.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(); + } ( CoreTrackerServices { @@ -105,12 +116,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 { diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index 913236483..302a4fbbe 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -62,9 +62,17 @@ pub struct HttpTrackerCoreServices { impl HttpTrackerCoreServices { #[must_use] pub fn initialize_from(tracker_core_container: &Arc) -> Arc { - 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); + // HTTP core stats + let keeper = statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let http_stats_repository = keeper.repository(); + + if tracker_core_container.core_config.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(); + } + let http_announce_service = Arc::new(AnnounceService::new( tracker_core_container.core_config.clone(), tracker_core_container.announce_handler.clone(), @@ -72,6 +80,7 @@ impl HttpTrackerCoreServices { 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(), diff --git a/packages/http-tracker-core/src/services/announce.rs b/packages/http-tracker-core/src/services/announce.rs index fa0c0c38c..a3014873e 100644 --- a/packages/http-tracker-core/src/services/announce.rs +++ b/packages/http-tracker-core/src/services/announce.rs @@ -252,10 +252,16 @@ 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 keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); + + if config.core.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(); + } ( CoreTrackerServices { diff --git a/packages/http-tracker-core/src/services/scrape.rs b/packages/http-tracker-core/src/services/scrape.rs index 5e8f54cc1..21308a6aa 100644 --- a/packages/http-tracker-core/src/services/scrape.rs +++ b/packages/http-tracker-core/src/services/scrape.rs @@ -272,8 +272,10 @@ mod tests { let configuration = configuration::ephemeral_public(); let core_config = Arc::new(configuration.core.clone()); - let (http_stats_event_sender, _http_stats_repository) = statistics::setup::factory(false); - let http_stats_event_sender = Arc::new(http_stats_event_sender); + // HTTP core stats + let keeper = statistics::setup::factory(false); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); let container = initialize_services_with_configuration(&configuration); @@ -462,8 +464,10 @@ mod tests { let container = initialize_services_with_configuration(&config); - let (http_stats_event_sender, _http_stats_repository) = statistics::setup::factory(false); - let http_stats_event_sender = Arc::new(http_stats_event_sender); + // HTTP core stats + let keeper = statistics::setup::factory(false); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); let info_hash = sample_info_hash(); let info_hashes = vec![info_hash]; diff --git a/packages/http-tracker-core/src/statistics/keeper.rs b/packages/http-tracker-core/src/statistics/keeper.rs index fad9382d7..4c0f7c916 100644 --- a/packages/http-tracker-core/src/statistics/keeper.rs +++ b/packages/http-tracker-core/src/statistics/keeper.rs @@ -39,11 +39,11 @@ impl Keeper { } #[must_use] - pub fn sender(&self) -> Option> { + pub fn sender(&self) -> Arc>> { if self.enable_sender { - Some(Box::new(self.broadcaster.clone())) + Arc::new(Some(Box::new(self.broadcaster.clone()))) } else { - None + Arc::new(None) } } diff --git a/packages/http-tracker-core/src/statistics/services.rs b/packages/http-tracker-core/src/statistics/services.rs index 2895d1b6d..172e7b9ab 100644 --- a/packages/http-tracker-core/src/statistics/services.rs +++ b/packages/http-tracker-core/src/statistics/services.rs @@ -88,7 +88,16 @@ 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); + // HTTP core stats + let keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let _http_stats_event_sender = keeper.sender(); + let http_stats_repository = keeper.repository(); + + if config.core.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(); + } let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), http_stats_repository).await; diff --git a/packages/http-tracker-core/src/statistics/setup.rs b/packages/http-tracker-core/src/statistics/setup.rs index e2b252c23..565e86fd4 100644 --- a/packages/http-tracker-core/src/statistics/setup.rs +++ b/packages/http-tracker-core/src/statistics/setup.rs @@ -5,7 +5,6 @@ use std::sync::Arc; use super::keeper::Keeper; use super::repository::Repository; -use crate::event; use crate::event::sender::Broadcaster; /// It builds the structs needed for handling the tracker metrics. @@ -21,16 +20,8 @@ use crate::event::sender::Broadcaster; /// 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>, Arc) { - let keeper = keeper_factory(tracker_usage_statistics); - - 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(); - } - - (keeper.sender(), keeper.repository()) +pub fn factory(tracker_usage_statistics: bool) -> Arc { + keeper_factory(tracker_usage_statistics) } #[must_use] @@ -48,17 +39,29 @@ mod test { async fn should_not_send_any_event_when_statistics_are_disabled() { let tracker_usage_statistics = false; - let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics); + // HTTP core stats + let keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); + + 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(); + } - assert!(stats_event_sender.is_none()); + assert!(http_stats_event_sender.is_none()); } #[tokio::test] async fn should_send_events_when_statistics_are_enabled() { let tracker_usage_statistics = true; - let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics); + // HTTP core stats + let keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = keeper.sender(); + let _http_stats_repository = keeper.repository(); - assert!(stats_event_sender.is_some()); + assert!(http_stats_event_sender.is_some()); } } diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index 744c8fd7c..ac8948e42 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/packages/rest-tracker-api-core/src/statistics/services.rs @@ -147,8 +147,15 @@ mod tests { let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP))); // HTTP core stats - let (_http_stats_event_sender, http_stats_repository) = - bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let _http_stats_event_sender = keeper.sender(); + let http_stats_repository = keeper.repository(); + + if config.core.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(); + } // UDP core stats let (_udp_stats_event_sender, _udp_stats_repository) = From b2cf5d9e921326e6d5f163023fa563ee49a583de Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 15:59:26 +0100 Subject: [PATCH 03/11] refactor: [#1444] rename variable --- packages/axum-http-tracker-server/src/server.rs | 9 +++++---- .../src/v1/handlers/announce.rs | 9 +++++---- .../src/v1/handlers/scrape.rs | 9 +++++---- packages/http-tracker-core/benches/helpers/util.rs | 8 ++++---- packages/http-tracker-core/src/container.rs | 8 ++++---- .../http-tracker-core/src/services/announce.rs | 8 ++++---- packages/http-tracker-core/src/services/scrape.rs | 12 ++++++------ .../http-tracker-core/src/statistics/services.rs | 8 ++++---- packages/http-tracker-core/src/statistics/setup.rs | 14 +++++++------- .../src/statistics/services.rs | 9 +++++---- 10 files changed, 49 insertions(+), 45 deletions(-) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index 40620674f..52085f822 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/packages/axum-http-tracker-server/src/server.rs @@ -271,14 +271,15 @@ mod tests { let http_tracker_config = Arc::new(http_tracker_config.clone()); // HTTP core stats - let keeper = bittorrent_http_tracker_core::statistics::setup::factory(configuration.core.tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let http_stats_repository = keeper.repository(); + let http_core_stats_keeper = + bittorrent_http_tracker_core::statistics::setup::factory(configuration.core.tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let http_stats_repository = http_core_stats_keeper.repository(); if configuration.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config)); diff --git a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs index 3729f5bdc..5c08e97eb 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs @@ -161,14 +161,15 @@ mod tests { )); // HTTP core stats - let keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = + bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); if config.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } let announce_service = Arc::new(AnnounceService::new( diff --git a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs index 9e5fafd46..76390ea0d 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs @@ -132,14 +132,15 @@ mod tests { let scrape_handler = Arc::new(ScrapeHandler::new(&whitelist_authorization, &in_memory_torrent_repository)); // HTTP core stats - let keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = + bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); if config.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } ( diff --git a/packages/http-tracker-core/benches/helpers/util.rs b/packages/http-tracker-core/benches/helpers/util.rs index 957f70444..3ef1ccf46 100644 --- a/packages/http-tracker-core/benches/helpers/util.rs +++ b/packages/http-tracker-core/benches/helpers/util.rs @@ -56,14 +56,14 @@ pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> ( )); // HTTP core stats - let keeper = statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); if config.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } ( diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index 302a4fbbe..0b8bd9337 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -63,14 +63,14 @@ impl HttpTrackerCoreServices { #[must_use] pub fn initialize_from(tracker_core_container: &Arc) -> Arc { // HTTP core stats - let keeper = statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let http_stats_repository = keeper.repository(); + let http_core_stats_keeper = statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let http_stats_repository = http_core_stats_keeper.repository(); if tracker_core_container.core_config.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } let http_announce_service = Arc::new(AnnounceService::new( diff --git a/packages/http-tracker-core/src/services/announce.rs b/packages/http-tracker-core/src/services/announce.rs index a3014873e..5e50ebd8f 100644 --- a/packages/http-tracker-core/src/services/announce.rs +++ b/packages/http-tracker-core/src/services/announce.rs @@ -253,14 +253,14 @@ mod tests { )); // HTTP core stats - let keeper = statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); if config.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } ( diff --git a/packages/http-tracker-core/src/services/scrape.rs b/packages/http-tracker-core/src/services/scrape.rs index 21308a6aa..7cd1a5991 100644 --- a/packages/http-tracker-core/src/services/scrape.rs +++ b/packages/http-tracker-core/src/services/scrape.rs @@ -273,9 +273,9 @@ mod tests { let core_config = Arc::new(configuration.core.clone()); // HTTP core stats - let keeper = statistics::setup::factory(false); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = statistics::setup::factory(false); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); let container = initialize_services_with_configuration(&configuration); @@ -465,9 +465,9 @@ mod tests { let container = initialize_services_with_configuration(&config); // HTTP core stats - let keeper = statistics::setup::factory(false); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = statistics::setup::factory(false); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); let info_hash = sample_info_hash(); let info_hashes = vec![info_hash]; diff --git a/packages/http-tracker-core/src/statistics/services.rs b/packages/http-tracker-core/src/statistics/services.rs index 172e7b9ab..94ade2e45 100644 --- a/packages/http-tracker-core/src/statistics/services.rs +++ b/packages/http-tracker-core/src/statistics/services.rs @@ -89,14 +89,14 @@ mod tests { let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default()); // HTTP core stats - let keeper = statistics::setup::factory(config.core.tracker_usage_statistics); - let _http_stats_event_sender = keeper.sender(); - let http_stats_repository = keeper.repository(); + let http_core_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let _http_stats_event_sender = http_core_stats_keeper.sender(); + let http_stats_repository = http_core_stats_keeper.repository(); if config.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), http_stats_repository).await; diff --git a/packages/http-tracker-core/src/statistics/setup.rs b/packages/http-tracker-core/src/statistics/setup.rs index 565e86fd4..a78c53f6d 100644 --- a/packages/http-tracker-core/src/statistics/setup.rs +++ b/packages/http-tracker-core/src/statistics/setup.rs @@ -40,14 +40,14 @@ mod test { let tracker_usage_statistics = false; // HTTP core stats - let keeper = factory(tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); 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(); + let _unused = http_core_stats_keeper.run_event_listener(); } assert!(http_stats_event_sender.is_none()); @@ -58,9 +58,9 @@ mod test { let tracker_usage_statistics = true; // HTTP core stats - let keeper = factory(tracker_usage_statistics); - let http_stats_event_sender = keeper.sender(); - let _http_stats_repository = keeper.repository(); + let http_core_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_core_stats_keeper.sender(); + let _http_stats_repository = http_core_stats_keeper.repository(); assert!(http_stats_event_sender.is_some()); } diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index ac8948e42..93c8951f7 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/packages/rest-tracker-api-core/src/statistics/services.rs @@ -147,14 +147,15 @@ mod tests { let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP))); // HTTP core stats - let keeper = bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); - let _http_stats_event_sender = keeper.sender(); - let http_stats_repository = keeper.repository(); + let http_core_stats_keeper = + bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + let _http_stats_event_sender = http_core_stats_keeper.sender(); + let http_stats_repository = http_core_stats_keeper.repository(); if config.core.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(); + let _unused = http_core_stats_keeper.run_event_listener(); } // UDP core stats From 5906037113a0d5a83b5451db790eb0fdcbc84ae6 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 16:09:31 +0100 Subject: [PATCH 04/11] refactor: [#1444] http core event listener start in app start. Step 3 --- packages/axum-http-tracker-server/src/server.rs | 1 + packages/http-tracker-core/src/container.rs | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index 52085f822..a169e2565 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/packages/axum-http-tracker-server/src/server.rs @@ -302,6 +302,7 @@ mod tests { HttpTrackerCoreContainer { tracker_core_container, http_tracker_config, + http_core_stats_keeper, http_stats_event_sender, http_stats_repository, announce_service, diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index 0b8bd9337..c41fac6dc 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -13,6 +13,7 @@ pub struct HttpTrackerCoreContainer { pub tracker_core_container: Arc, // `HttpTrackerCoreServices` + pub http_core_stats_keeper: Arc, pub http_stats_event_sender: Arc>>, pub http_stats_repository: Arc, pub announce_service: Arc, @@ -44,6 +45,7 @@ impl HttpTrackerCoreContainer { Arc::new(Self { tracker_core_container: tracker_core_container.clone(), http_tracker_config: http_tracker_config.clone(), + http_core_stats_keeper: http_tracker_core_services.http_core_stats_keeper.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(), @@ -53,6 +55,7 @@ impl HttpTrackerCoreContainer { } pub struct HttpTrackerCoreServices { + pub http_core_stats_keeper: Arc, pub http_stats_event_sender: Arc>>, pub http_stats_repository: Arc, pub http_announce_service: Arc, @@ -89,6 +92,7 @@ impl HttpTrackerCoreServices { )); Arc::new(Self { + http_core_stats_keeper, http_stats_event_sender, http_stats_repository, http_announce_service, From 6d49a1308384812563cd165879237e89ee3e5528 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 16:11:34 +0100 Subject: [PATCH 05/11] refactor: [#1444] rename fields --- .../axum-http-tracker-server/src/server.rs | 6 ++-- .../tests/server/v1/contract.rs | 35 +++---------------- packages/http-tracker-core/src/container.rs | 12 +++---- .../rest-tracker-api-core/src/container.rs | 2 +- 4 files changed, 15 insertions(+), 40 deletions(-) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index a169e2565..bf694de79 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/packages/axum-http-tracker-server/src/server.rs @@ -302,9 +302,9 @@ mod tests { HttpTrackerCoreContainer { tracker_core_container, http_tracker_config, - http_core_stats_keeper, - http_stats_event_sender, - http_stats_repository, + stats_keeper: http_core_stats_keeper, + stats_event_sender: http_stats_event_sender, + stats_repository: http_stats_repository, announce_service, scrape_service, } diff --git a/packages/axum-http-tracker-server/tests/server/v1/contract.rs b/packages/axum-http-tracker-server/tests/server/v1/contract.rs index ad5b5a482..37d96052f 100644 --- a/packages/axum-http-tracker-server/tests/server/v1/contract.rs +++ b/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); diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index c41fac6dc..0fcf6338c 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -13,9 +13,9 @@ pub struct HttpTrackerCoreContainer { pub tracker_core_container: Arc, // `HttpTrackerCoreServices` - pub http_core_stats_keeper: Arc, - pub http_stats_event_sender: Arc>>, - pub http_stats_repository: Arc, + pub stats_keeper: Arc, + pub stats_event_sender: Arc>>, + pub stats_repository: Arc, pub announce_service: Arc, pub scrape_service: Arc, } @@ -45,9 +45,9 @@ impl HttpTrackerCoreContainer { Arc::new(Self { tracker_core_container: tracker_core_container.clone(), http_tracker_config: http_tracker_config.clone(), - http_core_stats_keeper: http_tracker_core_services.http_core_stats_keeper.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(), + stats_keeper: http_tracker_core_services.http_core_stats_keeper.clone(), + stats_event_sender: http_tracker_core_services.http_stats_event_sender.clone(), + 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(), }) diff --git a/packages/rest-tracker-api-core/src/container.rs b/packages/rest-tracker-api-core/src/container.rs index 329c77eed..4451eb2c4 100644 --- a/packages/rest-tracker-api-core/src/container.rs +++ b/packages/rest-tracker-api-core/src/container.rs @@ -53,7 +53,7 @@ impl TrackerHttpApiCoreContainer { Arc::new(TrackerHttpApiCoreContainer { tracker_core_container: tracker_core_container.clone(), - http_stats_repository: http_tracker_core_container.http_stats_repository.clone(), + http_stats_repository: http_tracker_core_container.stats_repository.clone(), ban_service: udp_tracker_core_container.ban_service.clone(), udp_core_stats_repository: udp_tracker_core_container.udp_core_stats_repository.clone(), From 19bb37d980601a2b7ca1e135734cd2f764a880c0 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 16:14:19 +0100 Subject: [PATCH 06/11] refactor: [#1444] renaem variables --- packages/axum-http-tracker-server/src/server.rs | 10 +++++----- .../src/v1/handlers/announce.rs | 9 ++++----- .../src/v1/handlers/scrape.rs | 9 ++++----- packages/http-tracker-core/benches/helpers/util.rs | 8 ++++---- packages/http-tracker-core/src/container.rs | 14 +++++++------- .../http-tracker-core/src/services/announce.rs | 8 ++++---- packages/http-tracker-core/src/services/scrape.rs | 12 ++++++------ .../http-tracker-core/src/statistics/services.rs | 8 ++++---- packages/http-tracker-core/src/statistics/setup.rs | 14 +++++++------- .../src/statistics/services.rs | 9 ++++----- 10 files changed, 49 insertions(+), 52 deletions(-) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index bf694de79..95a13ab1c 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/packages/axum-http-tracker-server/src/server.rs @@ -271,15 +271,15 @@ mod tests { let http_tracker_config = Arc::new(http_tracker_config.clone()); // HTTP core stats - let http_core_stats_keeper = + let http_stats_keeper = bittorrent_http_tracker_core::statistics::setup::factory(configuration.core.tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let http_stats_repository = http_core_stats_keeper.repository(); + let http_stats_event_sender = http_stats_keeper.sender(); + let http_stats_repository = http_stats_keeper.repository(); if configuration.core.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 = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } let tracker_core_container = Arc::new(TrackerCoreContainer::initialize(&core_config)); @@ -302,7 +302,7 @@ mod tests { HttpTrackerCoreContainer { tracker_core_container, http_tracker_config, - stats_keeper: http_core_stats_keeper, + stats_keeper: http_stats_keeper, stats_event_sender: http_stats_event_sender, stats_repository: http_stats_repository, announce_service, diff --git a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs index 5c08e97eb..b4c54ce09 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs @@ -161,15 +161,14 @@ mod tests { )); // HTTP core stats - let http_core_stats_keeper = - bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + 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 { // todo: this should be started like the other jobs during `app::start` // and keep the join handle in a list of jobs. - let _unused = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } let announce_service = Arc::new(AnnounceService::new( diff --git a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs index 76390ea0d..e4ba6ed51 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs @@ -132,15 +132,14 @@ mod tests { let scrape_handler = Arc::new(ScrapeHandler::new(&whitelist_authorization, &in_memory_torrent_repository)); // HTTP core stats - let http_core_stats_keeper = - bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + 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 { // todo: this should be started like the other jobs during `app::start` // and keep the join handle in a list of jobs. - let _unused = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } ( diff --git a/packages/http-tracker-core/benches/helpers/util.rs b/packages/http-tracker-core/benches/helpers/util.rs index 3ef1ccf46..fc8969c10 100644 --- a/packages/http-tracker-core/benches/helpers/util.rs +++ b/packages/http-tracker-core/benches/helpers/util.rs @@ -56,14 +56,14 @@ pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> ( )); // HTTP core stats - let http_core_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + 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 { // todo: this should be started like the other jobs during `app::start` // and keep the join handle in a list of jobs. - let _unused = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } ( diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index 0fcf6338c..381d1f770 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -45,7 +45,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.http_core_stats_keeper.clone(), + stats_keeper: http_tracker_core_services.http_stats_keeper.clone(), stats_event_sender: http_tracker_core_services.http_stats_event_sender.clone(), stats_repository: http_tracker_core_services.http_stats_repository.clone(), announce_service: http_tracker_core_services.http_announce_service.clone(), @@ -55,7 +55,7 @@ impl HttpTrackerCoreContainer { } pub struct HttpTrackerCoreServices { - pub http_core_stats_keeper: Arc, + pub http_stats_keeper: Arc, pub http_stats_event_sender: Arc>>, pub http_stats_repository: Arc, pub http_announce_service: Arc, @@ -66,14 +66,14 @@ impl HttpTrackerCoreServices { #[must_use] pub fn initialize_from(tracker_core_container: &Arc) -> Arc { // HTTP core stats - let http_core_stats_keeper = statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let http_stats_repository = http_core_stats_keeper.repository(); + 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(); if tracker_core_container.core_config.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 = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } let http_announce_service = Arc::new(AnnounceService::new( @@ -92,7 +92,7 @@ impl HttpTrackerCoreServices { )); Arc::new(Self { - http_core_stats_keeper, + http_stats_keeper, http_stats_event_sender, http_stats_repository, http_announce_service, diff --git a/packages/http-tracker-core/src/services/announce.rs b/packages/http-tracker-core/src/services/announce.rs index 5e50ebd8f..07d576aca 100644 --- a/packages/http-tracker-core/src/services/announce.rs +++ b/packages/http-tracker-core/src/services/announce.rs @@ -253,14 +253,14 @@ mod tests { )); // HTTP core stats - let http_core_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + 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 { // todo: this should be started like the other jobs during `app::start` // and keep the join handle in a list of jobs. - let _unused = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } ( diff --git a/packages/http-tracker-core/src/services/scrape.rs b/packages/http-tracker-core/src/services/scrape.rs index 7cd1a5991..23f1566b3 100644 --- a/packages/http-tracker-core/src/services/scrape.rs +++ b/packages/http-tracker-core/src/services/scrape.rs @@ -273,9 +273,9 @@ mod tests { let core_config = Arc::new(configuration.core.clone()); // HTTP core stats - let http_core_stats_keeper = statistics::setup::factory(false); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + 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 container = initialize_services_with_configuration(&configuration); @@ -465,9 +465,9 @@ mod tests { let container = initialize_services_with_configuration(&config); // HTTP core stats - let http_core_stats_keeper = statistics::setup::factory(false); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + 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 info_hash = sample_info_hash(); let info_hashes = vec![info_hash]; diff --git a/packages/http-tracker-core/src/statistics/services.rs b/packages/http-tracker-core/src/statistics/services.rs index 94ade2e45..4a27b3267 100644 --- a/packages/http-tracker-core/src/statistics/services.rs +++ b/packages/http-tracker-core/src/statistics/services.rs @@ -89,14 +89,14 @@ mod tests { let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default()); // HTTP core stats - let http_core_stats_keeper = statistics::setup::factory(config.core.tracker_usage_statistics); - let _http_stats_event_sender = http_core_stats_keeper.sender(); - let http_stats_repository = http_core_stats_keeper.repository(); + 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 { // todo: this should be started like the other jobs during `app::start` // and keep the join handle in a list of jobs. - let _unused = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), http_stats_repository).await; diff --git a/packages/http-tracker-core/src/statistics/setup.rs b/packages/http-tracker-core/src/statistics/setup.rs index a78c53f6d..bac9303a6 100644 --- a/packages/http-tracker-core/src/statistics/setup.rs +++ b/packages/http-tracker-core/src/statistics/setup.rs @@ -40,14 +40,14 @@ mod test { let tracker_usage_statistics = false; // HTTP core stats - let http_core_stats_keeper = factory(tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + let http_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_stats_keeper.sender(); + let _http_stats_repository = http_stats_keeper.repository(); 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 = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } assert!(http_stats_event_sender.is_none()); @@ -58,9 +58,9 @@ mod test { let tracker_usage_statistics = true; // HTTP core stats - let http_core_stats_keeper = factory(tracker_usage_statistics); - let http_stats_event_sender = http_core_stats_keeper.sender(); - let _http_stats_repository = http_core_stats_keeper.repository(); + let http_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_stats_keeper.sender(); + let _http_stats_repository = http_stats_keeper.repository(); assert!(http_stats_event_sender.is_some()); } diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index 93c8951f7..93bbc7e1c 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/packages/rest-tracker-api-core/src/statistics/services.rs @@ -147,15 +147,14 @@ mod tests { let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP))); // HTTP core stats - let http_core_stats_keeper = - bittorrent_http_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); - let _http_stats_event_sender = http_core_stats_keeper.sender(); - let http_stats_repository = http_core_stats_keeper.repository(); + 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 { // todo: this should be started like the other jobs during `app::start` // and keep the join handle in a list of jobs. - let _unused = http_core_stats_keeper.run_event_listener(); + let _unused = http_stats_keeper.run_event_listener(); } // UDP core stats From 2d9af45fca91845af1446c8719f8f38626784d21 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 16:16:49 +0100 Subject: [PATCH 07/11] chore: [#1444] event listener has to be run manually on tests We are moving the execution of the event listener from AppContainer initialization to jobs start. However, in tests we still have to run it manually if we need it. --- packages/axum-http-tracker-server/src/server.rs | 2 -- packages/axum-http-tracker-server/src/v1/handlers/announce.rs | 2 -- packages/axum-http-tracker-server/src/v1/handlers/scrape.rs | 2 -- packages/http-tracker-core/benches/helpers/util.rs | 2 -- packages/http-tracker-core/src/services/announce.rs | 2 -- packages/http-tracker-core/src/statistics/services.rs | 2 -- packages/http-tracker-core/src/statistics/setup.rs | 2 -- packages/rest-tracker-api-core/src/statistics/services.rs | 2 -- 8 files changed, 16 deletions(-) diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index 95a13ab1c..3d7adfaf2 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/packages/axum-http-tracker-server/src/server.rs @@ -277,8 +277,6 @@ mod tests { let http_stats_repository = http_stats_keeper.repository(); if configuration.core.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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs index b4c54ce09..ddeff3ea4 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs @@ -166,8 +166,6 @@ mod tests { let _http_stats_repository = http_stats_keeper.repository(); if config.core.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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs index e4ba6ed51..67c75d6ed 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs +++ b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs @@ -137,8 +137,6 @@ mod tests { let _http_stats_repository = http_stats_keeper.repository(); if config.core.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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/http-tracker-core/benches/helpers/util.rs b/packages/http-tracker-core/benches/helpers/util.rs index fc8969c10..6bfbcffd6 100644 --- a/packages/http-tracker-core/benches/helpers/util.rs +++ b/packages/http-tracker-core/benches/helpers/util.rs @@ -61,8 +61,6 @@ pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> ( let _http_stats_repository = http_stats_keeper.repository(); if config.core.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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/http-tracker-core/src/services/announce.rs b/packages/http-tracker-core/src/services/announce.rs index 07d576aca..c4c94474f 100644 --- a/packages/http-tracker-core/src/services/announce.rs +++ b/packages/http-tracker-core/src/services/announce.rs @@ -258,8 +258,6 @@ mod tests { let _http_stats_repository = http_stats_keeper.repository(); if config.core.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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/http-tracker-core/src/statistics/services.rs b/packages/http-tracker-core/src/statistics/services.rs index 4a27b3267..7e4f03492 100644 --- a/packages/http-tracker-core/src/statistics/services.rs +++ b/packages/http-tracker-core/src/statistics/services.rs @@ -94,8 +94,6 @@ mod tests { let http_stats_repository = http_stats_keeper.repository(); if config.core.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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/http-tracker-core/src/statistics/setup.rs b/packages/http-tracker-core/src/statistics/setup.rs index bac9303a6..f1f907b2e 100644 --- a/packages/http-tracker-core/src/statistics/setup.rs +++ b/packages/http-tracker-core/src/statistics/setup.rs @@ -45,8 +45,6 @@ mod test { let _http_stats_repository = http_stats_keeper.repository(); 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 = http_stats_keeper.run_event_listener(); } diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index 93bbc7e1c..a299ccbaa 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/packages/rest-tracker-api-core/src/statistics/services.rs @@ -152,8 +152,6 @@ mod tests { let http_stats_repository = http_stats_keeper.repository(); if config.core.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 = http_stats_keeper.run_event_listener(); } From 07c58580e421ae3f32b32462d197db6732eb4ed6 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Thu, 24 Apr 2025 16:51:06 +0100 Subject: [PATCH 08/11] refactor: [#1444] http core event listener start in app start. Step 4 --- .../src/environment.rs | 37 ++++++++++++++++--- packages/http-tracker-core/src/container.rs | 6 --- .../http-tracker-core/src/statistics/setup.rs | 12 ------ src/app.rs | 21 +++++++++++ 4 files changed, 52 insertions(+), 24 deletions(-) diff --git a/packages/axum-http-tracker-server/src/environment.rs b/packages/axum-http-tracker-server/src/environment.rs index a89d9af08..30755b452 100644 --- a/packages/axum-http-tracker-server/src/environment.rs +++ b/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 { pub container: Arc, pub registar: Registar, pub server: HttpServer, + pub event_listener_job: Option>, } impl Environment { @@ -54,22 +56,32 @@ impl Environment { 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 { + // 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 + .unwrap(); + 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 { Environment::::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 { + // 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, } } diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index 381d1f770..e685dd521 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -70,12 +70,6 @@ impl HttpTrackerCoreServices { let http_stats_event_sender = http_stats_keeper.sender(); let http_stats_repository = http_stats_keeper.repository(); - if tracker_core_container.core_config.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 = http_stats_keeper.run_event_listener(); - } - let http_announce_service = Arc::new(AnnounceService::new( tracker_core_container.core_config.clone(), tracker_core_container.announce_handler.clone(), diff --git a/packages/http-tracker-core/src/statistics/setup.rs b/packages/http-tracker-core/src/statistics/setup.rs index f1f907b2e..09f077507 100644 --- a/packages/http-tracker-core/src/statistics/setup.rs +++ b/packages/http-tracker-core/src/statistics/setup.rs @@ -7,18 +7,6 @@ use super::keeper::Keeper; use super::repository::Repository; use crate::event::sender::Broadcaster; -/// It builds the structs needed for handling the tracker metrics. -/// -/// It returns: -/// -/// - An event [`Sender`](crate::event::sender::Sender) that allows you to send -/// events related to statistics. -/// - An statistics [`Repository`](crate::statistics::repository::Repository) -/// which is an in-memory repository for the tracker metrics. -/// -/// When the input argument `tracker_usage_statistics`is false the setup does -/// 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) -> Arc { keeper_factory(tracker_usage_statistics) diff --git a/src/app.rs b/src/app.rs index d394fe644..555900315 100644 --- a/src/app.rs +++ b/src/app.rs @@ -66,6 +66,7 @@ async fn load_data_from_database(config: &Configuration, app_container: &Arc) -> Vec> { let mut jobs: Vec> = Vec::new(); + start_http_core_event_listener(config, app_container); start_the_udp_instances(config, app_container, &mut jobs).await; start_the_http_instances(config, app_container, &mut jobs).await; start_the_http_api(config, app_container, &mut jobs).await; @@ -106,6 +107,26 @@ async fn load_whitelisted_torrents(config: &Configuration, app_container: &Arc) { + if config.core.tracker_usage_statistics { + let _job = app_container + .http_tracker_core_services + .http_stats_keeper + .run_event_listener(); + + // todo: this cannot be enabled otherwise the application never ends + // because the event listener never stops. You see this console message + // forever: + // + // !! shuting down in 90 seconds !! + // 2025-04-24T15:27:45.454101Z INFO graceful_shutdown: torrust_axum_server::signals: remaining alive connections: 0 + // + // Depends on: https://github.com/torrust/torrust-tracker/issues/1405 + + //jobs.push(job); + } +} + async fn start_the_udp_instances(config: &Configuration, app_container: &Arc, jobs: &mut Vec>) { if let Some(udp_trackers) = &config.udp_trackers { for udp_tracker_config in udp_trackers { From 2fa4e15d7e44859e9da51ab77305ca6d73f1ddd8 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Fri, 25 Apr 2025 11:24:53 +0100 Subject: [PATCH 09/11] refactor: [#1444] udp core event listener start in app start --- .../src/environment.rs | 4 +- .../src/statistics/services.rs | 4 +- .../udp-tracker-core/benches/helpers/sync.rs | 4 +- packages/udp-tracker-core/src/container.rs | 11 ++-- packages/udp-tracker-core/src/event/sender.rs | 1 + .../udp-tracker-core/src/services/connect.rs | 12 ++-- .../src/statistics/event/listener.rs | 4 +- .../udp-tracker-core/src/statistics/keeper.rs | 50 ++++++++++++---- .../src/statistics/services.rs | 5 +- .../udp-tracker-core/src/statistics/setup.rs | 59 +++++++++---------- .../udp-tracker-server/src/environment.rs | 51 +++++++++++----- .../src/handlers/announce.rs | 9 +-- .../src/handlers/connect.rs | 15 ++--- .../udp-tracker-server/src/handlers/mod.rs | 5 +- src/app.rs | 18 ++++++ 15 files changed, 155 insertions(+), 97 deletions(-) diff --git a/packages/axum-http-tracker-server/src/environment.rs b/packages/axum-http-tracker-server/src/environment.rs index 30755b452..f278ad29f 100644 --- a/packages/axum-http-tracker-server/src/environment.rs +++ b/packages/axum-http-tracker-server/src/environment.rs @@ -75,7 +75,7 @@ impl Environment { .server .start(self.container.http_tracker_core_container.clone(), self.registar.give_form()) .await - .unwrap(); + .expect("Failed to start the HTTP tracker server"); Environment { container: self.container.clone(), @@ -105,7 +105,7 @@ impl Environment { } // Stop the server - let server = self.server.stop().await.expect("Failed to stop the http tracker server"); + let server = self.server.stop().await.expect("Failed to stop the HTTP tracker server"); Environment { container: self.container, diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index a299ccbaa..093971b34 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/packages/rest-tracker-api-core/src/statistics/services.rs @@ -155,9 +155,7 @@ mod tests { let _unused = http_stats_keeper.run_event_listener(); } - // UDP core stats - let (_udp_stats_event_sender, _udp_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(config.core.tracker_usage_statistics); + // UDP core stats (not used in this test) // UDP server stats let (_udp_server_stats_event_sender, udp_server_stats_repository) = diff --git a/packages/udp-tracker-core/benches/helpers/sync.rs b/packages/udp-tracker-core/benches/helpers/sync.rs index b61204586..926916d61 100644 --- a/packages/udp-tracker-core/benches/helpers/sync.rs +++ b/packages/udp-tracker-core/benches/helpers/sync.rs @@ -14,8 +14,8 @@ pub async fn connect_once(samples: u64) -> Duration { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender)); let start = Instant::now(); diff --git a/packages/udp-tracker-core/src/container.rs b/packages/udp-tracker-core/src/container.rs index 79ce15d01..0a1bf54d4 100644 --- a/packages/udp-tracker-core/src/container.rs +++ b/packages/udp-tracker-core/src/container.rs @@ -16,6 +16,7 @@ pub struct UdpTrackerCoreContainer { pub tracker_core_container: Arc, // `UdpTrackerCoreServices` + pub stats_keeper: Arc, pub udp_core_stats_event_sender: Arc>>, pub udp_core_stats_repository: Arc, pub ban_service: Arc>, @@ -52,6 +53,7 @@ impl UdpTrackerCoreContainer { tracker_core_container: tracker_core_container.clone(), // `UdpTrackerCoreServices` + stats_keeper: udp_tracker_core_services.stats_keeper.clone(), udp_core_stats_event_sender: udp_tracker_core_services.udp_core_stats_event_sender.clone(), udp_core_stats_repository: udp_tracker_core_services.udp_core_stats_repository.clone(), ban_service: udp_tracker_core_services.udp_ban_service.clone(), @@ -63,6 +65,7 @@ impl UdpTrackerCoreContainer { } pub struct UdpTrackerCoreServices { + pub stats_keeper: Arc, pub udp_core_stats_event_sender: Arc>>, pub udp_core_stats_repository: Arc, pub udp_ban_service: Arc>, @@ -74,10 +77,9 @@ pub struct UdpTrackerCoreServices { impl UdpTrackerCoreServices { #[must_use] pub fn initialize_from(tracker_core_container: &Arc) -> Arc { - let (udp_core_stats_event_sender, udp_core_stats_repository) = - statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); - let udp_core_stats_repository = Arc::new(udp_core_stats_repository); + let keeper = statistics::setup::factory(tracker_core_container.core_config.tracker_usage_statistics); + let udp_core_stats_event_sender = keeper.sender(); + let udp_core_stats_repository = keeper.repository(); let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP))); let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender.clone())); let announce_service = Arc::new(AnnounceService::new( @@ -91,6 +93,7 @@ impl UdpTrackerCoreServices { )); Arc::new(Self { + stats_keeper: keeper, udp_core_stats_event_sender, udp_core_stats_repository, udp_ban_service: ban_service, diff --git a/packages/udp-tracker-core/src/event/sender.rs b/packages/udp-tracker-core/src/event/sender.rs index 511a381d0..b720926bb 100644 --- a/packages/udp-tracker-core/src/event/sender.rs +++ b/packages/udp-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, } diff --git a/packages/udp-tracker-core/src/services/connect.rs b/packages/udp-tracker-core/src/services/connect.rs index df3db6c4b..c6c1c098f 100644 --- a/packages/udp-tracker-core/src/services/connect.rs +++ b/packages/udp-tracker-core/src/services/connect.rs @@ -78,8 +78,8 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender)); @@ -98,8 +98,8 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender)); @@ -119,8 +119,8 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let connect_service = Arc::new(ConnectService::new(udp_core_stats_event_sender)); diff --git a/packages/udp-tracker-core/src/statistics/event/listener.rs b/packages/udp-tracker-core/src/statistics/event/listener.rs index 888fb8204..835283d1e 100644 --- a/packages/udp-tracker-core/src/statistics/event/listener.rs +++ b/packages/udp-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, UDP_TRACKER_LOG_TARGET}; -pub async fn dispatch_events(mut receiver: broadcast::Receiver, stats_repository: Repository) { +pub async fn dispatch_events(mut receiver: broadcast::Receiver, stats_repository: Arc) { loop { match receiver.recv().await { Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await, diff --git a/packages/udp-tracker-core/src/statistics/keeper.rs b/packages/udp-tracker-core/src/statistics/keeper.rs index d72dcb260..8acecc585 100644 --- a/packages/udp-tracker-core/src/statistics/keeper.rs +++ b/packages/udp-tracker-core/src/statistics/keeper.rs @@ -1,8 +1,10 @@ -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::UDP_TRACKER_LOG_TARGET; /// The service responsible for keeping tracker metrics (listening to statistics events and handle them). @@ -10,44 +12,70 @@ use crate::UDP_TRACKER_LOG_TARGET; /// 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, } 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) -> Self { Self { - repository: Repository::new(), + enable_sender, + broadcaster, + repository, + } + } + + #[must_use] + pub fn sender(&self) -> Arc>> { + if self.enable_sender { + Arc::new(Some(Box::new(self.broadcaster.clone()))) + } else { + Arc::new(None) } } - pub fn run_event_listener(&mut self, receiver: Receiver) { + #[must_use] + pub fn repository(&self) -> Arc { + 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: UDP_TRACKER_LOG_TARGET, "Starting UDP tracker core event listener"); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting HTTP tracker core event listener"); tokio::spawn(async move { dispatch_events(receiver, stats_repository).await; - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "UDP tracker core event listener finished"); - }); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "HTTP tracker core event listener finished"); + }) } } #[cfg(test)] mod tests { + use crate::statistics::keeper::Keeper; use crate::statistics::metrics::Metrics; #[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; diff --git a/packages/udp-tracker-core/src/statistics/services.rs b/packages/udp-tracker-core/src/statistics/services.rs index d9b016b0d..e1aa66f67 100644 --- a/packages/udp-tracker-core/src/statistics/services.rs +++ b/packages/udp-tracker-core/src/statistics/services.rs @@ -106,9 +106,8 @@ mod tests { let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default()); - let (_udp_core_stats_event_sender, udp_core_stats_repository) = - crate::statistics::setup::factory(config.core.tracker_usage_statistics); - let udp_core_stats_repository = Arc::new(udp_core_stats_repository); + let keeper = crate::statistics::setup::factory(config.core.tracker_usage_statistics); + let udp_core_stats_repository = keeper.repository(); let tracker_metrics = get_metrics(in_memory_torrent_repository.clone(), udp_core_stats_repository.clone()).await; diff --git a/packages/udp-tracker-core/src/statistics/setup.rs b/packages/udp-tracker-core/src/statistics/setup.rs index e2974e4c0..6466ac58b 100644 --- a/packages/udp-tracker-core/src/statistics/setup.rs +++ b/packages/udp-tracker-core/src/statistics/setup.rs @@ -1,38 +1,23 @@ //! Setup for the tracker statistics. //! //! The [`factory`] function builds the structs needed for handling the tracker metrics. -use crate::event::sender::Broadcaster; -use crate::{event, statistics}; - -/// It builds the structs needed for handling the tracker metrics. -/// -/// It returns: -/// -/// - An event [`Sender`](crate::event::sender::Sender) that allows you to send -/// events related to statistics. -/// - An statistics [`Repository`](crate::statistics::repository::Repository) -/// which is an in-memory repository for the tracker metrics. -/// -/// When the input argument `tracker_usage_statistics`is false the setup does -/// 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>, statistics::repository::Repository) { - let mut keeper = statistics::keeper::Keeper::new(); - - let opt_event_sender: Option> = if tracker_usage_statistics { - let broadcaster = Broadcaster::default(); +use std::sync::Arc; - keeper.run_event_listener(broadcaster.subscribe()); - - Some(Box::new(broadcaster)) - } else { - None - }; +use super::keeper::Keeper; +use super::repository::Repository; +use crate::event::sender::Broadcaster; - (opt_event_sender, keeper.repository) +#[must_use] +pub fn factory(tracker_usage_statistics: bool) -> Arc { + keeper_factory(tracker_usage_statistics) } +#[must_use] +pub fn keeper_factory(tracker_usage_statistics: bool) -> Arc { + let broadcaster = Broadcaster::default(); + let repository = Arc::new(Repository::new()); + Arc::new(Keeper::new(tracker_usage_statistics, broadcaster.clone(), repository.clone())) +} #[cfg(test)] mod test { use super::factory; @@ -41,17 +26,27 @@ mod test { async fn should_not_send_any_event_when_statistics_are_disabled() { let tracker_usage_statistics = false; - let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics); + // UDP core stats + let http_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_stats_keeper.sender(); + let _http_stats_repository = http_stats_keeper.repository(); + + if tracker_usage_statistics { + let _unused = http_stats_keeper.run_event_listener(); + } - assert!(stats_event_sender.is_none()); + assert!(http_stats_event_sender.is_none()); } #[tokio::test] async fn should_send_events_when_statistics_are_enabled() { let tracker_usage_statistics = true; - let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics); + // UDP core stats + let http_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_stats_keeper.sender(); + let _http_stats_repository = http_stats_keeper.repository(); - assert!(stats_event_sender.is_some()); + assert!(http_stats_event_sender.is_some()); } } diff --git a/packages/udp-tracker-server/src/environment.rs b/packages/udp-tracker-server/src/environment.rs index b97da90ad..3115d3b0b 100644 --- a/packages/udp-tracker-server/src/environment.rs +++ b/packages/udp-tracker-server/src/environment.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use bittorrent_primitives::info_hash::InfoHash; use bittorrent_tracker_core::container::TrackerCoreContainer; use bittorrent_udp_tracker_core::container::UdpTrackerCoreContainer; +use tokio::task::JoinHandle; use torrust_server_lib::registar::Registar; use torrust_tracker_configuration::{logging, Configuration, DEFAULT_TIMEOUT}; use torrust_tracker_primitives::peer; @@ -22,6 +23,7 @@ where pub container: Arc, pub registar: Registar, pub server: Server, + pub udp_core_event_listener_job: Option>, } impl Environment @@ -55,29 +57,38 @@ impl Environment { container, registar: Registar::default(), server, + udp_core_event_listener_job: None, } } + /// Starts the test environment and return a running environment. + /// /// # Panics /// /// Will panic if it cannot start the server. #[allow(dead_code)] pub async fn start(self) -> Environment { let cookie_lifetime = self.container.udp_tracker_core_container.udp_tracker_config.cookie_lifetime; + // Start the UDP tracker core event listener + let udp_core_event_listener_job = Some(self.container.udp_tracker_core_container.stats_keeper.run_event_listener()); + + // Start the UDP tracker server + let server = self + .server + .start( + self.container.udp_tracker_core_container.clone(), + self.container.udp_tracker_server_container.clone(), + self.registar.give_form(), + cookie_lifetime, + ) + .await + .expect("Failed to start the UDP tracker server"); Environment { container: self.container.clone(), registar: self.registar.clone(), - server: self - .server - .start( - self.container.udp_tracker_core_container.clone(), - self.container.udp_tracker_server_container.clone(), - self.registar.give_form(), - cookie_lifetime, - ) - .await - .unwrap(), + server, + udp_core_event_listener_job, } } } @@ -89,22 +100,34 @@ impl Environment { pub async fn new(configuration: &Arc) -> Self { tokio::time::timeout(DEFAULT_TIMEOUT, Environment::::new(configuration).start()) .await - .expect("it should create an environment within the timeout") + .expect("Failed to create a UDP tracker server running environment within the timeout") } + /// Stops the test environment and return a stopped environment. + /// /// # Panics /// /// Will panic if it cannot stop the service within the timeout. #[allow(dead_code)] pub async fn stop(self) -> Environment { - let stopped = tokio::time::timeout(DEFAULT_TIMEOUT, self.server.stop()) + // Stop the event listener + if let Some(udp_core_event_listener_job) = self.udp_core_event_listener_job { + // todo: send a message to the event listener to stop and wait for + // it to finish + udp_core_event_listener_job.abort(); + } + + // Stop the server + let server = tokio::time::timeout(DEFAULT_TIMEOUT, self.server.stop()) .await - .expect("it should stop the environment within the timeout"); + .expect("Failed to stop the UDP tracker server within the timeout") + .expect("Failed to stop the UDP tracker server"); Environment { container: self.container, registar: Registar::default(), - server: stopped.expect("it should stop the udp tracker service"), + server, + udp_core_event_listener_job: None, } } diff --git a/packages/udp-tracker-server/src/handlers/announce.rs b/packages/udp-tracker-server/src/handlers/announce.rs index 0167553f2..38b42a0b6 100644 --- a/packages/udp-tracker-server/src/handlers/announce.rs +++ b/packages/udp-tracker-server/src/handlers/announce.rs @@ -374,10 +374,6 @@ mod tests { core_tracker_services: Arc, core_udp_tracker_services: Arc, ) -> Response { - let (udp_core_stats_event_sender, _udp_core_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(false); - let _udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); @@ -710,9 +706,8 @@ mod tests { announce_handler: Arc, whitelist_authorization: Arc, ) -> Response { - let (udp_core_stats_event_sender, _udp_core_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); diff --git a/packages/udp-tracker-server/src/handlers/connect.rs b/packages/udp-tracker-server/src/handlers/connect.rs index aef8833b9..9ea36903c 100644 --- a/packages/udp-tracker-server/src/handlers/connect.rs +++ b/packages/udp-tracker-server/src/handlers/connect.rs @@ -81,9 +81,8 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); @@ -118,9 +117,8 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); @@ -155,9 +153,8 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); diff --git a/packages/udp-tracker-server/src/handlers/mod.rs b/packages/udp-tracker-server/src/handlers/mod.rs index f8ca9d8ea..bc39f63ae 100644 --- a/packages/udp-tracker-server/src/handlers/mod.rs +++ b/packages/udp-tracker-server/src/handlers/mod.rs @@ -284,9 +284,8 @@ pub(crate) mod tests { )); let scrape_handler = Arc::new(ScrapeHandler::new(&whitelist_authorization, &in_memory_torrent_repository)); - let (udp_core_stats_event_sender, _udp_core_stats_repository) = - bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = Arc::new(udp_core_stats_event_sender); + let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = keeper.sender(); let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); diff --git a/src/app.rs b/src/app.rs index 555900315..a0f63094b 100644 --- a/src/app.rs +++ b/src/app.rs @@ -67,6 +67,7 @@ async fn start_jobs(config: &Configuration, app_container: &Arc) - let mut jobs: Vec> = Vec::new(); start_http_core_event_listener(config, app_container); + start_udp_core_event_listener(config, app_container); start_the_udp_instances(config, app_container, &mut jobs).await; start_the_http_instances(config, app_container, &mut jobs).await; start_the_http_api(config, app_container, &mut jobs).await; @@ -127,6 +128,23 @@ fn start_http_core_event_listener(config: &Configuration, app_container: &Arc) { + if config.core.tracker_usage_statistics { + let _job = app_container.udp_tracker_core_services.stats_keeper.run_event_listener(); + + // todo: this cannot be enabled otherwise the application never ends + // because the event listener never stops. You see this console message + // forever: + // + // !! shuting down in 90 seconds !! + // 2025-04-24T15:27:45.454101Z INFO graceful_shutdown: torrust_axum_server::signals: remaining alive connections: 0 + // + // Depends on: https://github.com/torrust/torrust-tracker/issues/1405 + + //jobs.push(job); + } +} + async fn start_the_udp_instances(config: &Configuration, app_container: &Arc, jobs: &mut Vec>) { if let Some(udp_trackers) = &config.udp_trackers { for udp_tracker_config in udp_trackers { From 74e174d377264bee10b89c258f3917b9074d1215 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Fri, 25 Apr 2025 13:10:30 +0100 Subject: [PATCH 10/11] refactor: [#1444] udp server event listener start in app start --- .../src/statistics/services.rs | 4 +- packages/udp-tracker-server/src/container.rs | 11 ++-- .../udp-tracker-server/src/environment.rs | 23 ++++++- .../udp-tracker-server/src/event/sender.rs | 1 + .../src/handlers/announce.rs | 12 ++-- .../src/handlers/connect.rs | 24 ++++---- .../udp-tracker-server/src/handlers/mod.rs | 8 +-- .../udp-tracker-server/src/handlers/scrape.rs | 4 +- .../src/statistics/event/listener.rs | 4 +- .../src/statistics/keeper.rs | 52 ++++++++++++---- .../src/statistics/services.rs | 5 +- .../src/statistics/setup.rs | 61 +++++++++---------- src/app.rs | 21 +++++++ 13 files changed, 149 insertions(+), 81 deletions(-) diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index 093971b34..95e21633a 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/packages/rest-tracker-api-core/src/statistics/services.rs @@ -158,9 +158,9 @@ mod tests { // UDP core stats (not used in this test) // UDP server stats - let (_udp_server_stats_event_sender, udp_server_stats_repository) = + let udp_server_stats_keeper = torrust_udp_tracker_server::statistics::setup::factory(config.core.tracker_usage_statistics); - let udp_server_stats_repository = Arc::new(udp_server_stats_repository); + let udp_server_stats_repository = udp_server_stats_keeper.repository(); let tracker_metrics = get_metrics( in_memory_torrent_repository.clone(), diff --git a/packages/udp-tracker-server/src/container.rs b/packages/udp-tracker-server/src/container.rs index 2b1ce8c99..89740cf77 100644 --- a/packages/udp-tracker-server/src/container.rs +++ b/packages/udp-tracker-server/src/container.rs @@ -5,6 +5,7 @@ use torrust_tracker_configuration::Core; use crate::{event, statistics}; pub struct UdpTrackerServerContainer { + pub udp_server_stats_keeper: Arc, pub udp_server_stats_event_sender: Arc>>, pub udp_server_stats_repository: Arc, } @@ -15,6 +16,7 @@ impl UdpTrackerServerContainer { let udp_tracker_server_services = UdpTrackerServerServices::initialize(core_config); Arc::new(Self { + udp_server_stats_keeper: udp_tracker_server_services.udp_server_stats_keeper.clone(), udp_server_stats_event_sender: udp_tracker_server_services.udp_server_stats_event_sender.clone(), udp_server_stats_repository: udp_tracker_server_services.udp_server_stats_repository.clone(), }) @@ -22,6 +24,7 @@ impl UdpTrackerServerContainer { } pub struct UdpTrackerServerServices { + pub udp_server_stats_keeper: Arc, pub udp_server_stats_event_sender: Arc>>, pub udp_server_stats_repository: Arc, } @@ -29,12 +32,12 @@ pub struct UdpTrackerServerServices { impl UdpTrackerServerServices { #[must_use] pub fn initialize(core_config: &Arc) -> Arc { - let (udp_server_stats_event_sender, udp_server_stats_repository) = - statistics::setup::factory(core_config.tracker_usage_statistics); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); - let udp_server_stats_repository = Arc::new(udp_server_stats_repository); + let udp_server_stats_keeper = statistics::setup::factory(core_config.tracker_usage_statistics); + let udp_server_stats_event_sender = udp_server_stats_keeper.sender(); + let udp_server_stats_repository = udp_server_stats_keeper.repository(); Arc::new(Self { + udp_server_stats_keeper: udp_server_stats_keeper.clone(), udp_server_stats_event_sender: udp_server_stats_event_sender.clone(), udp_server_stats_repository: udp_server_stats_repository.clone(), }) diff --git a/packages/udp-tracker-server/src/environment.rs b/packages/udp-tracker-server/src/environment.rs index 3115d3b0b..2b31e78bd 100644 --- a/packages/udp-tracker-server/src/environment.rs +++ b/packages/udp-tracker-server/src/environment.rs @@ -24,6 +24,7 @@ where pub registar: Registar, pub server: Server, pub udp_core_event_listener_job: Option>, + pub udp_server_event_listener_job: Option>, } impl Environment @@ -58,6 +59,7 @@ impl Environment { registar: Registar::default(), server, udp_core_event_listener_job: None, + udp_server_event_listener_job: None, } } @@ -72,6 +74,14 @@ impl Environment { // Start the UDP tracker core event listener let udp_core_event_listener_job = Some(self.container.udp_tracker_core_container.stats_keeper.run_event_listener()); + // Start the UDP tracker server event listener + let udp_server_event_listener_job = Some( + self.container + .udp_tracker_server_container + .udp_server_stats_keeper + .run_event_listener(), + ); + // Start the UDP tracker server let server = self .server @@ -89,6 +99,7 @@ impl Environment { registar: self.registar.clone(), server, udp_core_event_listener_job, + udp_server_event_listener_job, } } } @@ -110,14 +121,21 @@ impl Environment { /// Will panic if it cannot stop the service within the timeout. #[allow(dead_code)] pub async fn stop(self) -> Environment { - // Stop the event listener + // Stop the UDP tracker core event listener if let Some(udp_core_event_listener_job) = self.udp_core_event_listener_job { // todo: send a message to the event listener to stop and wait for // it to finish udp_core_event_listener_job.abort(); } - // Stop the server + // Stop the UDP tracker server event listener + if let Some(udp_server_event_listener_job) = self.udp_server_event_listener_job { + // todo: send a message to the event listener to stop and wait for + // it to finish + udp_server_event_listener_job.abort(); + } + + // Stop the UDP tracker server let server = tokio::time::timeout(DEFAULT_TIMEOUT, self.server.stop()) .await .expect("Failed to stop the UDP tracker server within the timeout") @@ -128,6 +146,7 @@ impl Environment { registar: Registar::default(), server, udp_core_event_listener_job: None, + udp_server_event_listener_job: None, } } diff --git a/packages/udp-tracker-server/src/event/sender.rs b/packages/udp-tracker-server/src/event/sender.rs index 511a381d0..b720926bb 100644 --- a/packages/udp-tracker-server/src/event/sender.rs +++ b/packages/udp-tracker-server/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, } diff --git a/packages/udp-tracker-server/src/handlers/announce.rs b/packages/udp-tracker-server/src/handlers/announce.rs index 38b42a0b6..9dd7156b1 100644 --- a/packages/udp-tracker-server/src/handlers/announce.rs +++ b/packages/udp-tracker-server/src/handlers/announce.rs @@ -374,8 +374,8 @@ mod tests { core_tracker_services: Arc, core_udp_tracker_services: Arc, ) -> Response { - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = keeper.sender(); let client_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(126, 0, 0, 1)), 8080); let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); @@ -706,11 +706,11 @@ mod tests { announce_handler: Arc, whitelist_authorization: Arc, ) -> Response { - let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = keeper.sender(); + let core_keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = core_keeper.sender(); - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let server_keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = server_keeper.sender(); let client_ip_v4 = Ipv4Addr::new(126, 0, 0, 1); let client_ip_v6 = client_ip_v4.to_ipv6_compatible(); diff --git a/packages/udp-tracker-server/src/handlers/connect.rs b/packages/udp-tracker-server/src/handlers/connect.rs index 9ea36903c..fb05b3693 100644 --- a/packages/udp-tracker-server/src/handlers/connect.rs +++ b/packages/udp-tracker-server/src/handlers/connect.rs @@ -81,11 +81,11 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = keeper.sender(); + let core_keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = core_keeper.sender(); - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let server_keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = server_keeper.sender(); let request = ConnectRequest { transaction_id: TransactionId(0i32.into()), @@ -117,11 +117,11 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = keeper.sender(); + let core_keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = core_keeper.sender(); - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let server_keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = server_keeper.sender(); let request = ConnectRequest { transaction_id: TransactionId(0i32.into()), @@ -153,11 +153,11 @@ mod tests { let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); let server_service_binding = ServiceBinding::new(Protocol::UDP, server_socket_addr).unwrap(); - let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = keeper.sender(); + let core_keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = core_keeper.sender(); - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let server_keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = server_keeper.sender(); let request = ConnectRequest { transaction_id: TransactionId(0i32.into()), diff --git a/packages/udp-tracker-server/src/handlers/mod.rs b/packages/udp-tracker-server/src/handlers/mod.rs index bc39f63ae..0ad593bb2 100644 --- a/packages/udp-tracker-server/src/handlers/mod.rs +++ b/packages/udp-tracker-server/src/handlers/mod.rs @@ -284,11 +284,11 @@ pub(crate) mod tests { )); let scrape_handler = Arc::new(ScrapeHandler::new(&whitelist_authorization, &in_memory_torrent_repository)); - let keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); - let udp_core_stats_event_sender = keeper.sender(); + let core_keeper = bittorrent_udp_tracker_core::statistics::setup::factory(false); + let udp_core_stats_event_sender = core_keeper.sender(); - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let server_keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = server_keeper.sender(); let announce_service = Arc::new(AnnounceService::new( announce_handler.clone(), diff --git a/packages/udp-tracker-server/src/handlers/scrape.rs b/packages/udp-tracker-server/src/handlers/scrape.rs index 35b5ee65c..cef896d73 100644 --- a/packages/udp-tracker-server/src/handlers/scrape.rs +++ b/packages/udp-tracker-server/src/handlers/scrape.rs @@ -178,8 +178,8 @@ mod tests { core_tracker_services: Arc, core_udp_tracker_services: Arc, ) -> Response { - let (udp_server_stats_event_sender, _udp_server_stats_repository) = crate::statistics::setup::factory(false); - let udp_server_stats_event_sender = Arc::new(udp_server_stats_event_sender); + let keeper = crate::statistics::setup::factory(false); + let udp_server_stats_event_sender = keeper.sender(); let client_socket_addr = sample_ipv4_remote_addr(); let server_socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 196)), 6969); diff --git a/packages/udp-tracker-server/src/statistics/event/listener.rs b/packages/udp-tracker-server/src/statistics/event/listener.rs index cf348ea17..80c9f8d21 100644 --- a/packages/udp-tracker-server/src/statistics/event/listener.rs +++ b/packages/udp-tracker-server/src/statistics/event/listener.rs @@ -1,3 +1,5 @@ +use std::sync::Arc; + use bittorrent_udp_tracker_core::UDP_TRACKER_LOG_TARGET; use tokio::sync::broadcast; use torrust_tracker_clock::clock::Time; @@ -7,7 +9,7 @@ use crate::event::Event; use crate::statistics::repository::Repository; use crate::CurrentClock; -pub async fn dispatch_events(mut receiver: broadcast::Receiver, stats_repository: Repository) { +pub async fn dispatch_events(mut receiver: broadcast::Receiver, stats_repository: Arc) { loop { match receiver.recv().await { Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await, diff --git a/packages/udp-tracker-server/src/statistics/keeper.rs b/packages/udp-tracker-server/src/statistics/keeper.rs index c200b4cdf..1d525e7b3 100644 --- a/packages/udp-tracker-server/src/statistics/keeper.rs +++ b/packages/udp-tracker-server/src/statistics/keeper.rs @@ -1,56 +1,84 @@ +use std::sync::Arc; + use bittorrent_udp_tracker_core::UDP_TRACKER_LOG_TARGET; -use tokio::sync::broadcast::Receiver; +use tokio::task::JoinHandle; use super::event::listener::dispatch_events; use super::repository::Repository; -use crate::event::Event; +use crate::event::sender::{self, Broadcaster}; /// 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, } 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) -> Self { Self { - repository: Repository::new(), + enable_sender, + broadcaster, + repository, + } + } + + #[must_use] + pub fn sender(&self) -> Arc>> { + if self.enable_sender { + Arc::new(Some(Box::new(self.broadcaster.clone()))) + } else { + Arc::new(None) } } - pub fn run_event_listener(&mut self, receiver: Receiver) { + #[must_use] + pub fn repository(&self) -> Arc { + 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: UDP_TRACKER_LOG_TARGET, "Starting UDP tracker server event listener"); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting HTTP tracker core event listener"); tokio::spawn(async move { dispatch_events(receiver, stats_repository).await; - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "UDP tracker core server listener finished"); - }); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "HTTP tracker core event listener finished"); + }) } } #[cfg(test)] mod tests { + use crate::statistics::keeper::Keeper; use crate::statistics::metrics::Metrics; #[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; - assert_eq!(stats.udp4_requests, Metrics::default().udp4_requests); + assert_eq!(stats.udp4_announces_handled, Metrics::default().udp4_announces_handled); } } diff --git a/packages/udp-tracker-server/src/statistics/services.rs b/packages/udp-tracker-server/src/statistics/services.rs index b84bf4cd0..22f3f4754 100644 --- a/packages/udp-tracker-server/src/statistics/services.rs +++ b/packages/udp-tracker-server/src/statistics/services.rs @@ -127,9 +127,8 @@ mod tests { let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default()); let ban_service = Arc::new(RwLock::new(BanService::new(MAX_CONNECTION_ID_ERRORS_PER_IP))); - let (_udp_server_stats_event_sender, udp_server_stats_repository) = - statistics::setup::factory(config.core.tracker_usage_statistics); - let udp_server_stats_repository = Arc::new(udp_server_stats_repository); + let keeper = statistics::setup::factory(config.core.tracker_usage_statistics); + let udp_server_stats_repository = keeper.repository(); let tracker_metrics = get_metrics( in_memory_torrent_repository.clone(), diff --git a/packages/udp-tracker-server/src/statistics/setup.rs b/packages/udp-tracker-server/src/statistics/setup.rs index d8cc7bca9..09f077507 100644 --- a/packages/udp-tracker-server/src/statistics/setup.rs +++ b/packages/udp-tracker-server/src/statistics/setup.rs @@ -1,37 +1,22 @@ //! Setup for the tracker statistics. //! -//! The [`factory`] function builds the structs needed for handling the tracker -//! metrics. -use crate::event::sender::Broadcaster; -use crate::{event, statistics}; - -/// It builds the structs needed for handling the tracker metrics. -/// -/// It returns: -/// -/// - An event [`Sender`](crate::event::sender::Sender) that allows you to send -/// events related to statistics. -/// - An statistics [`Repository`](crate::statistics::repository::Repository) -/// which is an in-memory repository for the tracker metrics. -/// -/// When the input argument `tracker_usage_statistics`is false the setup does -/// 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>, statistics::repository::Repository) { - let mut keeper = statistics::keeper::Keeper::new(); +//! The [`factory`] function builds the structs needed for handling the tracker metrics. +use std::sync::Arc; - let opt_event_sender: Option> = if tracker_usage_statistics { - let broadcaster = Broadcaster::default(); - - keeper.run_event_listener(broadcaster.subscribe()); +use super::keeper::Keeper; +use super::repository::Repository; +use crate::event::sender::Broadcaster; - Some(Box::new(broadcaster)) - } else { - None - }; +#[must_use] +pub fn factory(tracker_usage_statistics: bool) -> Arc { + keeper_factory(tracker_usage_statistics) +} - (opt_event_sender, keeper.repository) +#[must_use] +pub fn keeper_factory(tracker_usage_statistics: bool) -> Arc { + let broadcaster = Broadcaster::default(); + let repository = Arc::new(Repository::new()); + Arc::new(Keeper::new(tracker_usage_statistics, broadcaster.clone(), repository.clone())) } #[cfg(test)] @@ -42,17 +27,27 @@ mod test { async fn should_not_send_any_event_when_statistics_are_disabled() { let tracker_usage_statistics = false; - let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics); + // HTTP core stats + let http_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_stats_keeper.sender(); + let _http_stats_repository = http_stats_keeper.repository(); + + if tracker_usage_statistics { + let _unused = http_stats_keeper.run_event_listener(); + } - assert!(stats_event_sender.is_none()); + assert!(http_stats_event_sender.is_none()); } #[tokio::test] async fn should_send_events_when_statistics_are_enabled() { let tracker_usage_statistics = true; - let (stats_event_sender, _stats_repository) = factory(tracker_usage_statistics); + // HTTP core stats + let http_stats_keeper = factory(tracker_usage_statistics); + let http_stats_event_sender = http_stats_keeper.sender(); + let _http_stats_repository = http_stats_keeper.repository(); - assert!(stats_event_sender.is_some()); + assert!(http_stats_event_sender.is_some()); } } diff --git a/src/app.rs b/src/app.rs index a0f63094b..67380d30d 100644 --- a/src/app.rs +++ b/src/app.rs @@ -68,6 +68,7 @@ async fn start_jobs(config: &Configuration, app_container: &Arc) - start_http_core_event_listener(config, app_container); start_udp_core_event_listener(config, app_container); + start_udp_server_event_listener(config, app_container); start_the_udp_instances(config, app_container, &mut jobs).await; start_the_http_instances(config, app_container, &mut jobs).await; start_the_http_api(config, app_container, &mut jobs).await; @@ -145,6 +146,26 @@ fn start_udp_core_event_listener(config: &Configuration, app_container: &Arc) { + if config.core.tracker_usage_statistics { + let _job = app_container + .udp_tracker_server_container + .udp_server_stats_keeper + .run_event_listener(); + + // todo: this cannot be enabled otherwise the application never ends + // because the event listener never stops. You see this console message + // forever: + // + // !! shuting down in 90 seconds !! + // 2025-04-24T15:27:45.454101Z INFO graceful_shutdown: torrust_axum_server::signals: remaining alive connections: 0 + // + // Depends on: https://github.com/torrust/torrust-tracker/issues/1405 + + //jobs.push(job); + } +} + async fn start_the_udp_instances(config: &Configuration, app_container: &Arc, jobs: &mut Vec>) { if let Some(udp_trackers) = &config.udp_trackers { for udp_tracker_config in udp_trackers { From 6d50a784239945299c63116fe5434da92fd0dd6e Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Fri, 25 Apr 2025 13:52:55 +0100 Subject: [PATCH 11/11] refactor: normalize container field names --- packages/http-tracker-core/src/container.rs | 30 +++++++------- .../rest-tracker-api-core/src/container.rs | 4 +- packages/udp-tracker-core/src/container.rs | 40 +++++++++---------- packages/udp-tracker-server/src/container.rs | 24 +++++------ .../udp-tracker-server/src/environment.rs | 7 +--- .../udp-tracker-server/src/handlers/mod.rs | 10 ++--- .../udp-tracker-server/src/server/launcher.rs | 11 ++--- .../src/server/processor.rs | 2 +- .../tests/server/contract.rs | 4 +- src/app.rs | 10 +---- src/container.rs | 8 ++-- 11 files changed, 67 insertions(+), 83 deletions(-) diff --git a/packages/http-tracker-core/src/container.rs b/packages/http-tracker-core/src/container.rs index e685dd521..496856494 100644 --- a/packages/http-tracker-core/src/container.rs +++ b/packages/http-tracker-core/src/container.rs @@ -45,21 +45,21 @@ 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.http_stats_keeper.clone(), - stats_event_sender: http_tracker_core_services.http_stats_event_sender.clone(), - 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_keeper: Arc, - pub http_stats_event_sender: Arc>>, - pub http_stats_repository: Arc, - pub http_announce_service: Arc, - pub http_scrape_service: Arc, + pub stats_keeper: Arc, + pub stats_event_sender: Arc>>, + pub stats_repository: Arc, + pub announce_service: Arc, + pub scrape_service: Arc, } impl HttpTrackerCoreServices { @@ -86,11 +86,11 @@ impl HttpTrackerCoreServices { )); Arc::new(Self { - http_stats_keeper, - 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, }) } } diff --git a/packages/rest-tracker-api-core/src/container.rs b/packages/rest-tracker-api-core/src/container.rs index 4451eb2c4..ec3786dfb 100644 --- a/packages/rest-tracker-api-core/src/container.rs +++ b/packages/rest-tracker-api-core/src/container.rs @@ -56,9 +56,9 @@ impl TrackerHttpApiCoreContainer { http_stats_repository: http_tracker_core_container.stats_repository.clone(), ban_service: udp_tracker_core_container.ban_service.clone(), - udp_core_stats_repository: udp_tracker_core_container.udp_core_stats_repository.clone(), + udp_core_stats_repository: udp_tracker_core_container.stats_repository.clone(), - udp_server_stats_repository: udp_tracker_server_container.udp_server_stats_repository.clone(), + udp_server_stats_repository: udp_tracker_server_container.stats_repository.clone(), http_api_config: http_api_config.clone(), }) diff --git a/packages/udp-tracker-core/src/container.rs b/packages/udp-tracker-core/src/container.rs index 0a1bf54d4..ef66e9b7e 100644 --- a/packages/udp-tracker-core/src/container.rs +++ b/packages/udp-tracker-core/src/container.rs @@ -17,8 +17,8 @@ pub struct UdpTrackerCoreContainer { // `UdpTrackerCoreServices` pub stats_keeper: Arc, - pub udp_core_stats_event_sender: Arc>>, - pub udp_core_stats_repository: Arc, + pub stats_event_sender: Arc>>, + pub stats_repository: Arc, pub ban_service: Arc>, pub connect_service: Arc, pub announce_service: Arc, @@ -54,24 +54,24 @@ impl UdpTrackerCoreContainer { // `UdpTrackerCoreServices` stats_keeper: udp_tracker_core_services.stats_keeper.clone(), - udp_core_stats_event_sender: udp_tracker_core_services.udp_core_stats_event_sender.clone(), - udp_core_stats_repository: udp_tracker_core_services.udp_core_stats_repository.clone(), - ban_service: udp_tracker_core_services.udp_ban_service.clone(), - connect_service: udp_tracker_core_services.udp_connect_service.clone(), - announce_service: udp_tracker_core_services.udp_announce_service.clone(), - scrape_service: udp_tracker_core_services.udp_scrape_service.clone(), + stats_event_sender: udp_tracker_core_services.stats_event_sender.clone(), + stats_repository: udp_tracker_core_services.stats_repository.clone(), + ban_service: udp_tracker_core_services.ban_service.clone(), + connect_service: udp_tracker_core_services.connect_service.clone(), + announce_service: udp_tracker_core_services.announce_service.clone(), + scrape_service: udp_tracker_core_services.scrape_service.clone(), }) } } pub struct UdpTrackerCoreServices { pub stats_keeper: Arc, - pub udp_core_stats_event_sender: Arc>>, - pub udp_core_stats_repository: Arc, - pub udp_ban_service: Arc>, - pub udp_connect_service: Arc, - pub udp_announce_service: Arc, - pub udp_scrape_service: Arc, + pub stats_event_sender: Arc>>, + pub stats_repository: Arc, + pub ban_service: Arc>, + pub connect_service: Arc, + pub announce_service: Arc, + pub scrape_service: Arc, } impl UdpTrackerCoreServices { @@ -94,12 +94,12 @@ impl UdpTrackerCoreServices { Arc::new(Self { stats_keeper: keeper, - udp_core_stats_event_sender, - udp_core_stats_repository, - udp_ban_service: ban_service, - udp_connect_service: connect_service, - udp_announce_service: announce_service, - udp_scrape_service: scrape_service, + stats_event_sender: udp_core_stats_event_sender, + stats_repository: udp_core_stats_repository, + ban_service, + connect_service, + announce_service, + scrape_service, }) } } diff --git a/packages/udp-tracker-server/src/container.rs b/packages/udp-tracker-server/src/container.rs index 89740cf77..64d01e754 100644 --- a/packages/udp-tracker-server/src/container.rs +++ b/packages/udp-tracker-server/src/container.rs @@ -5,9 +5,9 @@ use torrust_tracker_configuration::Core; use crate::{event, statistics}; pub struct UdpTrackerServerContainer { - pub udp_server_stats_keeper: Arc, - pub udp_server_stats_event_sender: Arc>>, - pub udp_server_stats_repository: Arc, + pub stats_keeper: Arc, + pub stats_event_sender: Arc>>, + pub stats_repository: Arc, } impl UdpTrackerServerContainer { @@ -16,17 +16,17 @@ impl UdpTrackerServerContainer { let udp_tracker_server_services = UdpTrackerServerServices::initialize(core_config); Arc::new(Self { - udp_server_stats_keeper: udp_tracker_server_services.udp_server_stats_keeper.clone(), - udp_server_stats_event_sender: udp_tracker_server_services.udp_server_stats_event_sender.clone(), - udp_server_stats_repository: udp_tracker_server_services.udp_server_stats_repository.clone(), + stats_keeper: udp_tracker_server_services.stats_keeper.clone(), + stats_event_sender: udp_tracker_server_services.stats_event_sender.clone(), + stats_repository: udp_tracker_server_services.stats_repository.clone(), }) } } pub struct UdpTrackerServerServices { - pub udp_server_stats_keeper: Arc, - pub udp_server_stats_event_sender: Arc>>, - pub udp_server_stats_repository: Arc, + pub stats_keeper: Arc, + pub stats_event_sender: Arc>>, + pub stats_repository: Arc, } impl UdpTrackerServerServices { @@ -37,9 +37,9 @@ impl UdpTrackerServerServices { let udp_server_stats_repository = udp_server_stats_keeper.repository(); Arc::new(Self { - udp_server_stats_keeper: udp_server_stats_keeper.clone(), - udp_server_stats_event_sender: udp_server_stats_event_sender.clone(), - udp_server_stats_repository: udp_server_stats_repository.clone(), + stats_keeper: udp_server_stats_keeper.clone(), + stats_event_sender: udp_server_stats_event_sender.clone(), + stats_repository: udp_server_stats_repository.clone(), }) } } diff --git a/packages/udp-tracker-server/src/environment.rs b/packages/udp-tracker-server/src/environment.rs index 2b31e78bd..cda8cd678 100644 --- a/packages/udp-tracker-server/src/environment.rs +++ b/packages/udp-tracker-server/src/environment.rs @@ -75,12 +75,7 @@ impl Environment { let udp_core_event_listener_job = Some(self.container.udp_tracker_core_container.stats_keeper.run_event_listener()); // Start the UDP tracker server event listener - let udp_server_event_listener_job = Some( - self.container - .udp_tracker_server_container - .udp_server_stats_keeper - .run_event_listener(), - ); + let udp_server_event_listener_job = Some(self.container.udp_tracker_server_container.stats_keeper.run_event_listener()); // Start the UDP tracker server let server = self diff --git a/packages/udp-tracker-server/src/handlers/mod.rs b/packages/udp-tracker-server/src/handlers/mod.rs index 0ad593bb2..8ef053684 100644 --- a/packages/udp-tracker-server/src/handlers/mod.rs +++ b/packages/udp-tracker-server/src/handlers/mod.rs @@ -98,7 +98,7 @@ pub(crate) async fn handle_packet( udp_request.from, server_service_binding, request_id, - &udp_tracker_server_container.udp_server_stats_event_sender, + &udp_tracker_server_container.stats_event_sender, cookie_time_values.valid_range.clone(), &error, Some(transaction_id), @@ -114,7 +114,7 @@ pub(crate) async fn handle_packet( udp_request.from, server_service_binding, request_id, - &udp_tracker_server_container.udp_server_stats_event_sender, + &udp_tracker_server_container.stats_event_sender, cookie_time_values.valid_range.clone(), &e, None, @@ -161,7 +161,7 @@ pub async fn handle_request( server_service_binding, &connect_request, &udp_tracker_core_container.connect_service, - &udp_tracker_server_container.udp_server_stats_event_sender, + &udp_tracker_server_container.stats_event_sender, cookie_time_values.issue_time, ) .await, @@ -174,7 +174,7 @@ pub async fn handle_request( server_service_binding, &announce_request, &udp_tracker_core_container.tracker_core_container.core_config, - &udp_tracker_server_container.udp_server_stats_event_sender, + &udp_tracker_server_container.stats_event_sender, cookie_time_values.valid_range, ) .await @@ -189,7 +189,7 @@ pub async fn handle_request( client_socket_addr, server_service_binding, &scrape_request, - &udp_tracker_server_container.udp_server_stats_event_sender, + &udp_tracker_server_container.stats_event_sender, cookie_time_values.valid_range, ) .await diff --git a/packages/udp-tracker-server/src/server/launcher.rs b/packages/udp-tracker-server/src/server/launcher.rs index d62a4d04e..02b9c8d74 100644 --- a/packages/udp-tracker-server/src/server/launcher.rs +++ b/packages/udp-tracker-server/src/server/launcher.rs @@ -182,8 +182,7 @@ impl Launcher { let client_socket_addr = req.from; - if let Some(udp_server_stats_event_sender) = udp_tracker_server_container.udp_server_stats_event_sender.as_deref() - { + if let Some(udp_server_stats_event_sender) = udp_tracker_server_container.stats_event_sender.as_deref() { udp_server_stats_event_sender .send_event(Event::UdpRequestReceived { context: ConnectionContext::new(client_socket_addr, server_service_binding.clone()), @@ -194,9 +193,7 @@ impl Launcher { if udp_tracker_core_container.ban_service.read().await.is_banned(&req.from.ip()) { tracing::debug!(target: UDP_TRACKER_LOG_TARGET, local_addr, "Udp::run_udp_server::loop continue: (banned ip)"); - if let Some(udp_server_stats_event_sender) = - udp_tracker_server_container.udp_server_stats_event_sender.as_deref() - { + if let Some(udp_server_stats_event_sender) = udp_tracker_server_container.stats_event_sender.as_deref() { udp_server_stats_event_sender .send_event(Event::UdpRequestBanned { context: ConnectionContext::new(client_socket_addr, server_service_binding.clone()), @@ -236,9 +233,7 @@ impl Launcher { if old_request_aborted { // Evicted task from active requests buffer was aborted. - if let Some(udp_server_stats_event_sender) = - udp_tracker_server_container.udp_server_stats_event_sender.as_deref() - { + if let Some(udp_server_stats_event_sender) = udp_tracker_server_container.stats_event_sender.as_deref() { udp_server_stats_event_sender .send_event(Event::UdpRequestAborted { context: ConnectionContext::new(client_socket_addr, server_service_binding), diff --git a/packages/udp-tracker-server/src/server/processor.rs b/packages/udp-tracker-server/src/server/processor.rs index 5e98b0361..297919bc3 100644 --- a/packages/udp-tracker-server/src/server/processor.rs +++ b/packages/udp-tracker-server/src/server/processor.rs @@ -115,7 +115,7 @@ impl Processor { } if let Some(udp_server_stats_event_sender) = - self.udp_tracker_server_container.udp_server_stats_event_sender.as_deref() + self.udp_tracker_server_container.stats_event_sender.as_deref() { udp_server_stats_event_sender .send_event(Event::UdpResponseSent { diff --git a/packages/udp-tracker-server/tests/server/contract.rs b/packages/udp-tracker-server/tests/server/contract.rs index 4cb23621d..860fd1f0b 100644 --- a/packages/udp-tracker-server/tests/server/contract.rs +++ b/packages/udp-tracker-server/tests/server/contract.rs @@ -268,7 +268,7 @@ mod receiving_an_announce_request { let udp_requests_banned_before = env .container .udp_tracker_server_container - .udp_server_stats_repository + .stats_repository .get_stats() .await .udp_requests_banned; @@ -284,7 +284,7 @@ mod receiving_an_announce_request { let udp_requests_banned_after = env .container .udp_tracker_server_container - .udp_server_stats_repository + .stats_repository .get_stats() .await .udp_requests_banned; diff --git a/src/app.rs b/src/app.rs index 67380d30d..41d8b67d1 100644 --- a/src/app.rs +++ b/src/app.rs @@ -111,10 +111,7 @@ async fn load_whitelisted_torrents(config: &Configuration, app_container: &Arc) { if config.core.tracker_usage_statistics { - let _job = app_container - .http_tracker_core_services - .http_stats_keeper - .run_event_listener(); + let _job = app_container.http_tracker_core_services.stats_keeper.run_event_listener(); // todo: this cannot be enabled otherwise the application never ends // because the event listener never stops. You see this console message @@ -148,10 +145,7 @@ fn start_udp_core_event_listener(config: &Configuration, app_container: &Arc) { if config.core.tracker_usage_statistics { - let _job = app_container - .udp_tracker_server_container - .udp_server_stats_keeper - .run_event_listener(); + let _job = app_container.udp_tracker_server_container.stats_keeper.run_event_listener(); // todo: this cannot be enabled otherwise the application never ends // because the event listener never stops. You see this console message diff --git a/src/container.rs b/src/container.rs index 537be2605..93f1fb4d7 100644 --- a/src/container.rs +++ b/src/container.rs @@ -130,10 +130,10 @@ impl AppContainer { TrackerHttpApiCoreContainer { tracker_core_container: self.tracker_core_container.clone(), http_api_config: http_api_config.clone(), - ban_service: self.udp_tracker_core_services.udp_ban_service.clone(), - http_stats_repository: self.http_tracker_core_services.http_stats_repository.clone(), - udp_core_stats_repository: self.udp_tracker_core_services.udp_core_stats_repository.clone(), - udp_server_stats_repository: self.udp_tracker_server_container.udp_server_stats_repository.clone(), + ban_service: self.udp_tracker_core_services.ban_service.clone(), + http_stats_repository: self.http_tracker_core_services.stats_repository.clone(), + udp_core_stats_repository: self.udp_tracker_core_services.stats_repository.clone(), + udp_server_stats_repository: self.udp_tracker_server_container.stats_repository.clone(), } .into() }