WebSocket em Rust: aplicações avançadas em tempo real

WebSocket em Rust: aplicações avançadas em tempo real

Rust é a linguagem ideal para criar aplicações de alto desempenho em tempo real, graças à sua segurança e abstrações de custo zero. Neste artigo, exploraremos técnicas avançadas de trabalho com WebSocket: desde escalabilidade até autorização e tratamento de erros. Você aprenderá como construir um servidor WebSocket pronto para produção, capaz de atender milhares de conexões simultâneas.

Arquitetura do servidor WebSocket com Tokio e Tungstenite

Para construir um servidor robusto, utilizamos a combinação de tokio (runtime assíncrono) e tokio-tungstenite (biblioteca WebSocket). A arquitetura avançada inclui pool de conexões, desligamento gracioso e tratamento de erros.

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("Erro no handshake");\n    \n    let (mut write, mut read) = ws_stream.split();\n    let (tx, mut rx) = mpsc::unbounded_channel::<Message>();\n    \n    // Registro no 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    // Iniciar tarefas\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    // Desligamento gracioso\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 demonstra a arquitetura básica: pool de conexões, comunicação duplex e remoção automática em caso de desconexão.

Escalabilidade: milhares de conexões e clusterização

Para processar milhares de conexões, utilizamos vários padrões-chave:

\n
• Escalabilidade horizontal — execução de várias instâncias do servidor com balanceador de carga
\n
• Redis Pub/Sub — troca de mensagens entre instâncias
\n
• Sharding — distribuição por user_id
\n

Exemplo de integração com Redis para troca 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 para o 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 distribuição de carga, usamos hashing consistente. Isso garante uma distribuição uniforme com