diff --git a/src/connection.rs b/src/connection.rs index 5c48f88..d74b333 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -2,6 +2,7 @@ use crate::query::CacheKey; use axum::response::sse::Event; use dashmap::DashMap; use std::collections::HashSet; +use std::sync::Arc; use tokio::sync::mpsc; use uuid::Uuid; @@ -106,26 +107,20 @@ impl ConnectionManager { /// This is the hot path of the reactive pipeline: the fanout task calls /// `get_connections_for_key` on every broadcast invalidation event to find /// which connections need rerendering. +/// +/// Clone is cheap (Arc pointer copy) and all clones share the same underlying +/// maps, so updates from any clone are visible to the fanout task. +#[derive(Clone)] pub struct SubscriptionRegistry { - // DashMap is Arc-backed internally, so clone is cheap. - key_to_conns: DashMap>, - conn_to_keys: DashMap>, -} - -impl Clone for SubscriptionRegistry { - fn clone(&self) -> Self { - Self { - key_to_conns: self.key_to_conns.clone(), - conn_to_keys: self.conn_to_keys.clone(), - } - } + key_to_conns: Arc>>, + conn_to_keys: Arc>>, } impl SubscriptionRegistry { pub fn new() -> Self { Self { - key_to_conns: DashMap::new(), - conn_to_keys: DashMap::new(), + key_to_conns: Arc::new(DashMap::new()), + conn_to_keys: Arc::new(DashMap::new()), } } diff --git a/src/server.rs b/src/server.rs index b6bddfd..9fd7b8b 100644 --- a/src/server.rs +++ b/src/server.rs @@ -109,9 +109,8 @@ pub fn spawn_fanout(state: SharedState) { let conn_ids = state.subscriptions.get_connections_for_key(&key); for conn_id in conn_ids { - let Some(signals_json) = state.connections.get_signals(&conn_id) else { - continue; - }; + let signals_json = state.connections.get_signals(&conn_id) + .unwrap_or_else(|| "{}".to_string()); let Some(sse_tx) = state.connections.get_sender(&conn_id) else { continue; }; @@ -145,17 +144,15 @@ pub fn spawn_fanout(state: SharedState) { Ok((html, keys)) => { state_ref.subscriptions.update(conn_id, keys); - // Build a single datastar-patch-elements SSE event. - // DataStar expects the event name set once, then one - // `data: elements ` entry per line of HTML. - // axum's Event::data() sets a single data field, so - // we pre-format the multi-line data payload manually - // and send it as one data value. + // Build a datastar-patch-elements SSE event. + // Each HTML line gets an `elements ` prefix; axum's + // Event::data() splits on \n and writes a `data:` + // line for each, which is correct SSE multiline format. let data_payload: String = html .lines() .map(|line| format!("elements {}", line)) .collect::>() - .join("\ndata: "); + .join("\n"); let event = Event::default() .event("datastar-patch-elements") @@ -241,7 +238,7 @@ pub async fn sse_handler( .lines() .map(|line| format!("elements {}", line)) .collect::>() - .join("\ndata: "); + .join("\n"); let event = Event::default() .event("datastar-patch-elements") .data(data_payload);