TcpTransport initial implementation NetService

core/Cargo.toml

@@ -30,14 +30,14 @@ pqc_kyber = { version = "0.7.1", features = ["std", "kyber512"] }
 rand_chacha = "0.3.1"
 pqc_dilithium_edit = "0.2.0"
 async-trait = "0.1.89"
-tokio = { version = "1.48.0", features = ["sync", "rt", "macros", "rt-multi-thread", "time", "fs", "io-util", "signal"] }
+tokio = { version = "1.48.0", features = ["sync", "rt", "macros", "rt-multi-thread", "time", "fs", "io-util", "signal", "net"] }
 bytes = { version = "1.11.0", features = ["serde"] }
 serde = { version = "1.0.228", features = ["derive"] }
-postcard = { version = "1.0", features = ["alloc"] }
+postcard = { version = "1.0", features = ["alloc", "use-std"] }
 heapless = "0.9.2"
 tokio-macros = "2.6.0"
 aes-gcm = { version= "0.10.3", features = ["rand_core"] }
-tokio-util = "0.7.17"
+tokio-util = { version = "0.7.17", features = ["join-map", "rt"] }
 thiserror = "2.0.17"
 serde-big-array = "0.5.1"
 tokio-stream = "0.1.18"

core/src/message.rs

@@ -5,7 +5,7 @@ use crate::{payload::{Payload, Query, Reply}, peer::PeerId};
 
 pub type MsgId = u64;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq)]
 pub struct IncomingMessage {
     pub from: PeerId,
     pub payload: Payload,
@@ -21,7 +21,7 @@ impl IncomingMessage {
     }
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq)]
 pub struct OutgoingMessage {
     pub to: PeerId,
     pub payload: Payload,

core/src/net/mod.rs

@@ -5,10 +5,12 @@ mod packet;
 mod types;
 mod client;
 mod session_store;
+mod service;
 
 pub use _session::*;
 pub use error::*;
 pub use packet::*;
 pub use types::*;
 pub use client::*;
-pub use session::*;
+pub use session::*;
+pub use service::*;

core/src/net/service.rs

@@ -0,0 +1,67 @@
+use tokio_util::sync::CancellationToken;
+use tokio::task::JoinError;
+use thiserror::Error;
+use crate::{net::{session::ActiveSession,error::{NetError,CryptoError}},message::{IncomingMessage,OutgoingMessage},payload::Payload,peer::{Peer,PeerId},utils::{SerdeError,ChannelError}, transport::{TransportError}};
+
+#[derive(Debug,Error)]
+pub enum NetServError {
+    #[error(transparent)]
+    Serde(#[from] SerdeError),
+
+    #[error(transparent)]
+    Channel(#[from] ChannelError),
+
+    #[error(transparent)]
+    Crypto(#[from] CryptoError),
+
+    #[error(transparent)]
+    Net(#[from] NetError),
+
+    #[error(transparent)]
+    Join(#[from] JoinError),
+
+    #[error(transparent)]
+    Transport(#[from] TransportError),
+
+    #[error("client not found")]
+    ClientNotFound,
+
+    #[error("sessions not found")]
+    SessionsNotFound,
+
+    #[error("Session already exists")]
+    SessionAlreadyExists,
+
+    #[error("peer not found")]
+    PeerNotFound,
+
+}
+
+
+
+pub trait NetService{
+    // Interface for handling connections out from a single peer/client
+
+    type Error: Into<NetServError>;
+
+    // Add fully formed sessions to the service
+    // Deal with handshakes prior to adding into service
+    // Fail on adding a session if the peer already exists
+    fn add_session(&mut self, client: (Peer,ActiveSession)) -> impl Future<Output = Result<(), Self::Error>> + Send;
+    // It is possible that you'd want multiple sessions between two peers. Currently this would be an error. 
+    // If not changed now, it will be hard to fix in the future.
+
+    // Close a session and drop it from the table
+    fn drop_session(&mut self, peer: &PeerId) -> 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;
+
+    // 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: PeerId, token: CancellationToken) -> impl Future<Output = Result<(), Self::Error>> + Send;
+}

core/src/net/session/session.rs

@@ -78,7 +78,7 @@ impl ActiveSession {
         aad
     }
 
-    pub fn send(&mut self, plaintext: &Vec<u8>) -> Result<Message, CryptoError> {
+    pub fn send(&mut self, plaintext: &[u8]) -> Result<Message, CryptoError> {
         self.seq += 1;
 
         let cipher = self.cipher();

core/src/payload/dht.rs

@@ -3,13 +3,13 @@ use serde::{Deserialize, Serialize};
 
 use crate::{dht::CID, payload::{Query, QueryError, Reply, ReplyError, TryFromQuery, TryFromReply}};
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum DhtQuery {
     Get(CID),
     Put(Bytes),
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum DhtReply {
     Return(Option<Bytes>),
 }

core/src/payload/mod.rs

@@ -9,20 +9,20 @@ pub use dht::*;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum Payload {
     Query(Query),
     Reply(Reply),
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum Query {
     Pow(PowQuery),
     Tag(TagQuery),
     Dht(DhtQuery),
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum Reply {
     Empty,
     Ok,
@@ -46,7 +46,7 @@ pub enum ReplyError {
 pub trait TryFromQuery: TryFrom<Query, Error = QueryError> {}
 pub trait TryFromReply: TryFrom<Reply, Error = ReplyError> {}
 
-#[derive(Serialize, Deserialize, Clone, Copy, Debug)]
+#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq)]
 #[repr(u16)]
 pub enum Action {
     PublishTag = 1,

core/src/payload/pow.rs

@@ -3,18 +3,18 @@ use thiserror::Error;
 
 use crate::{payload::{Action, Query, QueryError, Reply, ReplyError, TryFromQuery, TryFromReply}, pow::Pow};
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum PowQuery {
     Get(Action),
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug,PartialEq,Clone)]
 pub enum PowReply {
     Require(Pow),
     Err(PowReplyErr),
 }
 
-#[derive(Serialize, Deserialize, Debug, Error)]
+#[derive(Serialize, Deserialize, Debug, Error, PartialEq,Clone)]
 pub enum PowReplyErr {
     #[error("Incorrect nonce")]
     IncorrectNonce,

core/src/payload/tag.rs

@@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::{payload::{Query, QueryError, Reply, ReplyError, TryFromQuery, TryFromReply}, pow::Pow, tag::Tag};
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum TagQuery {
     Get,
     Publish {
@@ -12,7 +12,7 @@ pub enum TagQuery {
     },
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub enum TagReply {
     Return(Vec<Tag>),
 }

core/src/pow.rs

@@ -13,7 +13,7 @@ fn pow_input(secret: &[u8; 32], timestamp: u64, action: Action, random: &[u8; 16
     hasher.finalize().into()
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug, PartialEq,Clone)]
 pub struct Pow {
     pub input: [u8; 32],
     pub timestamp: u64,

core/src/tag/tag.rs

@@ -8,7 +8,7 @@ use bytes::{BufMut, Bytes, BytesMut};
 
 use crate::{VERSION, tag::TagError, utils::{self, deserialize, random_bytes, serialize}};
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub struct Tag {
     nonce: [u8; 12],
     content: Vec<u8>,

core/src/transport/error.rs

@@ -1,6 +1,7 @@
 use thiserror::Error;
+use tokio::{task::JoinError,sync::mpsc::error::{SendError},net::tcp::OwnedReadHalf};
 
-use crate::{net::SessionManagerDispatcherError, utils::ChannelError};
+use crate::{net::{SessionManagerDispatcherError,CryptoError,NetError,Message}, utils::ChannelError,peer::{PeerId}};
 
 #[derive(Debug, Error)]
 pub enum MockTransportError {
@@ -15,4 +16,54 @@ pub enum MockTransportError {
 
     #[error("peer not found")]
     PeerNotFound,
+}
+
+#[derive(Debug, Error)]
+pub enum TransportError {
+    // Make PeerId optional so that callers can add information without having to pass PeerId down to every function
+    #[error("session's read task has been cancelled")]
+    ReadCancelled(OwnedReadHalf,Option<PeerId>),
+
+    #[error("request was cancelled by token")]
+    Cancelled,
+
+    // Consider returning peer ID as data in this error
+    #[error("session not found")]
+    SessionNotFound(Option<PeerId>),
+
+    // Consider returning peer ID as data in this error
+    #[error("connection not found in active map")]
+    ConnectionNotInMap(Option<PeerId>),
+
+    // Consider returning peer ID as data in this error
+    #[error("no sessions available")]
+    NoSessions,
+
+    // Consider returning peer ID as data in this error
+    #[error("message channel has been closed")]
+    MessageChannelClosed,
+
+    #[error("peer already connected")]
+    PeerAlreadyConnected(Option<PeerId>),
+
+    #[error("connection closed")]
+    ConnectionClosed(Option<PeerId>),
+
+    #[error(transparent)]
+    Serialization(#[from] postcard::Error),
+
+    #[error(transparent)]
+    Reading(#[from] SendError::<(PeerId,Message)>),
+
+    #[error(transparent)]
+    IO(#[from] std::io::Error),
+
+    #[error(transparent)]
+    Join(#[from] JoinError),
+
+    #[error(transparent)]
+    Encrypt(#[from] CryptoError),
+
+    #[error(transparent)]
+    Decrypt(#[from] NetError),
 }

core/src/transport/mod.rs

@@ -2,6 +2,7 @@ mod mock;
 mod error;
 mod controls;
 mod participant;
+mod tcp;
 
 pub use mock::*;
 pub use error::*;