dashboard: subscribed nodes' messages in the console

Each tick's messages of a subscribed node, listed in the console with
this node's own, from its own stream.

Auth plan M3 (live data from every reachable node).

1 files changed, 22 insertions, 6 deletions

diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 291a76b..8220041 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -9,7 +9,7 @@
 //! sent (the firewall plays a lossy network) but not as received.
 
 use std::collections::{HashMap, VecDeque};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
 
 use anyhow::Result;
@@ -156,7 +156,8 @@ const LOGGED_TEXT: usize = 8 * 1024;
 /// handed.
 #[derive(Debug, Default)]
 pub(crate) struct MessageLog {
-    watching: AtomicBool,
+    /// How many look (`watch`).
+    watching: AtomicUsize,
     inner: Mutex<LogInner>,
 }
 
@@ -230,14 +231,29 @@ fn related(peer: &str, message: &ParsedMessage) -> Option<String> {
 // Read by the dashboard only.
 #[cfg_attr(not(feature = "dashboard"), allow(dead_code))]
 impl MessageLog {
-    /// Keeps messages from now on (`true`), or no more, forgetting those kept.
+    /// One more looks (`true`): messages are kept from now on; or one less (`false`):
+    /// once none looks, no more, and those kept are forgotten.
     pub(crate) fn watch(&self, on: bool) {
-        self.watching.store(on, Ordering::Relaxed);
-        if !on {
+        if on {
+            self.watching.fetch_add(1, Ordering::Relaxed);
+            return;
+        }
+        let was = self
+            .watching
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| {
+                Some(n.saturating_sub(1))
+            })
+            .unwrap_or(0);
+        if was <= 1 {
             self.inner.lock_or_recover().entries.clear();
         }
     }
 
+    /// The latest message's number (`Logged::seq`): those after it are new.
+    pub(crate) fn latest(&self) -> u64 {
+        self.inner.lock_or_recover().next
+    }
+
     pub(crate) fn note(
         &self,
         way: &'static str,
@@ -245,7 +261,7 @@ impl MessageLog {
         message: &ParsedMessage,
         bytes: usize,
     ) {
-        if !self.watching.load(Ordering::Relaxed) {
+        if self.watching.load(Ordering::Relaxed) == 0 {
             return;
         }
         let at = std::time::SystemTime::now()