1
0
mirror of https://github.com/actix/examples synced 2025-03-15 01:06:25 +01:00
2022-10-26 16:51:27 +01:00

85 lines
2.3 KiB
Rust

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));
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();
let mut ok_clients = Vec::new();
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> {
let (tx, rx) = sse::channel(10);
tx.send(sse::Data::new("connected")).await.unwrap();
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;
}
}