Server-Sent Events (SSE)

Волга включает встроенную поддержку Server-Sent Events (SSE), что позволяет вам реализовать одностороннюю связь в реальном времени от сервера к клиенту в ваших веб-приложениях. Эта функция доступна по умолчанию и совместима с протоколами HTTP/1 и HTTP/2.

Базовая реализация

В примере ниже показано, как создать простую конечную точку SSE. Она сопоставляет запрос GET с маршрутом /events, устанавливает Content Type как text/event-stream и непрерывно отправляет сообщение "Hello, world!" раз в секунду, пока клиент не отключится:

use volga::{App, http::sse::Message, sse_stream};
use std::time::Duration;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut app = App::new();

    app.map_get("/events", || async {
        // Создаем поток сообщений с отправкой раз в секунду
        sse_stream! {
            loop {
                yield Message::new().data("Hello, world!");
            
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    });

    app.run().await
}

Настройка сообщений

Волга предоставляет структуру Message, которая поможет вам создавать и настраивать сообщения SSE.

Для простых текстовых сообщений используйте метод data(), как показано выше. Если вам необходимо отправить структурированные данные, такие как JSON, используйте метод json(), который принимает любой тип, реализующий типаж serde::Serialize:

use volga::http::sse::Message;
use serde::Serialize;

#[derive(Serialize)]
struct SseData {
    data: String,
}

let payload = SseData { data: "Hello, world!".into() };
Message::new().json(payload);

В дополнение к этому, Message также поддерживает настройку имени события (поле event), идентификатора (поле id), комментариев (поле comment) и интервала попыток повторного подключения клиента (поле retry). Подробную информацию о формате сообщения SSE см. в руководстве MDN.

Завершение потока при остановке сервера

SSE-поток, который сам по себе не заканчивается, держит соединение открытым, пока его не закроет таймаут завершения. Чтобы завершить поток сразу, как только началась остановка, примите экстрактор ShutdownHandle и оборвите поток по его future cancelled():

use std::time::Duration;
use futures_util::StreamExt;
use volga::{App, ShutdownHandle, http::sse::{Message, SseStream}, sse_stream};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mut app = App::new();

    app.map_get("/events", |shutdown: ShutdownHandle| async move {
        let events = sse_stream! {
            loop {
                yield Message::new().data("Hello, world!");
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        };
        SseStream::new(events.take_until(shutdown.cancelled()))
    });

    app.run().await
}

Полный пример использования можно так же найти по по ссылке.

Последнее обновление: