post webrtc intégration
This commit is contained in:
@@ -0,0 +1,140 @@
|
|||||||
|
---
|
||||||
|
sessionId: session-260912-132834-fuse
|
||||||
|
---
|
||||||
|
|
||||||
|
# Requirements
|
||||||
|
|
||||||
|
### Overview & Goals
|
||||||
|
Replace the current placeholder raw-UDP voice relay (`src/udp/*`) with a real WebRTC media stack built on the `rustrtc` crate (already declared in `Cargo.toml`). The server becomes a centralized **SFU** (Selective Forwarding Unit): each connected client opens **one `PeerConnection`** with the server, the server decrypts/receives that client's audio (and later video) track and forwards it to every other member of the same voice `Channel`.
|
||||||
|
|
||||||
|
### Scope
|
||||||
|
**In Scope**
|
||||||
|
- Remove `src/udp/server.rs`, `src/udp/router.rs` raw-socket logic (metrics module is kept/adapted).
|
||||||
|
- Add a `voice` module hosting: `PeerConnection` lifecycle management via `rustrtc`, SDP offer/answer exchange, ICE candidate exchange, and SFU-style track forwarding scoped by `channel_id`.
|
||||||
|
- Reuse the existing WebSocket gateway (`GatewayManager` / `RealtimeRouter` / `on_message` in `src/routes/gateway/mod.rs`) as the signaling transport — add a `Voice` namespace to the existing JSON event envelope (`GatewayEvent { namespace, action, content }`) instead of introducing a new endpoint.
|
||||||
|
- Reuse `channel::ChannelType::Voice`, `channel_user`, and the `JOIN_VOICE` / `SPEAK` permission bits already defined in `src/permissions.rs` to authorize who may join/publish in a voice channel.
|
||||||
|
- Update `config.toml` / `NetworkConfig` so the single `udp_port` continues to be the one UDP port used, bound through `rustrtc`'s `ice_udp_mux` (single-port ICE) instead of a raw `UdpSocket`.
|
||||||
|
- Update `AppMetrics`/`core::mod::App::run` wiring so the new voice service starts/stops alongside the HTTP server, replacing `UdpServer`.
|
||||||
|
|
||||||
|
**Out of Scope**
|
||||||
|
- TURN/STUN server configuration (kept minimal for LAN/local network scenarios for now).
|
||||||
|
- Video/screen-share specific handling (structure will allow it later, but only audio forwarding is implemented now).
|
||||||
|
- Client-side (frontend) implementation details beyond the signaling contract needed by the server.
|
||||||
|
|
||||||
|
### User Stories
|
||||||
|
- As a user with `JOIN_VOICE` permission on a voice channel, I want to connect and have my microphone audio heard by other members of that same channel.
|
||||||
|
- As a user already in a voice channel, I want to hear every other member's audio forwarded by the server without opening a direct connection to each peer.
|
||||||
|
- As a server operator, I want the previous ad-hoc UDP relay removed and replaced by a maintained WebRTC stack (rustrtc) so voice traffic is encrypted (SRTP) and NAT-traversal (ICE) works out of the box.
|
||||||
|
|
||||||
|
### Functional Requirements
|
||||||
|
- A client sends an SDP offer through the gateway (`namespace: "Voice"`), scoped to a `channel_id`; the server validates channel membership/permissions, creates a `rustrtc::PeerConnection`, and returns an SDP answer through the same gateway connection.
|
||||||
|
- ICE candidates generated by either side are exchanged as additional `Voice` gateway events.
|
||||||
|
- Once connected, the server subscribes the client's inbound audio track to every other `PeerConnection` currently joined to the same voice channel (SFU fan-out), and adds new joiners' tracks to previously-connected peers as they arrive.
|
||||||
|
- When a user disconnects (gateway socket closes) or leaves the voice channel, their `PeerConnection` is closed and their track is removed from all other peers' forwarding sets.
|
||||||
|
- Existing `UdpMetrics`-style counters (packets/bytes in/out, errors) are preserved in spirit, adapted for voice metrics and reported by the periodic `reporter`.
|
||||||
|
|
||||||
|
|
||||||
|
# Technical Design
|
||||||
|
|
||||||
|
### Current Implementation
|
||||||
|
- `src/udp/server.rs`: binds a raw `tokio::net::UdpSocket` on `NetworkConfig.udp_port`, loops on `recv_from`, and blindly forwards datagrams to registered peers under a hard-coded `"default"` channel without protocol isolation or encryption.
|
||||||
|
- `src/udp/router.rs`: `RoutingTable` maps `ChannelId -> Vec<SocketAddr>`; dead placeholder API.
|
||||||
|
- `src/udp/metrics.rs`: atomic counters (`packets_received`, `bytes_sent`, etc.) + `spawn_reporter` logging every interval, plugged into `AppMetrics.udp` (`src/metrics/mod.rs`).
|
||||||
|
- `src/core/mod.rs`: `App::build`/`App::run` constructs `UdpServer::new(&config.network, udp_metrics)` and `tokio::spawn(udp_server.run())`, joined into shutdown-broadcast pattern.
|
||||||
|
- `src/routes/gateway/mod.rs`: `GatewayManager` tracks client WebSocket connections; `GatewayClient::on_message` handles incoming messages and acts as the extension point for voice signaling.
|
||||||
|
- `src/services/realtime_registry.rs`: maintains `channel_id -> HashSet<user_id>` membership computed from `computed_permission` (READ_CHANNEL).
|
||||||
|
- `src/permissions.rs`: `JOIN_VOICE`, `SPEAK`, `STREAM`, `MOVE_OTHERS`, `DISCONNECT_OTHERS`, `MANAGE_VOICE_CHANNEL` bits already defined.
|
||||||
|
- `src/models/channel.rs`: `ChannelType::Voice` variant already exists.
|
||||||
|
|
||||||
|
### Key Decisions
|
||||||
|
1. **Topology: centralized SFU** — one `rustrtc::PeerConnection` per connected client per voice channel; the server forwards each publisher's RTP track to every other subscriber in that channel. No direct peer-to-peer mesh.
|
||||||
|
2. **Signaling transport: existing WebSocket gateway** — SDP offer/answer and ICE candidates are carried as `Voice` namespace events inside the current `GatewayEvent` envelope, handled in `GatewayClient::on_message` and dispatched to `VoiceService`.
|
||||||
|
3. **Single UDP port reuse** — `rustrtc`'s `RtcConfiguration` is set up with `ice_udp_mux = true` and bound to `NetworkConfig.udp_port`, preserving existing firewall/network port configurations.
|
||||||
|
4. **No TURN/STUN for now** — `RtcConfiguration.ice_servers` left default/empty for local and direct network setups.
|
||||||
|
5. **Voice presence tracked separately from `RealtimeRegistry`** — an in-memory registry (`VoiceRoom` per `channel_id`) tracks active WebRTC connections independently from general WebSocket channel presence.
|
||||||
|
|
||||||
|
### Proposed Changes
|
||||||
|
- Replace `src/udp` module with `src/voice`:
|
||||||
|
- `voice/mod.rs`: public API, exports `VoiceService`.
|
||||||
|
- `voice/service.rs`: `VoiceService` owns `rustrtc::RtcConfiguration`, a map `channel_id -> VoiceRoom`, and handler methods for offer, ICE candidates, and channel leaves.
|
||||||
|
- `voice/room.rs`: `VoiceRoom` holds `HashMap<user_id, Arc<PeerConnection>>` for a channel and orchestrates track fan-out across peers.
|
||||||
|
- `voice/metrics.rs`: tracks voice counters (packets/bytes/errors), exposed through `AppMetrics.voice` and the existing reporter cadence.
|
||||||
|
- Extend `GatewayEvent` handling: add `"Voice"` namespace with actions `offer`, `answer`, `ice-candidate`, `leave`.
|
||||||
|
- Permission check on `offer`: verify `ChannelPermission::JOIN_VOICE` and `SPEAK` before initializing a `PeerConnection`.
|
||||||
|
- Update `src/core/mod.rs`: replace `UdpServer` initialization with `VoiceService` held inside `AppState`.
|
||||||
|
- Update `src/config.rs`: keep `NetworkConfig.udp_port` feeding `rustrtc`'s `ice_udp_mux_port`.
|
||||||
|
|
||||||
|
### Data Models / Contracts
|
||||||
|
```rust
|
||||||
|
// Gateway JSON contract additions (namespace = "Voice")
|
||||||
|
{ "namespace": "Voice", "action": "offer", "content": { "channel_id": Uuid, "sdp": String } }
|
||||||
|
{ "namespace": "Voice", "action": "answer", "content": { "channel_id": Uuid, "sdp": String } }
|
||||||
|
{ "namespace": "Voice", "action": "ice-candidate", "content": { "channel_id": Uuid, "candidate": String } }
|
||||||
|
{ "namespace": "Voice", "action": "leave", "content": { "channel_id": Uuid } }
|
||||||
|
```
|
||||||
|
|
||||||
|
```rust
|
||||||
|
pub struct VoiceService {
|
||||||
|
config: rustrtc::RtcConfiguration,
|
||||||
|
rooms: RwLock<HashMap<Uuid, VoiceRoom>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl VoiceService {
|
||||||
|
pub async fn handle_offer(&self, user_id: Uuid, channel_id: Uuid, sdp: String) -> anyhow::Result<String>;
|
||||||
|
pub async fn handle_ice_candidate(&self, user_id: Uuid, channel_id: Uuid, candidate: String) -> anyhow::Result<()>;
|
||||||
|
pub fn leave(&self, user_id: Uuid, channel_id: Uuid);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Architecture Diagram
|
||||||
|
```mermaid
|
||||||
|
graph TD
|
||||||
|
ClientA[Client A] -- WebSocket: Voice offer/answer/ICE --> Gateway[GatewayManager / on_message]
|
||||||
|
ClientB[Client B] -- WebSocket: Voice offer/answer/ICE --> Gateway
|
||||||
|
Gateway -- dispatch --> VoiceService
|
||||||
|
VoiceService -- creates/owns --> RoomA[VoiceRoom per channel_id]
|
||||||
|
RoomA -- PeerConnection A --> PCA[rustrtc PeerConnection A]
|
||||||
|
RoomA -- PeerConnection B --> PCB[rustrtc PeerConnection B]
|
||||||
|
ClientA == RTP/SRTP media on UDP mux port ==> PCA
|
||||||
|
ClientB == RTP/SRTP media on UDP mux port ==> PCB
|
||||||
|
PCA -- forward audio track --> PCB
|
||||||
|
PCB -- forward audio track --> PCA
|
||||||
|
```
|
||||||
|
|
||||||
|
### Risks
|
||||||
|
- `rustrtc` API details: verify track forwarding primitives against `rustrtc 0.3.133` during implementation and encapsulate track re-subscription in `voice/room.rs`.
|
||||||
|
- Permission enforcement: ensure permission checks take place before creating peer connections to prevent unauthorized audio relay.
|
||||||
|
- Signaling message volume: ensure gateway handling remains non-blocking for high frequency ICE candidates.
|
||||||
|
|
||||||
|
|
||||||
|
# Delivery Steps
|
||||||
|
|
||||||
|
### Step 1: Scaffold the voice module and RtcConfiguration from NetworkConfig
|
||||||
|
A new `src/voice` module exists with a `VoiceService` capable of building a `rustrtc::RtcConfiguration` from the app's network config, replacing the old `src/udp` module's role in `core/mod.rs`.
|
||||||
|
- Create `src/voice/mod.rs`, `src/voice/service.rs`, `src/voice/room.rs`, and `src/voice/metrics.rs`.
|
||||||
|
- Port `UdpMetrics`-style counters into `voice/metrics.rs` (packets/bytes in/out, errors), keeping the `Metrics`/`MetricsSnapshot` trait implementations used by `crate::metrics::reporter`.
|
||||||
|
- Build `RtcConfiguration`/`RtcConfigurationBuilder` in `VoiceService::new(&NetworkConfig, metrics)` using `ice_udp_mux = true` and `ice_udp_mux_port = network.udp_port`, with no external ICE servers configured.
|
||||||
|
- Remove `src/udp/server.rs` and `src/udp/router.rs`; delete unused `RoutingTable`.
|
||||||
|
- Update `src/core/mod.rs` (`App::build`/`App::run`) to construct `VoiceService` instead of `UdpServer`, store it on `AppState`, and drop unneeded `udp_shutdown_tx`/`udp_handle` tasks.
|
||||||
|
- Update `src/metrics/mod.rs` (`AppMetrics`) to reference the new voice metrics type.
|
||||||
|
|
||||||
|
### Step 2: Implement per-channel PeerConnection lifecycle in VoiceRoom
|
||||||
|
`VoiceService`/`VoiceRoom` can accept an SDP offer for a given channel, create a `rustrtc::PeerConnection`, and return an SDP answer, tracking connections per `channel_id`.
|
||||||
|
- Implement `VoiceRoom` (`channel_id -> HashMap<user_id, Arc<PeerConnection>>`).
|
||||||
|
- Implement `VoiceService::handle_offer(user_id, channel_id, sdp)`: looks up or creates the room, creates a `PeerConnection` via `rustrtc`, invokes `set_remote_description`/`create_answer`/`set_local_description`, and returns the answer SDP.
|
||||||
|
- Implement `VoiceService::handle_ice_candidate` and `VoiceService::leave`, closing and removing the `PeerConnection` from its room.
|
||||||
|
- Enforce `ChannelPermission::JOIN_VOICE` (and `SPEAK` for publishing) before creating a `PeerConnection`, reusing the `computed_permission` lookup pattern from `src/services/realtime_registry.rs`.
|
||||||
|
|
||||||
|
### Step 3: Wire SFU track forwarding between peers in the same channel
|
||||||
|
Audio published by one connected client in a voice channel is forwarded by the server to every other client connected to the same channel.
|
||||||
|
- On `PeerConnection::on_track` for a given user's connection, register the inbound track on the owning `VoiceRoom`.
|
||||||
|
- For every other `PeerConnection` already in that `VoiceRoom`, add and forward the new track (SFU fan-out) using `rustrtc`'s track/`MediaCapabilities` APIs.
|
||||||
|
- When a new peer joins an existing room, subscribe it to all tracks already being forwarded by other room members.
|
||||||
|
- On `leave` or disconnect, remove the peer's published track from all other peers' forwarding sets and close its `PeerConnection`.
|
||||||
|
|
||||||
|
### Step 4: Expose Voice signaling through the existing WebSocket gateway
|
||||||
|
Clients can perform the full offer/answer/ICE handshake over the existing gateway WebSocket connection using a new `Voice` namespace, with no new HTTP/WS endpoint introduced.
|
||||||
|
- Extend `GatewayClient::on_message` in `src/routes/gateway/mod.rs` to parse `GatewayEvent{namespace: "Voice", action, content}` messages (`offer`, `answer`, `ice-candidate`, `leave`).
|
||||||
|
- Dispatch parsed messages to `VoiceService` (accessed via `AppState`) and send the resulting answer/ICE-candidate events back through the client's existing `mpsc::UnboundedSender<Message>`.
|
||||||
|
- On gateway disconnect (`GatewayClient::on_disconnect`), call `VoiceService::leave` for any channel the user was actively connected to in voice.
|
||||||
|
- Update `config.toml`'s `DEFAULT_CONFIG_TOML` comment for `udp_port` to reflect its new role as the rustrtc ICE/media mux port.
|
||||||
@@ -0,0 +1,134 @@
|
|||||||
|
---
|
||||||
|
sessionId: session-260912-132945-1get
|
||||||
|
---
|
||||||
|
|
||||||
|
# Requirements
|
||||||
|
|
||||||
|
### Overview & Goals
|
||||||
|
Separate WebRTC voice signaling from the main real-time chat gateway by introducing a **dedicated Voice WebSocket endpoint** (`/ws/voice`). The voice signaling protocol (SDP offer/answer exchange, ICE candidates, room joins/leaves) is isolated from chat events, mirroring the architecture of production communication platforms (e.g. Discord's separate Gateway and Voice Gateway).
|
||||||
|
|
||||||
|
### Scope
|
||||||
|
**In Scope**
|
||||||
|
- Implement a dedicated WebSocket route `GET /ws/voice` (nested alongside `/ws/gateway` under `ws_routes`).
|
||||||
|
- Authenticate incoming voice WebSocket connections using the existing `CurrentUser` extractor (supporting Authorization headers, cookies, and `?token=` query parameters).
|
||||||
|
- Define and implement a lightweight, dedicated JSON signaling protocol for voice sessions (`offer`, `answer`, `ice-candidate`, `leave`, `error`).
|
||||||
|
- Bind the voice connection lifecycle directly to the WebSocket lifecycle: connecting/joining initiates the session; closing the socket cleanly tears down the `PeerConnection` and removes the participant from the `VoiceRoom`.
|
||||||
|
- Remove voice signaling handlers (`namespace: "Voice"`) from the general chat gateway (`src/routes/gateway/mod.rs`), restoring clean separation of concerns.
|
||||||
|
- Maintain full compatibility with `VoiceService` and `rustrtc` SFU track fan-out.
|
||||||
|
|
||||||
|
**Out of Scope**
|
||||||
|
- Changing the underlying `rustrtc` SFU forwarding engine or UDP multiplexing port (`udp_port`).
|
||||||
|
- Implementing TURN/STUN relay configurations.
|
||||||
|
- Video/screen-sharing UI controls (handled in future iterations).
|
||||||
|
|
||||||
|
### User Stories
|
||||||
|
- As a client joining a voice channel, I want to open a dedicated WebSocket connection for voice so that media signaling traffic (large SDP payloads, ICE candidates) does not delay or congest my chat message stream.
|
||||||
|
- As a client leaving a voice channel, I want closing my voice WebSocket connection to immediately and cleanly release all server WebRTC resources without affecting my general gateway connection.
|
||||||
|
- As a backend developer, I want voice signaling isolated in its own module/handler so the main gateway router remains focused strictly on application domain events.
|
||||||
|
|
||||||
|
### Functional Requirements
|
||||||
|
- **Dedicated Route**: `GET /ws/voice` upgrades to a WebSocket connection authenticated via `CurrentUser`.
|
||||||
|
- **Signaling Exchange**:
|
||||||
|
- Client sends `offer` with `{ "channel_id": Uuid, "sdp": String }`. The server validates permissions (`JOIN_VOICE`), creates/updates the `rustrtc::PeerConnection`, and returns an `answer` event with the server SDP.
|
||||||
|
- Client and server exchange `ice-candidate` events with `{ "channel_id": Uuid, "candidate": String }`.
|
||||||
|
- Client sends `leave` or closes the socket; server removes peer from `VoiceRoom` and unsubscribes tracks.
|
||||||
|
- **Error Handling**: Server returns `{ "action": "error", "message": String }` on permission denial, invalid payload, or WebRTC negotiation failure.
|
||||||
|
- **Gateway Decoupling**: `/ws/gateway` no longer processes `"Voice"` namespace events.
|
||||||
|
|
||||||
|
### Non-Functional Requirements
|
||||||
|
- **Performance**: High throughput and low latency signaling with asynchronous message handling via Tokio tasks.
|
||||||
|
- **Reliability**: Deterministic cleanup on socket disconnection preventing orphaned WebRTC sessions or memory leaks.
|
||||||
|
|
||||||
|
# Technical Design
|
||||||
|
|
||||||
|
### Current Implementation
|
||||||
|
- `src/routes/mod.rs`: Mounts `ws_routes` containing `/ws/gateway`.
|
||||||
|
- `src/routes/gateway/mod.rs`: `GatewayClient::on_message` inspects JSON events; if `namespace == "Voice"`, it dispatches `offer`, `ice-candidate`, and `leave` to `AppState.voice`.
|
||||||
|
- `src/voice/service.rs`: Manages `RtcConfiguration`, `VoiceRoom` instances, SDP answers, ICE candidates, and permissions.
|
||||||
|
- `src/http/middleware.rs` & `src/http/context.rs`: `auth_middleware` extracts JWTs from Bearer headers, cookies, or `?token=` query parameters and populates `CurrentUser`.
|
||||||
|
|
||||||
|
### Key Decisions
|
||||||
|
1. **Dedicated WebSocket Route (`/ws/voice`)**: Voice signaling is completely extracted into its own endpoint and handler module (`src/routes/voice/` or `src/voice/ws.rs`), eliminating protocol coupling in `src/routes/gateway/`.
|
||||||
|
2. **Connection-bound Voice Lifecycle**: The dedicated WebSocket lifetime directly reflects voice room occupancy. When the WebSocket connection drops, `VoiceService::leave` is triggered automatically.
|
||||||
|
3. **Simplified Signaling Protocol**: Elimination of the generic `GatewayEvent` envelope wrapper (`{ namespace, action, content }`) in favor of direct, typed voice signaling messages (`{ action, ... }`).
|
||||||
|
4. **Standardized Authentication**: Reuse of `CurrentUser` extractor ensures uniform JWT validation across REST, Gateway, and Voice WebSocket endpoints.
|
||||||
|
|
||||||
|
### Proposed Changes
|
||||||
|
- **New Route & Handler**:
|
||||||
|
- Create `src/routes/voice/` with `mod.rs`, `handlers.rs`, and `routes.rs` (or integrate under `src/voice/` and expose via `src/routes/mod.rs`).
|
||||||
|
- Add `GET /voice` route to `ws_routes` in `src/routes/mod.rs` (resulting in `/ws/voice`).
|
||||||
|
- Implement `voice_ws_handler` accepting `WebSocketUpgrade`, `State(AppState)`, and `CurrentUser(user)`.
|
||||||
|
- **Voice WebSocket Protocol**:
|
||||||
|
- Define incoming message types: `Offer { channel_id, sdp }`, `IceCandidate { channel_id, candidate }`, `Leave { channel_id }`.
|
||||||
|
- Define outgoing message types: `Answer { channel_id, sdp }`, `IceCandidate { channel_id, candidate }`, `Error { message }`.
|
||||||
|
- **Gateway Cleanup**:
|
||||||
|
- Remove `namespace == "Voice"` handling from `GatewayClient::on_message`.
|
||||||
|
- Remove voice-specific imports from `src/routes/gateway/mod.rs`.
|
||||||
|
- **VoiceService Integration**:
|
||||||
|
- Update `VoiceService::handle_offer` to take the dedicated voice WebSocket sender channel for emitting server ICE candidates.
|
||||||
|
|
||||||
|
### Data Models / Contracts
|
||||||
|
```rust
|
||||||
|
// Voice WebSocket Client -> Server messages
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
#[serde(tag = "action", rename_all = "kebab-case")]
|
||||||
|
pub enum VoiceClientMessage {
|
||||||
|
Offer {
|
||||||
|
channel_id: Uuid,
|
||||||
|
sdp: String,
|
||||||
|
},
|
||||||
|
IceCandidate {
|
||||||
|
channel_id: Uuid,
|
||||||
|
candidate: String,
|
||||||
|
},
|
||||||
|
Leave {
|
||||||
|
channel_id: Uuid,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
// Voice WebSocket Server -> Client messages
|
||||||
|
#[derive(Debug, Serialize)]
|
||||||
|
#[serde(tag = "action", rename_all = "kebab-case")]
|
||||||
|
pub enum VoiceServerMessage {
|
||||||
|
Answer {
|
||||||
|
channel_id: Uuid,
|
||||||
|
sdp: String,
|
||||||
|
},
|
||||||
|
IceCandidate {
|
||||||
|
channel_id: Uuid,
|
||||||
|
candidate: String,
|
||||||
|
},
|
||||||
|
Error {
|
||||||
|
message: String,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Components
|
||||||
|
- `VoiceWsHandler` (`src/routes/voice/handlers.rs`): Handles WebSocket upgrades, authentication, event loop, message serialization/deserialization, and lifecycle teardown.
|
||||||
|
- `VoiceRoutes` (`src/routes/voice/routes.rs`): Defines Axum router for `/voice`.
|
||||||
|
- `VoiceService` (`src/voice/service.rs`): Coordinates `rustrtc::PeerConnection` instances, SFU track forwarding, and room management.
|
||||||
|
- `GatewayManager` (`src/routes/gateway/mod.rs`): Retains responsibility solely for domain event broadcasting (chat, server, channel updates).
|
||||||
|
|
||||||
|
### File Structure
|
||||||
|
- `src/routes/mod.rs`: Register `voice::routes::router()` under `/ws`.
|
||||||
|
- `src/routes/voice/mod.rs`: Public module definition.
|
||||||
|
- `src/routes/voice/routes.rs`: Voice WebSocket route definition.
|
||||||
|
- `src/routes/voice/handlers.rs`: Voice WebSocket connection handler and message loop.
|
||||||
|
- `src/routes/voice/messages.rs`: Strongly typed signaling message definitions.
|
||||||
|
- `src/routes/gateway/mod.rs`: Cleaned up to remove voice signaling logic.
|
||||||
|
|
||||||
|
### Architecture Diagram
|
||||||
|
```mermaid
|
||||||
|
graph TD
|
||||||
|
Client[Client App]
|
||||||
|
Client -- Chat events / presence --> GW["/ws/gateway (GatewayManager)"]
|
||||||
|
Client -- Voice signaling (SDP/ICE) --> VWS["/ws/voice (VoiceWsHandler)"]
|
||||||
|
VWS -- Manage connections & rooms --> VS[VoiceService]
|
||||||
|
VS -- SFU media fan-out --> RR[rustrtc / VoiceRoom]
|
||||||
|
Client == SRTP media over single UDP port ==> RR
|
||||||
|
```
|
||||||
|
|
||||||
|
### Risks
|
||||||
|
- **Socket Disconnect Race Conditions**: Ensure that client disconnection promptly removes the user from `VoiceRoom` without deadlocking `RwLock` guards.
|
||||||
|
- **Message Deserialization Errors**: Malformed payloads must result in an `Error` response over the WebSocket rather than terminating the connection loop prematurely.
|
||||||
@@ -0,0 +1,138 @@
|
|||||||
|
---
|
||||||
|
sessionId: session-260912-101455-1pxv
|
||||||
|
---
|
||||||
|
|
||||||
|
# Requirements
|
||||||
|
|
||||||
|
### Overview & Goals
|
||||||
|
Replace the current placeholder raw-UDP voice relay (`src/udp/*`) with a real WebRTC media stack built on the `rustrtc` crate (already declared in `Cargo.toml`). The server becomes a centralized **SFU** (Selective Forwarding Unit): each connected client opens **one `PeerConnection`** with the server, the server decrypts/receives that client's audio (and later video) track and re-encodes/forwards it to every other member of the same voice `Channel`.
|
||||||
|
|
||||||
|
### Scope
|
||||||
|
**In Scope**
|
||||||
|
- Remove `src/udp/server.rs`, `src/udp/router.rs` raw-socket logic (metrics module is kept/adapted).
|
||||||
|
- Add a `voice` module hosting: `PeerConnection` lifecycle management via `rustrtc`, SDP offer/answer exchange, ICE candidate exchange, and SFU-style track forwarding scoped by `channel_id`.
|
||||||
|
- Reuse the existing WebSocket gateway (`GatewayManager` / `RealtimeRouter` / `on_message` in `src/routes/gateway/mod.rs`) as the signaling transport — add a `Voice` namespace to the existing JSON event envelope (`GatewayEvent { namespace, action, content }`) instead of introducing a new endpoint, per explicit request to avoid breaking the current structure.
|
||||||
|
- Reuse `channel::ChannelType::Voice`, `channel_user`, and the `JOIN_VOICE` / `SPEAK` permission bits already defined in `src/permissions.rs` to authorize who may join/publish in a voice channel.
|
||||||
|
- Update `config.toml` / `NetworkConfig` so the single `udp_port` continues to be the one UDP port used, now bound through `rustrtc`'s `ice_udp_mux` (single-port ICE) instead of a raw `UdpSocket`.
|
||||||
|
- Update `AppMetrics`/`core::mod::App::run` wiring so the new voice service starts/stops alongside the HTTP server, replacing `UdpServer`.
|
||||||
|
|
||||||
|
**Out of Scope**
|
||||||
|
- TURN/STUN server configuration (kept minimal per user's answer — LAN/local network scenario for now).
|
||||||
|
- Video/screen-share specific handling (structure will allow it later, but only audio forwarding is implemented now).
|
||||||
|
- Client-side (frontend) implementation details beyond the signaling contract needed by the server.
|
||||||
|
|
||||||
|
### User Stories
|
||||||
|
- As a user with `JOIN_VOICE` permission on a voice channel, I want to connect and have my microphone audio heard by other members of that same channel.
|
||||||
|
- As a user already in a voice channel, I want to hear every other member's audio mixed/forwarded by the server without opening a direct connection to each peer.
|
||||||
|
- As a server operator, I want the previous ad-hoc UDP relay removed and replaced by a maintained WebRTC stack (rustrtc) so voice traffic is encrypted (SRTP) and NAT-traversal (ICE) works out of the box.
|
||||||
|
|
||||||
|
### Functional Requirements
|
||||||
|
- A client sends an SDP offer through the gateway (`namespace: "Voice"`), scoped to a `channel_id`; the server validates channel membership/permissions, creates a `rustrtc::PeerConnection`, and returns an SDP answer through the same gateway connection.
|
||||||
|
- ICE candidates generated by either side are exchanged as additional `Voice` gateway events.
|
||||||
|
- Once connected, the server subscribes the client's inbound audio track to every other `PeerConnection` currently joined to the same voice channel (SFU fan-out), and adds new joiners' tracks to previously-connected peers as they arrive.
|
||||||
|
- When a user disconnects (gateway socket closes) or leaves the voice channel, their `PeerConnection` is closed and their track is removed from all other peers' forwarding sets.
|
||||||
|
- Existing `UdpMetrics`-style counters (packets/bytes in/out, errors) are preserved in spirit, adapted to whatever counters `rustrtc` exposes (or wrapped manually) for the periodic `reporter`.
|
||||||
|
|
||||||
|
# Technical Design
|
||||||
|
|
||||||
|
### Current Implementation
|
||||||
|
- `src/udp/server.rs`: binds a raw `tokio::net::UdpSocket` on `NetworkConfig.udp_port`, loops on `recv_from`, and blindly forwards every datagram to all peers registered in a `RoutingTable` under a **hard-coded** `"default"` channel — there is no real protocol, no per-voice-channel isolation, no encryption.
|
||||||
|
- `src/udp/router.rs`: `RoutingTable` maps `ChannelId -> Vec<SocketAddr>`; `join`/`leave`/`routing_table_mut()` exist but are **never called** anywhere in the codebase — dead placeholder API.
|
||||||
|
- `src/udp/metrics.rs`: atomic counters (`packets_received`, `bytes_sent`, …) + `spawn_reporter` logging every interval. Plugged into `AppMetrics.udp` (`src/metrics/mod.rs`).
|
||||||
|
- `src/core/mod.rs`: `App::build`/`App::run` construct `UdpServer::new(&config.network, udp_metrics)` and `tokio::spawn(udp_server.run())`, joined into the same `tokio::select!`/shutdown-broadcast pattern as `HttpServer`.
|
||||||
|
- `src/routes/gateway/mod.rs`: `GatewayManager` keeps `ConnectionKey{user_id, connection_id} -> GatewayClient{sender: mpsc::UnboundedSender<Message>}`; events are pushed as JSON `GatewayEvent{namespace, action, content}`. `GatewayClient::on_message` currently just logs incoming text — this is the extension point for voice signaling.
|
||||||
|
- `src/services/realtime_registry.rs`: already maintains `channel_id -> HashSet<user_id>` membership computed from `computed_permission` (READ_CHANNEL) — reusable to know who is allowed in a channel, but voice-specific "currently connected to voice" state does not exist yet and must be tracked separately (WebSocket presence in a channel != actively broadcasting audio).
|
||||||
|
- `src/permissions.rs`: `JOIN_VOICE`, `SPEAK`, `STREAM`, `MOVE_OTHERS`, `DISCONNECT_OTHERS`, `MANAGE_VOICE_CHANNEL` bits already defined but unused by any voice logic today.
|
||||||
|
- `src/models/channel.rs`: `ChannelType::Voice` variant already exists.
|
||||||
|
|
||||||
|
### Key Decisions
|
||||||
|
1. **Topology: centralized SFU** (confirmed by user) — one `rustrtc::PeerConnection` per connected client per voice channel; the server forwards each publisher's decoded RTP track to every other subscriber in that channel. No client-to-client PeerConnections.
|
||||||
|
2. **Signaling transport: existing WebSocket gateway** (confirmed by user, to avoid restructuring) — SDP offer/answer and ICE candidates are carried as new `Voice` namespace events inside the current `GatewayEvent` envelope, handled in `GatewayClient::on_message` (currently a stub) and dispatched to a new `VoiceService`.
|
||||||
|
3. **Single UDP port reuse** — `rustrtc`'s `RtcConfiguration` will be set up with `ice_udp_mux = true` and bound to the existing `NetworkConfig.udp_port`, so the media/ICE traffic keeps using the same single port previously owned by the raw `UdpServer`, minimizing config/infra changes (firewall rules, `config.toml` stay compatible).
|
||||||
|
4. **No TURN/STUN for now** (confirmed by user) — `RtcConfiguration.ice_servers` left empty/default; can be added later via `config.toml` without further architecture changes.
|
||||||
|
5. **Voice presence tracked separately from `RealtimeRegistry`** — a new lightweight in-memory registry (`VoiceRoom` per `channel_id`) tracks which `PeerConnection`s are actively publishing/subscribing in a voice channel, since being subscribed to gateway events (`RealtimeRegistry`) is not the same as being connected to voice media.
|
||||||
|
|
||||||
|
### Proposed Changes
|
||||||
|
- Replace the `src/udp` module with a new `src/voice` module:
|
||||||
|
- `voice/mod.rs`: public API, exports `VoiceService`.
|
||||||
|
- `voice/service.rs`: `VoiceService` owns a `rustrtc::RtcConfiguration` (built from `NetworkConfig`), a map `channel_id -> VoiceRoom`, and methods `handle_offer(user_id, channel_id, sdp) -> answer_sdp`, `handle_ice_candidate(...)`, `leave(user_id, channel_id)`.
|
||||||
|
- `voice/room.rs`: `VoiceRoom` holds `HashMap<user_id, Arc<PeerConnection>>` for one channel; implements track fan-out — on receiving a remote track from peer A, it calls into every other peer's `PeerConnection` to add/forward that track (rustrtc `PeerConnection` API, per its `Usage` example: create connection, `set_remote_description`, `create_answer`, `set_local_description`, subscribe to `on_track`).
|
||||||
|
- `voice/metrics.rs`: keep counters analogous to today's `UdpMetrics` (packets/bytes/errors), fed by hooks around track forwarding, still exposed through `AppMetrics.udp` (renamed `AppMetrics.voice` where feasible) and the existing `reporter::spawn_reporter` cadence.
|
||||||
|
- Extend `GatewayEvent` handling: add `"Voice"` namespace with actions like `offer`, `answer`, `ice-candidate`, `leave`; `GatewayClient::on_message` parses these and calls `VoiceService` through `AppState`, then pushes the answer/ICE-candidate response back on the same `mpsc::UnboundedSender<Message>` used for all other gateway events (no protocol reinvention).
|
||||||
|
- Permission check on `offer`: verify the user has `ChannelPermission::JOIN_VOICE` (and `SPEAK` to publish) on the target channel, reusing the same `computed_permission` lookups already used by `RealtimeRegistry`/`http/permissions.rs`.
|
||||||
|
- Update `src/core/mod.rs`: drop `UdpServer::new` / `udp_server.run()` / `udp_shutdown_tx`; construct `VoiceService` instead and store it in `AppState` (alongside `gateway`, `services`) so gateway handlers can call it; no UDP socket bind/run task is spawned by `App::run` anymore — `rustrtc` manages its own I/O internally once configured.
|
||||||
|
- Update `src/config.rs`: keep `NetworkConfig.udp_port` (renamed conceptually to "voice/media port" in comments) feeding `rustrtc`'s `ice_udp_mux_port`; update `DEFAULT_CONFIG_TOML` comment accordingly.
|
||||||
|
- Remove `src/udp/` directory entirely once `voice/` fully replaces its responsibilities; update `Cargo.toml`/module declarations (`lib.rs`) accordingly.
|
||||||
|
|
||||||
|
### Data Models / Contracts
|
||||||
|
```rust
|
||||||
|
// Gateway JSON contract additions (namespace = "Voice")
|
||||||
|
{ "namespace": "Voice", "action": "offer", "content": { "channel_id": Uuid, "sdp": String } }
|
||||||
|
{ "namespace": "Voice", "action": "answer", "content": { "channel_id": Uuid, "sdp": String } }
|
||||||
|
{ "namespace": "Voice", "action": "ice-candidate", "content": { "channel_id": Uuid, "candidate": String } }
|
||||||
|
{ "namespace": "Voice", "action": "leave", "content": { "channel_id": Uuid } }
|
||||||
|
```
|
||||||
|
```rust
|
||||||
|
pub struct VoiceService {
|
||||||
|
config: rustrtc::RtcConfiguration,
|
||||||
|
rooms: RwLock<HashMap<Uuid, VoiceRoom>>, // channel_id -> room
|
||||||
|
}
|
||||||
|
|
||||||
|
impl VoiceService {
|
||||||
|
pub async fn handle_offer(&self, user_id: Uuid, channel_id: Uuid, sdp: String) -> anyhow::Result<String>;
|
||||||
|
pub async fn handle_ice_candidate(&self, user_id: Uuid, channel_id: Uuid, candidate: String) -> anyhow::Result<()>;
|
||||||
|
pub fn leave(&self, user_id: Uuid, channel_id: Uuid);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Architecture Diagram
|
||||||
|
```mermaid
|
||||||
|
graph TD
|
||||||
|
ClientA[Client A - browser] -- WebSocket gateway: Voice offer/answer/ICE --> Gateway[GatewayManager / on_message]
|
||||||
|
ClientB[Client B - browser] -- WebSocket gateway: Voice offer/answer/ICE --> Gateway
|
||||||
|
Gateway -- dispatch --> VoiceService
|
||||||
|
VoiceService -- creates/owns --> RoomA[VoiceRoom per channel_id]
|
||||||
|
RoomA -- PeerConnection A --> PCA[rustrtc PeerConnection A]
|
||||||
|
RoomA -- PeerConnection B --> PCB[rustrtc PeerConnection B]
|
||||||
|
ClientA == RTP/SRTP media, single UDP port ==> PCA
|
||||||
|
ClientB == RTP/SRTP media, single UDP port ==> PCB
|
||||||
|
PCA -- forward audio track --> PCB
|
||||||
|
PCB -- forward audio track --> PCA
|
||||||
|
```
|
||||||
|
|
||||||
|
### Risks
|
||||||
|
- `rustrtc` is a young/fast-moving crate (frequent point releases per the benchmark notes found) — API surface for track-forwarding/SFU usage should be validated against the pinned `0.3.133` docs during implementation; if a needed primitive (e.g., explicit track re-publishing helper) is missing, a thin adapter layer will be needed inside `voice/room.rs`.
|
||||||
|
- Moving from "no real protocol" to full SDP/ICE negotiation is a larger surface than the previous placeholder; permission checks (`JOIN_VOICE`/`SPEAK`) must be enforced before any `PeerConnection` is created to avoid unauthorized channel joins.
|
||||||
|
- Since signaling now flows through the same WebSocket used for all other realtime events, malformed/large SDP payloads must not block the `GatewayClient` message loop — the voice service calls will be dispatched without blocking other event types.
|
||||||
|
|
||||||
|
# Delivery Steps
|
||||||
|
|
||||||
|
### * Step 1: Scaffold the voice module and RtcConfiguration from NetworkConfig
|
||||||
|
A new `src/voice` module exists with a `VoiceService` capable of building a `rustrtc::RtcConfiguration` from the app's network config, replacing the old `src/udp` module's role in `core/mod.rs`.
|
||||||
|
- Create `src/voice/mod.rs`, `src/voice/service.rs`, `src/voice/room.rs`, `src/voice/metrics.rs`.
|
||||||
|
- Port `UdpMetrics`-style counters into `voice/metrics.rs` (packets/bytes in/out, errors), keeping the `Metrics`/`MetricsSnapshot` trait implementations used by `crate::metrics::reporter`.
|
||||||
|
- Build `RtcConfiguration`/`RtcConfigurationBuilder` in `VoiceService::new(&NetworkConfig, metrics)` using `ice_udp_mux = true` and `ice_udp_mux_port = network.udp_port`, no ICE servers configured.
|
||||||
|
- Remove `src/udp/server.rs` and `src/udp/router.rs`; delete unused `RoutingTable` (dead code confirmed unused elsewhere).
|
||||||
|
- Update `src/core/mod.rs` (`App::build`/`App::run`) to construct `VoiceService` instead of `UdpServer`, store it on `AppState`, and drop the now-unneeded `udp_shutdown_tx`/`udp_handle` spawn/join wiring.
|
||||||
|
- Update `src/metrics/mod.rs` (`AppMetrics`) to reference the new voice metrics type.
|
||||||
|
|
||||||
|
### Step 2: Implement per-channel PeerConnection lifecycle in VoiceRoom
|
||||||
|
`VoiceService`/`VoiceRoom` can accept an SDP offer for a given channel, create a `rustrtc::PeerConnection`, and return an SDP answer, tracking connections per `channel_id`.
|
||||||
|
- Implement `VoiceRoom` (`channel_id -> HashMap<user_id, Arc<PeerConnection>>`).
|
||||||
|
- Implement `VoiceService::handle_offer(user_id, channel_id, sdp)`: looks up/creates the room, creates a `PeerConnection` via `rustrtc`, calls `set_remote_description`/`create_answer`/`set_local_description`, returns the answer SDP.
|
||||||
|
- Implement `VoiceService::handle_ice_candidate` and `VoiceService::leave`, closing and removing the `PeerConnection` from its room.
|
||||||
|
- Enforce `ChannelPermission::JOIN_VOICE` (and `SPEAK` for publishing) before creating a `PeerConnection`, reusing the `computed_permission` lookup pattern from `src/services/realtime_registry.rs`.
|
||||||
|
|
||||||
|
### Step 3: Wire SFU track forwarding between peers in the same channel
|
||||||
|
Audio published by one connected client in a voice channel is forwarded by the server to every other client connected to the same channel.
|
||||||
|
- On `PeerConnection::on_track` for a given user's connection, register the inbound track on the owning `VoiceRoom`.
|
||||||
|
- For every other `PeerConnection` already in that `VoiceRoom`, add/forward the new track (SFU fan-out) using `rustrtc`'s track/`MediaCapabilities` APIs.
|
||||||
|
- When a new peer joins an existing room, subscribe it to all tracks already being forwarded by other room members.
|
||||||
|
- On `leave`/disconnect, remove the peer's published track from all other peers' forwarding sets and close its `PeerConnection`.
|
||||||
|
|
||||||
|
### Step 4: Expose Voice signaling through the existing WebSocket gateway
|
||||||
|
Clients can perform the full offer/answer/ICE handshake over the existing gateway WebSocket connection using a new `Voice` namespace, with no new HTTP/WS endpoint introduced.
|
||||||
|
- Extend `GatewayClient::on_message` in `src/routes/gateway/mod.rs` to parse `GatewayEvent{namespace: "Voice", action, content}` messages (`offer`, `answer`, `ice-candidate`, `leave`).
|
||||||
|
- Dispatch parsed messages to `VoiceService` (accessed via `AppState`) and send the resulting answer/ICE-candidate events back through the client's existing `mpsc::UnboundedSender<Message>`.
|
||||||
|
- On gateway disconnect (`GatewayClient::on_disconnect`), call `VoiceService::leave` for any channel the user was actively connected to in voice.
|
||||||
|
- Update `config.toml`'s `DEFAULT_CONFIG_TOML` comment for `udp_port` to reflect its new role as the rustrtc ICE/media mux port.
|
||||||
Generated
+953
-55
File diff suppressed because it is too large
Load Diff
+5
-5
@@ -21,19 +21,19 @@ event_bus = { path = "event_bus" }
|
|||||||
parking_lot = "0.12.5"
|
parking_lot = "0.12.5"
|
||||||
serde = "1.0.229"
|
serde = "1.0.229"
|
||||||
serde_json = "1.0.151"
|
serde_json = "1.0.151"
|
||||||
toml = "1.1.4"
|
toml = "1.1.6"
|
||||||
uuid = { version = "1.26.0", features = ["v4", "v7", "fast-rng", "serde"] }
|
uuid = { version = "1.26.1", features = ["v4", "v7", "fast-rng", "serde"] }
|
||||||
tracing = "0.1.44"
|
tracing = "0.1.44"
|
||||||
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "time"] }
|
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "time"] }
|
||||||
thiserror = "2"
|
thiserror = "2"
|
||||||
utoipa = { version = "5", features = ["uuid", "chrono"] }
|
utoipa = { version = "5", features = ["uuid", "chrono"] }
|
||||||
utoipa-swagger-ui = { version = "9", features = ["axum"] }
|
utoipa-swagger-ui = { version = "9", features = ["axum"] }
|
||||||
log = "0.4"
|
log = "0.4"
|
||||||
bitflags = "2.13.1"
|
bitflags = "2.13.2"
|
||||||
argon2 = { version = "0.6.0", features = ["password-hash"] }
|
argon2 = { version = "0.6.0", features = ["password-hash"] }
|
||||||
jsonwebtoken = { version = "11.0.0", features = ["aws_lc_rs"] }
|
jsonwebtoken = { version = "11.0.0", features = ["aws_lc_rs"] }
|
||||||
tower = { version = "0.5", features = ["util"] }
|
tower = { version = "0.5", features = ["util"] }
|
||||||
tower-http = { version = "0.7.0", features = ["catch-panic", "cors", "trace"] }
|
tower-http = { version = "0.7.1", features = ["catch-panic", "cors", "trace"] }
|
||||||
chrono = "0.4.45"
|
chrono = "0.4.45"
|
||||||
validator = { version = "0.21.0", features = ["derive"] }
|
validator = { version = "0.21.0", features = ["derive"] }
|
||||||
async-trait = "0.1.92"
|
async-trait = "0.1.92"
|
||||||
@@ -42,4 +42,4 @@ futures-util = "0.3"
|
|||||||
form_urlencoded = "1.2.2"
|
form_urlencoded = "1.2.2"
|
||||||
time = "0.3.55"
|
time = "0.3.55"
|
||||||
sha2 = "0.11.0"
|
sha2 = "0.11.0"
|
||||||
|
rustrtc = "0.3.133"
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ harness = false
|
|||||||
tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] }
|
tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] }
|
||||||
parking_lot = "0.12.5"
|
parking_lot = "0.12.5"
|
||||||
tracing = "0.1"
|
tracing = "0.1"
|
||||||
uuid = { version = "1.26.0", features = ["v4"] }
|
uuid = { version = "1.26.1", features = ["v4"] }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
tokio = { version = "1.53.1", default-features = false, features = ["rt", "rt-multi-thread", "macros", "time", "sync"] }
|
tokio = { version = "1.53.1", default-features = false, features = ["rt", "rt-multi-thread", "macros", "time", "sync"] }
|
||||||
|
|||||||
+1
-1
@@ -48,7 +48,7 @@ host = "0.0.0.0"
|
|||||||
# TCP and UDP port can be the same
|
# TCP and UDP port can be the same
|
||||||
# HTTP port
|
# HTTP port
|
||||||
tcp_port = 8080
|
tcp_port = 8080
|
||||||
# Voice/Video port
|
# WebRTC ICE/Media UDP multiplexing port
|
||||||
udp_port = 8080
|
udp_port = 8080
|
||||||
|
|
||||||
[database]
|
[database]
|
||||||
|
|||||||
+7
-14
@@ -7,7 +7,7 @@ use crate::metrics::{AppMetrics, reporter};
|
|||||||
use crate::repositories::Repositories;
|
use crate::repositories::Repositories;
|
||||||
use crate::routes::gateway::{GatewayManager, RealtimeRouter};
|
use crate::routes::gateway::{GatewayManager, RealtimeRouter};
|
||||||
use crate::services::Services;
|
use crate::services::Services;
|
||||||
use crate::udp::server::UdpServer;
|
use crate::voice::VoiceService;
|
||||||
use event_bus::EventBus;
|
use event_bus::EventBus;
|
||||||
use migration::{Migrator, MigratorTrait};
|
use migration::{Migrator, MigratorTrait};
|
||||||
pub use state::AppState;
|
pub use state::AppState;
|
||||||
@@ -65,6 +65,8 @@ impl App {
|
|||||||
};
|
};
|
||||||
|
|
||||||
let metrics = AppMetrics::new();
|
let metrics = AppMetrics::new();
|
||||||
|
let voice_metrics = Arc::clone(&metrics.voice);
|
||||||
|
let voice = Arc::new(VoiceService::new(&config.network, voice_metrics));
|
||||||
|
|
||||||
let services = Arc::new(Services::new(repositories.clone(), event_bus.clone()));
|
let services = Arc::new(Services::new(repositories.clone(), event_bus.clone()));
|
||||||
services.permission_sync.start_listen_event().await;
|
services.permission_sync.start_listen_event().await;
|
||||||
@@ -90,6 +92,7 @@ impl App {
|
|||||||
gateway,
|
gateway,
|
||||||
event_bus,
|
event_bus,
|
||||||
services,
|
services,
|
||||||
|
voice,
|
||||||
};
|
};
|
||||||
|
|
||||||
Ok(Self { state })
|
Ok(Self { state })
|
||||||
@@ -116,28 +119,20 @@ impl App {
|
|||||||
// Initialize HTTP Server
|
// Initialize HTTP Server
|
||||||
let (http_server, http_shutdown_tx) = HttpServer::new(&config.network, self.state.clone());
|
let (http_server, http_shutdown_tx) = HttpServer::new(&config.network, self.state.clone());
|
||||||
|
|
||||||
// Initialize UDP service
|
|
||||||
let udp_metrics = Arc::clone(&self.state.metrics.udp);
|
|
||||||
let (udp_server, udp_shutdown_tx) = UdpServer::new(&config.network, udp_metrics);
|
|
||||||
|
|
||||||
// Lance le reporter central de métriques toutes les 30 secondes
|
// Lance le reporter central de métriques toutes les 30 secondes
|
||||||
reporter::spawn_reporter(
|
reporter::spawn_reporter(
|
||||||
Arc::new(self.state.metrics.clone()),
|
Arc::new(self.state.metrics.clone()),
|
||||||
Duration::from_secs(30),
|
Duration::from_secs(30),
|
||||||
);
|
);
|
||||||
|
|
||||||
// On lance les serveurs dans des tâches séparées
|
// On lance le serveur HTTP dans une tâche séparée
|
||||||
let mut http_handle = tokio::spawn(http_server.run());
|
let mut http_handle = tokio::spawn(http_server.run());
|
||||||
let mut udp_handle = tokio::spawn(udp_server.run());
|
|
||||||
|
|
||||||
// On arbitre : soit un signal arrive, soit une tâche se termine (erreur/crash)
|
// On arbitre : soit un signal arrive, soit une tâche se termine (erreur/crash)
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
res = &mut http_handle => {
|
res = &mut http_handle => {
|
||||||
tracing::error!("HTTP server stopped unexpectedly: {:?}", res);
|
tracing::error!("HTTP server stopped unexpectedly: {:?}", res);
|
||||||
}
|
}
|
||||||
res = &mut udp_handle => {
|
|
||||||
tracing::error!("UDP server stopped unexpectedly: {:?}", res);
|
|
||||||
}
|
|
||||||
_ = Self::shutdown_signal() => {
|
_ = Self::shutdown_signal() => {
|
||||||
tracing::info!("Shutdown signal received, initiating graceful shutdown...");
|
tracing::info!("Shutdown signal received, initiating graceful shutdown...");
|
||||||
}
|
}
|
||||||
@@ -145,11 +140,9 @@ impl App {
|
|||||||
|
|
||||||
// Dans tous les cas (Ctrl-C ou crash d'un service), on demande l'arrêt global
|
// Dans tous les cas (Ctrl-C ou crash d'un service), on demande l'arrêt global
|
||||||
let _ = http_shutdown_tx.send(());
|
let _ = http_shutdown_tx.send(());
|
||||||
let _ = udp_shutdown_tx.send(());
|
|
||||||
|
|
||||||
// On attend que tout le monde ait fini de nettoyer
|
// On attend que la tâche HTTP termine
|
||||||
// (Note: join! supporte les handles déjà terminés ou annulés)
|
let _ = http_handle.await;
|
||||||
let _ = tokio::join!(http_handle, udp_handle);
|
|
||||||
|
|
||||||
Database::checkpoint_wal(&self.state.db).await?;
|
Database::checkpoint_wal(&self.state.db).await?;
|
||||||
Database::close(&self.state.db).await?;
|
Database::close(&self.state.db).await?;
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ use crate::models::server;
|
|||||||
use crate::repositories::Repositories;
|
use crate::repositories::Repositories;
|
||||||
use crate::routes::gateway::GatewayManager;
|
use crate::routes::gateway::GatewayManager;
|
||||||
use crate::services::Services;
|
use crate::services::Services;
|
||||||
|
use crate::voice::VoiceService;
|
||||||
use event_bus::EventBus;
|
use event_bus::EventBus;
|
||||||
use sea_orm::DatabaseConnection;
|
use sea_orm::DatabaseConnection;
|
||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, RwLock};
|
||||||
@@ -19,6 +20,7 @@ pub struct AppState {
|
|||||||
pub gateway: Arc<GatewayManager>,
|
pub gateway: Arc<GatewayManager>,
|
||||||
pub event_bus: Arc<EventBus>,
|
pub event_bus: Arc<EventBus>,
|
||||||
pub services: Arc<Services>,
|
pub services: Arc<Services>,
|
||||||
|
pub voice: Arc<VoiceService>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AppState {}
|
impl AppState {}
|
||||||
|
|||||||
+1
-1
@@ -7,7 +7,7 @@ pub mod core;
|
|||||||
pub mod permissions;
|
pub mod permissions;
|
||||||
pub mod repositories;
|
pub mod repositories;
|
||||||
pub mod routes;
|
pub mod routes;
|
||||||
pub mod udp;
|
pub mod voice;
|
||||||
|
|
||||||
pub mod auth;
|
pub mod auth;
|
||||||
pub mod metrics;
|
pub mod metrics;
|
||||||
|
|||||||
+3
-3
@@ -4,7 +4,7 @@ use std::sync::Arc;
|
|||||||
use std::time::Instant;
|
use std::time::Instant;
|
||||||
|
|
||||||
use crate::http::metrics::HttpMetrics;
|
use crate::http::metrics::HttpMetrics;
|
||||||
use crate::udp::metrics::UdpMetrics;
|
use crate::voice::metrics::VoiceMetrics;
|
||||||
|
|
||||||
/// Contrat minimal pour un jeu de compteurs métriques.
|
/// Contrat minimal pour un jeu de compteurs métriques.
|
||||||
pub trait Metrics {
|
pub trait Metrics {
|
||||||
@@ -26,14 +26,14 @@ pub trait MetricsSnapshot: Clone {
|
|||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct AppMetrics {
|
pub struct AppMetrics {
|
||||||
pub http: Arc<HttpMetrics>,
|
pub http: Arc<HttpMetrics>,
|
||||||
pub udp: Arc<UdpMetrics>,
|
pub voice: Arc<VoiceMetrics>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AppMetrics {
|
impl AppMetrics {
|
||||||
pub fn new() -> Self {
|
pub fn new() -> Self {
|
||||||
Self {
|
Self {
|
||||||
http: HttpMetrics::new(),
|
http: HttpMetrics::new(),
|
||||||
udp: UdpMetrics::new(),
|
voice: VoiceMetrics::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+11
-11
@@ -8,23 +8,23 @@ use crate::metrics::AppMetrics;
|
|||||||
/// Lance une tâche tokio unique qui reporte toutes les métriques à intervalle régulier.
|
/// Lance une tâche tokio unique qui reporte toutes les métriques à intervalle régulier.
|
||||||
pub fn spawn_reporter(metrics: Arc<AppMetrics>, interval: Duration) {
|
pub fn spawn_reporter(metrics: Arc<AppMetrics>, interval: Duration) {
|
||||||
let metrics_http = Arc::clone(&metrics.http);
|
let metrics_http = Arc::clone(&metrics.http);
|
||||||
let metrics_udp = Arc::clone(&metrics.udp);
|
let metrics_voice = Arc::clone(&metrics.voice);
|
||||||
|
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let mut ticker = tokio::time::interval(interval);
|
let mut ticker = tokio::time::interval(interval);
|
||||||
ticker.tick().await;
|
ticker.tick().await;
|
||||||
|
|
||||||
let mut prev_http = metrics_http.snapshot();
|
let mut prev_http = metrics_http.snapshot();
|
||||||
let mut prev_udp = metrics_udp.snapshot();
|
let mut prev_voice = metrics_voice.snapshot();
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
ticker.tick().await;
|
ticker.tick().await;
|
||||||
|
|
||||||
let current_http = metrics_http.snapshot();
|
let current_http = metrics_http.snapshot();
|
||||||
let current_udp = metrics_udp.snapshot();
|
let current_voice = metrics_voice.snapshot();
|
||||||
|
|
||||||
let http_rates = current_http.rates_since(&prev_http);
|
let http_rates = current_http.rates_since(&prev_http);
|
||||||
let udp_rates = current_udp.rates_since(&prev_udp);
|
let voice_rates = current_voice.rates_since(&prev_voice);
|
||||||
|
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
// ── HTTP ──
|
// ── HTTP ──
|
||||||
@@ -34,17 +34,17 @@ pub fn spawn_reporter(metrics: Arc<AppMetrics>, interval: Duration) {
|
|||||||
http_responses_5xx = current_http.responses_5xx,
|
http_responses_5xx = current_http.responses_5xx,
|
||||||
http_req_per_sec = format_args!("{:.2}", http_rates.requests_per_sec),
|
http_req_per_sec = format_args!("{:.2}", http_rates.requests_per_sec),
|
||||||
http_avg_latency_ms = format_args!("{:.1}", http_rates.avg_latency_ms),
|
http_avg_latency_ms = format_args!("{:.1}", http_rates.avg_latency_ms),
|
||||||
// ── UDP ──
|
// ── Voice ──
|
||||||
udp_pkts_rx = current_udp.packets_received,
|
voice_pkts_rx = current_voice.packets_received,
|
||||||
udp_pkts_tx = current_udp.packets_sent,
|
voice_pkts_tx = current_voice.packets_sent,
|
||||||
udp_pkts_dropped = current_udp.packets_dropped,
|
voice_pkts_dropped = current_voice.packets_dropped,
|
||||||
udp_pkts_rx_s = format_args!("{:.1}", udp_rates.packets_received_per_sec),
|
voice_pkts_rx_s = format_args!("{:.1}", voice_rates.packets_received_per_sec),
|
||||||
udp_pkts_tx_s = format_args!("{:.1}", udp_rates.packets_sent_per_sec),
|
voice_pkts_tx_s = format_args!("{:.1}", voice_rates.packets_sent_per_sec),
|
||||||
"App metrics"
|
"App metrics"
|
||||||
);
|
);
|
||||||
|
|
||||||
prev_http = current_http;
|
prev_http = current_http;
|
||||||
prev_udp = current_udp;
|
prev_voice = current_voice;
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -56,9 +56,10 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) {
|
|||||||
|
|
||||||
// Task pour recevoir les messages du WebSocket
|
// Task pour recevoir les messages du WebSocket
|
||||||
let client_clone = client.clone();
|
let client_clone = client.clone();
|
||||||
|
let state_clone = state.clone();
|
||||||
let mut recv_task = tokio::spawn(async move {
|
let mut recv_task = tokio::spawn(async move {
|
||||||
while let Some(Ok(message)) = receiver.next().await {
|
while let Some(Ok(message)) = receiver.next().await {
|
||||||
client_clone.on_message(message).await;
|
client_clone.on_message(message, &state_clone).await;
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -70,5 +71,5 @@ async fn handle_socket(socket: WebSocket, state: AppState, user: User) {
|
|||||||
|
|
||||||
state.gateway.remove_client(&client);
|
state.gateway.remove_client(&client);
|
||||||
// // Déconnexion (Disconnect)
|
// // Déconnexion (Disconnect)
|
||||||
client.on_disconnect().await;
|
client.on_disconnect(&state).await;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
use crate::core::AppState;
|
||||||
use crate::domain::events::channel::{
|
use crate::domain::events::channel::{
|
||||||
ChannelCreatedEvent, ChannelDeletedEvent, ChannelUpdatedEvent,
|
ChannelCreatedEvent, ChannelDeletedEvent, ChannelUpdatedEvent,
|
||||||
};
|
};
|
||||||
@@ -472,16 +473,18 @@ impl GatewayClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn on_connect(&mut self) {
|
pub async fn on_connect(&mut self) {
|
||||||
tracing::info!(user_id = %self.user.id, "Client connected");
|
tracing::info!(user_id = %self.user.id, "Client connected");
|
||||||
}
|
}
|
||||||
async fn on_disconnect(&mut self) {
|
|
||||||
|
pub async fn on_disconnect(&mut self, state: &AppState) {
|
||||||
tracing::info!(user_id = %self.user.id, "Client disconnected");
|
tracing::info!(user_id = %self.user.id, "Client disconnected");
|
||||||
|
state.voice.leave_all(self.user.id).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn on_message(&self, message: Message) {
|
pub async fn on_message(&self, message: Message, _state: &AppState) {
|
||||||
if let Message::Text(content) = message {
|
if let Message::Text(content) = message {
|
||||||
tracing::info!(user_id = %self.user.id, "Received text message: {}", content);
|
tracing::debug!(user_id = %self.user.id, "Received text message: {}", content);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-1
@@ -18,6 +18,7 @@ pub mod role;
|
|||||||
pub mod server;
|
pub mod server;
|
||||||
pub mod server_item_order;
|
pub mod server_item_order;
|
||||||
pub mod user;
|
pub mod user;
|
||||||
|
pub mod voice;
|
||||||
|
|
||||||
pub fn router() -> OxRouter {
|
pub fn router() -> OxRouter {
|
||||||
// Routes nécessitant une authentification
|
// Routes nécessitant une authentification
|
||||||
@@ -41,7 +42,9 @@ pub fn router() -> OxRouter {
|
|||||||
.merge(core::routes::router());
|
.merge(core::routes::router());
|
||||||
let public_attachment_routes = attachment::routes::public_router();
|
let public_attachment_routes = attachment::routes::public_router();
|
||||||
|
|
||||||
let ws_routes = Router::new().merge(gateway::routes::router());
|
let ws_routes = Router::new()
|
||||||
|
.merge(gateway::routes::router())
|
||||||
|
.merge(voice::routes::router());
|
||||||
|
|
||||||
Router::new()
|
Router::new()
|
||||||
.nest("/api", api_routes)
|
.nest("/api", api_routes)
|
||||||
|
|||||||
@@ -0,0 +1,94 @@
|
|||||||
|
use super::messages::{VoiceClientMessage, VoiceServerMessage};
|
||||||
|
use crate::{core::AppState, http::context::CurrentUser};
|
||||||
|
use axum::{
|
||||||
|
extract::{
|
||||||
|
State,
|
||||||
|
ws::{Message, WebSocket, WebSocketUpgrade},
|
||||||
|
},
|
||||||
|
response::IntoResponse,
|
||||||
|
};
|
||||||
|
use futures_util::{SinkExt, StreamExt};
|
||||||
|
use tokio::sync::mpsc;
|
||||||
|
|
||||||
|
pub async fn ws_handler(
|
||||||
|
ws: WebSocketUpgrade,
|
||||||
|
State(state): State<AppState>,
|
||||||
|
CurrentUser(user): CurrentUser,
|
||||||
|
) -> impl IntoResponse {
|
||||||
|
ws.on_upgrade(move |socket| handle_socket(socket, state, user))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_socket(socket: WebSocket, state: AppState, user: crate::models::user::Model) {
|
||||||
|
let (mut sender, mut receiver) = socket.split();
|
||||||
|
let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
|
||||||
|
let send_task = tokio::spawn(async move {
|
||||||
|
while let Some(message) = rx.recv().await {
|
||||||
|
if sender.send(message).await.is_err() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
while let Some(Ok(message)) = receiver.next().await {
|
||||||
|
let Message::Text(text) = message else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
let parsed = match serde_json::from_str::<VoiceClientMessage>(&text) {
|
||||||
|
Ok(message) => message,
|
||||||
|
Err(error) => {
|
||||||
|
send_error(&tx, format!("Invalid voice message: {error}"));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
match parsed {
|
||||||
|
VoiceClientMessage::Offer { channel_id, sdp } => {
|
||||||
|
match state
|
||||||
|
.voice
|
||||||
|
.handle_offer(
|
||||||
|
user.id,
|
||||||
|
channel_id,
|
||||||
|
sdp,
|
||||||
|
&state.repositories,
|
||||||
|
Some(tx.clone()),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(answer) => send(
|
||||||
|
&tx,
|
||||||
|
VoiceServerMessage::Answer {
|
||||||
|
channel_id,
|
||||||
|
sdp: answer,
|
||||||
|
},
|
||||||
|
),
|
||||||
|
Err(error) => send_error(&tx, error.to_string()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
VoiceClientMessage::IceCandidate {
|
||||||
|
channel_id,
|
||||||
|
candidate,
|
||||||
|
} => {
|
||||||
|
if let Err(error) = state
|
||||||
|
.voice
|
||||||
|
.handle_ice_candidate(user.id, channel_id, candidate)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
send_error(&tx, error.to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
VoiceClientMessage::Leave { channel_id } => {
|
||||||
|
state.voice.leave(user.id, channel_id).await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
state.voice.leave_all(user.id).await;
|
||||||
|
send_task.abort();
|
||||||
|
}
|
||||||
|
|
||||||
|
fn send(tx: &mpsc::UnboundedSender<Message>, message: VoiceServerMessage) {
|
||||||
|
if let Ok(json) = serde_json::to_string(&message) {
|
||||||
|
let _ = tx.send(Message::Text(json.into()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn send_error(tx: &mpsc::UnboundedSender<Message>, message: String) {
|
||||||
|
send(tx, VoiceServerMessage::Error { message });
|
||||||
|
}
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
#[derive(Debug, Deserialize)]
|
||||||
|
#[serde(tag = "action", rename_all = "kebab-case")]
|
||||||
|
pub enum VoiceClientMessage {
|
||||||
|
Offer { channel_id: Uuid, sdp: String },
|
||||||
|
IceCandidate { channel_id: Uuid, candidate: String },
|
||||||
|
Leave { channel_id: Uuid },
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Serialize)]
|
||||||
|
#[serde(tag = "action", rename_all = "kebab-case")]
|
||||||
|
pub enum VoiceServerMessage {
|
||||||
|
Answer { channel_id: Uuid, sdp: String },
|
||||||
|
IceCandidate { channel_id: Uuid, candidate: String },
|
||||||
|
Error { message: String },
|
||||||
|
}
|
||||||
@@ -0,0 +1,3 @@
|
|||||||
|
pub mod handlers;
|
||||||
|
pub mod messages;
|
||||||
|
pub mod routes;
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
use super::handlers;
|
||||||
|
use crate::core::AppState;
|
||||||
|
use axum::{Router, routing::get};
|
||||||
|
|
||||||
|
pub fn router() -> Router<AppState> {
|
||||||
|
Router::new().route("/voice", get(handlers::ws_handler))
|
||||||
|
}
|
||||||
@@ -1,244 +0,0 @@
|
|||||||
//! Métrologie du serveur UDP.
|
|
||||||
//!
|
|
||||||
//! Ce module expose :
|
|
||||||
//! - [`UdpMetrics`] : compteurs atomiques lock-free (pas de contention dans la
|
|
||||||
//! boucle de routage).
|
|
||||||
//! - [`UdpMetricsSnapshot`] : lecture cohérente de tous les compteurs à un
|
|
||||||
//! instant T, utilisable pour calculer des deltas.
|
|
||||||
//! - [`UdpRates`] : taux moyens par seconde calculés entre deux snapshots.
|
|
||||||
//! - [`spawn_reporter`] : tâche tokio de reporting périodique via `tracing`.
|
|
||||||
//!
|
|
||||||
//! # Métriques collectées
|
|
||||||
//!
|
|
||||||
//! | Compteur | Description |
|
|
||||||
//! |--------------------|-----------------------------------------------|
|
|
||||||
//! | `packets_received` | Datagrammes reçus |
|
|
||||||
//! | `bytes_received` | Octets reçus (payload uniquement) |
|
|
||||||
//! | `packets_sent` | Datagrammes retransmis vers des abonnés |
|
|
||||||
//! | `bytes_sent` | Octets retransmis |
|
|
||||||
//! | `packets_dropped` | Paquets ignorés (canal sans abonnés) |
|
|
||||||
//! | `send_errors` | Échecs `send_to` |
|
|
||||||
//! | `recv_errors` | Échecs `recv_from` (avant erreur fatale) |
|
|
||||||
//!
|
|
||||||
//! Chaque métrique est également disponible en taux moyen par seconde via
|
|
||||||
//! [`UdpMetricsSnapshot::rates_since`].
|
|
||||||
|
|
||||||
use std::sync::Arc;
|
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
|
||||||
use std::time::{Duration, Instant};
|
|
||||||
|
|
||||||
use crate::metrics::{Metrics, MetricsSnapshot};
|
|
||||||
|
|
||||||
// ── Compteurs ────────────────────────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Compteurs atomiques du serveur UDP.
|
|
||||||
///
|
|
||||||
/// Partagé via [`Arc`] entre la boucle de routage et le reporter périodique.
|
|
||||||
/// Tous les accès utilisent [`Ordering::Relaxed`] : on accepte que les lectures
|
|
||||||
/// voient des valeurs légèrement décalées entre compteurs (suffisant pour de la
|
|
||||||
/// métrologie), ce qui évite tout overhead de synchronisation.
|
|
||||||
#[derive(Debug, Default)]
|
|
||||||
pub struct UdpMetrics {
|
|
||||||
/// Nombre total de datagrammes reçus.
|
|
||||||
pub packets_received: AtomicU64,
|
|
||||||
/// Volume total d'octets reçus (payload des datagrammes).
|
|
||||||
pub bytes_received: AtomicU64,
|
|
||||||
/// Nombre total de datagrammes retransmis (somme sur tous les abonnés).
|
|
||||||
pub packets_sent: AtomicU64,
|
|
||||||
/// Volume total d'octets retransmis.
|
|
||||||
pub bytes_sent: AtomicU64,
|
|
||||||
/// Paquets ignorés car le canal ne possède aucun abonné.
|
|
||||||
pub packets_dropped: AtomicU64,
|
|
||||||
/// Nombre d'erreurs `send_to` (non fatales).
|
|
||||||
pub send_errors: AtomicU64,
|
|
||||||
/// Nombre d'erreurs `recv_from` enregistrées avant arrêt du serveur.
|
|
||||||
pub recv_errors: AtomicU64,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl UdpMetrics {
|
|
||||||
/// Crée un jeu de métriques vide enroulé dans un [`Arc`].
|
|
||||||
pub fn new() -> Arc<Self> {
|
|
||||||
Arc::new(Self::default())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Enregistre la réception d'un datagramme 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 datagramme de `bytes` octets vers un client.
|
|
||||||
#[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é (canal sans abonnés).
|
|
||||||
#[inline]
|
|
||||||
pub fn inc_dropped(&self) {
|
|
||||||
self.packets_dropped.fetch_add(1, Ordering::Relaxed);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Enregistre un échec `send_to` non fatal.
|
|
||||||
#[inline]
|
|
||||||
pub fn inc_send_error(&self) {
|
|
||||||
self.send_errors.fetch_add(1, Ordering::Relaxed);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Enregistre un échec `recv_from`.
|
|
||||||
#[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) -> UdpMetricsSnapshot {
|
|
||||||
UdpMetricsSnapshot {
|
|
||||||
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 UdpMetrics {
|
|
||||||
type Snapshot = UdpMetricsSnapshot;
|
|
||||||
|
|
||||||
fn snapshot(&self) -> UdpMetricsSnapshot {
|
|
||||||
self.snapshot()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── Snapshot ─────────────────────────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Lecture cohérente de l'ensemble des compteurs à un instant T.
|
|
||||||
///
|
|
||||||
/// Permet de calculer des deltas et des taux entre deux points dans le temps
|
|
||||||
/// sans bloquer la boucle de routage.
|
|
||||||
#[derive(Debug, Clone, Copy)]
|
|
||||||
pub struct UdpMetricsSnapshot {
|
|
||||||
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 UdpMetricsSnapshot {
|
|
||||||
/// Calcule les taux moyens par seconde depuis un snapshot précédent.
|
|
||||||
pub fn rates_since(&self, previous: &Self) -> UdpRates {
|
|
||||||
let secs = self
|
|
||||||
.taken_at
|
|
||||||
.duration_since(previous.taken_at)
|
|
||||||
.as_secs_f64()
|
|
||||||
.max(f64::EPSILON);
|
|
||||||
|
|
||||||
UdpRates {
|
|
||||||
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 UdpMetricsSnapshot {
|
|
||||||
fn taken_at(&self) -> Instant {
|
|
||||||
self.taken_at
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── Taux ─────────────────────────────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Taux moyens par seconde calculés entre deux [`UdpMetricsSnapshot`].
|
|
||||||
#[derive(Debug, Clone, Copy)]
|
|
||||||
pub struct UdpRates {
|
|
||||||
/// Paquets reçus par seconde.
|
|
||||||
pub packets_received_per_sec: f64,
|
|
||||||
/// Octets reçus par seconde.
|
|
||||||
pub bytes_received_per_sec: f64,
|
|
||||||
/// Paquets envoyés par seconde.
|
|
||||||
pub packets_sent_per_sec: f64,
|
|
||||||
/// Octets envoyés par seconde.
|
|
||||||
pub bytes_sent_per_sec: f64,
|
|
||||||
/// Paquets ignorés par seconde.
|
|
||||||
pub packets_dropped_per_sec: f64,
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── Reporter périodique ───────────────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Lance une tâche Tokio qui logue les métriques toutes les `interval`.
|
|
||||||
///
|
|
||||||
/// Chaque rapport inclut les compteurs cumulatifs **et** les taux moyens sur
|
|
||||||
/// la fenêtre écoulée depuis le rapport précédent.
|
|
||||||
///
|
|
||||||
/// # Exemple
|
|
||||||
/// ```no_run
|
|
||||||
/// use std::time::Duration;
|
|
||||||
/// use std::sync::Arc;
|
|
||||||
/// use oxspeak_server_lib::udp::metrics::{UdpMetrics, spawn_reporter};
|
|
||||||
///
|
|
||||||
/// #[tokio::main]
|
|
||||||
/// async fn main() {
|
|
||||||
/// let metrics = UdpMetrics::new();
|
|
||||||
/// spawn_reporter(Arc::clone(&metrics), Duration::from_secs(5));
|
|
||||||
/// }
|
|
||||||
/// ```
|
|
||||||
pub fn spawn_reporter(metrics: Arc<UdpMetrics>, interval: Duration) {
|
|
||||||
tokio::spawn(async move {
|
|
||||||
let mut ticker = tokio::time::interval(interval);
|
|
||||||
// Le premier tick est immédiat ; on le consomme pour démarrer à t=0.
|
|
||||||
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!(
|
|
||||||
// ── Cumulatifs ──
|
|
||||||
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,
|
|
||||||
// ── Taux / s ──
|
|
||||||
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),
|
|
||||||
"UDP metrics"
|
|
||||||
);
|
|
||||||
|
|
||||||
prev_snapshot = current;
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
@@ -1,3 +0,0 @@
|
|||||||
pub mod metrics;
|
|
||||||
pub mod router;
|
|
||||||
pub mod server;
|
|
||||||
@@ -1,61 +0,0 @@
|
|||||||
use std::collections::HashMap;
|
|
||||||
use std::net::SocketAddr;
|
|
||||||
|
|
||||||
/// Identifiant d'un canal de routage (ex: room ID, channel name…).
|
|
||||||
pub type ChannelId = String;
|
|
||||||
|
|
||||||
/// Table de routage UDP.
|
|
||||||
///
|
|
||||||
/// Associe un identifiant de canal à la liste des clients (adresses IP/port)
|
|
||||||
/// actuellement abonnés à ce canal. Le serveur UDP utilise cette table pour
|
|
||||||
/// décider où retransmettre les paquets entrants.
|
|
||||||
///
|
|
||||||
/// # Note future
|
|
||||||
/// Ce type est intentionnellement simple pour démarrer. La prochaine étape
|
|
||||||
/// sera d'y associer une logique de dispatch (ex: forwarding sélectif,
|
|
||||||
/// authentification du client, etc.).
|
|
||||||
#[derive(Debug, Default)]
|
|
||||||
pub struct RoutingTable {
|
|
||||||
channels: HashMap<ChannelId, Vec<SocketAddr>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl RoutingTable {
|
|
||||||
/// Crée une table de routage vide.
|
|
||||||
pub fn new() -> Self {
|
|
||||||
Self::default()
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Inscrit un client dans un canal.
|
|
||||||
///
|
|
||||||
/// Si le canal n'existe pas encore, il est créé automatiquement.
|
|
||||||
/// Si le client est déjà inscrit dans ce canal, l'appel est sans effet.
|
|
||||||
pub fn join(&mut self, channel: impl Into<ChannelId>, client: SocketAddr) {
|
|
||||||
let clients = self.channels.entry(channel.into()).or_default();
|
|
||||||
if !clients.contains(&client) {
|
|
||||||
clients.push(client);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Retire un client d'un canal.
|
|
||||||
///
|
|
||||||
/// Si le canal devient vide après le retrait, il est supprimé de la table.
|
|
||||||
pub fn leave(&mut self, channel: &str, client: &SocketAddr) {
|
|
||||||
if let Some(clients) = self.channels.get_mut(channel) {
|
|
||||||
clients.retain(|c| c != client);
|
|
||||||
if clients.is_empty() {
|
|
||||||
self.channels.remove(channel);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Retourne la liste des clients abonnés à un canal, ou `None` si le
|
|
||||||
/// canal n'existe pas.
|
|
||||||
pub fn subscribers(&self, channel: &str) -> Option<&[SocketAddr]> {
|
|
||||||
self.channels.get(channel).map(Vec::as_slice)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Retourne tous les canaux connus et leurs abonnés.
|
|
||||||
pub fn channels(&self) -> &HashMap<ChannelId, Vec<SocketAddr>> {
|
|
||||||
&self.channels
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,178 +0,0 @@
|
|||||||
use std::net::SocketAddr;
|
|
||||||
use std::sync::Arc;
|
|
||||||
use tokio::net::UdpSocket;
|
|
||||||
use tokio::sync::broadcast;
|
|
||||||
|
|
||||||
use super::metrics::UdpMetrics;
|
|
||||||
use super::router::RoutingTable;
|
|
||||||
use crate::config::NetworkConfig;
|
|
||||||
|
|
||||||
/// Taille du buffer de lecture pour chaque datagramme UDP entrant.
|
|
||||||
///
|
|
||||||
/// La RFC 768 limite les datagrammes UDP à 65 507 octets (payload max avec
|
|
||||||
/// en-têtes IP+UDP). Pour de la voix/vidéo compressée, les paquets réels
|
|
||||||
/// seront bien plus petits, mais on alloue le maximum une seule fois pour
|
|
||||||
/// éviter toute troncature silencieuse.
|
|
||||||
const UDP_READ_BUFFER_SIZE: usize = 65_507;
|
|
||||||
|
|
||||||
/// Erreurs pouvant survenir pendant l'opération du serveur UDP.
|
|
||||||
#[derive(Debug, thiserror::Error)]
|
|
||||||
pub enum UdpServerError {
|
|
||||||
#[error("failed to bind UDP socket to {addr}: {source}")]
|
|
||||||
Bind {
|
|
||||||
addr: SocketAddr,
|
|
||||||
#[source]
|
|
||||||
source: std::io::Error,
|
|
||||||
},
|
|
||||||
#[error("I/O error: {0}")]
|
|
||||||
Io(#[from] std::io::Error),
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Serveur UDP asynchrone agissant comme routeur de paquets.
|
|
||||||
///
|
|
||||||
/// Reçoit des datagrammes entrants et les route vers les clients inscrits
|
|
||||||
/// dans les canaux correspondants via une [`RoutingTable`].
|
|
||||||
///
|
|
||||||
/// La configuration réseau est fournie par [`NetworkConfig`] (issue de
|
|
||||||
/// [`AppConfig`][crate::config::AppConfig]), qui centralise la lecture du
|
|
||||||
/// fichier TOML.
|
|
||||||
///
|
|
||||||
/// Les métriques sont collectées dans un [`UdpMetrics`] partageable via
|
|
||||||
/// [`Arc`] — passez-le à [`metrics::spawn_reporter`][super::metrics::spawn_reporter]
|
|
||||||
/// pour un reporting périodique automatique.
|
|
||||||
///
|
|
||||||
/// # Exemple
|
|
||||||
/// ```no_run
|
|
||||||
/// use std::time::Duration;
|
|
||||||
/// use oxspeak_server_lib::config::{AppConfig, NetworkConfig};
|
|
||||||
/// use oxspeak_server_lib::udp::server::UdpServer;
|
|
||||||
/// use oxspeak_server_lib::udp::metrics::{UdpMetrics, spawn_reporter};
|
|
||||||
///
|
|
||||||
/// #[tokio::main]
|
|
||||||
/// async fn main() {
|
|
||||||
/// let config = AppConfig::load().unwrap();
|
|
||||||
/// let metrics = UdpMetrics::new();
|
|
||||||
/// spawn_reporter(metrics.clone(), Duration::from_secs(10));
|
|
||||||
/// let (server, _shutdown_tx) = UdpServer::new(config.network, metrics);
|
|
||||||
/// server.run().await.unwrap();
|
|
||||||
/// }
|
|
||||||
/// ```
|
|
||||||
pub struct UdpServer {
|
|
||||||
bind_addr: SocketAddr,
|
|
||||||
routing_table: RoutingTable,
|
|
||||||
metrics: Arc<UdpMetrics>,
|
|
||||||
shutdown_rx: broadcast::Receiver<()>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl UdpServer {
|
|
||||||
/// Crée un nouveau [`UdpServer`] depuis la configuration réseau globale.
|
|
||||||
///
|
|
||||||
/// Retourne le serveur et un [`broadcast::Sender`] pour déclencher le
|
|
||||||
/// shutdown gracieux.
|
|
||||||
pub fn new(network: &NetworkConfig, metrics: Arc<UdpMetrics>) -> (Self, broadcast::Sender<()>) {
|
|
||||||
let bind_addr = SocketAddr::new(network.host.into(), network.udp_port);
|
|
||||||
let (shutdown_tx, shutdown_rx) = broadcast::channel(1);
|
|
||||||
(
|
|
||||||
Self {
|
|
||||||
bind_addr,
|
|
||||||
routing_table: RoutingTable::new(),
|
|
||||||
metrics,
|
|
||||||
shutdown_rx,
|
|
||||||
},
|
|
||||||
shutdown_tx,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Expose la table de routage de façon mutable pour y inscrire des
|
|
||||||
/// clients avant ou pendant l'exécution (via partage d'état ou messages).
|
|
||||||
pub fn routing_table_mut(&mut self) -> &mut RoutingTable {
|
|
||||||
&mut self.routing_table
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Retourne une référence aux métriques du serveur.
|
|
||||||
pub fn metrics(&self) -> &Arc<UdpMetrics> {
|
|
||||||
&self.metrics
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Bind le socket et démarre la boucle de routage.
|
|
||||||
///
|
|
||||||
/// Pour chaque datagramme reçu, le paquet est retransmis inline (sans
|
|
||||||
/// spawn de tâche) vers tous les clients abonnés au canal identifié.
|
|
||||||
/// La future se résout lorsqu'un signal de shutdown est reçu ou qu'une
|
|
||||||
/// erreur I/O fatale survient.
|
|
||||||
pub async fn run(mut self) -> Result<(), UdpServerError> {
|
|
||||||
let socket =
|
|
||||||
UdpSocket::bind(self.bind_addr)
|
|
||||||
.await
|
|
||||||
.map_err(|source| UdpServerError::Bind {
|
|
||||||
addr: self.bind_addr,
|
|
||||||
source,
|
|
||||||
})?;
|
|
||||||
|
|
||||||
tracing::info!(addr = %self.bind_addr, "UDP server listening");
|
|
||||||
|
|
||||||
let mut buf = vec![0u8; UDP_READ_BUFFER_SIZE];
|
|
||||||
|
|
||||||
loop {
|
|
||||||
tokio::select! {
|
|
||||||
result = socket.recv_from(&mut buf) => {
|
|
||||||
match result {
|
|
||||||
Ok((len, peer)) => {
|
|
||||||
self.metrics.inc_received(len as u64);
|
|
||||||
self.route_packet(&socket, &buf[..len], peer).await;
|
|
||||||
}
|
|
||||||
Err(err) => {
|
|
||||||
self.metrics.inc_recv_error();
|
|
||||||
tracing::error!(%err, "recv_from failed");
|
|
||||||
return Err(UdpServerError::Io(err));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_ = self.shutdown_rx.recv() => {
|
|
||||||
tracing::info!("UDP server shutting down");
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Route un paquet entrant vers les abonnés du canal correspondant.
|
|
||||||
///
|
|
||||||
/// # Logique actuelle (placeholder)
|
|
||||||
/// En l'absence de protocole applicatif défini, tous les paquets reçus
|
|
||||||
/// sont logués. Branche ici la logique d'identification du canal
|
|
||||||
/// (ex: lire un header de paquet pour extraire le `channel_id`).
|
|
||||||
async fn route_packet(&self, socket: &UdpSocket, data: &[u8], sender: SocketAddr) {
|
|
||||||
tracing::debug!(%sender, bytes = data.len(), "datagram received");
|
|
||||||
|
|
||||||
// TODO: extraire le channel_id depuis le header du paquet applicatif.
|
|
||||||
// Pour l'instant on utilise un canal de démonstration statique.
|
|
||||||
let channel_id = "default";
|
|
||||||
|
|
||||||
match self.routing_table.subscribers(channel_id) {
|
|
||||||
Some(clients) => {
|
|
||||||
for &client in clients {
|
|
||||||
// Ne pas renvoyer au sender lui-même.
|
|
||||||
if client == sender {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
match socket.send_to(data, client).await {
|
|
||||||
Ok(_) => {
|
|
||||||
self.metrics.inc_sent(data.len() as u64);
|
|
||||||
}
|
|
||||||
Err(err) => {
|
|
||||||
self.metrics.inc_send_error();
|
|
||||||
tracing::warn!(%client, %err, "failed to forward packet");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
None => {
|
|
||||||
self.metrics.inc_dropped();
|
|
||||||
tracing::debug!(%sender, channel = channel_id, "no subscribers, packet dropped");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -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<Self> {
|
||||||
|
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<VoiceMetrics>, 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;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
@@ -0,0 +1,12 @@
|
|||||||
|
pub mod metrics;
|
||||||
|
pub mod room;
|
||||||
|
pub mod service;
|
||||||
|
|
||||||
|
pub use metrics::VoiceMetrics;
|
||||||
|
pub use room::VoiceRoom;
|
||||||
|
pub use service::VoiceService;
|
||||||
|
|
||||||
|
/// Nom générique du moteur média. `VoiceService` reste exporté pour compatibilité
|
||||||
|
/// avec le code de démarrage existant, mais ce service est destiné à l'audio,
|
||||||
|
/// la vidéo et les autres pistes WebRTC.
|
||||||
|
pub type MediaService = VoiceService;
|
||||||
@@ -0,0 +1,27 @@
|
|||||||
|
use rustrtc::media::track::MediaRelay;
|
||||||
|
use rustrtc::peer_connection::PeerConnection;
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
/// Représente une salle vocale correspondant à un canal vocal (`ChannelType::Voice`).
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct VoiceRoom {
|
||||||
|
pub channel_id: Uuid,
|
||||||
|
pub peers: HashMap<Uuid, Arc<PeerConnection>>,
|
||||||
|
pub relays: HashMap<Uuid, Arc<MediaRelay>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl VoiceRoom {
|
||||||
|
pub fn new(channel_id: Uuid) -> Self {
|
||||||
|
Self {
|
||||||
|
channel_id,
|
||||||
|
peers: HashMap::new(),
|
||||||
|
relays: HashMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn is_empty(&self) -> bool {
|
||||||
|
self.peers.is_empty()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,431 @@
|
|||||||
|
use std::collections::HashMap;
|
||||||
|
use std::fmt;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use tokio::sync::{RwLock, mpsc};
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use axum::extract::ws::Message;
|
||||||
|
use rustrtc::config::{RtcConfiguration, RtcConfigurationBuilder};
|
||||||
|
use rustrtc::media::track::MediaRelay;
|
||||||
|
use rustrtc::peer_connection::{PeerConnection, PeerConnectionEvent, RtpCodecParameters};
|
||||||
|
use rustrtc::sdp::{SdpType, SessionDescription};
|
||||||
|
use rustrtc::transports::ice::IceCandidate;
|
||||||
|
|
||||||
|
use crate::config::NetworkConfig;
|
||||||
|
use crate::models::{channel::ChannelType, channel_user, computed_permission::PermissionScopeType};
|
||||||
|
use crate::permissions::ChannelPermission;
|
||||||
|
use crate::repositories::Repositories;
|
||||||
|
use crate::routes::voice::messages::VoiceServerMessage;
|
||||||
|
use crate::voice::metrics::VoiceMetrics;
|
||||||
|
use crate::voice::room::VoiceRoom;
|
||||||
|
|
||||||
|
/// Paramètres de codec audio par défaut (Opus).
|
||||||
|
pub fn opus_codec() -> RtpCodecParameters {
|
||||||
|
RtpCodecParameters {
|
||||||
|
payload_type: 111,
|
||||||
|
name: "opus".to_string(),
|
||||||
|
clock_rate: 48000,
|
||||||
|
channels: 2,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Moteur média WebRTC central (SFU).
|
||||||
|
///
|
||||||
|
/// Le WebSocket ne fait que transporter l'offre SDP, les réponses et les
|
||||||
|
/// candidats ICE. Toute la vie des PeerConnections et des pistes média est
|
||||||
|
/// gérée ici, ce qui permettra d'ajouter la vidéo sans mélanger les protocoles.
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct VoiceService {
|
||||||
|
pub config: RtcConfiguration,
|
||||||
|
pub rooms: Arc<RwLock<HashMap<Uuid, VoiceRoom>>>,
|
||||||
|
pub metrics: Arc<VoiceMetrics>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Debug for VoiceService {
|
||||||
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
|
f.debug_struct("VoiceService")
|
||||||
|
.field("config", &self.config)
|
||||||
|
.field("metrics", &self.metrics)
|
||||||
|
.finish()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl VoiceService {
|
||||||
|
/// Crée une nouvelle instance de [`VoiceService`] configurée avec les paramètres réseau.
|
||||||
|
pub fn new(network: &NetworkConfig, metrics: Arc<VoiceMetrics>) -> 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,
|
||||||
|
rooms: Arc::new(RwLock::new(HashMap::new())),
|
||||||
|
metrics,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Récupère les métriques de la voix.
|
||||||
|
pub fn metrics(&self) -> &Arc<VoiceMetrics> {
|
||||||
|
&self.metrics
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Vérifie si l'utilisateur possède la permission de rejoindre le canal vocal.
|
||||||
|
pub async fn check_permission(
|
||||||
|
repositories: &Repositories,
|
||||||
|
user_id: Uuid,
|
||||||
|
channel_id: Uuid,
|
||||||
|
) -> anyhow::Result<bool> {
|
||||||
|
let channel = repositories
|
||||||
|
.channel
|
||||||
|
.get_by_id(channel_id)
|
||||||
|
.await?
|
||||||
|
.ok_or_else(|| anyhow::anyhow!("Channel not found"))?;
|
||||||
|
|
||||||
|
if channel.channel_type == ChannelType::DM {
|
||||||
|
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
|
||||||
|
let is_member = channel_user::Entity::find()
|
||||||
|
.filter(channel_user::Column::ChannelId.eq(channel_id))
|
||||||
|
.filter(channel_user::Column::UserId.eq(user_id))
|
||||||
|
.one(&repositories.channel.context.db)
|
||||||
|
.await?
|
||||||
|
.is_some();
|
||||||
|
return Ok(is_member);
|
||||||
|
}
|
||||||
|
|
||||||
|
if channel.channel_type != ChannelType::Voice {
|
||||||
|
return Ok(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
let permissions = repositories.computed_permission.get_all().await?;
|
||||||
|
for perm in permissions {
|
||||||
|
if perm.user_id == user_id
|
||||||
|
&& perm.scope_type == PermissionScopeType::Channel
|
||||||
|
&& perm.resource_id == channel_id
|
||||||
|
{
|
||||||
|
let chan_perm = ChannelPermission::from_bits_retain(perm.permissions as u64);
|
||||||
|
if chan_perm.contains(ChannelPermission::JOIN_VOICE) {
|
||||||
|
return Ok(true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(false)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Traite une offre SDP entrante d'un client pour un canal donné.
|
||||||
|
pub async fn handle_offer(
|
||||||
|
&self,
|
||||||
|
user_id: Uuid,
|
||||||
|
channel_id: Uuid,
|
||||||
|
offer_sdp: String,
|
||||||
|
repositories: &Repositories,
|
||||||
|
ice_sender: Option<mpsc::UnboundedSender<Message>>,
|
||||||
|
) -> anyhow::Result<String> {
|
||||||
|
let allowed = Self::check_permission(repositories, user_id, channel_id).await?;
|
||||||
|
if !allowed {
|
||||||
|
anyhow::bail!("Permission denied: cannot join voice channel");
|
||||||
|
}
|
||||||
|
|
||||||
|
let pc = Arc::new(PeerConnection::new(self.config.clone()));
|
||||||
|
|
||||||
|
// Abonner la nouvelle PeerConnection à tous les flux déjà présents dans la room
|
||||||
|
let existing_relays = {
|
||||||
|
let rooms = self.rooms.read().await;
|
||||||
|
rooms
|
||||||
|
.get(&channel_id)
|
||||||
|
.map(|r| r.relays.clone())
|
||||||
|
.unwrap_or_default()
|
||||||
|
};
|
||||||
|
|
||||||
|
for (&other_user_id, relay) in &existing_relays {
|
||||||
|
if other_user_id != user_id {
|
||||||
|
let sub_track = relay.subscribe();
|
||||||
|
if let Err(err) = pc.add_track(sub_track, opus_codec()) {
|
||||||
|
tracing::warn!(%err, "Failed to subscribe new peer to existing track");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let offer_desc = SessionDescription::parse(SdpType::Offer, &offer_sdp)?;
|
||||||
|
pc.set_remote_description(offer_desc).await?;
|
||||||
|
|
||||||
|
let answer_desc = pc.create_answer().await?;
|
||||||
|
let answer_sdp = answer_desc.to_sdp_string();
|
||||||
|
pc.set_local_description(answer_desc)?;
|
||||||
|
|
||||||
|
// Enregistrer la peer connection dans la room
|
||||||
|
{
|
||||||
|
let mut rooms = self.rooms.write().await;
|
||||||
|
let room = rooms
|
||||||
|
.entry(channel_id)
|
||||||
|
.or_insert_with(|| VoiceRoom::new(channel_id));
|
||||||
|
if let Some(old_pc) = room.peers.insert(user_id, Arc::clone(&pc)) {
|
||||||
|
old_pc.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Relayer les candidats ICE locaux vers le client via le WebSocket
|
||||||
|
if let Some(sender) = ice_sender {
|
||||||
|
let mut ice_rx = pc.subscribe_ice_candidates();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
while let Ok(candidate) = ice_rx.recv().await {
|
||||||
|
let cand_sdp = candidate.to_sdp();
|
||||||
|
let event = VoiceServerMessage::IceCandidate {
|
||||||
|
channel_id,
|
||||||
|
candidate: cand_sdp,
|
||||||
|
};
|
||||||
|
if let Ok(json) = serde_json::to_string(&event) {
|
||||||
|
if sender.send(Message::Text(json.into())).is_err() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
self.start_peer_event_loop(pc, user_id, channel_id);
|
||||||
|
|
||||||
|
self.metrics.inc_received(1);
|
||||||
|
Ok(answer_sdp)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Starts the media event loop for one WebRTC connection.
|
||||||
|
///
|
||||||
|
/// ```text
|
||||||
|
/// Client microphone/camera
|
||||||
|
/// |
|
||||||
|
/// | SRTP/RTP packets over ICE/UDP
|
||||||
|
/// v
|
||||||
|
/// rustrtc internal transport <- raw packets are read here
|
||||||
|
/// |
|
||||||
|
/// | PeerConnectionEvent::Track
|
||||||
|
/// v
|
||||||
|
/// this function (the server's media entry point)
|
||||||
|
/// |
|
||||||
|
/// | MediaRelay::subscribe() for every other peer
|
||||||
|
/// v
|
||||||
|
/// other PeerConnections -> their clients
|
||||||
|
/// ```
|
||||||
|
///
|
||||||
|
/// There is deliberately no `UdpSocket::recv` in this service. `rustrtc`
|
||||||
|
/// owns the UDP mux, ICE, SRTP decryption and RTP parsing. `recv()` below
|
||||||
|
/// receives the resulting high-level WebRTC events, especially tracks.
|
||||||
|
fn start_peer_event_loop(&self, pc: Arc<PeerConnection>, user_id: Uuid, channel_id: Uuid) {
|
||||||
|
let rooms = Arc::clone(&self.rooms);
|
||||||
|
let metrics = Arc::clone(&self.metrics);
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
while let Some(event) = pc.recv().await {
|
||||||
|
match event {
|
||||||
|
PeerConnectionEvent::Track(transceiver) => {
|
||||||
|
Self::forward_incoming_track(
|
||||||
|
rooms.as_ref(),
|
||||||
|
metrics.as_ref(),
|
||||||
|
user_id,
|
||||||
|
channel_id,
|
||||||
|
transceiver,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
PeerConnectionEvent::DataChannel(_) => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Converts one received track into a relay and subscribes every other
|
||||||
|
/// peer in the same room to it. This is the SFU fan-out operation.
|
||||||
|
async fn forward_incoming_track(
|
||||||
|
rooms: &RwLock<HashMap<Uuid, VoiceRoom>>,
|
||||||
|
metrics: &VoiceMetrics,
|
||||||
|
user_id: Uuid,
|
||||||
|
channel_id: Uuid,
|
||||||
|
transceiver: Arc<rustrtc::peer_connection::RtpTransceiver>,
|
||||||
|
) {
|
||||||
|
let Some(receiver) = transceiver.receiver() else { return };
|
||||||
|
let relay = Arc::new(MediaRelay::new(receiver.track()));
|
||||||
|
let mut rooms = rooms.write().await;
|
||||||
|
let Some(room) = rooms.get_mut(&channel_id) else { return };
|
||||||
|
room.relays.insert(user_id, Arc::clone(&relay));
|
||||||
|
for (&other_user_id, other_pc) in &room.peers {
|
||||||
|
if other_user_id == user_id { continue; }
|
||||||
|
if let Err(error) = other_pc.add_track(relay.subscribe(), opus_codec()) {
|
||||||
|
tracing::warn!(%error, "Failed to fan-out incoming media track");
|
||||||
|
metrics.inc_send_error();
|
||||||
|
} else {
|
||||||
|
metrics.inc_sent(1);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Traite un candidat ICE reçu d'un client.
|
||||||
|
pub async fn handle_ice_candidate(
|
||||||
|
&self,
|
||||||
|
user_id: Uuid,
|
||||||
|
channel_id: Uuid,
|
||||||
|
candidate_sdp: String,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
let pc = {
|
||||||
|
let rooms = self.rooms.read().await;
|
||||||
|
rooms
|
||||||
|
.get(&channel_id)
|
||||||
|
.and_then(|r| r.peers.get(&user_id))
|
||||||
|
.cloned()
|
||||||
|
};
|
||||||
|
|
||||||
|
if let Some(pc) = pc {
|
||||||
|
let candidate = IceCandidate::from_sdp(&candidate_sdp)?;
|
||||||
|
pc.add_ice_candidate(candidate)?;
|
||||||
|
self.metrics.inc_received(1);
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Quitte un canal vocal pour un utilisateur.
|
||||||
|
pub async fn leave(&self, user_id: Uuid, channel_id: Uuid) {
|
||||||
|
let pc_to_close = {
|
||||||
|
let mut rooms = self.rooms.write().await;
|
||||||
|
if let Some(room) = rooms.get_mut(&channel_id) {
|
||||||
|
let pc = room.peers.remove(&user_id);
|
||||||
|
room.relays.remove(&user_id);
|
||||||
|
if room.is_empty() {
|
||||||
|
rooms.remove(&channel_id);
|
||||||
|
}
|
||||||
|
pc
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if let Some(pc) = pc_to_close {
|
||||||
|
pc.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Quitte tous les canaux vocaux pour un utilisateur (ex. déconnexion WebSocket).
|
||||||
|
pub async fn leave_all(&self, user_id: Uuid) {
|
||||||
|
let mut pcs_to_close = Vec::new();
|
||||||
|
{
|
||||||
|
let mut rooms = self.rooms.write().await;
|
||||||
|
let mut empty_rooms = Vec::new();
|
||||||
|
for (&channel_id, room) in rooms.iter_mut() {
|
||||||
|
if let Some(pc) = room.peers.remove(&user_id) {
|
||||||
|
pcs_to_close.push(pc);
|
||||||
|
room.relays.remove(&user_id);
|
||||||
|
}
|
||||||
|
if room.is_empty() {
|
||||||
|
empty_rooms.push(channel_id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for channel_id in empty_rooms {
|
||||||
|
rooms.remove(&channel_id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for pc in pcs_to_close {
|
||||||
|
pc.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::net::Ipv4Addr;
|
||||||
|
|
||||||
|
fn test_network_config() -> NetworkConfig {
|
||||||
|
NetworkConfig {
|
||||||
|
host: Ipv4Addr::new(127, 0, 0, 1),
|
||||||
|
hostv6: None,
|
||||||
|
tcp_port: 8080,
|
||||||
|
udp_port: 9000,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_voice_service_creation_and_room_lifecycle() {
|
||||||
|
let network = test_network_config();
|
||||||
|
let metrics = VoiceMetrics::new();
|
||||||
|
let service = VoiceService::new(&network, metrics);
|
||||||
|
|
||||||
|
let user1 = Uuid::new_v4();
|
||||||
|
let user2 = Uuid::new_v4();
|
||||||
|
let channel_id = Uuid::new_v4();
|
||||||
|
|
||||||
|
// Check room creation and empty initially
|
||||||
|
assert!(service.rooms.read().await.is_empty());
|
||||||
|
|
||||||
|
// Insert peer connections into room
|
||||||
|
let pc1 = Arc::new(PeerConnection::new(service.config.clone()));
|
||||||
|
let pc2 = Arc::new(PeerConnection::new(service.config.clone()));
|
||||||
|
|
||||||
|
{
|
||||||
|
let mut rooms = service.rooms.write().await;
|
||||||
|
let room = rooms
|
||||||
|
.entry(channel_id)
|
||||||
|
.or_insert_with(|| VoiceRoom::new(channel_id));
|
||||||
|
room.peers.insert(user1, Arc::clone(&pc1));
|
||||||
|
room.peers.insert(user2, Arc::clone(&pc2));
|
||||||
|
}
|
||||||
|
|
||||||
|
assert_eq!(service.rooms.read().await.len(), 1);
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.rooms
|
||||||
|
.read()
|
||||||
|
.await
|
||||||
|
.get(&channel_id)
|
||||||
|
.unwrap()
|
||||||
|
.peers
|
||||||
|
.len(),
|
||||||
|
2
|
||||||
|
);
|
||||||
|
|
||||||
|
// User 1 leaves
|
||||||
|
service.leave(user1, channel_id).await;
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.rooms
|
||||||
|
.read()
|
||||||
|
.await
|
||||||
|
.get(&channel_id)
|
||||||
|
.unwrap()
|
||||||
|
.peers
|
||||||
|
.len(),
|
||||||
|
1
|
||||||
|
);
|
||||||
|
|
||||||
|
// User 2 leaves via leave_all
|
||||||
|
service.leave_all(user2).await;
|
||||||
|
assert!(service.rooms.read().await.is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_opus_codec_parameters() {
|
||||||
|
let codec = opus_codec();
|
||||||
|
assert_eq!(codec.name, "opus");
|
||||||
|
assert_eq!(codec.clock_rate, 48000);
|
||||||
|
assert_eq!(codec.channels, 2);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_voice_metrics_counters_and_rates() {
|
||||||
|
let metrics = VoiceMetrics::new();
|
||||||
|
metrics.inc_received(500);
|
||||||
|
metrics.inc_sent(1000);
|
||||||
|
metrics.inc_dropped();
|
||||||
|
metrics.inc_send_error();
|
||||||
|
metrics.inc_recv_error();
|
||||||
|
|
||||||
|
let snap = metrics.snapshot();
|
||||||
|
assert_eq!(snap.packets_received, 1);
|
||||||
|
assert_eq!(snap.bytes_received, 500);
|
||||||
|
assert_eq!(snap.packets_sent, 1);
|
||||||
|
assert_eq!(snap.bytes_sent, 1000);
|
||||||
|
assert_eq!(snap.packets_dropped, 1);
|
||||||
|
assert_eq!(snap.send_errors, 1);
|
||||||
|
assert_eq!(snap.recv_errors, 1);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user