dashboard: allow message inspection

1 files changed, 33 insertions, 2 deletions

diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index 732c95b..fdc9b4a 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -72,6 +72,9 @@ struct Hub {
     /// Where the document changed since the last tick, to the clients connected:
     /// `{"paths": [...]}`, or `{"all": true}` if too many or not known.
     documents: broadcast::Sender<Arc<String>>,
+    /// The messages exchanged since the last tick, one by one (`traffic::MessageLog`),
+    /// to the clients connected: a JSON list.
+    messages: broadcast::Sender<Arc<String>>,
     control: Mutex<Control>,
     /// Someone asked for it now (`refresh`): a tick at once, members asked again.
     poke: Notify,
@@ -94,11 +97,13 @@ impl Hub {
         let (latest, _) = watch::channel(None);
         let (patches, _) = broadcast::channel(64);
         let (documents, _) = broadcast::channel(64);
+        let (messages, _) = broadcast::channel(64);
         Arc::new(Self {
             state,
             latest,
             patches,
             documents,
+            messages,
             poke: Notify::new(),
             live: AtomicBool::new(true),
             control: Mutex::new(Control {
@@ -146,6 +151,10 @@ 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.
+        let log = &self.state.traffic.log;
+        log.watch(true);
+        let mut logged = 0u64;
         loop {
             let live = self.live.load(Ordering::Relaxed);
             if Instant::now() >= graph_due {
@@ -182,11 +191,19 @@ impl Hub {
             if let Some(changed) = document_changes(&mut changes) {
                 let _ = self.documents.send(Arc::new(changed.to_string()));
             }
+            let messages = log.after(logged, MESSAGES_AT_MOST);
+            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
             };
             if *self.closing.borrow() {
+                log.watch(false);
                 self.control.lock_or_recover().running = false;
                 return;
             }
@@ -198,6 +215,7 @@ impl Hub {
                 }
                 let mut control = self.control.lock_or_recover();
                 if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
+                    log.watch(false);
                     control.running = false;
                     return;
                 }
@@ -210,6 +228,8 @@ impl Hub {
 
 /// Paths changed in a tick listed at most; more: `all`.
 const MAX_CHANGED: usize = 256;
+/// Messages sent to the clients per tick at most, the latest (the console keeps fewer).
+const MESSAGES_AT_MOST: usize = 500;
 
 /// Where the document changed since last asked (`Hub::documents`), if anywhere.
 fn document_changes(
@@ -240,6 +260,7 @@ fn document_changes(
 enum Next {
     Patch(Result<Arc<Published>, broadcast::error::RecvError>),
     Document(Result<Arc<String>, broadcast::error::RecvError>),
+    Messages(Result<Arc<String>, broadcast::error::RecvError>),
     Closing,
 }
 
@@ -507,12 +528,14 @@ async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
 /// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a
 /// `patch` whenever something changed (each with its number as the event's ID; a
 /// client that fell behind gets a `snapshot` again), a `document` where the document
-/// changed (`Hub::documents`, at most every tick), and `offline` when the node stops,
-/// which ends the stream.
+/// changed (`Hub::documents`, at most every tick), `messages` with the messages since
+/// the last tick (`Hub::messages`), and `offline` when the node stops, which ends the
+/// stream.
 async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
     // Subscribed before the snapshot is read: no patch after it is missed.
     let mut patches = hub.patches.subscribe();
     let mut documents = hub.documents.subscribe();
+    let mut messages = hub.messages.subscribe();
     hub.want();
     let first = hub.current().await;
     let stream = async_stream::stream! {
@@ -526,6 +549,7 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
             let next = tokio::select! {
                 received = patches.recv() => Next::Patch(received),
                 changed = documents.recv() => Next::Document(changed),
+                logged = messages.recv() => Next::Messages(logged),
                 () = hub.closed() => Next::Closing,
             };
             let received = match next {
@@ -539,6 +563,13 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
                     continue;
                 }
                 Next::Document(Err(broadcast::error::RecvError::Closed)) => break,
+                Next::Messages(Ok(logged)) => {
+                    yield Ok(Event::default().event("messages").data(logged.as_str()));
+                    continue;
+                }
+                // Missed ones are only missed in the console.
+                Next::Messages(Err(broadcast::error::RecvError::Lagged(_))) => continue,
+                Next::Messages(Err(broadcast::error::RecvError::Closed)) => break,
                 Next::Closing => {
                     yield Ok(offline_event());
                     break;