dashboard: subscribed nodes' messages as dots

The page reads a subscribed node's message counts from it too, and
draws them as dots on the map, as this node's own.

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

1 files changed, 48 insertions, 0 deletions

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.