diff --git a/src/lib.rs b/src/lib.rs index 0d300da..87d15a5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,3 +16,5 @@ pub mod domain; pub mod services; pub mod utils; + +pub mod rtc; diff --git a/src/routes/mod.rs b/src/routes/mod.rs index 5327d05..cce075b 100644 --- a/src/routes/mod.rs +++ b/src/routes/mod.rs @@ -15,6 +15,7 @@ pub mod gateway; pub mod message; pub mod openapi; pub mod role; +pub mod rtc; pub mod server; pub mod server_item_order; pub mod user; @@ -44,7 +45,8 @@ pub fn router() -> OxRouter { let ws_routes = Router::new() .merge(gateway::routes::router()) - .merge(voice::routes::router()); + .merge(voice::routes::router()) + .merge(rtc::routes::router()); Router::new() .nest("/api", api_routes) diff --git a/src/routes/rtc/handlers.rs b/src/routes/rtc/handlers.rs new file mode 100644 index 0000000..b337e27 --- /dev/null +++ b/src/routes/rtc/handlers.rs @@ -0,0 +1,13 @@ +use crate::core::AppState; +use crate::http::context::CurrentUser; +use crate::rtc::ws_entrypoint::ws_entrypoint_handler; +use axum::extract::{State, WebSocketUpgrade}; +use axum::response::IntoResponse; + +pub async fn ws_handler( + ws: WebSocketUpgrade, + State(state): State, + CurrentUser(user): CurrentUser, +) -> impl IntoResponse { + ws.on_upgrade(move |socket| ws_entrypoint_handler(socket, state, user)) +} diff --git a/src/routes/rtc/mod.rs b/src/routes/rtc/mod.rs new file mode 100644 index 0000000..76b7f5c --- /dev/null +++ b/src/routes/rtc/mod.rs @@ -0,0 +1,2 @@ +mod handlers; +pub mod routes; diff --git a/src/routes/rtc/routes.rs b/src/routes/rtc/routes.rs new file mode 100644 index 0000000..cfc4175 --- /dev/null +++ b/src/routes/rtc/routes.rs @@ -0,0 +1,8 @@ +use super::handlers; +use crate::core::AppState; +use axum::Router; +use axum::routing::get; + +pub fn router() -> Router { + Router::new().route("/rtc", get(handlers::ws_handler)) +} diff --git a/src/routes/voice/README.md b/src/routes/voice/README.md new file mode 100644 index 0000000..a54301b --- /dev/null +++ b/src/routes/voice/README.md @@ -0,0 +1,2 @@ +Recréer ce module de 0, afin de le comprendre à 100% (dans le dossier RTC) + diff --git a/src/rtc/metrics.rs b/src/rtc/metrics.rs new file mode 100644 index 0000000..8eb1d33 --- /dev/null +++ b/src/rtc/metrics.rs @@ -0,0 +1,178 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use crate::metrics::{Metrics, MetricsSnapshot}; + +/// Compteurs atomiques pour les métriques de la voix / WebRTC. +#[derive(Debug, Default)] +pub struct VoiceMetrics { + /// Nombre total de datagrammes / paquets reçus. + pub packets_received: AtomicU64, + /// Volume total d'octets reçus. + pub bytes_received: AtomicU64, + /// Nombre total de datagrammes / paquets retransmis. + pub packets_sent: AtomicU64, + /// Volume total d'octets retransmis. + pub bytes_sent: AtomicU64, + /// Paquets ignorés ou rejetés. + pub packets_dropped: AtomicU64, + /// Nombre d'erreurs d'émission. + pub send_errors: AtomicU64, + /// Nombre d'erreurs de réception. + pub recv_errors: AtomicU64, +} + +impl VoiceMetrics { + /// Crée un jeu de métriques vide enveloppé dans un [`Arc`]. + pub fn new() -> Arc { + Arc::new(Self::default()) + } + + /// Enregistre la réception d'un paquet de `bytes` octets. + #[inline] + pub fn inc_received(&self, bytes: u64) { + self.packets_received.fetch_add(1, Ordering::Relaxed); + self.bytes_received.fetch_add(bytes, Ordering::Relaxed); + } + + /// Enregistre l'émission d'un paquet de `bytes` octets. + #[inline] + pub fn inc_sent(&self, bytes: u64) { + self.packets_sent.fetch_add(1, Ordering::Relaxed); + self.bytes_sent.fetch_add(bytes, Ordering::Relaxed); + } + + /// Enregistre un paquet ignoré. + #[inline] + pub fn inc_dropped(&self) { + self.packets_dropped.fetch_add(1, Ordering::Relaxed); + } + + /// Enregistre un échec d'émission non fatal. + #[inline] + pub fn inc_send_error(&self) { + self.send_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Enregistre un échec de réception. + #[inline] + pub fn inc_recv_error(&self) { + self.recv_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Prend un instantané cohérent de tous les compteurs. + pub fn snapshot(&self) -> VoiceMetricsSnapshot { + VoiceMetricsSnapshot { + taken_at: Instant::now(), + packets_received: self.packets_received.load(Ordering::Relaxed), + bytes_received: self.bytes_received.load(Ordering::Relaxed), + packets_sent: self.packets_sent.load(Ordering::Relaxed), + bytes_sent: self.bytes_sent.load(Ordering::Relaxed), + packets_dropped: self.packets_dropped.load(Ordering::Relaxed), + send_errors: self.send_errors.load(Ordering::Relaxed), + recv_errors: self.recv_errors.load(Ordering::Relaxed), + } + } +} + +impl Metrics for VoiceMetrics { + type Snapshot = VoiceMetricsSnapshot; + + fn snapshot(&self) -> VoiceMetricsSnapshot { + self.snapshot() + } +} + +/// Lecture cohérente de l'ensemble des compteurs à un instant T. +#[derive(Debug, Clone, Copy)] +pub struct VoiceMetricsSnapshot { + pub taken_at: Instant, + pub packets_received: u64, + pub bytes_received: u64, + pub packets_sent: u64, + pub bytes_sent: u64, + pub packets_dropped: u64, + pub send_errors: u64, + pub recv_errors: u64, +} + +impl VoiceMetricsSnapshot { + /// Calcule les taux moyens par seconde depuis un snapshot précédent. + pub fn rates_since(&self, previous: &Self) -> VoiceRates { + let secs = self + .taken_at + .duration_since(previous.taken_at) + .as_secs_f64() + .max(f64::EPSILON); + + VoiceRates { + packets_received_per_sec: self + .packets_received + .saturating_sub(previous.packets_received) + as f64 + / secs, + bytes_received_per_sec: self.bytes_received.saturating_sub(previous.bytes_received) + as f64 + / secs, + packets_sent_per_sec: self.packets_sent.saturating_sub(previous.packets_sent) as f64 + / secs, + bytes_sent_per_sec: self.bytes_sent.saturating_sub(previous.bytes_sent) as f64 / secs, + packets_dropped_per_sec: self + .packets_dropped + .saturating_sub(previous.packets_dropped) + as f64 + / secs, + } + } +} + +impl MetricsSnapshot for VoiceMetricsSnapshot { + fn taken_at(&self) -> Instant { + self.taken_at + } +} + +/// Taux moyens par seconde calculés entre deux [`VoiceMetricsSnapshot`]. +#[derive(Debug, Clone, Copy)] +pub struct VoiceRates { + pub packets_received_per_sec: f64, + pub bytes_received_per_sec: f64, + pub packets_sent_per_sec: f64, + pub bytes_sent_per_sec: f64, + pub packets_dropped_per_sec: f64, +} + +pub fn spawn_reporter(metrics: Arc, interval: Duration) { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + ticker.tick().await; + + let mut prev_snapshot = metrics.snapshot(); + + loop { + ticker.tick().await; + + let current = metrics.snapshot(); + let rates = current.rates_since(&prev_snapshot); + + tracing::info!( + pkts_rx = current.packets_received, + bytes_rx = current.bytes_received, + pkts_tx = current.packets_sent, + bytes_tx = current.bytes_sent, + pkts_dropped = current.packets_dropped, + send_errors = current.send_errors, + recv_errors = current.recv_errors, + pkts_rx_s = format!("{:.1}", rates.packets_received_per_sec), + bytes_rx_s = format!("{:.0}", rates.bytes_received_per_sec), + pkts_tx_s = format!("{:.1}", rates.packets_sent_per_sec), + bytes_tx_s = format!("{:.0}", rates.bytes_sent_per_sec), + pkts_dropped_s = format!("{:.1}", rates.packets_dropped_per_sec), + "Voice metrics" + ); + + prev_snapshot = current; + } + }); +} diff --git a/src/rtc/mod.rs b/src/rtc/mod.rs new file mode 100644 index 0000000..4676b4f --- /dev/null +++ b/src/rtc/mod.rs @@ -0,0 +1,22 @@ +mod metrics; +pub mod ws_entrypoint; + +use crate::config::NetworkConfig; +use metrics::VoiceMetrics; +use rustrtc::{RtcConfiguration, RtcConfigurationBuilder}; +use std::sync::Arc; + +pub struct RTCManager { + pub config: RtcConfiguration, +} + +impl RTCManager { + pub fn new(network: &NetworkConfig, metrics: Arc) -> Self { + let builder = RtcConfigurationBuilder::new() + .ice_udp_mux(true) + .ice_udp_mux_port(network.udp_port) + .bind_ip(network.host.to_string()); + let rtc_config = builder.build(); + Self { config: rtc_config } + } +} diff --git a/src/rtc/ws_entrypoint.rs b/src/rtc/ws_entrypoint.rs new file mode 100644 index 0000000..56138d7 --- /dev/null +++ b/src/rtc/ws_entrypoint.rs @@ -0,0 +1,7 @@ +// This is the first point when WebRTC ask for a connection + +use crate::core::AppState; +use crate::models::user; +use axum::extract::ws::WebSocket; + +pub async fn ws_entrypoint_handler(socket: WebSocket, state: AppState, user: user::Model) {} diff --git a/src/voice/README.md b/src/voice/README.md new file mode 100644 index 0000000..a54301b --- /dev/null +++ b/src/voice/README.md @@ -0,0 +1,2 @@ +Recréer ce module de 0, afin de le comprendre à 100% (dans le dossier RTC) +