feat: make use of sn_messaging crate for messaging serialisation/dese…

…rialisation

Cargo.toml

@@ -22,6 +22,7 @@ lru_time_cache = "~0.11.0"
 qp2p = "~0.9.10"
 rand = "~0.7.3"
 rand_chacha = "~0.2.2"
+sn_messaging = { git = "https://github.com/bochaco/sn_messaging.git", branch = "support-node-ping-infrastructure-types" }
 thiserror = "1.0.23"
 xor_name = "1.1.0"
 resource_proof = "0.8.0"
@@ -59,6 +60,7 @@ structopt = "~0.3.17"
 tracing-appender = "~0.1.2"
 tracing-subscriber = "~0.2.15"
 yansi = "~0.5.0"
+sn_data_types = "0.14.1"
 
   [dev-dependencies.rand]
   version = "~0.7.3"

examples/minimal.rs

@@ -277,10 +277,8 @@ async fn handle_event(index: usize, node: &mut Routing, event: Event) -> bool {
             return false;
         }
         Event::ClientMessageReceived { content, src, .. } => info!(
-            "Node #{} received message from client: {:?}, content: {}",
-            index,
-            src,
-            HexFmt(&content)
+            "Node #{} received message from client: {:?}, content: {:?}",
+            index, src, content
         ),
         Event::ClientLost(addr) => info!("Node #{} received ClientLost({:?})", index, addr),
     }

src/error.rs

@@ -19,11 +19,11 @@ pub enum Error {
     FailedSignature,
     #[error("Cannot route.")]
     CannotRoute,
-    #[error("Network layer error: {}", .0)]
+    #[error("Network layer error: {0}")]
     Network(#[from] qp2p::Error),
     #[error("The node is not in a state to handle the action.")]
     InvalidState,
-    #[error("Bincode error: {}", .0)]
+    #[error("Bincode error: {0}")]
     Bincode(#[from] bincode::Error),
     #[error("Invalid source location.")]
     InvalidSrcLocation,
@@ -39,4 +39,8 @@ pub enum Error {
     FailedSend,
     #[error("Invalid vote.")]
     InvalidVote,
+    #[error("Messaging protocol error: {0}")]
+    Messaging(#[from] sn_messaging::Error),
+    #[error("Node messaging error: {0}")]
+    NodeMessaging(#[from] sn_messaging::node::Error),
 }

src/event.rs

@@ -12,6 +12,7 @@ use bytes::Bytes;
 use ed25519_dalek::Keypair;
 use hex_fmt::HexFmt;
 pub use qp2p::{RecvStream, SendStream};
+use sn_messaging::client::MsgEnvelope;
 use std::{
     collections::BTreeSet,
     fmt::{self, Debug, Formatter},
@@ -39,8 +40,6 @@ pub enum NodeElderChange {
 ///
 /// `Request` and `Response` events from section locations are only raised once the majority has
 /// been reached, i.e. enough members of the section have sent the same message.
-// FIXME - See https://maidsafe.atlassian.net/browse/MAID-2026 for info on removing this exclusion.
-#[allow(clippy::large_enum_variant)]
 pub enum Event {
     /// Received a message.
     MessageReceived {
@@ -100,13 +99,14 @@ pub enum Event {
     /// Received a message from a client node.
     ClientMessageReceived {
         /// The content of the message.
-        content: Bytes,
+        content: Box<MsgEnvelope>,
         /// The address of the client that sent the message.
         src: SocketAddr,
-        /// Stream to send messages back to the client that sent the message
-        send: SendStream,
         /// Stream to receive more messages from the client on the same channel
         recv: RecvStream,
+        /// Stream to send messages back to the client that sent
+        /// the message if it was received on a bi-directional stream
+        send: Option<SendStream>,
     },
     /// Failed in sending a message to client, or connection to client is lost
     ClientLost(SocketAddr),
@@ -167,9 +167,8 @@ impl Debug for Event {
             Self::RestartRequired => write!(formatter, "RestartRequired"),
             Self::ClientMessageReceived { content, src, .. } => write!(
                 formatter,
-                "ClientMessageReceived {{ content: \"{:<8}\", src: {:?} }}",
-                HexFmt(content),
-                src,
+                "ClientMessageReceived {{ content: {:?}, src: {:?} }}",
+                content, src,
             ),
             Self::ClientLost(addr) => write!(formatter, "ClientLost({:?})", addr),
         }

src/lib.rs

@@ -74,7 +74,6 @@ pub use self::{
     error::{Error, Result},
     event::{Event, NodeElderChange, SendStream},
     location::{DstLocation, SrcLocation},
-    messages::{InfrastructureQuery, MessageKind},
     routing::{Config, EventStream, Routing},
     section::{SectionProofChain, MIN_AGE},
 };

src/messages/mod.rs

@@ -6,32 +6,23 @@
 // KIND, either express or implied. Please review the Licences for the specific language governing
 // permissions and limitations relating to use of the SAFE Network Software.
 
-mod envelope;
 mod hash;
-mod infrastructure_query;
 mod plain_message;
 mod src_authority;
 mod variant;
 
+pub use self::{hash::MessageHash, src_authority::SrcAuthority};
 pub(crate) use self::{
-    envelope::Envelope,
     plain_message::PlainMessage,
     variant::{JoinRequest, ResourceProofResponse, Variant},
 };
-pub use self::{
-    envelope::MessageKind,
-    hash::MessageHash,
-    infrastructure_query::{GetSectionResponse, InfrastructureQuery},
-    src_authority::SrcAuthority,
-};
 use crate::{
     crypto::{self, name, Verifier},
     error::{Error, Result},
     location::DstLocation,
     node::Node,
     section::{ExtendError, SectionProofChain, TrustStatus},
 };
-
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 use std::fmt::{self, Debug, Formatter};
@@ -65,8 +56,8 @@ pub(crate) struct Message {
 
 impl Message {
     /// Deserialize the message. Only called on message receipt.
-    pub(crate) fn from_bytes(bytes: &Bytes) -> Result<Self, CreateError> {
-        let mut msg: Message = bincode::deserialize(&bytes[..])?;
+    pub(crate) fn from_bytes(msg_bytes: Bytes) -> Result<Self, CreateError> {
+        let mut msg: Message = bincode::deserialize(&msg_bytes)?;
 
         let signed_bytes = bincode::serialize(&SignableView {
             dst: &msg.dst,
@@ -96,8 +87,8 @@ impl Message {
             }
         }
 
-        msg.serialized = bytes.clone();
-        msg.hash = MessageHash::from_bytes(bytes);
+        msg.serialized = msg_bytes.clone();
+        msg.hash = MessageHash::from_bytes(&msg_bytes);
 
         Ok(msg)
     }

src/relocation.rs

@@ -11,13 +11,14 @@
 use crate::{
     crypto::{self, Keypair, Signature, Verifier},
     error::Error,
-    messages::{Envelope, Message, Variant},
+    messages::{Message, Variant},
     network::Network,
     peer::Peer,
     section::{MemberInfo, Section},
 };
 use bytes::Bytes;
 use serde::{de::Error as SerdeDeError, Deserialize, Deserializer, Serialize, Serializer};
+use sn_messaging::MessageType;
 use std::net::SocketAddr;
 use tokio::sync::mpsc;
 use xor_name::XorName;
@@ -110,11 +111,11 @@ impl RelocateDetails {
 #[derive(Clone, Eq, PartialEq, Debug)]
 pub(crate) struct SignedRelocateDetails {
     /// Signed message whose content is Variant::Relocate
-    signed_msg: Message,
+    signed_msg: Box<Message>,
 }
 
 impl SignedRelocateDetails {
-    pub fn new(signed_msg: Message) -> Result<Self, Error> {
+    pub fn new(signed_msg: Box<Message>) -> Result<Self, Error> {
         if let Variant::Relocate(_) = signed_msg.variant() {
             Ok(Self { signed_msg })
         } else {
@@ -208,7 +209,7 @@ pub(crate) enum RelocateState {
     // will exchange it for an actual `Relocate` message.
     Delayed(Bytes),
     // Relocation in progress. The sender is used to pass messages to the bootstrap task.
-    InProgress(mpsc::Sender<(Envelope, SocketAddr)>),
+    InProgress(mpsc::Sender<(MessageType, SocketAddr)>),
 }
 
 /// Action to relocate a node.