WebSocket на Rust: создаём real-time приложения

Введение: почему WebSocket на Rust — это мощный тандем

Современные веб-приложения всё чаще требуют мгновенной передачи данных: чаты, онлайн-игры, биржевые котировки, уведомления. Традиционный HTTP с его запрос-ответной моделью здесь неэффективен — он создаёт лишнюю нагрузку и задержки. На помощь приходит протокол WebSocket, который устанавливает постоянное двустороннее соединение между клиентом и сервером.

Язык Rust с его акцентом на производительность, безопасность памяти и нулевую стоимость абстракций идеально подходит для создания высоконагруженных real-time систем. В этой статье мы разберём, как построить WebSocket-сервер на Rust с использованием библиотеки tokio-tungstenite, и рассмотрим ключевые паттерны для real-time приложений.

Основы WebSocket и подготовка окружения

Что такое WebSocket?

WebSocket — это протокол, работающий поверх TCP. Он позволяет серверу и клиенту обмениваться сообщениями в реальном времени без необходимости повторных HTTP-рукопожатий. Соединение начинается как HTTP-запрос с апгрейдом до WebSocket, после чего данные передаются в фреймах.

Установка зависимостей

Для работы нам понадобится tokio (асинхронный рантайм) и tokio-tungstenite (реализация WebSocket). Создайте новый проект:

cargo new websocket_demo
cd websocket_demo

Добавьте зависимости в Cargo.toml:

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-tungstenite = "0.21"
futures-util = "0.3"

Создание простого WebSocket-сервера

Базовый эхо-сервер

Начнём с простого сервера, который принимает соединения и отправляет обратно полученные сообщения (эхо-режим). Это классический «Hello, World» для WebSocket.

use tokio::net::TcpListener;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
use futures_util::StreamExt;

#[tokio::main]
async fn main() {
    let addr = "127.0.0.1:8080";
    let listener = TcpListener::bind(addr).await.expect("Не удалось привязать адрес");
    println!("Сервер запущен на {}", addr);

    while let Ok((stream, peer)) = listener.accept().await {
        println!("Новое подключение: {}", peer);
        tokio::spawn(handle_connection(stream));
    }
}

async fn handle_connection(stream: tokio::net::TcpStream) {
    let ws_stream = accept_async(stream)
        .await
        .expect("Ошибка при handshake WebSocket");

    let (_, mut read) = ws_stream.split();

    while let Some(msg) = read.next().await {
        match msg {
            Ok(Message::Text(text)) => {
                println!("Получено: {}", text);
                // Здесь можно отправить ответ, но для простоты пропустим
            }
            Ok(Message::Close(_)) => break,
            _ => {}
        }
    }
}

Этот код запускает TCP-сервер на порту 8080. Для каждого нового соединения создаётся асинхронная задача handle_connection, которая выполняет WebSocket-рукопожатие и читает входящие сообщения.

Добавляем отправку сообщений

Чтобы сделать полноценный эхо-сервер, нам нужно разделить поток на чтение и запись. Используем futures_util::StreamExt и SinkExt:

use futures_util::SinkExt;

async fn handle_connection(stream: tokio::net::TcpStream) {
    let ws_stream = accept_async(stream)
        .await
        .expect("Ошибка при handshake WebSocket");

    let (mut write, mut read) = ws_stream.split();

    while let Some(msg) = read.next().await {
        match msg {
            Ok(Message::Text(text)) => {
                let echo = format!("Эхо: {}", text);
                write.send(Message::Text(echo.into())).await.unwrap();
            }
            Ok(Message::Close(_)) => break,
            _ => {}
        }
    }
}

Теперь клиент получает обратно своё сообщение с префиксом «Эхо:». Это основа, на которой строятся более сложные приложения.

Real-time чат: управление подключениями

Архитектура с общим состоянием

Для чата нам нужно хранить список всех активных подключений и рассылать сообщения всем участникам. Используем tokio::sync::broadcast — канал, который отправляет данные всем подписчикам.

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let addr = "127.0.0.1:8080";
    let listener = TcpListener::bind(addr).await.unwrap();
    let (tx, _rx) = broadcast::channel::<String>(100);

    println!("Чат-сервер запущен на {}", addr);

    while let Ok((stream, peer)) = listener.accept().await {
        let tx = tx.clone();
        let rx = tx.subscribe();
        tokio::spawn(handle_chat_client(stream, peer, tx, rx));
    }
}

Обработка клиента чата

Каждый клиент получает свой экземпляр приёмника rx и передатчик tx. Задача: читать сообщения от клиента и отправлять их в канал, а также слушать канал и отправлять сообщения клиенту.

async fn handle_chat_client(
    stream: tokio::net::TcpStream,
    peer: std::net::SocketAddr,
    tx: broadcast::Sender<String>,
    mut rx: broadcast::Receiver<String>,
) {
    let ws_stream = accept_async(stream).await.unwrap();
    let (mut write, mut read) = ws_stream.split();

    // Приветственное сообщение
    let welcome = format!("{} присоединился к чату", peer);
    let _ = tx.send(welcome);

    loop {
        tokio::select! {
            // Пришло сообщение от клиента
            msg = read.next() => {
                match msg {
                    Some(Ok(Message::Text(text))) => {
                        let broadcast_msg = format!("{}: {}", peer, text);
                        let _ = tx.send(broadcast_msg);
                    }
                    Some(Ok(Message::Close(_))) | None => break,
                    _ => {}
                }
            }
            // Пришло сообщение от других клиентов
            msg = rx.recv() => {
                match msg {
                    Ok(text) => {
                        if write.send(Message::Text(text.into())).await.is_err() {
                            break;
                        }
                    }
                    Err(_) => break,
                }
            }
        }
    }

    // Уведомление о выходе
    let goodbye = format!("{}