WebSocket en Rust : applications temps réel avancées

WebSocket en Rust : applications temps réel avancées

Rust est le langage idéal pour créer des applications temps réel performantes grâce à sa sécurité et à ses abstractions à coût zéro. Dans cet article, nous aborderons les techniques avancées de travail avec WebSocket : de la mise à l'échelle à l'authentification et à la gestion des erreurs. Vous apprendrez à construire un serveur WebSocket prêt pour la production, capable de gérer des milliers de connexions simultanées.

Architecture du serveur WebSocket avec Tokio et Tungstenite

Pour construire un serveur fiable, nous utilisons la combinaison tokio (runtime asynchrone) et tokio-tungstenite (bibliothèque WebSocket). L'architecture avancée comprend un pool de connexions, un arrêt gracieux et une gestion des erreurs.

use tokio::net::TcpListener;\nuse tokio_tungstenite::accept_async;\nuse futures_util::{SinkExt, StreamExt};\nuse std::sync::Arc;\nuse tokio::sync::Mutex;\n\ntype PeerMap = Arc<Mutex<HashMap<String, PeerConnection>>>;\n\n#[derive(Clone)]\nstruct PeerConnection {\n    sender: mpsc::UnboundedSender<Message>,\n    user_id: String,\n}\n\nasync fn handle_connection(\n    peer_map: PeerMap,\n    raw_stream: TcpStream,\n    addr: SocketAddr,\n) {\n    let ws_stream = accept_async(raw_stream)\n        .await\n        .expect("Erreur lors de la poignée de main");\n    \n    let (mut write, mut read) = ws_stream.split();\n    let (tx, mut rx) = mpsc::unbounded_channel::<Message>();\n    \n    // Enregistrement dans le pool\n    let user_id = Uuid::new_v4().to_string();\n    peer_map.lock().await.insert(\n        user_id.clone(),\n        PeerConnection { sender: tx, user_id: user_id.clone() }\n    );\n    \n    // Lancement des tâches\n    let send_task = tokio::spawn(async move {\n        while let Some(msg) = rx.recv().await {\n            if write.send(msg).await.is_err() {\n                break;\n            }\n        }\n    });\n    \n    let recv_task = tokio::spawn(async move {\n        while let Some(Ok(msg)) = read.next().await {\n            broadcast_message(&peer_map, &user_id, &msg).await;\n        }\n    });\n    \n    // Arrêt gracieux\n    tokio::select! {\n        _ = send_task => {},\n        _ = recv_task => {},\n    }\n    \n    peer_map.lock().await.remove(&user_id);\n    println!("Client {} déconnecté", addr);\n}\n

Ce code illustre l'architecture de base : un pool de connexions, une communication duplex et une suppression automatique en cas de déconnexion.

Mise à l'échelle : des milliers de connexions et clustering

Pour gérer des milliers de connexions, nous utilisons plusieurs modèles clés :

\n
• Mise à l'échelle horizontale — lancement de plusieurs instances du serveur avec un équilibreur de charge
\n
• Redis Pub/Sub — échange de messages entre instances
\n
• Sharding — répartition par user_id
\n

Exemple d'intégration avec Redis pour l'échange interserveur :

use redis::AsyncCommands;\nuse serde::{Serialize, Deserialize};\n\n#[derive(Serialize, Deserialize)]\nstruct ClusterMessage {\n    from: String,\n    to: String,\n    payload: String,\n    timestamp: u64,\n}\n\nasync fn handle_cluster_message(\n    redis_conn: &mut redis::aio::Connection,\n    peer_map: PeerMap,\n    channel: String,\n) {\n    let mut pubsub = redis_conn.as_pubsub();\n    pubsub.subscribe(channel).await.unwrap();\n    \n    let mut stream = pubsub.on_message();\n    \n    while let Some(msg) = stream.next().await {\n        let payload: String = msg.get_payload().unwrap();\n        let cluster_msg: ClusterMessage = serde_json::from_str(&payload).unwrap();\n        \n        // Envoi au client local\n        if let Some(peer) = peer_map.lock().await.get(&cluster_msg.to) {\n            let _ = peer.sender.send(Message::Text(cluster_msg.payload.into()));\n        }\n    }\n}\n

Pour la répartition de la charge, nous utilisons le hachage cohérent. Cela assure une distribution uniforme de la