add traffic counter

This introduces a new wrapper that counts messages and payload bytes per
peer, split by kind (hello, helloReply, sync, ping, pong), each sent,
received and refused.

The traffic counter goes around the firewall, which means messages
dropped outbound are still counted as sent, and messages dropped inbound
don't count as received.

1 files changed, 12 insertions, 4 deletions

diff --git a/src/main.rs b/src/main.rs
index 972c325..8b6f823 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -34,6 +34,7 @@ use swarm::{bootstrap, peer, sync};
 use transport::Transport;
 use transport::firewall::{Firewall, FirewallTransport};
 use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig};
+use transport::traffic::{CountingTransport, Traffic};
 
 #[derive(Parser, Debug)]
 #[command(version, about = "continuum messaging system", long_about = None)]
@@ -167,6 +168,7 @@ async fn main() -> Result<()> {
     debug!(epoch = %initial_epoch, "Local epoch");
 
     let firewall = Arc::new(Firewall::default());
+    let traffic = Arc::new(Traffic::default());
 
     let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?);
     debug!(local_id = mycelium_transport.local_id(), "Transport ready");
@@ -178,11 +180,17 @@ async fn main() -> Result<()> {
             discovered_swarm_id: None,
         })),
         peer_states: Arc::new(RwLock::new(HashMap::new())),
-        transport: Arc::new(FirewallTransport::new(
-            mycelium_transport,
-            Arc::clone(&firewall),
+        // Counted outside the firewall: what it drops outbound counts as sent (it plays a
+        // lossy network), what it drops inbound never arrived.
+        transport: Arc::new(CountingTransport::new(
+            Arc::new(FirewallTransport::new(
+                mycelium_transport,
+                Arc::clone(&firewall),
+            )),
+            Arc::clone(&traffic),
         )),
         firewall,
+        traffic,
         sync_events: sync_tx,
         change_tx,
         initial_sync_done: Arc::new(AtomicBool::new(false)),
@@ -208,7 +216,7 @@ async fn main() -> Result<()> {
         let inner_state = consumer_state.clone();
         let _ = consumer_state
             .transport
-            .start_consumer(Box::new(move |sender, msg| {
+            .start_consumer(Box::new(move |sender, msg, _payload_len| {
                 let state_clone = inner_state.clone();
                 Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await })
             }))