Notify + Axum SSE

[ekanna]

Hi

I want to use https://github.com/notify-rs/notify for file change notifications and use Axum Server Sent Events for communicating with client. Now how to send file change notification via SSE? Here is my pseudo code.

use axum::{
    Router,
    routing::get,
    response::sse::{Event, KeepAlive, Sse},
};
use std::{time::Duration, convert::Infallible, path::Path};
use tokio_stream::StreamExt as _ ;
use futures_util::stream::{self, Stream};

use notify::{Watcher, RecommendedWatcher, RecursiveMode, Result as NResult};

#[tokio::main]
async fn main() {

    // file change notification code
    fn event_fn(res: NResult<notify::Event>) {
        match res {
           // How to pass this event information to SSE handler? 
           Ok(event) => println!("event: {:?}", event),
           Err(e) => println!("watch error: {:?}", e),
        }
    }
    let mut watcher = notify::recommended_watcher(event_fn)?;
    watcher.watch(Path::new("."), RecursiveMode::Recursive)?;


    // build our application with a single route
    let app = Router::new().route("/sse", get(sse_handler));

    // run it with hyper on localhost:3000
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

async fn sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    // A `Stream` that repeats an event every second
    let stream = stream::repeat_with(|| Event::default().data("hi!"))
        .map(Ok)
        .throttle(Duration::from_secs(1));
    Sse::new(stream).keep_alive(KeepAlive::default())

    // how to send file change notifications based on events rather than time?

}

Thanks

[davidpdrsn]

Send on a channel Sender when a file changes and make an SSE stream from the Receiver.

[ekanna]

Thanks. My working example.

use axum::{
    Router,
    routing::get,
    response::sse::{Event, KeepAlive, Sse},
    extract::State,
};
use tower_http::services::ServeDir;

use std::path::Path;
use futures_util::stream::Stream;

use notify::{Watcher, RecursiveMode, Result as NResult};

use tokio::sync::broadcast;
use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};

#[derive(Clone)]
struct AppState {
    // Channel used to send messages to all connected clients.
    tx: broadcast::Sender<Event>,
}

#[tokio::main]
async fn main() {

    let (tx, _) = broadcast::channel::<Event>(100);
    let app_state = AppState{tx: tx.clone()};

    // file change notification code
    let mut watcher = notify::recommended_watcher(move |res: NResult<notify::Event>| {
        match res {
            // How to pass this event information to SSE handler? 
            Ok(event) => {
             println!("event: {:?}", event);
             let k = Event::default().data(r#"{"Id": 12, "Type": "text", "Data": "hello"}"#);
             _ = tx.send(k);
            },
            Err(e) => println!("watch error: {:?}", e),
         }
    }).expect("watcher");

    // watch for changes in current directory
    watcher.watch(Path::new("."), RecursiveMode::Recursive).expect("watch");

    // serve current directory
    let static_files_service = ServeDir::new(".").append_index_html_on_directories(true);

    // build our application with a single route
    let app = Router::new()
        .fallback_service(static_files_service)
        .route("/sse", get(sse_handler))
        .with_state(app_state);

    println!("server running at http://0.0.0.0:3000");
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

async fn sse_handler(State(app_state): State<AppState>) -> Sse<impl Stream<Item = Result<Event, BroadcastStreamRecvError>>> {
    let rx = app_state.tx.subscribe();

    let mystream = BroadcastStream::new(rx);

    Sse::new(mystream).keep_alive(KeepAlive::default())
}

[jakswa]

Thanks for this thread. Was off lost in compilation errors trying to coerce my Receiver into a Stream, didn't know about BroadcastStream.

[ekanna]

Same is the case with me. I have learnt it from one of fasterthanlime videos https://www.youtube.com/watch?v=c_5Jy_AVDaM