Add Messages and Message

src/connections.rs

@@ -6,12 +6,13 @@ use std::{
     time::{Duration, Instant},
 };
 
+use serde::{Serialize, de::DeserializeOwned};
 use thiserror::Error;
 
 use crate::{
     PeerConfig, PeerMessage,
-    listener::{DATAGRAM_SIZE, Datagram, ListenerError},
+    listener::{DATAGRAM_SIZE, ListenerError},
-    package::Package,
+    package::{Message, Package},
 };
 
 /// Represents an error during sending a package for any number of reason
@@ -39,17 +40,18 @@ pub enum ConnectionError {
 }
 
 /// Manages connections for the Peer
-pub(crate) struct ConnectionManager {
+pub(crate) struct ConnectionManager<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static>
+{
     /// Wraps UdpSocket with some helper methods
     pub udp: UdpWrapper,
-    connections: HashMap<SocketAddr, Connection>,
+    connections: HashMap<SocketAddr, Connection<T>>,
     closing_since: Option<Instant>,
     config: PeerConfig,
 }
 
-impl ConnectionManager {
+impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> ConnectionManager<T> {
     /// Creates a new ConnectionManager with the given socket and config
-    pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager {
+    pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager<T> {
         ConnectionManager {
             udp: UdpWrapper::new(socket),
             connections: HashMap::new(),
@@ -94,7 +96,7 @@ impl ConnectionManager {
                 if duration > self.config.ping_interval {
                     match conn.state {
                         ConnectionState::ReceivingConnection | ConnectionState::Connecting => {
-                            match self.udp.send_to(*addr, Package::Hello) {
+                            match self.udp.send_to(*addr, Package::<T>::Hello) {
                                 Ok(_) => {
                                     conn.last_sent_ping = now;
                                 }
@@ -106,7 +108,7 @@ impl ConnectionManager {
                             }
                         }
                         ConnectionState::Connected | ConnectionState::ConnectingNearlyReady => {
-                            match self.udp.send_to(*addr, Package::Ping) {
+                            match self.udp.send_to(*addr, Package::<T>::Ping) {
                                 Ok(_) => {
                                     conn.last_sent_ping = now;
                                 }
@@ -118,7 +120,7 @@ impl ConnectionManager {
                             }
                         }
                         ConnectionState::Closing | ConnectionState::ReceivingClosing => {
-                            match self.udp.send_to(*addr, Package::Close) {
+                            match self.udp.send_to(*addr, Package::<T>::Close) {
                                 Ok(_) => {
                                     conn.last_sent_ping = now;
                                 }
@@ -159,7 +161,11 @@ impl ConnectionManager {
 
     /// Handle Package from a remote peer
     #[must_use]
-    pub fn handle_package(&mut self, package: Package, addr: &SocketAddr) -> Vec<PeerMessage> {
+    pub fn handle_package(
+        &mut self,
+        package: Package<T>,
+        addr: &SocketAddr,
+    ) -> Vec<PeerMessage<T>> {
         let mut messages = Vec::new();
 
         match package {
@@ -186,7 +192,7 @@ impl ConnectionManager {
 
                     if conn.state == ConnectionState::Connected {
                         conn.last_recv_ping = Instant::now();
-                        match self.udp.send_to(*addr, Package::Pong) {
+                        match self.udp.send_to(*addr, Package::<T>::Pong) {
                             Ok(_) => {}
                             Err(err) => {
                                 conn.closing_since = Instant::now();
@@ -230,6 +236,7 @@ impl ConnectionManager {
                     self.connections.remove(addr);
                 }
             }
+            Package::Messages(messages) => todo!(),
         }
 
         messages
@@ -247,7 +254,7 @@ impl ConnectionManager {
     }
 
     /// Remove any connections that have closed or errored.
-    pub fn purge_old_connections(&mut self) -> Vec<PeerMessage> {
+    pub fn purge_old_connections(&mut self) -> Vec<PeerMessage<T>> {
         let mut messages = Vec::new();
 
         let now = Instant::now();
@@ -298,7 +305,11 @@ impl UdpWrapper {
     }
 
     /// Send a Package to a remote peer at addr
-    fn send_to(&self, addr: SocketAddr, package: Package) -> Result<(), SendError> {
+    fn send_to<T: Serialize>(
+        &self,
+        addr: SocketAddr,
+        package: Package<T>,
+    ) -> Result<(), SendError> {
         if let Some(socket) = &self.socket {
             let mut buf = Vec::new();
             let cursor = Cursor::new(&mut buf);
@@ -318,7 +329,7 @@ impl UdpWrapper {
 }
 
 /// Represents a single connection from the peer to another peer
-pub struct Connection {
+pub struct Connection<T: Clone> {
     /// Address of the connection
     pub address: SocketAddr,
     /// "ping", as in the time between the last sent ping and the last received
@@ -330,12 +341,14 @@ pub struct Connection {
     last_recv_close: Instant,
     closing_since: Instant,
 
+    reliable_queue: Vec<Message<T>>,
+
     /// What state is the connection currently in?
     pub state: ConnectionState,
     error: Option<ConnectionError>,
 }
 
-impl Clone for Connection {
+impl<T: Clone> Clone for Connection<T> {
     fn clone(&self) -> Self {
         Self {
             address: self.address.clone(),
@@ -344,14 +357,15 @@ impl Clone for Connection {
             last_recv_ping: self.last_recv_ping.clone(),
             last_recv_close: self.last_recv_close.clone(),
             closing_since: self.closing_since.clone(),
+            reliable_queue: self.reliable_queue.clone(),
             state: self.state.clone(),
             error: None,
         }
     }
 }
 
-impl Connection {
+impl<T: Clone> Connection<T> {
-    pub fn from(address: SocketAddr, state: ConnectionState) -> Connection {
+    pub fn from(address: SocketAddr, state: ConnectionState) -> Connection<T> {
         Connection {
             address,
             ping: Duration::default(),
@@ -359,6 +373,7 @@ impl Connection {
             last_recv_ping: Instant::now(),
             last_recv_close: Instant::now(),
             closing_since: Instant::now(),
+            reliable_queue: Vec::new(),
             state,
             error: None,
         }

src/lib.rs

@@ -10,6 +10,7 @@ use std::{
     time::Duration,
 };
 
+use serde::{Serialize, de::DeserializeOwned};
 use thiserror::*;
 
 use crate::{
@@ -31,11 +32,11 @@ pub enum PeerError {
 }
 
 /// Possible messages from occurring events from the peer
-pub enum PeerMessage {
+pub enum PeerMessage<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
     /// A new connection has been connected
-    NewConnection(Connection),
+    NewConnection(Connection<T>),
     /// An existing connection has disconnected, with an optional error
-    Disconnected(Connection, Option<ConnectionError>),
+    Disconnected(Connection<T>, Option<ConnectionError>),
     /// The Peer has been closed
     Closed,
 }
@@ -85,17 +86,17 @@ impl PeerConfig {
     }
 }
 
-pub struct Peer {
+pub struct Peer<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
-    connection_mgr: ConnectionManager,
+    connection_mgr: ConnectionManager<T>,
     closed: Arc<AtomicBool>,
-    receiver: Receiver<ListenerMessage>,
+    receiver: Receiver<ListenerMessage<T>>,
-    messages: VecDeque<PeerMessage>,
+    messages: VecDeque<PeerMessage<T>>,
 }
 
-impl Peer {
+impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Peer<T> {
     /// Bind given port to listen to incoming messages. Creates a new Peer that
     /// can establish new connections.
-    pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer, PeerError> {
+    pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer<T>, PeerError> {
         let socket = Arc::new(
             UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0], port.unwrap_or(0))))
                 .map_err(|e| PeerError::BindError(e))?,
@@ -133,7 +134,7 @@ impl Peer {
     /// possible events, such as connections and messages. Should be called as
     /// often as possible.
     #[must_use]
-    pub fn poll(&mut self) -> Result<Option<PeerMessage>, PeerError> {
+    pub fn poll(&mut self) -> Result<Option<PeerMessage<T>>, PeerError> {
         if self.connection_mgr.udp.is_closed() {
             if self.closed.load(Ordering::Acquire) {
                 self.messages.push_back(PeerMessage::Closed);

src/listener.rs

@@ -1,9 +1,11 @@
 use std::{
     io::Cursor,
+    marker::PhantomData,
     net::{SocketAddr, UdpSocket},
     sync::{Arc, mpsc::Sender},
 };
 
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
 use thiserror::Error;
 
 use crate::package::Package;
@@ -12,14 +14,19 @@ pub const DATAGRAM_SIZE: usize = 102400;
 pub type Datagram = [u8; DATAGRAM_SIZE];
 
 /// Listener thread for Peer's UdpSocket
-pub struct Listener {
+pub struct Listener<T: Clone + Serialize + DeserializeOwned> {
     socket: Arc<UdpSocket>,
-    sender: Sender<ListenerMessage>,
+    sender: Sender<ListenerMessage<T>>,
+    pd: PhantomData<T>,
 }
 
-impl Listener {
+impl<'a, T: Clone + Serialize + DeserializeOwned> Listener<T> {
-    pub fn new(socket: Arc<UdpSocket>, sender: Sender<ListenerMessage>) -> Listener {
+    pub fn new(socket: Arc<UdpSocket>, sender: Sender<ListenerMessage<T>>) -> Listener<T> {
-        Listener { socket, sender }
+        Listener {
+            socket,
+            sender,
+            pd: PhantomData::default(),
+        }
     }
 
     /// Read from the socket, parse packages and emit messages.
@@ -29,7 +36,8 @@ impl Listener {
             Ok((num_bytes, from_addr)) => {
                 let mut ciborium_buf: Datagram = [0; _];
                 let bytes = Cursor::new(&mut buffer[..num_bytes]);
-                let res = ciborium::from_reader_with_buffer::<Package, _>(bytes, &mut ciborium_buf);
+                let res =
+                    ciborium::from_reader_with_buffer::<Package<T>, _>(bytes, &mut ciborium_buf);
                 match res {
                     Ok(pkg) => self.sender.send(ListenerMessage::Package(pkg, from_addr)),
                     Err(_) => self.sender.send(ListenerMessage::PackageError(
@@ -52,8 +60,8 @@ pub enum ListenerError {
     ParseError,
 }
 
-pub enum ListenerMessage {
+pub enum ListenerMessage<T: Clone + Serialize + DeserializeOwned> {
-    Package(Package, SocketAddr),
+    Package(Package<T>, SocketAddr),
     PackageError(ListenerError, SocketAddr),
     Error(std::io::Error),
 }

src/package.rs

@@ -1,9 +1,23 @@
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Serialize, Deserialize)]
-pub enum Package {
+pub enum Package<T> {
     Hello,
     Ping,
     Pong,
     Close,
+    Messages(Messages<T>),
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Messages<T> {
+    pub ack: u64,
+    pub reliable: Vec<Message<T>>,
+    pub unreliable: Vec<Message<T>>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct Message<T> {
+    pub message_id: u64,
+    pub message: T,
 }