Rust:axum學習筆記(6) SSE(Server Send Event)服務端推送

接上一篇繼續，SSE也就是服務端推送技術，自html5推出以來基本上各大瀏覽器都已支持，axum自然也支持，參考下面的代碼：

async fn sse_handler(
    TypedHeader(user_agent): TypedHeader<headers::UserAgent>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    println!("`{}` connected", user_agent.as_str());

    let mut i = 0;
    // A `Stream` that repeats an event every second
    let stream = stream::repeat_with(move || {
        i += 1;
        Event::default().data(format!("hi,{}", &i))
    })
    .map(Ok)
    .throttle(Duration::from_secs(3)); //每3秒，向瀏覽器發1次消息

    //每隔1秒發1次保活
    Sse::new(stream).keep_alive(
        axum::response::sse::KeepAlive::new()
            .interval(Duration::from_secs(1))
            .text("keep-alive-text"),
    )
}

上面的代碼，表示每3秒向瀏覽器發1次消息，每秒發1次keep-alive保活，完整代碼如下：

cargo.toml

[package]
name = "sse"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
axum =  {version = "0.4.3", features = ["headers"] }
tokio = { version = "1.0", features = ["full"] }
tower-http = { version = "0.2.0", features = ["fs", "trace"] }
futures = "0.3"
tokio-stream = "0.1"
headers = "0.3"

 main.rs

use axum::{
    extract::TypedHeader,
    response::sse::{Event, Sse},
    routing::get,
    Router,
};
use futures::stream::{self, Stream};
use std::{convert::Infallible, net::SocketAddr, time::Duration};
use tokio_stream::StreamExt as _;

#[tokio::main]
async fn main() {
    // build our application with a route
    let app = Router::new()
        .route("/sse", get(sse_handler))
        .route("/", get(|| async { "Hello, World!" }));

    // run it
    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    axum::Server::bind(&addr)
        .serve(app.into_make_service())
        .await
        .unwrap();
}

async fn sse_handler(
    TypedHeader(user_agent): TypedHeader<headers::UserAgent>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    println!("`{}` connected", user_agent.as_str());

    let mut i = 0;
    // A `Stream` that repeats an event every second
    let stream = stream::repeat_with(move || {
        i += 1;
        Event::default().data(format!("hi,{}", &i))
    })
    .map(Ok)
    .throttle(Duration::from_secs(3)); //每3秒，向瀏覽器發1次消息

    //每隔1秒發1次保活
    Sse::new(stream).keep_alive(
        axum::response::sse::KeepAlive::new()
            .interval(Duration::from_secs(1))
            .text("keep-alive-text"),
    )
}

運行效果：

先訪問http://localhost:3000/ 然后在瀏覽器的console控制台，輸入以下js:

var eventSource = new EventSource('/sse');

eventSource.onmessage = function(event) {
    console.log('Message from server ', event.data);
}

順利的話，就能看到控制台不斷輸出服務端推送過來的數據：

切換到Network面板，可以看到/sse返回的content-type為text/event-stream

如果是chrome瀏覽器，直接訪問/sse，還能看到keep-alive的動態輸出