src/connections.rs

@@ -100,14 +100,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
 
         if let Some(_) = &self.udp.socket {
             for (addr, conn) in &mut self.connections {
-                let interval = if conn.last_sent_ping > conn.last_recv_pong {
-                    Duration::from_millis(2000)
-                } else {
-                    self.config.ping_interval
-                };
-
                 let duration = now - conn.last_sent_ping;
-                if duration > interval {
+                if duration > self.config.ping_interval {
                     match conn.state {
                         ConnectionState::ReceivingConnection | ConnectionState::Connecting => {
                             match self
@@ -185,7 +179,8 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
         }
         match self.udp.send_delayed_messages() {
             Ok(_) => {}
-            Err(_) => {
+            Err(err) => {
+                println!("Error: {}", err);
                 self.close();
             }
         }
@@ -329,7 +324,6 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
             Package::Pong => {
                 if let Some(conn) = self.connections.get_mut(addr) {
                     if conn.state == ConnectionState::Connected {
-                        conn.last_recv_pong = Instant::now();
                         conn.ping = Instant::now() - conn.last_sent_ping;
                     }
                 }
@@ -536,28 +530,24 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
     }
 
     fn send_delayed_messages(&mut self) -> Result<(), SendError> {
-        let now = Instant::now();
+        let queue_iter = self.delayed_queue.clone().into_iter();
+        let treshold = Instant::now() - self.additional_ping;
+        let messages = queue_iter.take_while(|m| treshold >= m.created);
+        self.delayed_queue.retain(|m| treshold < m.created);
 
         if let Some(socket) = &self.socket {
-            for i in (0..self.delayed_queue.len()).rev() {
-                if let Some(message) = self.delayed_queue.get(i) {
-                    if now < (message.created + self.additional_ping) {
-                        continue;
-                    }
-                }
-                if let Some(message) = self.delayed_queue.remove(i) {
-                    let mut buf = Vec::new();
-                    let cursor = Cursor::new(&mut buf);
-                    let encoder = GzEncoder::new(cursor, Compression::fast());
-                    ciborium::into_writer(&message.package, encoder)
-                        .map_err(|err| SendError::SerializationError(err))?;
-                    if buf.len() > DATAGRAM_SIZE {
-                        return Err(SendError::DatagramTooLarge(buf.len()));
-                    }
-                    socket
-                        .send_to(&buf, message.addr)
-                        .map_err(|e| SendError::SendError(e))?;
+            for message in messages {
+                let mut buf = Vec::new();
+                let cursor = Cursor::new(&mut buf);
+                let encoder = GzEncoder::new(cursor, Compression::fast());
+                ciborium::into_writer(&message.package, encoder)
+                    .map_err(|err| SendError::SerializationError(err))?;
+                if buf.len() > DATAGRAM_SIZE {
+                    return Err(SendError::DatagramTooLarge(buf.len()));
                 }
+                socket
+                    .send_to(&buf, message.addr)
+                    .map_err(|e| SendError::SendError(e))?;
             }
             Ok(())
         } else {
@@ -576,7 +566,6 @@ pub struct Connection<T: Clone> {
 
     last_sent_ping: Instant,
     last_recv_ping: Instant,
-    last_recv_pong: Instant,
     last_recv_close: Instant,
     closing_since: Instant,
 
@@ -604,7 +593,6 @@ impl<T: Clone> Clone for Connection<T> {
             ping: self.ping.clone(),
             last_sent_ping: self.last_sent_ping.clone(),
             last_recv_ping: self.last_recv_ping.clone(),
-            last_recv_pong: self.last_recv_pong.clone(),
             last_recv_close: self.last_recv_close.clone(),
             closing_since: self.closing_since.clone(),
 
@@ -633,7 +621,6 @@ impl<T: Clone> Connection<T> {
             ping: Duration::default(),
             last_sent_ping: Instant::now() - Duration::from_hours(1),
             last_recv_ping: Instant::now(),
-            last_recv_pong: Instant::now(),
             last_recv_close: Instant::now(),
             closing_since: Instant::now(),
 

src/lib.rs

@@ -51,7 +51,6 @@ pub enum PeerMessage<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'st
 #[derive(Debug, Clone)]
 pub struct PeerConfig {
     ping_interval: Duration,
-    max_ping_interval: Duration,
     timeout: Duration,
     disconnect_timeout: Duration,
     message_retry: Duration,
@@ -63,8 +62,7 @@ impl Default for PeerConfig {
     fn default() -> Self {
         Self {
             ping_interval: Duration::from_millis(100),
-            max_ping_interval: Duration::from_millis(2000),
-            timeout: Duration::from_millis(5000),
+            timeout: Duration::from_millis(2000),
             disconnect_timeout: Duration::from_millis(500),
             message_retry: Duration::from_millis(100),
             identifier: "ExamplePeer".to_string(),
@@ -74,8 +72,7 @@ impl Default for PeerConfig {
 }
 
 impl PeerConfig {
-    /// Set the desired interval between "ping"s (assuming pongs arrive in due
-    /// time)
+    /// Set the desired interval between "ping"s
     pub fn with_ping_interval(self, interval: Duration) -> PeerConfig {
         PeerConfig {
             ping_interval: interval,
@@ -83,15 +80,6 @@ impl PeerConfig {
         }
     }
 
-    /// Set the desired maximum interval between "ping"s (assuming pongs aren't
-    /// being received in due time)
-    pub fn with_max_ping_interval(self, interval: Duration) -> PeerConfig {
-        PeerConfig {
-            max_ping_interval: interval,
-            ..self
-        }
-    }
-
     /// Set the length of timeout; that is to say the duration of time which is
     /// acceptable to occur between pings before a Timeout-error occurs
     pub fn with_timeout(self, timeout: Duration) -> PeerConfig {
@@ -131,7 +119,7 @@ impl PeerConfig {
     /// connections.
     pub fn with_additional_ping(self, ping: Duration) -> PeerConfig {
         PeerConfig {
-            additional_ping: ping / 2,
+            additional_ping: ping,
             ..self
         }
     }
@@ -216,7 +204,7 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
         let now = Instant::now();
         for i in (0..self.delayed_messages.len()).rev() {
             if let Some(msg) = self.delayed_messages.get(i) {
-                if now >= (msg.created + self.config.additional_ping) {
+                if (msg.created + self.config.additional_ping) > now {
                     if let Some(msg) = self.delayed_messages.remove(i) {
                         self.messages.extend(self.connection_mgr.handle_package(
                             msg.package,