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, 26 insertions, 1 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 19c4126..b2f5c9d 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -11,6 +11,7 @@ pub use crate::document::replica::Epoch;
 
 pub mod firewall;
 pub mod mycelium;
+pub mod traffic;
 
 /// Random identifier of one run of a node (from start to exit).
 pub type RunId = [u8; 16];
@@ -101,7 +102,27 @@ pub enum ParsedMessage {
     Pong(Keepalive),
 }
 
+/// What a message is, without its contents.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MessageKind {
+    Hello,
+    HelloReply,
+    Sync,
+    Ping,
+    Pong,
+}
+
 impl ParsedMessage {
+    pub fn kind(&self) -> MessageKind {
+        match self {
+            ParsedMessage::Hello(_) => MessageKind::Hello,
+            ParsedMessage::HelloReply(_) => MessageKind::HelloReply,
+            ParsedMessage::Sync(_) => MessageKind::Sync,
+            ParsedMessage::Ping(_) => MessageKind::Ping,
+            ParsedMessage::Pong(_) => MessageKind::Pong,
+        }
+    }
+
     /// Every message carries the sender's epoch.
     pub fn epoch(&self) -> Epoch {
         match self {
@@ -161,9 +182,13 @@ pub struct SyncEnvelope {
 pub struct EncodedMessage {
     pub peer: String,
     pub body: bytes::Bytes,
+    pub kind: MessageKind,
+    /// Bytes the message takes on the wire (without the transport's own framing).
+    pub payload_len: usize,
 }
 
-pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
+/// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`).
+pub type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
     + Send
     + Sync
     + 'static;