3 files changed, 82 insertions, 3 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 96f7651..b8b57af 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -409,8 +409,16 @@ pub(crate) struct LinksRequest {
     pub(crate) nonce: u64,
     /// For this long (seconds), a LINKS_REPLY again whenever the links change (a
     /// subscription; the next LINKS renews it). 0: this reply only, and a subscription
-    /// ends.
+    /// ends (a relayed one runs out instead).
     pub(crate) subscribe_secs: u32,
+    /// Members the receiver is to ask on the sender's behalf, which the sender can't
+    /// reach, passing their links on (relayed, `swarm::links`); none: the receiver's own.
+    pub(crate) of: Vec<String>,
+    /// How long relayed links may be held to send several in one LINKS_REPLY, by the
+    /// receiver and the hops after it together (milliseconds; `swarm::links`).
+    pub(crate) batch_ms: u32,
+    /// How many more nodes a relayed request may go through, after the receiver.
+    pub(crate) hops: u8,
 }
 
 /// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them.
@@ -422,7 +430,18 @@ pub(crate) struct LinksReply {
     pub(crate) epoch_seq: u32,
     /// The request's.
     pub(crate) nonce: u64,
-    /// At most `swarm::links::MAX_LINKS`.
+    /// The sender's own, if asked for; at most `swarm::links::MAX_LINKS`.
+    pub(crate) links: Vec<Link>,
+    /// Other members' links, passed on (`LinksRequest::of`).
+    pub(crate) relayed: Vec<RelayedLinks>,
+}
+
+/// A member's links as a LINKS_REPLY passes them on.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct RelayedLinks {
+    pub(crate) member: String,
+    /// How long ago the member reported them (milliseconds).
+    pub(crate) age_ms: u32,
     pub(crate) links: Vec<Link>,
 }
 
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index bf9438f..b7907fb 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -30,6 +30,11 @@ impl Counter {
         self.messages += 1;
         self.bytes += bytes as u64;
     }
+
+    fn absorb(&mut self, other: Counter) {
+        self.messages += other.messages;
+        self.bytes += other.bytes;
+    }
 }
 
 #[derive(Debug, Default, Clone, Copy, Serialize)]
@@ -51,6 +56,37 @@ pub(crate) struct ByKind {
 }
 
 impl ByKind {
+    fn absorb(&mut self, other: &ByKind) {
+        for kind in [
+            MessageKind::Hello,
+            MessageKind::HelloReply,
+            MessageKind::Sync,
+            MessageKind::Status,
+            MessageKind::StatusReply,
+            MessageKind::Blob,
+            MessageKind::Who,
+            MessageKind::Ping,
+            MessageKind::Links,
+        ] {
+            let theirs = *other.get(kind);
+            self.get_mut(kind).absorb(theirs);
+        }
+    }
+
+    fn get(&self, kind: MessageKind) -> &Counter {
+        match kind {
+            MessageKind::Hello => &self.hello,
+            MessageKind::HelloReply => &self.hello_reply,
+            MessageKind::Sync => &self.sync,
+            MessageKind::Status => &self.status,
+            MessageKind::StatusReply => &self.status_reply,
+            MessageKind::Blob => &self.blob,
+            MessageKind::Who | MessageKind::Me => &self.identity,
+            MessageKind::Ping | MessageKind::Pong => &self.paths,
+            MessageKind::Links | MessageKind::LinksReply => &self.links,
+        }
+    }
+
     fn get_mut(&mut self, kind: MessageKind) -> &mut Counter {
         match kind {
             MessageKind::Hello => &mut self.hello,
@@ -282,6 +318,22 @@ impl Traffic {
         self.lock().peers.get(peer).copied()
     }
 
+    /// A node's traffic: counted under it, and under `addresses`, where it is (the
+    /// layers below the node count what they send and receive on their own there: WHO
+    /// and ME, the paths layer's PINGs and PONGs).
+    pub(crate) fn node(&self, node: &str, addresses: &[String]) -> PeerTraffic {
+        let table = self.lock();
+        let mut traffic = table.peers.get(node).copied().unwrap_or_default();
+        for address in addresses.iter().filter(|address| address.as_str() != node) {
+            if let Some(there) = table.peers.get(address) {
+                traffic.sent.absorb(&there.sent);
+                traffic.received.absorb(&there.received);
+                traffic.refused.absorb(&there.refused);
+            }
+        }
+        traffic
+    }
+
     pub(crate) fn total(&self) -> PeerTraffic {
         self.lock().total
     }
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 7717200..0769e9a 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 14;
+pub(crate) const VERSION: u8 = 15;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -428,6 +428,9 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             epoch_seq: 1,
             nonce: 3,
             subscribe_secs: 60,
+            of: vec!["q".into()],
+            batch_ms: 250,
+            hops: 2,
         }),
         ParsedMessage::LinksReply(crate::transport::LinksReply {
             epoch,
@@ -450,6 +453,11 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
                     within_ms: 40,
                 }),
             }],
+            relayed: vec![crate::transport::RelayedLinks {
+                member: "q".into(),
+                age_ms: 1200,
+                links: Vec::new(),
+            }],
         }),
     ]
 }