dev: stream impl for processing udp requests

.cargo/config.toml

@@ -23,4 +23,3 @@ rustflags = [
     "-D",
     "unused",
 ]
-

Cargo.toml

@@ -63,13 +63,15 @@ r2d2_sqlite = { version = "0", features = ["bundled"] }
 rand = "0"
 reqwest = { version = "0", features = ["json"] }
 ringbuf = "0"
+rstest = "0"
 serde = { version = "1", features = ["derive"] }
 serde_bencode = "0"
 serde_bytes = "0"
 serde_json = { version = "1", features = ["preserve_order"] }
 serde_repr = "0"
 thiserror = "1"
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio-stream = "0"
 torrust-tracker-clock = { version = "3.0.0-alpha.12-develop", path = "packages/clock" }
 torrust-tracker-configuration = { version = "3.0.0-alpha.12-develop", path = "packages/configuration" }
 torrust-tracker-contrib-bencode = { version = "3.0.0-alpha.12-develop", path = "contrib/bencode" }
@@ -80,6 +82,7 @@ tower = { version = "0.4.13", features = ["timeout"] }
 tower-http = { version = "0", features = ["compression-full", "cors", "propagate-header", "request-id", "trace"] }
 trace = "0"
 tracing = "0"
+#tracing-subscriber = { version = "0", features = ["json"] }
 url = "2"
 uuid = { version = "1", features = ["v4"] }
 zerocopy = "0.7.33"

cSpell.json

@@ -34,10 +34,12 @@
         "codecov",
         "codegen",
         "completei",
+        "Condvar",
         "connectionless",
         "Containerfile",
         "conv",
         "curr",
+        "cvar",
         "Cyberneering",
         "dashmap",
         "datagram",

src/servers/udp/handlers.rs

@@ -11,7 +11,6 @@ use aquatic_udp_protocol::{
     ScrapeRequest, ScrapeResponse, TorrentScrapeStatistics, TransactionId,
 };
 use log::debug;
-use tokio::net::UdpSocket;
 use torrust_tracker_located_error::DynError;
 use torrust_tracker_primitives::info_hash::InfoHash;
 use uuid::Uuid;
@@ -35,13 +34,12 @@ use crate::shared::bit_torrent::common::MAX_SCRAPE_TORRENTS;
 /// type.
 ///
 /// It will return an `Error` response if the request is invalid.
-pub(crate) async fn handle_packet(udp_request: UdpRequest, tracker: &Arc<Tracker>, socket: Arc<UdpSocket>) -> Response {
+pub(crate) async fn handle_packet(udp_request: UdpRequest, tracker: &Arc<Tracker>, addr: SocketAddr) -> Response {
     debug!("Handling Packets: {udp_request:?}");
 
     let start_time = Instant::now();
 
     let request_id = RequestId::make(&udp_request);
-    let server_socket_addr = socket.local_addr().expect("Could not get local_addr for socket.");
 
     match Request::parse_bytes(&udp_request.payload[..udp_request.payload.len()], MAX_SCRAPE_TORRENTS).map_err(|e| {
         Error::InternalServer {
@@ -50,7 +48,7 @@ pub(crate) async fn handle_packet(udp_request: UdpRequest, tracker: &Arc<Tracker
         }
     }) {
         Ok(request) => {
-            log_request(&request, &request_id, &server_socket_addr);
+            log_request(&request, &request_id, &addr);
 
             let transaction_id = match &request {
                 Request::Connect(connect_request) => connect_request.transaction_id,
@@ -65,7 +63,7 @@ pub(crate) async fn handle_packet(udp_request: UdpRequest, tracker: &Arc<Tracker
 
             let latency = start_time.elapsed();
 
-            log_response(&response, &transaction_id, &request_id, &server_socket_addr, latency);
+            log_response(&response, &transaction_id, &request_id, &addr, latency);
 
             response
         }