3 files changed, 81 insertions, 2 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 52b544a..8cbf412 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -251,6 +251,12 @@ pub(crate) enum ParsedMessage {
     Prune(TreeLink),
     /// Asks the recipient to send a writer's changes at once (`swarm::plumtree`).
     Graft(TreeLink),
+    /// Asks the recipient to be neighbours (`swarm::hyparview`).
+    Neighbour(NeighbourRequest),
+    /// Answers a NEIGHBOUR.
+    NeighbourReply(NeighbourAnswer),
+    /// Tells the recipient we are neighbours no more (`swarm::hyparview`).
+    Disconnect(NeighbourAnswer),
 }
 
 /// A node's proof that it holds the key of its node ID (`identity`), sending from its
@@ -397,6 +403,9 @@ pub(crate) enum MessageKind {
     ProbeReply,
     Prune,
     Graft,
+    Neighbour,
+    NeighbourReply,
+    Disconnect,
 }
 
 impl ParsedMessage {
@@ -418,6 +427,9 @@ impl ParsedMessage {
             ParsedMessage::ProbeReply(_) => MessageKind::ProbeReply,
             ParsedMessage::Prune(_) => MessageKind::Prune,
             ParsedMessage::Graft(_) => MessageKind::Graft,
+            ParsedMessage::Neighbour(_) => MessageKind::Neighbour,
+            ParsedMessage::NeighbourReply(_) => MessageKind::NeighbourReply,
+            ParsedMessage::Disconnect(_) => MessageKind::Disconnect,
         }
     }
 
@@ -441,6 +453,10 @@ impl ParsedMessage {
             ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => {
                 Some((link.epoch, link.epoch_seq))
             }
+            ParsedMessage::Neighbour(request) => Some((request.epoch, request.epoch_seq)),
+            ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => {
+                Some((answer.epoch, answer.epoch_seq))
+            }
             ParsedMessage::Who(_)
             | ParsedMessage::Me(_)
             | ParsedMessage::Ping(_)
@@ -461,6 +477,29 @@ pub(crate) struct TreeLink {
     pub(crate) writer: u64,
 }
 
+/// NEIGHBOUR: the sender asks the recipient to be neighbours (`swarm::hyparview`), as
+/// HyParView's NEIGHBOR; `high` when the sender has none (then the recipient makes room
+/// if it must).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct NeighbourRequest {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    pub(crate) high: bool,
+}
+
+/// NEIGHBOUR_REPLY (whether the recipient of a NEIGHBOUR agreed) and DISCONNECT (never
+/// agreed: neighbours no more) (`swarm::hyparview`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct NeighbourAnswer {
+    #[serde(with = "crate::transport::wire::codec::epoch")]
+    pub(crate) epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub(crate) epoch_seq: u32,
+    pub(crate) agreed: bool,
+}
+
 /// PROBE: whether the recipient reaches `member` (`swarm::working_links`), as SWIM
 /// asks others before it suspects a member. Answered from what the recipient knows,
 /// without contacting the member.
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 501c77b..291a76b 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -57,6 +57,8 @@ pub(crate) struct ByKind {
     pub(crate) probe: Counter,
     /// PRUNE and GRAFT (`swarm::plumtree`).
     pub(crate) plumtree: Counter,
+    /// NEIGHBOUR, NEIGHBOUR_REPLY and DISCONNECT (`swarm::hyparview`).
+    pub(crate) hyparview: Counter,
 }
 
 impl ByKind {
@@ -73,6 +75,7 @@ impl ByKind {
             MessageKind::Links,
             MessageKind::Probe,
             MessageKind::Prune,
+            MessageKind::Neighbour,
         ] {
             let theirs = *other.get(kind);
             self.get_mut(kind).absorb(theirs);
@@ -92,6 +95,9 @@ impl ByKind {
             MessageKind::Links | MessageKind::LinksReply => &self.links,
             MessageKind::Probe | MessageKind::ProbeReply => &self.probe,
             MessageKind::Prune | MessageKind::Graft => &self.plumtree,
+            MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => {
+                &self.hyparview
+            }
         }
     }
 
@@ -108,6 +114,9 @@ impl ByKind {
             MessageKind::Links | MessageKind::LinksReply => &mut self.links,
             MessageKind::Probe | MessageKind::ProbeReply => &mut self.probe,
             MessageKind::Prune | MessageKind::Graft => &mut self.plumtree,
+            MessageKind::Neighbour | MessageKind::NeighbourReply | MessageKind::Disconnect => {
+                &mut self.hyparview
+            }
         }
     }
 }
@@ -302,6 +311,9 @@ fn names(kind: MessageKind) -> (&'static str, &'static str) {
         MessageKind::ProbeReply => ("PROBE_REPLY", "probe"),
         MessageKind::Prune => ("PRUNE", "plumtree"),
         MessageKind::Graft => ("GRAFT", "plumtree"),
+        MessageKind::Neighbour => ("NEIGHBOUR", "hyparview"),
+        MessageKind::NeighbourReply => ("NEIGHBOUR_REPLY", "hyparview"),
+        MessageKind::Disconnect => ("DISCONNECT", "hyparview"),
     }
 }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 729acd3..0733d9c 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 = 37;
+pub(crate) const VERSION: u8 = 38;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -35,6 +35,9 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::ProbeReply => 13,
         MessageKind::Prune => 14,
         MessageKind::Graft => 15,
+        MessageKind::Neighbour => 16,
+        MessageKind::NeighbourReply => 17,
+        MessageKind::Disconnect => 18,
     }
 }
 
@@ -94,6 +97,10 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
         ParsedMessage::Prune(link) | ParsedMessage::Graft(link) => {
             rmp_serde::encode::write(&mut payload, link)?
         }
+        ParsedMessage::Neighbour(request) => rmp_serde::encode::write(&mut payload, request)?,
+        ParsedMessage::NeighbourReply(answer) | ParsedMessage::Disconnect(answer) => {
+            rmp_serde::encode::write(&mut payload, answer)?
+        }
     }
     Ok(payload)
 }
@@ -171,6 +178,12 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::Prune(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::Graft) {
         ParsedMessage::Graft(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Neighbour) {
+        ParsedMessage::Neighbour(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::NeighbourReply) {
+        ParsedMessage::NeighbourReply(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Disconnect) {
+        ParsedMessage::Disconnect(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -551,6 +564,21 @@ pub(crate) fn samples() -> Vec<ParsedMessage> {
             epoch_seq: 1,
             writer: 5,
         }),
+        ParsedMessage::Neighbour(crate::transport::NeighbourRequest {
+            epoch,
+            epoch_seq: 1,
+            high: true,
+        }),
+        ParsedMessage::NeighbourReply(crate::transport::NeighbourAnswer {
+            epoch,
+            epoch_seq: 1,
+            agreed: true,
+        }),
+        ParsedMessage::Disconnect(crate::transport::NeighbourAnswer {
+            epoch,
+            epoch_seq: 1,
+            agreed: false,
+        }),
     ]
 }
 
@@ -582,7 +610,7 @@ mod tests {
     #[test]
     fn every_kind_round_trips() {
         let samples = samples();
-        assert_eq!(samples.len(), 16, "a kind is missing from `samples`");
+        assert_eq!(samples.len(), 19, "a kind is missing from `samples`");
         for message in samples {
             let bytes = encode(&message).unwrap();
             let decoded = decode(&bytes).unwrap();