refactor: [torrust#1319] add UDP server events

We will splot UDP stats events into:

- UDP core events
- UDP server event

This is step 1 in the refactor:

- Step 1. Create UDP server events.
- Step 2. Remove UDP server events from core events.

Author: josecelano

packages/udp-tracker-server/src/lib.rs

@@ -638,6 +638,7 @@ pub mod environment;
 pub mod error;
 pub mod handlers;
 pub mod server;
+pub mod statistics;
 
 use std::net::SocketAddr;
 

packages/udp-tracker-server/src/statistics/event/handler.rs

@@ -0,0 +1,184 @@
+use crate::statistics::event::{Event, UdpResponseKind};
+use crate::statistics::repository::Repository;
+
+pub async fn handle_event(event: Event, stats_repository: &Repository) {
+    match event {
+        // UDP
+        Event::UdpRequestAborted => {
+            stats_repository.increase_udp_requests_aborted().await;
+        }
+        Event::UdpRequestBanned => {
+            stats_repository.increase_udp_requests_banned().await;
+        }
+
+        // UDP4
+        Event::Udp4Request { kind } => {
+            stats_repository.increase_udp4_requests().await;
+            match kind {
+                UdpResponseKind::Connect => {
+                    stats_repository.increase_udp4_connections().await;
+                }
+                UdpResponseKind::Announce => {
+                    stats_repository.increase_udp4_announces().await;
+                }
+                UdpResponseKind::Scrape => {
+                    stats_repository.increase_udp4_scrapes().await;
+                }
+                UdpResponseKind::Error => {}
+            }
+        }
+        Event::Udp4Response {
+            kind,
+            req_processing_time,
+        } => {
+            stats_repository.increase_udp4_responses().await;
+
+            match kind {
+                UdpResponseKind::Connect => {
+                    stats_repository
+                        .recalculate_udp_avg_connect_processing_time_ns(req_processing_time)
+                        .await;
+                }
+                UdpResponseKind::Announce => {
+                    stats_repository
+                        .recalculate_udp_avg_announce_processing_time_ns(req_processing_time)
+                        .await;
+                }
+                UdpResponseKind::Scrape => {
+                    stats_repository
+                        .recalculate_udp_avg_scrape_processing_time_ns(req_processing_time)
+                        .await;
+                }
+                UdpResponseKind::Error => {}
+            }
+        }
+        Event::Udp4Error => {
+            stats_repository.increase_udp4_errors().await;
+        }
+
+        // UDP6
+        Event::Udp6Request => {
+            stats_repository.increase_udp6_requests().await;
+        }
+        Event::Udp6Response {
+            kind: _,
+            req_processing_time: _,
+        } => {
+            stats_repository.increase_udp6_responses().await;
+        }
+        Event::Udp6Error => {
+            stats_repository.increase_udp6_errors().await;
+        }
+    }
+
+    tracing::debug!("stats: {:?}", stats_repository.get_stats().await);
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::statistics::event::handler::handle_event;
+    use crate::statistics::event::{Event, UdpResponseKind};
+    use crate::statistics::repository::Repository;
+
+    #[tokio::test]
+    async fn should_increase_the_udp_abort_counter_when_it_receives_a_udp_abort_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::UdpRequestAborted, &stats_repository).await;
+        let stats = stats_repository.get_stats().await;
+        assert_eq!(stats.udp_requests_aborted, 1);
+    }
+    #[tokio::test]
+    async fn should_increase_the_udp_ban_counter_when_it_receives_a_udp_banned_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::UdpRequestBanned, &stats_repository).await;
+        let stats = stats_repository.get_stats().await;
+        assert_eq!(stats.udp_requests_banned, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_requests_counter_when_it_receives_a_udp4_request_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(
+            Event::Udp4Request {
+                kind: UdpResponseKind::Connect,
+            },
+            &stats_repository,
+        )
+        .await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_requests, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_responses_counter_when_it_receives_a_udp4_response_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(
+            Event::Udp4Response {
+                kind: crate::statistics::event::UdpResponseKind::Announce,
+                req_processing_time: std::time::Duration::from_secs(1),
+            },
+            &stats_repository,
+        )
+        .await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_responses, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp4_errors_counter_when_it_receives_a_udp4_error_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp4Error, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp4_errors_handled, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp6_requests_counter_when_it_receives_a_udp6_request_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp6Request, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp6_requests, 1);
+    }
+
+    #[tokio::test]
+    async fn should_increase_the_udp6_response_counter_when_it_receives_a_udp6_response_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(
+            Event::Udp6Response {
+                kind: crate::statistics::event::UdpResponseKind::Announce,
+                req_processing_time: std::time::Duration::from_secs(1),
+            },
+            &stats_repository,
+        )
+        .await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp6_responses, 1);
+    }
+    #[tokio::test]
+    async fn should_increase_the_udp6_errors_counter_when_it_receives_a_udp6_error_event() {
+        let stats_repository = Repository::new();
+
+        handle_event(Event::Udp6Error, &stats_repository).await;
+
+        let stats = stats_repository.get_stats().await;
+
+        assert_eq!(stats.udp6_errors_handled, 1);
+    }
+}

packages/udp-tracker-server/src/statistics/event/listener.rs

@@ -0,0 +1,11 @@
+use tokio::sync::mpsc;
+
+use super::handler::handle_event;
+use super::Event;
+use crate::statistics::repository::Repository;
+
+pub async fn dispatch_events(mut receiver: mpsc::Receiver<Event>, stats_repository: Repository) {
+    while let Some(event) = receiver.recv().await {
+        handle_event(event, &stats_repository).await;
+    }
+}

packages/udp-tracker-server/src/statistics/event/mod.rs

@@ -0,0 +1,43 @@
+use std::time::Duration;
+
+pub mod handler;
+pub mod listener;
+pub mod sender;
+
+/// An statistics event. It is used to collect tracker metrics.
+///
+/// - `Tcp` prefix means the event was triggered by the HTTP tracker
+/// - `Udp` prefix means the event was triggered by the UDP tracker
+/// - `4` or `6` prefixes means the IP version used by the peer
+/// - Finally the event suffix is the type of request: `announce`, `scrape` or `connection`
+///
+/// > NOTE: HTTP trackers do not use `connection` requests.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Event {
+    // code-review: consider one single event for request type with data: Event::Announce { scheme: HTTPorUDP, ip_version: V4orV6 }
+    // Attributes are enums too.
+    UdpRequestAborted,
+    UdpRequestBanned,
+    Udp4Request {
+        kind: UdpResponseKind,
+    },
+    Udp4Response {
+        kind: UdpResponseKind,
+        req_processing_time: Duration,
+    },
+    Udp4Error,
+    Udp6Request,
+    Udp6Response {
+        kind: UdpResponseKind,
+        req_processing_time: Duration,
+    },
+    Udp6Error,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum UdpResponseKind {
+    Connect,
+    Announce,
+    Scrape,
+    Error,
+}

packages/udp-tracker-server/src/statistics/event/sender.rs

@@ -0,0 +1,29 @@
+use futures::future::BoxFuture;
+use futures::FutureExt;
+#[cfg(test)]
+use mockall::{automock, predicate::str};
+use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::SendError;
+
+use super::Event;
+
+/// A trait to allow sending statistics events
+#[cfg_attr(test, automock)]
+pub trait Sender: Sync + Send {
+    fn send_event(&self, event: Event) -> BoxFuture<'_, Option<Result<(), SendError<Event>>>>;
+}
+
+/// An [`statistics::EventSender`](crate::statistics::event::sender::Sender) implementation.
+///
+/// It uses a channel sender to send the statistic events. The channel is created by a
+/// [`statistics::Keeper`](crate::statistics::keeper::Keeper)
+#[allow(clippy::module_name_repetitions)]
+pub struct ChannelSender {
+    pub(crate) sender: mpsc::Sender<Event>,
+}
+
+impl Sender for ChannelSender {
+    fn send_event(&self, event: Event) -> BoxFuture<'_, Option<Result<(), SendError<Event>>>> {
+        async move { Some(self.sender.send(event).await) }.boxed()
+    }
+}

packages/udp-tracker-server/src/statistics/keeper.rs

@@ -0,0 +1,81 @@
+use tokio::sync::mpsc;
+
+use super::event::listener::dispatch_events;
+use super::event::sender::{ChannelSender, Sender};
+use super::event::Event;
+use super::repository::Repository;
+
+const CHANNEL_BUFFER_SIZE: usize = 65_535;
+
+/// 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,
+}
+
+impl Default for Keeper {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Keeper {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            repository: Repository::new(),
+        }
+    }
+
+    #[must_use]
+    pub fn new_active_instance() -> (Box<dyn Sender>, Repository) {
+        let mut stats_tracker = Self::new();
+
+        let stats_event_sender = stats_tracker.run_event_listener();
+
+        (stats_event_sender, stats_tracker.repository)
+    }
+
+    pub fn run_event_listener(&mut self) -> Box<dyn Sender> {
+        let (sender, receiver) = mpsc::channel::<Event>(CHANNEL_BUFFER_SIZE);
+
+        let stats_repository = self.repository.clone();
+
+        tokio::spawn(async move { dispatch_events(receiver, stats_repository).await });
+
+        Box::new(ChannelSender { sender })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::statistics::event::{Event, UdpResponseKind};
+    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 = stats_tracker.repository.get_stats().await;
+
+        assert_eq!(stats.udp4_requests, Metrics::default().udp4_requests);
+    }
+
+    #[tokio::test]
+    async fn should_create_an_event_sender_to_send_statistical_events() {
+        let mut stats_tracker = Keeper::new();
+
+        let event_sender = stats_tracker.run_event_listener();
+
+        let result = event_sender
+            .send_event(Event::Udp4Request {
+                kind: UdpResponseKind::Connect,
+            })
+            .await;
+
+        assert!(result.is_some());
+    }
+}