3 files changed, 166 insertions, 60 deletions

diff --git a/src/api/cors.rs b/src/api/cors.rs
index 313d331..2d5fdc3 100644
--- a/src/api/cors.rs
+++ b/src/api/cors.rs
@@ -1,5 +1,6 @@
 //! Pages of other origins reading this node: the dashboards of other members, which a
-//! page subscribes to for their own links (`/admin/dashboard/links/own`), refreshing its
+//! page subscribes to for their own links and traffic (`/admin/dashboard/links/own`,
+//! `/admin/dashboard/traffic/own/events`), refreshing its
 //! token on any node (`/auth/refresh`, `/auth/whoami`). Only those endpoints, only for
 //! the origins of the dashboards the member list names and those of `--cors-origins`,
 //! and without credentials: a page sends its token itself (`Authorization: Bearer`),
@@ -21,6 +22,7 @@ fn open(path: &str) -> bool {
         path,
         "/admin/dashboard/links/own"
             | "/admin/dashboard/links/own/events"
+            | "/admin/dashboard/traffic/own/events"
             | "/auth/refresh"
             | "/auth/whoami"
     )
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index 8e57363..efbf892 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -2298,43 +2298,69 @@ function follow(id, delay = 2000) {
   } catch (_) {
     return fail("its dashboard's address isn't one");
   }
-  (async () => {
-    const response = await fetch(url, { signal: controller.signal, cache: "no-store",
-      headers: { Authorization: "Bearer " + narrowToken.token, Accept: "text/event-stream" } });
-    if (!response.ok) {
-      const answer = await response.json().catch(() => ({}));
-      throw new Error(response.status + " " + (answer.error || response.statusText));
-    }
+  read(url, controller.signal, () => {
     entry.status = "live";
     entry.why = null;
     delay = 2000;
     render();
-    const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
-    let buffer = "";
-    for (;;) {
-      const { value, done } = await reader.read();
-      if (done) throw new Error("its stream ended");
-      buffer += value.replace(/\r/g, "");
-      let end;
-      while ((end = buffer.indexOf("\n\n")) >= 0) {
-        const block = buffer.slice(0, end);
-        buffer = buffer.slice(end + 2);
-        let kind = "message";
-        const data = [];
-        for (const line of block.split("\n")) {
-          if (line.startsWith("event:")) kind = line.slice(6).trim();
-          else if (line.startsWith("data:")) data.push(line.slice(5).replace(/^ /, ""));
-        }
-        if (kind === "links") {
-          entry.report = JSON.parse(data.join("\n"));
-          entry.at = Date.now();
-          render();
-        } else if (kind === "offline") {
-          throw new Error("it stopped");
-        }
+  }, (kind, data) => {
+    if (kind === "links") {
+      entry.report = JSON.parse(data);
+      entry.at = Date.now();
+      render();
+    }
+  }).catch(error => fail(error.message || String(error)));
+  // Its traffic, for its messages on the map: while it goes, quietly; a failure of its
+  // own leaves the links be.
+  let trafficUrl;
+  try {
+    trafficUrl = new URL("dashboard/traffic/own/events", url).href;
+  } catch (_) {
+    return;
+  }
+  const followTraffic = (wait = 2000) => {
+    if (controller.signal.aborted) return;
+    read(trafficUrl, controller.signal, () => { entry.trafficLive = true; wait = 2000; },
+      (kind, data) => { if (kind === "traffic") remoteTrafficArrived(id, entry, JSON.parse(data)); })
+      .catch(() => {
+        entry.trafficLive = false;
+        entry.traffic = null;
+        if (!controller.signal.aborted) setTimeout(() => followTraffic(Math.min(60000, wait * 2)), wait);
+      });
+  };
+  followTraffic();
+}
+// Server-sent events from `url`, read with this page's narrow token (`EventSource`
+// can't send one): `opened` once answered, `event(kind, data)` for each; rejects when it
+// fails or ends (`offline`: the node stopped).
+async function read(url, signal, opened, event) {
+  const response = await fetch(url, { signal, cache: "no-store",
+    headers: { Authorization: "Bearer " + narrowToken.token, Accept: "text/event-stream" } });
+  if (!response.ok) {
+    const answer = await response.json().catch(() => ({}));
+    throw new Error(response.status + " " + (answer.error || response.statusText));
+  }
+  opened();
+  const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
+  let buffer = "";
+  for (;;) {
+    const { value, done } = await reader.read();
+    if (done) throw new Error("its stream ended");
+    buffer += value.replace(/\r/g, "");
+    let end;
+    while ((end = buffer.indexOf("\n\n")) >= 0) {
+      const block = buffer.slice(0, end);
+      buffer = buffer.slice(end + 2);
+      let kind = "message";
+      const data = [];
+      for (const line of block.split("\n")) {
+        if (line.startsWith("event:")) kind = line.slice(6).trim();
+        else if (line.startsWith("data:")) data.push(line.slice(5).replace(/^ /, ""));
       }
+      if (kind === "offline") throw new Error("it stopped");
+      event(kind, data.join("\n"));
     }
-  })().catch(error => fail(error.message || String(error)));
+  }
 }
 // A node's menu (right-click): subscribing to its links, or no longer; its dashboard.
 function nodeMenu(id, x, y) {
@@ -2618,8 +2644,8 @@ graph.addEventListener("mousemove", event => { if (event.target === graph) hover
 graph.addEventListener("mouseleave", () => hover(null));
 
 // --- Messages on the map: this node's traffic with each peer (`/peers`, `traffic`),
-// as it grew since the last update, drawn as dots running along the link the way the
-// messages went. Counts come with each update (every 2 s), so the dots are spread
+// and each subscribed node's with its peers (`remoteTrafficArrived`), as it grew since
+// the last update, drawn as dots running along the link the way the messages went. Counts come with each update (every 2 s), so the dots are spread
 // over the next 2 s, a few at most per link and way: a picture of the traffic, not a
 // record of it. All at one speed (`flowSpeed`), so over a longer link they arrive
 // later. Kept above everything else on the map (`flows`, after what `show` places). ---
@@ -2763,9 +2789,7 @@ function trafficArrived() {
   // arriving at once with it, each of which would add its dots to the same 2 s.
   const wall = Date.now(), gap = wall - lastTrafficAt;
   lastTrafficAt = wall;
-  const timely = gap >= FLOW_SPREAD_MS / 4 && gap <= FLOW_SPREAD_MS * 2.5;
-  const dots = timely && flowsShown && !calm.matches && !document.hidden &&
-    !document.getElementById("map").classList.contains("collapsed");
+  const dots = dotsNow(gap);
   const current = new Map();
   for (const [id, peer] of Object.entries(state.peers.peers)) {
     if (peer.traffic) {
@@ -2773,35 +2797,67 @@ function trafficArrived() {
         received: messages(peer.traffic.received) });
     }
   }
-  if (lastTraffic) {
-    for (const [id, counts] of current) {
-      const before = lastTraffic.get(id);
-      if (!before) continue;
-      for (const [way, from, to] of [["sent", self, id], ["received", id, self]]) {
-        const kinds = {};
-        let total = 0;
-        for (const [kind, count] of Object.entries(counts[way])) {
-          const was = before[way][kind] || { messages: 0, bytes: 0 };
-          const grown = Math.max(0, count.messages - was.messages);
-          if (grown > 0 && (linksShown || !isLinks(kind))) {
-            kinds[kind] = { messages: grown, bytes: Math.max(0, count.bytes - was.bytes) };
-            total += grown;
-          }
+  if (lastTraffic && dots) addFlows(self, current, lastTraffic, () => false);
+  measureRates(current);
+  lastTraffic = current;
+}
+// Dots for `node`'s traffic with each peer, as it grew from `before` to `current` (by
+// peer: `sent`, `received`, each messages and bytes by kind), spread over the next 2 s;
+// but where `skip(way, peer)` (drawn already from another's).
+function addFlows(node, current, before, skip) {
+  const now = performance.now();
+  for (const [id, counts] of current) {
+    const was = before.get(id);
+    if (!was) continue;
+    for (const [way, from, to] of [["sent", node, id], ["received", id, node]]) {
+      if (skip(way, id)) continue;
+      const kinds = {};
+      let total = 0;
+      for (const [kind, count] of Object.entries(counts[way])) {
+        const earlier = was[way][kind] || { messages: 0, bytes: 0 };
+        const grown = Math.max(0, count.messages - earlier.messages);
+        if (grown > 0 && (linksShown || !isLinks(kind))) {
+          kinds[kind] = { messages: grown, bytes: Math.max(0, count.bytes - earlier.bytes) };
+          total += grown;
         }
-        if (total === 0) continue;
-        if (!dots) continue;
-        const shown = dotsFor(kinds, total);
-        shown.forEach((dot, i) => {
-          flows.push({ from, to, start: now + i * FLOW_SPREAD_MS / shown.length,
-            kind: dot.kind, r: dotRadius(dot.bytes) });
-        });
       }
+      if (total === 0) continue;
+      const shown = dotsFor(kinds, total);
+      shown.forEach((dot, i) => {
+        flows.push({ from, to, start: now + i * FLOW_SPREAD_MS / shown.length,
+          kind: dot.kind, r: dotRadius(dot.bytes) });
+      });
     }
   }
-  measureRates(current);
-  lastTraffic = current;
   if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
 }
+// Whether dots are drawn now, for an update whose traffic came `gap` ms after the last:
+// in its time (not the first after a gap, the computer asleep or the stream stalled,
+// whose counts grew over all of it), shown, and seen.
+function dotsNow(gap) {
+  return gap >= FLOW_SPREAD_MS / 4 && gap <= FLOW_SPREAD_MS * 2.5 && flowsShown &&
+    !calm.matches && !document.hidden &&
+    !document.getElementById("map").classList.contains("collapsed");
+}
+// A subscribed node's traffic (`followTraffic`), as `trafficArrived` this node's: its
+// messages with this node are this node's to draw, and what it received from another
+// subscribed node that one's (as what it sent).
+function remoteTrafficArrived(id, entry, traffic) {
+  if (!state) return;
+  const current = new Map();
+  for (const [peer, counts] of Object.entries(traffic)) {
+    current.set(peer, { sent: messages(counts.sent), received: messages(counts.received) });
+  }
+  const wall = Date.now(), gap = wall - (entry.trafficAt || 0);
+  entry.trafficAt = wall;
+  if (entry.traffic && dotsNow(gap)) {
+    const self = state.node.id;
+    addFlows(id, current, entry.traffic, (way, peer) => peer === self ||
+      (way === "received" && peer !== id && subscriptions.has(peer) &&
+        (subscribed.get(peer) || {}).trafficLive));
+  }
+  entry.traffic = current;
+}
 
 // --- The console's rates: messages and bytes per second sent and received, per peer
 // and in all, averaged as load averages are, over 1, 5 and 15 minutes (each sample
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
index d5f0ed8..1d7dbf7 100644
--- a/src/api/dashboard/mod.rs
+++ b/src/api/dashboard/mod.rs
@@ -539,6 +539,10 @@ pub(crate) fn router(state: AppState) -> Router<AppState> {
         .route("/admin/dashboard/tree", get(writer_tree))
         .route("/admin/dashboard/links/own", get(own_links_now))
         .route("/admin/dashboard/links/own/events", get(own_links_events))
+        .route(
+            "/admin/dashboard/traffic/own/events",
+            get(own_traffic_events),
+        )
         .layer(Extension(Hub::new(state, closing)))
 }
 
@@ -594,6 +598,50 @@ async fn own_links_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoRespo
     )
 }
 
+/// This node's traffic with each peer: by node ID, what it sent and received, by kind
+/// (messages and bytes counted since it started, as `/peers` has them).
+async fn own_traffic(state: &AppState) -> JsonValue {
+    let peers = state.peers_snapshot().await;
+    let traffic: Map<String, JsonValue> = peers
+        .iter()
+        .map(|(peer, _)| {
+            let counted = state
+                .traffic
+                .node(peer, &state.transport.identities_of(peer));
+            (
+                peer.clone(),
+                json!({ "sent": counted.sent, "received": counted.received }),
+            )
+        })
+        .collect();
+    JsonValue::Object(traffic)
+}
+
+/// `GET /admin/dashboard/traffic/own/events`: server-sent events, `traffic` with this
+/// node's traffic with each peer (`own_traffic`) every `TICK`, for a page of another
+/// member subscribed to this node to draw its messages on the map, as a page of this
+/// node draws its own from `/peers`; `offline` when the node stops, which ends it.
+async fn own_traffic_events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse {
+    let stream = async_stream::stream! {
+        let mut every = tokio::time::interval(TICK);
+        loop {
+            tokio::select! {
+                _ = every.tick() => {}
+                () = hub.closed() => {
+                    yield Ok::<_, Infallible>(offline_event());
+                    break;
+                }
+            }
+            let traffic = own_traffic(&hub.state).await;
+            yield Ok(Event::default().event("traffic").data(traffic.to_string()));
+        }
+    };
+    (
+        [(super::UNBUFFERED.0, super::UNBUFFERED.1)],
+        Sse::new(stream).keep_alive(KeepAlive::default()),
+    )
+}
+
 /// The page as built into this binary.
 const PAGE: &str = include_str!("dashboard.html");
 /// Where the page says which version of it it is.