node-data: add protocol version to Message

Resolves #2197

node-data/Cargo.toml

@@ -24,6 +24,7 @@ bs58 = { version = "0.4" }
 tracing = "0.1"
 anyhow = "1"
 thiserror = "1"
+varint-rs = "2.2"
 
 
 [dev-dependencies]

node-data/src/message.rs

@@ -24,6 +24,42 @@ use async_channel::TrySendError;
 
 use self::payload::{Candidate, Ratification, Validation};
 
+use varint_rs::VarintReader;
+use varint_rs::VarintWriter;
+
+pub const PROTOCOL_VERSION: Version = Version(1, 0, 0);
+
+#[derive(Debug, Clone)]
+/// Represent version (major, minor, patch)
+pub struct Version(pub u16, pub u16, pub u16);
+
+impl Default for Version {
+    fn default() -> Self {
+        PROTOCOL_VERSION
+    }
+}
+
+impl crate::Serializable for Version {
+    fn write<W: Write>(&self, w: &mut W) -> io::Result<()> {
+        let Version(maj, min, patch) = self;
+        w.write_u16_varint(*maj)?;
+        w.write_u16_varint(*min)?;
+        w.write_u16_varint(*patch)?;
+
+        Ok(())
+    }
+
+    fn read<R: Read>(r: &mut R) -> io::Result<Self>
+    where
+        Self: Sized,
+    {
+        let maj = r.read_u16_varint()?;
+        let min = r.read_u16_varint()?;
+        let patch = r.read_u16_varint()?;
+        Ok(Self(maj, min, patch))
+    }
+}
+
 /// Topic field position in the message binary representation
 pub const TOPIC_FIELD_POS: usize = 8 + 8 + 4;
 
@@ -46,6 +82,7 @@ impl From<Ordering> for Status {
 /// Message definition
 #[derive(Debug, Default, Clone)]
 pub struct Message {
+    version: Version,
     topic: Topics,
     pub header: ConsensusHeader,
     pub payload: Payload,
@@ -95,6 +132,15 @@ impl Message {
             _ => StepName::Proposal.to_step(self.header.iteration),
         }
     }
+
+    pub fn version(&self) -> &Version {
+        &self.version
+    }
+
+    pub(crate) fn with_version(mut self, v: Version) -> Self {
+        self.version = v;
+        self
+    }
 }
 
 /// Defines a transport-related properties that determines how the message
@@ -107,6 +153,7 @@ pub struct Metadata {
 
 impl Serializable for Message {
     fn write<W: Write>(&self, w: &mut W) -> io::Result<()> {
+        self.version.write(w)?;
         w.write_all(&[self.topic as u8])?;
 
         match &self.payload {
@@ -128,6 +175,8 @@ impl Serializable for Message {
     where
         Self: Sized,
     {
+        let version = Version::read(r)?;
+
         // Read topic
         let topic = Topics::from(Self::read_u8(r)?);
         let message: Message = match topic {
@@ -149,7 +198,7 @@ impl Serializable for Message {
             }
         };
 
-        Ok(message)
+        Ok(message.with_version(version))
     }
 }