Server-Sent Events in Rust

What is a Server-Sent Events?

It is a standard that enables the servers to send a stream of events to the subscribed clients.

Why do we need SSE?

SSE helps to get real-time data from the server. Web sockets can do the same but they are bidirectional and the client can also send the data to the server.

They can be overkill for the applications which need to send updates to the clients only. So we need SSE for that purpose.

Some of the Uses Cases

• Twitter feeds
• Score Updates
• Stocks rates

How does SSE works?

When a client requests, it forms an EventSource Object with the url of the server endpoint over the regular HTTP request.

The server responds back with an event stream and keeps the connection until it feels that it has no events to send (or stable) or the client explicitly close the connection.

How to implement SSE with actix-web?

Let's follow the easy steps:-

Libraries required:-

• actix-web:- A powerful, pragmatic, and extremely fast web framework for Rust
• actix-web-lab:- Experimental extractors, middleware, and other extras for possible inclusion in Actix Web.

1. Setting Up Actix-web

Initializing the Rust Project

Start a new project with the following cargo new <file-name>.

Implementing basic actix server

Let's clone the sample code from actix official docs.

Add dependency in cargo.toml

....
[dependencies]
actix-web = "4"
#sse
actix-web-lab = "0.18.5"
parking_lot = "0.12.1"
futures-util = { version = "0.3.25", default-features = false, features = ["std"] }
....

Let's write code for actix server.

use actix_web::{web, App, HttpServer};

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .route("/hello", web::get().to(|| async { "Hello World!" }))
    })
    .bind(("127.0.0.1", 8080))?
    .run()
    .await
}

Implementing SSE using actix-web-lab

Broadcaster

• Initialise the client struct
  

  #[derive(Debug, Clone, Default)]
  struct BroadcasterInner {
      clients: Vec<sse::Sender>,
  }
  

  • The Debug trait is used for printing or debugging
  • The Clone trait let you clone the BroadcasterInner/ clients
  • Default trait lets you assign default value on initialisation

• Initialise the Broadcaster struct
  
  

    pub struct Broadcaster {
        inner: Mutex<BroadcasterInner>,
    }

We have used the Mutex to mutate the value safely across the threads.

• Creating methods to implement ping, create a new instance, remove stale client, create new client, and broadcast message.

  • Create a new broadcaster instance
    

    impl Broadcaster {
        ....
        pub fn create() -> Arc<Self> {
            let this = Arc::new(Broadcaster {
                inner: Mutex::new(BroadcasterInner::default()),
            });
            Broadcaster::spawn_ping(Arc::clone(&this));
            this
        }
        ....
    }
    

    This method will create a broadcaster instance which will ping the clients.

  • ping client
    

    fn spawn_ping(this: Arc<Self>) {
        actix_web::rt::spawn(async move {
            let mut interval = interval(Duration::from_secs(10));
    
            loop {
                interval.tick().await;
                this.remove_stale_clients().await;
            }
        });
    }
    

    This ping method is used to know the active clients and remove the stable ones. You can modify the logic as per your requirement.

  • Remove stale clients
    

    async fn remove_stale_clients(&self) {
        let clients = self.inner.lock().clients.clone();
        println!("active client {:?}",clients);
    
        let mut ok_clients = Vec::new();
    
        println!("okay active client {:?}",ok_clients);
    
        for client in clients {
            if client
                .send(sse::Event::Comment("ping".into()))
                .await
                .is_ok()
            {
                ok_clients.push(client.clone());
            }
        }
    
        self.inner.lock().clients = ok_clients;
    }
    

    This method will revalidate the client list and removes stale one.

  • Create a new client
    

    pub async fn new_client(&self) -> Sse<ChannelStream> {
        println!("starting creation");
        let (tx, rx) = sse::channel(10);
    
        tx.send(sse::Data::new("connected")).await.unwrap();
        println!("creating new clients success {:?}",tx);
        self.inner.lock().clients.push(tx);
        rx
    }
    

    sse::channel will create sender and receiver pair. The channel will take a buffer parameter which is the count of unsend messages stored without waiting.

    The first part -> sender can be cloned and shared across threads.

    The second part -> receiver is the regular event stream.

  • Broadcast new message
    

    pub async fn broadcast(&self, msg: &str) {
        let clients = self.inner.lock().clients.clone();
    
        let send_futures = clients
            .iter()
            .map(|client| client.send(sse::Data::new(msg)));
    
        // try to send to all clients, ignoring failures
        // disconnected clients will get swept up by `remove_stale_clients`
        let _ = future::join_all(send_futures).await;
    }
    

    It will broadcast messages to the active clients.

  • Complete Broadcaster
    

    use std::{sync::Arc, time::Duration};
    use actix_web::rt::time::interval;
    use actix_web_lab::sse::{self, ChannelStream, Sse};
    use futures_util::future;
    use parking_lot::Mutex;
    
    pub struct Broadcaster {
        inner: Mutex<BroadcasterInner>,
    }
    
    #[derive(Debug, Clone, Default)]
    struct BroadcasterInner {
        clients: Vec<sse::Sender>,
    }
    
    impl Broadcaster {
        /// Constructs new broadcaster and spawns ping loop.
        pub fn create() -> Arc<Self> {
            let this = Arc::new(Broadcaster {
                inner: Mutex::new(BroadcasterInner::default()),
            });
            Broadcaster::spawn_ping(Arc::clone(&this));
            // println!("created success");
    
            this
        }
    
        /// Pings clients every 10 seconds to see if they are alive and remove them from the broadcast list if not.
        fn spawn_ping(this: Arc<Self>) {
            actix_web::rt::spawn(async move {
                let mut interval = interval(Duration::from_secs(10));
    
                loop {
                    interval.tick().await;
                    this.remove_stale_clients().await;
                }
            });
        }
    
        /// Removes all non-responsive clients from broadcast list.
        async fn remove_stale_clients(&self) {
            let clients = self.inner.lock().clients.clone();
            println!("active client {:?}",clients);
    
            let mut ok_clients = Vec::new();
    
            println!("okay active client {:?}",ok_clients);
    
            for client in clients {
                if client
                    .send(sse::Event::Comment("ping".into()))
                    .await
                    .is_ok()
                {
                    ok_clients.push(client.clone());
                }
            }
    
            self.inner.lock().clients = ok_clients;
        }
    
        /// Registers client with broadcaster, returning an SSE response body.
        pub async fn new_client(&self) -> Sse<ChannelStream> {
            println!("starting creation");
            let (tx, rx) = sse::channel(10);
    
            tx.send(sse::Data::new("connected")).await.unwrap();
            println!("creating new clients success {:?}",tx);
            self.inner.lock().clients.push(tx);
            rx
        }
    
        /// Broadcasts `msg` to all clients.
        pub async fn broadcast(&self, msg: &str) {
            let clients = self.inner.lock().clients.clone();
    
            let send_futures = clients
                .iter()
                .map(|client| client.send(sse::Data::new(msg)));
    
            // try to send to all clients, ignoring failures
            // disconnected clients will get swept up by `remove_stale_clients`
            let _ = future::join_all(send_futures).await;
        }
    }
    

• Creating a route for the sse stream and broadcast message
  
  

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let broadcaster = Broadcaster::create();

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(AppState {
                broadcaster: Arc::clone(&broadcaster)
            }))
            // This route is used to listen events/ sse events
            .route("/events{_:/?}", web::get().to(sse_client))
            // This route will create notification
            .route("/events/{msg}", web::get().to(broadcast_msg))
    })
    .bind(format!("{}:{}","127.0.0.1", "8000"))?
    .run()
    .await
}

• Let's implement the sse_client and broadcast_msg

  • sse_client
    

    pub async fn sse_client(state: web::Data<AppState>) -> impl Responder {
        state.broadcaster.new_client().await
    }
    
    

  • broadcast_msg
    

    pub async fn broadcast_msg(
        state: web::Data<AppState>,
        Path((msg,)): Path<(String,)>,
    ) -> impl Responder {
        state.broadcaster.broadcast(&msg).await;
        HttpResponse::Ok().body("msg sent")
    }
    

  • Complete main.rs
    

    use actix_web::HttpResponse;
    use actix_web::Responder;
    use actix_web::{web, App, HttpServer};
    mod broadcast;
    use self::broadcast::Broadcaster;
    use std::{io, sync::Arc};
    use actix_web_lab::extract::Path;
    
    pub struct  AppState{
        broadcaster:Arc<Broadcaster>
    }
    
    // SSE
    pub async fn sse_client(state: web::Data<AppState>) -> impl Responder {
        state.broadcaster.new_client().await
    }
    
    pub async fn broadcast_msg(
        state: web::Data<AppState>,
        Path((msg,)): Path<(String,)>,
    ) -> impl Responder {
        state.broadcaster.broadcast(&msg).await;
        HttpResponse::Ok().body("msg sent")
    }
    
    #[actix_web::main]
    async fn main() -> std::io::Result<()> {
        let broadcaster = Broadcaster::create();
    
        HttpServer::new(move || {
            App::new()
                .app_data(web::Data::new(AppState {
                    broadcaster: Arc::clone(&broadcaster)
                }))
                // This route is used to listen to events/ sse events
                .route("/events{_:/?}", web::get().to(sse_client))
                // This route will create a notification
                .route("/events/{msg}", web::get().to(broadcast_msg))
        })
        .bind(format!("{}:{}","127.0.0.1", "8000"))?
        .run()
        .await
    }
    

• Testing the SSE

  • Open the /events to subscribe to the event stream
  • Then hit the /events/ to send a message to the clients.

Hurray! we have finally implemented the Server-Sent Events.

Feel free to ask queries and make pull request for changes and suggestion in GitHub.

Source Code GitHub

Happy Hacking
Rustaceans!