diff options
Diffstat (limited to 'src/api/dashboard/mod.rs')
| -rw-r--r-- | src/api/dashboard/mod.rs | 80 |
1 files changed, 57 insertions, 23 deletions
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index d6485d7..1d63f38 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -37,6 +37,11 @@ use crate::swarm::{members, topology}; use crate::transport::Hex; use links::Link; +/// How long messages logged are gathered before they are sent on, after the first: a +/// busy node sends about ten events a second at most, a quiet one each message soon +/// after it went. +const MESSAGES_GATHERED: Duration = Duration::from_millis(100); + /// How often the snapshot is built and compared, while anyone looks. const TICK: Duration = Duration::from_secs(2); /// How often the member list's graph is recomputed (it costs about `members²`). @@ -91,8 +96,8 @@ struct Hub { /// `{"paths": [...]}`, or `{"all": true}` if too many or not known. Shared by every page /// open; each is told of the paths its user may read at least partly (`changed_for`). documents: broadcast::Sender<Arc<String>>, - /// The messages exchanged since the last tick, one by one (`traffic::MessageLog`), - /// to the clients connected: a JSON list. + /// The messages exchanged, one by one (`traffic::MessageLog`), to the clients + /// connected as they come (`forward_messages`): a JSON list. messages: broadcast::Sender<Arc<String>>, control: Mutex<Control>, /// Someone asked for it now (`refresh`): a tick at once, the graph recomputed. @@ -168,10 +173,11 @@ impl Hub { let mut graph = Graph::default(); let mut graph_due = Instant::now(); let mut changes = self.state.change_events.subscribe(); - // The messages one by one, from now on, while it runs. + // The messages one by one, from now on, while it runs: sent on as they come. let log = &self.state.traffic.log; log.watch(true); - let mut logged = 0u64; + let forwarding = tokio::spawn(Arc::clone(&self).forward_messages()); + let _forwarding = AbortOnDrop(forwarding); loop { if Instant::now() >= graph_due { graph.refresh(&self.state); @@ -204,14 +210,6 @@ impl Hub { if let Some(changed) = document_changes(&mut changes) { let _ = self.documents.send(Arc::new(changed.to_string())); } - let mut messages = log.after(logged, MESSAGES_AT_MOST); - by_node(&self.state, &mut messages).await; - if let Some(last) = messages.last() { - logged = last.seq; - if let Ok(json) = serde_json::to_string(&messages) { - let _ = self.messages.send(Arc::new(json)); - } - } let unwanted = || { self.patches.receiver_count() == 0 && Instant::now() >= self.control.lock_or_recover().wanted_until @@ -236,6 +234,38 @@ impl Hub { } } +impl Hub { + /// The messages logged, to every client connected (`messages`), as they come: + /// gathered for `MESSAGES_GATHERED` after the first, the latest `MESSAGES_AT_MOST`. + /// For as long as the producer runs. + async fn forward_messages(self: Arc<Self>) { + let log = &self.state.traffic.log; + let mut logged = log.latest(); + loop { + log.logged_after(logged).await; + tokio::time::sleep(MESSAGES_GATHERED).await; + let mut messages = log.after(logged, MESSAGES_AT_MOST); + let Some(last) = messages.last() else { + continue; + }; + logged = last.seq; + by_node(&self.state, &mut messages).await; + if let Ok(json) = serde_json::to_string(&messages) { + let _ = self.messages.send(Arc::new(json)); + } + } + } +} + +/// A task stopped when this is dropped. +struct AbortOnDrop(tokio::task::JoinHandle<()>); + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } +} + /// Paths changed in a tick listed at most; more: `all`. /// The messages the paths layer logged under an address (its own PINGs and PONGs) under /// the node there instead, where one is known: the address it proved itself at, or is @@ -536,9 +566,9 @@ impl Drop for Watching { } /// `GET /admin/dashboard/messages/own/events`: server-sent events, `messages` with the -/// messages this node sent and received since the last (every `TICK`, those of the -/// tick, none too, as the page's own `messages` events have them, without each message -/// written out), for a page of another member subscribed to this node: its console lists them, +/// messages this node sent and received since the last, as they come (gathered for +/// `MESSAGES_GATHERED` after the first), as the page's own `messages` events have them, +/// without each message written out, for a page of another member subscribed to this node: its console lists them, /// and its map runs them along this node's links. `offline` when the node stops, which /// ends it. The message log is kept while it is open. async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse { @@ -548,21 +578,25 @@ async fn own_messages_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRe let mut logged = log.latest(); let stream = async_stream::stream! { let _watching = watching; - let mut every = tokio::time::interval(TICK); loop { + // As they come, gathered a moment (`MESSAGES_GATHERED`). + let log = &hub.state.traffic.log; tokio::select! { - _ = every.tick() => {} + () = async { + log.logged_after(logged).await; + tokio::time::sleep(MESSAGES_GATHERED).await; + } => {} () = hub.closed() => { yield Ok::<_, Infallible>(offline_event()); break; } } - // Every tick, none too: the page times its dots by them. - let mut messages = hub.state.traffic.log.after(logged, MESSAGES_AT_MOST); - if let Some(last) = messages.last() { - logged = last.seq; - by_node(&hub.state, &mut messages).await; - } + let mut messages = log.after(logged, MESSAGES_AT_MOST); + let Some(last) = messages.last() else { + continue; + }; + logged = last.seq; + by_node(&hub.state, &mut messages).await; let messages: Vec<JsonValue> = messages .into_iter() .filter_map(|message| { |