diff --git a/src/connections.rs b/src/connections.rs index fb3b956..0603b14 100644 --- a/src/connections.rs +++ b/src/connections.rs @@ -1,5 +1,5 @@ use std::{ - collections::HashMap, + collections::{HashMap, VecDeque}, io::Cursor, net::{SocketAddr, UdpSocket}, sync::Arc, @@ -13,7 +13,7 @@ use thiserror::Error; use crate::{ PeerConfig, PeerMessage, listener::{DATAGRAM_SIZE, ListenerError}, - package::{CloseReason, Message, Messages, Package}, + package::{CloseReason, DelayedPackage, Message, Messages, Package}, stats::NetStats, }; @@ -46,7 +46,7 @@ pub(crate) struct ConnectionManager< T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static, > { /// Wraps UdpSocket with some helper methods - pub udp: UdpWrapper, + pub udp: UdpWrapper, connections: HashMap>, closing_since: Option, config: PeerConfig, @@ -60,7 +60,7 @@ impl, config: PeerConfig) -> ConnectionManager { ConnectionManager { - udp: UdpWrapper::new(socket), + udp: UdpWrapper::new(socket, config.additional_ping), connections: HashMap::new(), closing_since: None, config, @@ -177,6 +177,13 @@ impl {} + Err(err) => { + println!("Error: {}", err); + self.close(); + } + } } pub fn send(&mut self, addr: &SocketAddr, message: T, reliable: bool) { @@ -453,14 +460,22 @@ impl { socket: Option>, + delayed_queue: VecDeque>, + additional_ping: Duration, } -impl UdpWrapper { - pub fn new(socket: Arc) -> UdpWrapper { +impl + UdpWrapper +{ + pub fn new(socket: Arc, additional_ping: Duration) -> UdpWrapper { UdpWrapper { socket: Some(socket), + delayed_queue: VecDeque::new(), + additional_ping, } } @@ -475,16 +490,32 @@ impl UdpWrapper { } /// Send a Package to a remote peer at addr - fn send_to( - &self, - addr: SocketAddr, - package: Package, - ) -> Result { + fn send_to(&mut self, addr: SocketAddr, package: Package) -> Result { + let delayed = DelayedPackage { + package, + created: Instant::now(), + addr, + size: 0, + }; + if Instant::now() < (delayed.created + self.additional_ping) { + let mut buf = Vec::new(); + let cursor = Cursor::new(&mut buf); + let encoder = GzEncoder::new(cursor, Compression::fast()); + ciborium::into_writer(&delayed.package, encoder) + .map_err(|err| SendError::SerializationError(err))?; + if buf.len() > DATAGRAM_SIZE { + return Err(SendError::DatagramTooLarge(buf.len())); + } + + self.delayed_queue.push_back(delayed); + return Ok(buf.len()); + } + if let Some(socket) = &self.socket { let mut buf = Vec::new(); let cursor = Cursor::new(&mut buf); let encoder = GzEncoder::new(cursor, Compression::fast()); - ciborium::into_writer(&package, encoder) + ciborium::into_writer(&delayed.package, encoder) .map_err(|err| SendError::SerializationError(err))?; if buf.len() > DATAGRAM_SIZE { return Err(SendError::DatagramTooLarge(buf.len())); @@ -497,6 +528,32 @@ impl UdpWrapper { Err(SendError::SocketDisconnected) } } + + fn send_delayed_messages(&mut self) -> 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); + + 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())); + } + socket + .send_to(&buf, message.addr) + .map_err(|e| SendError::SendError(e))?; + } + Ok(()) + } else { + Err(SendError::SocketDisconnected) + } + } } /// Represents a single connection from the peer to another peer diff --git a/src/lib.rs b/src/lib.rs index 202631c..c337f5d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,7 +7,7 @@ use std::{ mpsc::{Receiver, channel}, }, thread::spawn, - time::Duration, + time::{Duration, Instant}, }; use serde::{Serialize, de::DeserializeOwned}; @@ -16,7 +16,7 @@ use thiserror::*; use crate::{ connections::{Connection, ConnectionError, ConnectionManager}, listener::{Listener, ListenerMessage}, - package::CloseReason, + package::{CloseReason, DelayedPackage, Package}, stats::NetStats, }; @@ -55,6 +55,7 @@ pub struct PeerConfig { disconnect_timeout: Duration, message_retry: Duration, identifier: String, + additional_ping: Duration, } impl Default for PeerConfig { @@ -65,6 +66,7 @@ impl Default for PeerConfig { disconnect_timeout: Duration::from_millis(500), message_retry: Duration::from_millis(100), identifier: "ExamplePeer".to_string(), + additional_ping: Duration::from_nanos(0), } } } @@ -112,12 +114,23 @@ impl PeerConfig { ..self } } + + /// Sets additional ping for this peer, useful for debugging high-latency + /// connections. + pub fn with_additional_ping(self, ping: Duration) -> PeerConfig { + PeerConfig { + additional_ping: ping, + ..self + } + } } pub struct Peer { connection_mgr: ConnectionManager, + config: PeerConfig, closed: Arc, receiver: Receiver>, + delayed_messages: VecDeque>, messages: VecDeque>, } impl Drop @@ -159,9 +172,11 @@ impl now { + if let Some(msg) = self.delayed_messages.remove(i) { + self.messages.extend(self.connection_mgr.handle_package( + msg.package, + &msg.addr, + msg.size, + )); + } + } + } + } + let mut msg; while { msg = match self.receiver.try_recv() { @@ -196,11 +227,21 @@ impl { - self.messages.extend(self.connection_mgr.handle_package( - package, - &socket_addr, - bytes, - )); + let received = Instant::now(); + if Instant::now() > (received + self.config.additional_ping) { + self.messages.extend(self.connection_mgr.handle_package( + package, + &socket_addr, + bytes, + )); + } else { + self.delayed_messages.push_back(DelayedPackage { + package, + created: received, + addr: socket_addr, + size: bytes, + }); + } } ListenerMessage::PackageError(listener_error, socket_addr) => { println!("Error: {}", listener_error); diff --git a/src/listener.rs b/src/listener.rs index 0907c12..edd73af 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -1,8 +1,10 @@ use std::{ + collections::VecDeque, io::{self, Cursor}, marker::PhantomData, net::{SocketAddr, UdpSocket}, sync::{Arc, mpsc::Sender}, + time::{Duration, Instant}, }; use flate2::read::GzDecoder; diff --git a/src/package.rs b/src/package.rs index 8bcd375..f637841 100644 --- a/src/package.rs +++ b/src/package.rs @@ -1,7 +1,9 @@ -use serde::{Deserialize, Serialize}; +use std::{net::SocketAddr, time::Instant}; -#[derive(Debug, Serialize, Deserialize)] -pub(crate) enum Package { +use serde::{Deserialize, Serialize, de::DeserializeOwned}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) enum Package { Hello(String), Ping, Pong, @@ -33,7 +35,7 @@ impl std::fmt::Display for CloseReason { } } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct Messages { pub ack: u64, pub messages_sent: usize, @@ -46,3 +48,13 @@ pub(crate) struct Message { pub message_id: u64, pub message: T, } + +#[derive(Clone)] +pub(crate) struct DelayedPackage< + T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static, +> { + pub package: Package, + pub created: Instant, + pub addr: SocketAddr, + pub size: usize, +}