Messages

.gitignore

@@ -1 +1,2 @@
-/target
+/target
+/core/*

crates/net/src/message.rs

@@ -7,50 +7,100 @@ use crate::{payload::{Payload, Query, Reply}};
 pub type MsgId = u64;
 
 #[derive(Serialize, Deserialize, Debug, PartialEq)]
-pub struct IncomingMessage {
+pub struct Message {
+    pub id: MsgId,
     pub from: Id,
     pub payload: Payload,
-    pub id: MsgId,
+    pub to: Id,
 }
 
-impl IncomingMessage {
-    pub fn receive(from: Id, msg: OutgoingMessage) -> Self {
-        Self { from, payload: msg.payload, id: msg.id }
+impl Message {
+
+    pub fn new(from: &Id, to: &Id, payload: Payload) -> Self {
+        Message{
+            from: from.clone(),
+            id: OsRng.next_u64(),
+            payload,
+            to: to.clone(),
+        }
     }
-    pub fn reply(self, reply: Reply) -> OutgoingMessage {
-        OutgoingMessage { to: self.from, payload: Payload::Reply(reply), id: self.id }
+    pub fn query(from: &Id, to: &Id, q: impl Into<Query>) -> Self {
+        Self::new(from, to, Payload::Query(q.into()))
     }
-}
 
-#[derive(Serialize, Deserialize, Debug, PartialEq)]
-pub struct OutgoingMessage {
-    pub to: Id,
-    pub payload: Payload,
-    pub id: MsgId,
+    pub fn reply(&self, reply: impl Into<Reply>) -> Self {
+        // At the moment I think it only makes sense to reply to an existing message
+        // I don't think we need to consume the original message - but that's up for debate
+        Self { from: self.to.clone(), to: self.from.clone(), payload: Payload::Reply(reply.into()), id: self.id }
+    }
 }
 
-impl OutgoingMessage {
-    pub fn new(to: &Id, payload: Payload) -> Self {
-        Self { to: to.clone(), payload, id: OsRng.next_u64() }
-    }
-    pub fn query(to: &Id, q: impl Into<Query>) -> Self {
-        Self::new(to, Payload::Query(q.into()))
-    }
-    pub fn reply(to: &Id, r: impl Into<Reply>) -> Self {
-        Self::new(to, Payload::Reply(r.into()))
+#[cfg(test)]
+mod test{
+    use super::*;
+    use crate::payload::{TestQuery,TestReply};
+
+    #[test]
+    fn new_message() {
+        let alice = Id::new([0u8; 32]);
+        let bob = Id::new([1u8; 32]);
+        let payload = Payload::Query(Query::Mock(TestQuery::Ping));
+        let expect = Message{
+            from: alice.clone(),
+            id: 1234567890,
+            payload: payload.clone(),
+            to: bob.clone(),
+        };
+
+        let got = Message::new(&alice, &bob, payload);
+
+        // ID is assigned randomly so can't compare structs directly
+        assert_eq!(expect.from,got.from,"From should match");
+        assert_eq!(expect.to,got.to,"To should match");
+        assert_eq!(expect.payload,got.payload,"Payload should match");
     }
-}
 
-pub struct OutgoingMessageBuilder {
-    to: Id,
-    id: MsgId,
-}
+    #[test]
+    fn test_query() {
+        let alice = Id::new([0u8; 32]);
+        let bob = Id::new([1u8; 32]);
+        let payload = Payload::Query(Query::Mock(TestQuery::Ping));
+        let expect = Message{
+            from: alice.clone(),
+            id: 1234567890,
+            payload: payload.clone(),
+            to: bob.clone(),
+        };
+
+        let got = Message::new(&alice, &bob, payload);
 
-impl OutgoingMessageBuilder {
-    pub fn new(incoming: &IncomingMessage) -> Self {
-        Self { to: incoming.from.clone(), id: incoming.id }
+        // ID is assigned randomly so can't compare structs directly
+        assert_eq!(expect.from,got.from,"From should match");
+        assert_eq!(expect.to,got.to,"To should match");
+        assert_eq!(expect.payload,got.payload,"Payload should match");
     }
-    pub fn reply(self, reply: Reply) -> OutgoingMessage {
-        OutgoingMessage { to: self.to, payload: Payload::Reply(reply), id: self.id }
+
+    #[test]
+    fn test_reply() {
+        let alice = Id::new([0u8; 32]);
+        let bob = Id::new([1u8; 32]);
+        let payload = Payload::Query(Query::Mock(TestQuery::Ping));
+        let message = Message{
+            from: alice.clone(),
+            id: 1234567890,
+            payload,
+            to: bob.clone(),
+        };
+
+        let expect = Message {
+            from: bob.clone(),
+            id: 1234567890,
+            payload: Payload::Reply(Reply::Mock(TestReply::Pong)),
+            to: alice.clone(),
+        };
+
+        // ID should match so we can compare directly
+        let got = message.reply(Reply::Mock(TestReply::Pong));
+        assert_eq!(expect,got,"Reply should match");
     }
 }

crates/net/src/packet.rs

@@ -31,5 +31,5 @@ pub type SessionId = Id;
 pub enum Packet {
     Handshake(Handshake),
     HandshakeAck(HandshakeAck),
-    Message(WireMessage),
+    WireMessage(WireMessage),
 }

crates/net/src/payload/mod.rs

@@ -10,21 +10,21 @@ pub use test::*;
 
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
 pub enum Payload {
     Query(Query),
     Reply(Reply),
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
 pub enum Query {
     // Pow(PowQuery),
     // Tag(TagQuery),
     // Dht(DhtQuery),
     Mock(TestQuery),
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
 pub enum Reply {
     // Empty,
     // Ok,

crates/net/src/payload/test.rs

@@ -2,12 +2,12 @@ use serde::{Deserialize, Serialize};
 
 use crate::payload::{Query, Reply};
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
 pub enum TestQuery {
     Ping,
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
 pub enum TestReply {
     Pong,
 }

crates/net/src/service.rs

@@ -2,7 +2,7 @@ use crypto::{error::CryptoError, id::Id};
 use tokio_util::sync::CancellationToken;
 use tokio::task::JoinError;
 use thiserror::Error;
-use crate::{/*net::session::ActiveSession, */error::NetError,message::IncomingMessage,payload::Payload,peer::Peer,/*utils::{SerdeError,ChannelError}, */transport::error::TransportError};
+use crate::{/*net::session::ActiveSession, */error::NetError,message::Message,payload::Payload,peer::Peer,/*utils::{SerdeError,ChannelError}, */transport::error::TransportError};
 
 #[derive(Debug,Error)]
 pub enum NetServError {
@@ -56,13 +56,12 @@ pub trait NetService{
     fn drop_session(&mut self, peer: &Id) -> impl Future<Output = Result<(), Self::Error>> + Send;
 
     // Listen for incoming messages from all peers
-    fn listen(&mut self, token: CancellationToken) -> impl Future<Output = Result<IncomingMessage, Self::Error>> + Send;
+    fn listen(&mut self, token: CancellationToken) -> impl Future<Output = Result<Message, Self::Error>> + Send;
 
     // Broadcast messages to all sessions
     // Responsible for encrypting for each peer
     fn broadcast(&mut self, msg: Payload, token: CancellationToken) -> impl Future<Output = Result<(), Self::Error>> + Send;
 
     // Transmit messages to a specific session
-    // OutgoingMessage has its own PeerID
     fn transmit(&mut self, msg: Payload,target: Id, token: CancellationToken) -> impl Future<Output = Result<(), Self::Error>> + Send;
 }

crates/net/src/transport/controls.rs

@@ -1,24 +1,24 @@
 use tokio::sync::mpsc;
 
-use crate::{message::{IncomingMessage, OutgoingMessage}, utils::ChannelError};
+use crate::{message::{Message}, utils::ChannelError};
 
 const CHAN_SIZE: usize = 128;
 
 pub struct TransportDispatcher {
-    pub(super) rx: mpsc::Receiver<IncomingMessage>,
-    pub(super) tx: mpsc::Sender<OutgoingMessage>,
+    pub(super) rx: mpsc::Receiver<Message>,
+    pub(super) tx: mpsc::Sender<Message>,
 }
 
 pub struct TransportHandler {
-    pub(super) rx: mpsc::Receiver<OutgoingMessage>,
-    pub(super) tx: mpsc::Sender<IncomingMessage>,
+    pub(super) rx: mpsc::Receiver<Message>,
+    pub(super) tx: mpsc::Sender<Message>,
 }
 
 impl TransportDispatcher {
-    pub async fn send(&self, msg: OutgoingMessage) -> Result<(), ChannelError> {
+    pub async fn send(&self, msg: Message) -> Result<(), ChannelError> {
         self.tx.send(msg).await.map_err(|_| ChannelError::Closed)
     }
-    pub async fn recv(&mut self) -> Option<IncomingMessage> {
+    pub async fn recv(&mut self) -> Option<Message> {
         self.rx.recv().await
     }
 }
@@ -29,10 +29,10 @@ impl TransportHandler {
         let (recv_tx, recv_rx) = mpsc::channel(CHAN_SIZE);
         (Self { rx: send_rx, tx: recv_tx }, TransportDispatcher { rx: recv_rx, tx: send_tx })
     }
-    pub async fn send(&self, msg: IncomingMessage) -> Result<(), ChannelError> {
+    pub async fn send(&self, msg: Message) -> Result<(), ChannelError> {
         self.tx.send(msg).await.map_err(|_| ChannelError::Closed)
     }
-    pub async fn recv(&mut self) -> Option<OutgoingMessage> {
+    pub async fn recv(&mut self) -> Option<Message> {
         self.rx.recv().await
     }
 }

crates/net/src/transport/error.rs

@@ -2,7 +2,7 @@ use crypto::{error::CryptoError, id::Id};
 use thiserror::Error;
 use tokio::{net::tcp::OwnedReadHalf, sync::mpsc::error::SendError, task::JoinError};
 
-use crate::{error::NetError, packet::Message};
+use crate::{error::NetError, packet::WireMessage};
 
 // use crate::{net::SessionManagerDispatcherError, utils::ChannelError};
 
@@ -56,7 +56,7 @@ pub enum TransportError {
     Serialization(#[from] postcard::Error),
 
     #[error(transparent)]
-    Reading(#[from] SendError::<(Id,Message)>),
+    Reading(#[from] SendError::<(Id,WireMessage)>),
 
     #[error(transparent)]
     IO(#[from] std::io::Error),

crates/net/src/transport/mock.rs

@@ -92,7 +92,7 @@ mod tests {
     use tokio::time::timeout;
     use tokio_util::sync::CancellationToken;
 
-    use crate::{message::OutgoingMessage, net::{NetClient, SessionManager, SessionManagerDispatcher}, payload::{Payload, Reply, TagQuery}, peer::{PeerTable, PeerTableDispatcher}, service::Service, transport::{MockTransport, TransportHandler}, utils::random_bytes};
+    use crate::{message::Message, net::{NetClient, SessionManager, SessionManagerDispatcher}, payload::{Payload, Reply, TagQuery}, peer::{PeerTable, PeerTableDispatcher}, service::Service, transport::{MockTransport, TransportHandler}, utils::random_bytes};
 
     fn setup_peer_table() -> PeerTableDispatcher {
         let (service, dispatcher) = PeerTable::new();
@@ -126,12 +126,15 @@ mod tests {
         let (alice_handler, mut alice) = TransportHandler::new();
         let (bob_handler, mut bob) = TransportHandler::new();
 
+        // Alice needs an ID for Bob to be able to reply back
+        // TODO - NetIdentity has been removed. In favour of crypto::id?
+        // let alice_id = Id::new([0u8;32]);
         transport.add_participant(alice_sessions, None, alice_handler).await.expect("add participant failed");
         transport.add_participant(bob_sessions, Some(bob_identity), bob_handler).await.expect("add participant failed");
 
         tokio::spawn(async { transport.run(CancellationToken::new()).await.unwrap() });
 
-        alice.send(OutgoingMessage::query(&bob_id, TagQuery::Get)).await.expect("alice query failed");
+        alice.send(Message::query(&alice_id,&bob_id, TagQuery::Get)).await.expect("alice query failed");
 
         let incoming = timeout(TIMEOUT, bob.recv()).await.expect("timeout").expect("channel should not be closed");
 

crates/net/src/transport/tcp.rs

@@ -9,7 +9,7 @@ use bytes::BytesMut;
 use postcard::{to_stdvec,take_from_bytes,from_bytes};
 
 use crypto::id::Id;
-use crate::{/*net::{ActiveSession, PendingSession, NetClient},*/ message::{IncomingMessage,OutgoingMessage}, packet::Message, payload::Payload, peer::Peer, service::NetService, transport::error::TransportError};
+use crate::{/*net::{ActiveSession, PendingSession, NetClient},*/ message::{Message}, packet::Message, payload::Payload, peer::Peer, service::NetService, transport::error::TransportError};
 
 const CHANNELSIZE: usize = 128;
 // Max message len current 226
@@ -141,7 +141,7 @@ impl NetService for TcpTransport {
         Ok(())
     }
 
-    async fn listen(&mut self, token: CancellationToken) -> Result<IncomingMessage, Self::Error> {
+    async fn listen(&mut self, token: CancellationToken) -> Result<Message, Self::Error> {
         loop {
             tokio::select!{
                 _ = token.cancelled() => { return Err(TransportError::Cancelled); },
@@ -176,8 +176,7 @@ impl NetService for TcpTransport {
                     if let Some((id,encrypted)) = msg {
                         let session = self.sessions.get_mut(&id).ok_or(TransportError::SessionNotFound(Some(id)))?;
                         let decrypted = session.receive(encrypted)?;
-                        let outgoing = from_bytes::<OutgoingMessage>(&decrypted)?;
-                        return Ok(IncomingMessage::receive(id, outgoing))
+                        return Ok(from_bytes::<Message>(&decrypted)?);
                     } else {
                         // Channel has been closed
                         return Err(TransportError::MessageChannelClosed)
@@ -232,7 +231,7 @@ impl NetService for TcpTransport {
 #[cfg(test)]
 mod test {    
     use super::*;
-    use crate::{message::OutgoingMessage,payload::{Action, Query, Reply, TagQuery}, pow::Pow, tag::{Tag,TagPayload}};
+    use crate::{message::Message,payload::{Action, Query, Reply, TagQuery}, pow::Pow, tag::{Tag,TagPayload}};
     use std::{io::Write, sync::mpsc::{Sender, channel}};
     use tokio::time::timeout;
     use std::{time::{Duration}};
@@ -420,15 +419,17 @@ mod test {
             let peer = Peer::new(client.identity().expect("Expect static ID"), addr.to_string());
 
             expect_messages.push(
-                IncomingMessage{
+                Message{
+                    to: transport.client.identity().expect("Expected an ID").peer_id(),
                     from: peer.id.clone(),
                     payload: msg.clone(),
                     id: 0,
                 }
             );
 
-            let send_message = OutgoingMessage{
+            let send_message = Message{
                 to: transport.client.identity().expect("Expected an ID").peer_id(),
+                from: peer.id.clone(),
                 payload: msg.clone(),
                 id: 0,
             };