Add sending messages

src/connections.rs

@@ -12,7 +12,7 @@ use thiserror::Error;
 use crate::{
     PeerConfig, PeerMessage,
     listener::{DATAGRAM_SIZE, ListenerError},
-    package::{Message, Package},
+    package::{Message, Messages, Package},
 };
 
 /// Represents an error during sending a package for any number of reason
@@ -138,6 +138,68 @@ impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Connection
         }
     }
 
+    pub fn send_queued_messages(&mut self) {
+        let now = Instant::now();
+        for (addr, conn) in &mut self.connections {
+            if now - conn.last_msg_send > self.config.message_retry {
+                let messages = Messages {
+                    ack: conn.ack,
+                    reliable: conn.reliable_queue.clone(),
+                    unreliable: Vec::new(),
+                };
+                match self.udp.send_to(*addr, Package::Messages(messages)) {
+                    Ok(_) => {
+                        conn.last_msg_send = now;
+                    }
+                    Err(err) => {
+                        conn.closing_since = Instant::now();
+                        conn.state = ConnectionState::Error;
+                        conn.error = Some(ConnectionError::SendError(err));
+                    }
+                }
+            }
+        }
+    }
+
+    pub fn send(&mut self, addr: &SocketAddr, message: T, reliable: bool) {
+        if let Some(conn) = self.connections.get_mut(addr) {
+            if conn.state == ConnectionState::Connected {
+                let message_id = conn.message_counter;
+                conn.message_counter += 1;
+
+                let message = Message {
+                    message_id,
+                    message,
+                };
+
+                let mut reliables = conn.reliable_queue.clone();
+                let mut unreliables = Vec::new();
+                if reliable {
+                    reliables.push(message.clone());
+                    conn.reliable_queue.push(message);
+                } else {
+                    unreliables.push(message);
+                }
+
+                let messages = Messages {
+                    ack: conn.ack,
+                    reliable: reliables,
+                    unreliable: unreliables,
+                };
+                match self.udp.send_to(*addr, Package::Messages(messages)) {
+                    Ok(_) => {
+                        conn.last_msg_send = Instant::now();
+                    }
+                    Err(err) => {
+                        conn.closing_since = Instant::now();
+                        conn.state = ConnectionState::Error;
+                        conn.error = Some(ConnectionError::SendError(err));
+                    }
+                }
+            }
+        }
+    }
+
     /// Close the connection to another peer at the given address
     pub fn close_connection(&mut self, addr: &SocketAddr) {
         if let Some(conn) = self.connections.get_mut(addr) {
@@ -236,7 +298,38 @@ impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Connection
                     self.connections.remove(addr);
                 }
             }
-            Package::Messages(messages) => todo!(),
+            Package::Messages(msgs) => {
+                if let Some(conn) = self.connections.get_mut(addr) {
+                    // Clear all messages from the queue that the other peer has
+                    // already seen
+                    conn.reliable_queue.retain(|v| v.message_id > msgs.ack);
+
+                    // Process all messages that have not been previously seen
+                    // before, that is to say the connection's ack value is
+                    // lower than the message's id
+                    for reliable_msg in &msgs.reliable {
+                        if reliable_msg.message_id >= conn.ack {
+                            messages.push(PeerMessage::Message(reliable_msg.message.clone()));
+                        }
+                    }
+
+                    // Set connection's ack value to be at least one higher than
+                    // the highest message_id that was received, so that we
+                    // inform the remote peer not to send these messages again.
+                    conn.ack = msgs
+                        .reliable
+                        .iter()
+                        .map(|r| r.message_id + 1)
+                        .max()
+                        .unwrap_or(0)
+                        .max(conn.ack);
+
+                    // Always process all unreliable messages
+                    for unreliable_msg in &msgs.unreliable {
+                        messages.push(PeerMessage::Message(unreliable_msg.message.clone()));
+                    }
+                }
+            }
         }
 
         messages
@@ -341,7 +434,10 @@ pub struct Connection<T: Clone> {
     last_recv_close: Instant,
     closing_since: Instant,
 
+    message_counter: u64,
+    ack: u64,
     reliable_queue: Vec<Message<T>>,
+    last_msg_send: Instant,
 
     /// What state is the connection currently in?
     pub state: ConnectionState,
@@ -357,7 +453,12 @@ impl<T: Clone> Clone for Connection<T> {
             last_recv_ping: self.last_recv_ping.clone(),
             last_recv_close: self.last_recv_close.clone(),
             closing_since: self.closing_since.clone(),
+
+            message_counter: self.message_counter,
+            ack: self.ack,
             reliable_queue: self.reliable_queue.clone(),
+            last_msg_send: self.last_msg_send.clone(),
+
             state: self.state.clone(),
             error: None,
         }
@@ -373,7 +474,12 @@ impl<T: Clone> Connection<T> {
             last_recv_ping: Instant::now(),
             last_recv_close: Instant::now(),
             closing_since: Instant::now(),
+
+            message_counter: 0,
+            ack: 0,
             reliable_queue: Vec::new(),
+            last_msg_send: Instant::now() - Duration::from_hours(1),
+
             state,
             error: None,
         }

src/lib.rs

@@ -39,6 +39,8 @@ pub enum PeerMessage<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'st
     Disconnected(Connection<T>, Option<ConnectionError>),
     /// The Peer has been closed
     Closed,
+    /// A single message of type T
+    Message(T),
 }
 
 /// Optional configuration available for peers, mainly configuration of
@@ -48,6 +50,7 @@ pub struct PeerConfig {
     ping_interval: Duration,
     timeout: Duration,
     disconnect_timeout: Duration,
+    message_retry: Duration,
 }
 
 impl Default for PeerConfig {
@@ -56,6 +59,7 @@ impl Default for PeerConfig {
             ping_interval: Duration::from_millis(100),
             timeout: Duration::from_millis(2000),
             disconnect_timeout: Duration::from_millis(500),
+            message_retry: Duration::from_millis(100),
         }
     }
 }
@@ -84,6 +88,16 @@ impl PeerConfig {
             ..self
         }
     }
+
+    /// Sets the duration which is awaited before any queued reliable messages
+    /// are sent again. This timer is also reset every time any new messages are
+    /// sent manually.
+    pub fn with_retry(self, retry: Duration) -> PeerConfig {
+        PeerConfig {
+            message_retry: retry,
+            ..self
+        }
+    }
 }
 
 pub struct Peer<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
@@ -143,6 +157,7 @@ impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Peer<T> {
         }
 
         self.connection_mgr.send_pings();
+        self.connection_mgr.send_queued_messages();
 
         let mut msg;
         while {
@@ -181,6 +196,17 @@ impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Peer<T> {
         Ok(self.messages.pop_front())
     }
 
+    /// Send a reliable message, which is retried until the remote peer
+    /// acknowledges it.
+    pub fn send_reliable(&mut self, to: &SocketAddr, message: T) {
+        self.connection_mgr.send(to, message, true);
+    }
+
+    /// Sends an unreliable message, which is never retried again.
+    pub fn send_unreliable(&mut self, to: &SocketAddr, message: T) {
+        self.connection_mgr.send(to, message, false);
+    }
+
     /// Close the Peer, disconnecting all connections.
     pub fn close(&mut self) {
         self.connection_mgr.close();