diff --git a/src/connections.rs b/src/connections.rs index 0603b14..5fdc6cb 100644 --- a/src/connections.rs +++ b/src/connections.rs @@ -100,8 +100,14 @@ impl 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 {} - Err(err) => { - println!("Error: {}", err); + Err(_) => { self.close(); } } @@ -324,6 +329,7 @@ impl { 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,24 +536,28 @@ impl Result<(), SendError> { - 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); + let now = Instant::now(); if let Some(socket) = &self.socket { - 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())); + 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))?; } - socket - .send_to(&buf, message.addr) - .map_err(|e| SendError::SendError(e))?; } Ok(()) } else { @@ -566,6 +576,7 @@ pub struct Connection { last_sent_ping: Instant, last_recv_ping: Instant, + last_recv_pong: Instant, last_recv_close: Instant, closing_since: Instant, @@ -593,6 +604,7 @@ impl Clone for Connection { 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 Connection { 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(), diff --git a/src/lib.rs b/src/lib.rs index c337f5d..af33425 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -51,6 +51,7 @@ pub enum PeerMessage 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 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,