Fix ping interval

src/connections.rs

@@ -100,8 +100,14 @@ 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 > self.config.ping_interval {
+                if duration > interval {
                     match conn.state {
                         ConnectionState::ReceivingConnection | ConnectionState::Connecting => {
                             match self
@@ -179,8 +185,7 @@ 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();
             }
         }
@@ -324,6 +329,7 @@ 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;
                     }
                 }
@@ -530,13 +536,16 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
     }
 
     fn send_delayed_messages(&mut self) -> Result<(), SendError> {
-        let queue_iter = self.delayed_queue.clone().into_iter();
+        let now = Instant::now();
-        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 message in messages {
+            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());
@@ -549,6 +558,7 @@ impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + '
                         .send_to(&buf, message.addr)
                         .map_err(|e| SendError::SendError(e))?;
                 }
+            }
             Ok(())
         } else {
             Err(SendError::SocketDisconnected)
@@ -566,6 +576,7 @@ pub struct Connection<T: Clone> {
 
     last_sent_ping: Instant,
     last_recv_ping: Instant,
+    last_recv_pong: Instant,
     last_recv_close: Instant,
     closing_since: Instant,
 
@@ -593,6 +604,7 @@ 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(),
 
@@ -621,6 +633,7 @@ 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,6 +51,7 @@ 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,
@@ -62,7 +63,8 @@ impl Default for PeerConfig {
     fn default() -> Self {
         Self {
             ping_interval: Duration::from_millis(100),
-            timeout: Duration::from_millis(2000),
+            max_ping_interval: Duration::from_millis(2000),
+            timeout: Duration::from_millis(5000),
             disconnect_timeout: Duration::from_millis(500),
             message_retry: Duration::from_millis(100),
             identifier: "ExamplePeer".to_string(),
@@ -72,7 +74,8 @@ impl Default for PeerConfig {
 }
 
 impl PeerConfig {
-    /// Set the desired interval between "ping"s
+    /// Set the desired interval between "ping"s (assuming pongs arrive in due
+    /// time)
     pub fn with_ping_interval(self, interval: Duration) -> PeerConfig {
         PeerConfig {
             ping_interval: interval,
@@ -80,6 +83,15 @@ 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 {
@@ -204,7 +216,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 (msg.created + self.config.additional_ping) > now {
+                if now >= (msg.created + self.config.additional_ping) {
                     if let Some(msg) = self.delayed_messages.remove(i) {
                         self.messages.extend(self.connection_mgr.handle_package(
                             msg.package,