From cdab0bb448a1dd4d04750c4798c9424c2cf76d37 Mon Sep 17 00:00:00 2001 From: Sofia Date: Thu, 16 Jul 2026 04:44:06 +0300 Subject: [PATCH] Add sending messages --- src/connections.rs | 110 ++++++++++++++++++++++++++++++++++++++++++++- src/lib.rs | 26 +++++++++++ 2 files changed, 134 insertions(+), 2 deletions(-) diff --git a/src/connections.rs b/src/connections.rs index d8ca1a5..a32ef42 100644 --- a/src/connections.rs +++ b/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 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 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 { last_recv_close: Instant, closing_since: Instant, + message_counter: u64, + ack: u64, reliable_queue: Vec>, + last_msg_send: Instant, /// What state is the connection currently in? pub state: ConnectionState, @@ -357,7 +453,12 @@ impl Clone for Connection { 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 Connection { 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, } diff --git a/src/lib.rs b/src/lib.rs index 9501c55..519a736 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -39,6 +39,8 @@ pub enum PeerMessage, Option), /// 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 { @@ -143,6 +157,7 @@ impl Peer { } self.connection_mgr.send_pings(); + self.connection_mgr.send_queued_messages(); let mut msg; while { @@ -181,6 +196,17 @@ impl Peer { 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();