Server-Sent Events на Rust: отправляем данные в реальном времени без WebSocket

Когда WebSocket — это слишком, или почему SSE снова в моде

Представьте: вам нужно отправлять данные с сервера клиенту в реальном времени. Курсы валют, уведомления о новых сообщениях, логи сервера, прогресс обработки файла. Первое, что приходит в голову — WebSocket. Но что, если вам не нужен двусторонний канал? Что, если клиент только слушает, а сервер только говорит?

Здесь на сцену выходят Server-Sent Events (SSE) — протокол, который позволяет серверу отправлять данные клиенту через обычное HTTP-соединение. Без лишних хэндшейков, без сложных протоколов, без головной боли с прокси и балансировщиками. И да, на Rust это работает просто божественно.

В этой статье мы разберём, как поднять SSE на Rust с нуля, используя популярные фреймворки. Вы увидите рабочие примеры, узнаете о подводных камнях и поймёте, когда SSE — ваш лучший выбор.

Что такое Server-Sent Events и почему Rust?

SSE — это стандарт W3C (EventSource API), который позволяет серверу отправлять потоковые данные клиенту по HTTP. Клиент подключается к эндпоинту, и сервер начинает слать сообщения в формате text/event-stream. Всё работает поверх обычного HTTP/1.1 или HTTP/2, не требует специальных прокси и поддерживается всеми современными браузерами.

Почему Rust? Потому что Rust даёт нам:

• Асинхронность без сюрпризов — Tokio и async-std позволяют держать тысячи SSE-коннектов на одном потоке
• Безопасность памяти — никаких утечек, никаких race conditions в многопоточном стриминге
• Минимальное потребление ресурсов — одно SSE-соединение потребляет единицы килобайт
• Экосистему — Actix Web, Rocket, Axum — все поддерживают SSE из коробки или с минимальными расширениями

Как работает SSE: разбираем протокол

Клиент создаёт EventSource в браузере, указывая URL. Сервер отвечает с заголовком Content-Type: text/event-stream и начинает слать строки в формате:

data: Привет, мир!\n\n

Каждое сообщение — это блок, где data: содержит полезную нагрузку. Сообщения разделяются двумя символами новой строки. Можно передавать JSON, числа, строки — что угодно. Дополнительно есть поля id, event и retry.

Вот минимальный пример SSE-сообщения:

id: 1
event: update
data: {"status": "processing", "progress": 42}

Клиент на JavaScript получает это как событие update с полем data, содержащим JSON.

Пишем SSE-сервер на Rust с Actix Web

Начнём с самого популярного фреймворка — Actix Web. Установим зависимости:

[dependencies]
actix-web = "4"
actix-rt = "2"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

Теперь напишем простой эндпоинт, который будет слать текущее время каждые 2 секунды:

use actix_web::{get, web, App, HttpRequest, HttpResponse, HttpServer, middleware};
use actix_web::rt::time;
use futures::stream::StreamExt;
use std::time::Duration;

#[get("/events")]
async fn stream_events(req: HttpRequest) -> HttpResponse {
    let mut rx = tokio::sync::broadcast::channel::<String>(100).1;

    let stream = async_stream::stream! {
        loop {
            let message = format!("data: {}\n\n", chrono::Local::now());
            yield Ok::<_, actix_web::Error>(
                web::Bytes::from(message)
            );
            time::sleep(Duration::from_secs(2)).await;
        }
    };

    HttpResponse::Ok()
        .insert_header(("Content-Type", "text/event-stream"))
        .insert_header(("Cache-Control", "no-cache"))
        .insert_header(("Connection", "keep-alive"))
        .streaming(stream)
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .service(stream_events)
    })
    .bind("127.0.0.1:8080")?
    .run()
    .await
}

Это база. Но в реальном приложении вы, скорее всего, захотите отправлять не время, а данные из бизнес-логики. Давай улучшим пример.

Отправляем структурированные данные через SSE

Допустим, мы отслеживаем статус задач. Создадим канал, через который разные части приложения могут отправлять обновления всем подключённым клиентам:

use tokio::sync::broadcast;
use serde::Serialize;

#[derive(Serialize, Clone)]
struct TaskUpdate {
    task_id: String,
    status: String,
    progress: u8,
}

#[get("/tasks/stream")]
async fn task_stream(rx: web::Data<broadcast::Sender<TaskUpdate>>) -> HttpResponse {
    let mut rx = rx.subscribe();

    let stream = async_stream::stream! {
        while let Ok(update) = rx.recv().await {
            let json = serde_json::to_string(&update).unwrap();
            let message = format!("event: task_update\ndata: {}\n\n", json);
            yield Ok::<_, actix_web::Error>(web::Bytes::from(message));
        }
    };

    HttpResponse::Ok()
        .insert_header(("Content-Type", "text/event-stream"))
        .streaming(stream)
}

Теперь любой модуль вашего приложения может сделать tx.send(update), и все клиенты получат обновление мгновенно.

SSE на Rocket: элегантный подход

Rocket — ещё один мощный фреймворк для Rust. Он славится своей эргономикой. Для SSE нам понадобится крейт rocket с поддержкой потоков:

[dependencies]
rocket = { version = "0.5", features = ["secrets"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

Пример эндпоинта:

use rocket::{get, routes, State, Shutdown};
use rocket::response::stream::{Event, EventStream};
use rocket::tokio::sync::broadcast;
use rocket::tokio::select;

#[get("/events")]
async fn events(tx: &State<broadcast::Sender<String>>, mut end: Shutdown) -> EventStream![] {
    let mut rx = tx.subscribe();
    EventStream! {
        loop {
            select! {
                msg = rx.recv() => {
                    match msg {
                        Ok(data) => yield Event::data(data),
                        Err(_) => break,
                    }
                }
                _ = &mut end => break,
            }
        }
    }
}

Красота Rocket в том, что Shutdown обрабатывается автоматически — при завершении сервера все стримы корректно закрываются.

Продвинутые техники: heartbeat, реконнект