WebSocket en Rust: aplicaciones en tiempo real avanzadas

WebSocket en Rust: aplicaciones en tiempo real avanzadas

Rust es el lenguaje ideal para crear aplicaciones en tiempo real de alto rendimiento gracias a su seguridad y abstracciones de costo cero. En este artículo exploraremos técnicas avanzadas para trabajar con WebSocket: desde el escalado hasta la autorización y el manejo de errores. Aprenderás a construir un servidor WebSocket listo para producción, capaz de atender miles de conexiones simultáneas.

Arquitectura del servidor WebSocket con Tokio y Tungstenite

Para construir un servidor robusto utilizamos la combinación de tokio (entorno de ejecución asíncrono) y tokio-tungstenite (biblioteca WebSocket). La arquitectura avanzada incluye un pool de conexiones, apagado controlado y manejo de errores.

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("Error en el handshake");\n    \n    let (mut write, mut read) = ws_stream.split();\n    let (tx, mut rx) = mpsc::unbounded_channel::<Message>();\n    \n    // Registro en el 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    // Inicio de tareas\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    // Apagado controlado\n    tokio::select! {\n        _ = send_task => {},\n        _ = recv_task => {},\n    }\n    \n    peer_map.lock().await.remove(&user_id);\n    println!("Cliente {} desconectado", addr);\n}\n

Este código muestra la arquitectura básica: pool de conexiones, comunicación dúplex y eliminación automática al desconectarse.

Escalado: miles de conexiones y agrupación en clústeres

Para manejar miles de conexiones utilizamos varios patrones clave:

\n
• Escalado horizontal — ejecución de múltiples instancias del servidor con un balanceador de carga
\n
• Redis Pub/Sub — intercambio de mensajes entre instancias
\n
• Sharding — distribución por user_id
\n

Ejemplo de integración con Redis para comunicación entre servidores:

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        // Enviar al cliente 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

Para la distribución de la carga utilizamos consistent hashing. Esto garantiza una distribución uniforme de