From ad697b21b75a25b5ee379a88c4320a30f6bd3d65 Mon Sep 17 00:00:00 2001
From: Sofia <sofia@teascade.net>
Date: Thu, 16 Jul 2026 16:29:49 +0300
Subject: [PATCH] Fix reliable messages being sent

---
 src/connections.rs | 73 +++++++++++++++++++++++++++-------------------
 src/lib.rs         |  6 ++--
 2 files changed, 46 insertions(+), 33 deletions(-)

diff --git a/src/connections.rs b/src/connections.rs
index 686ce87..8c56cbf 100644
--- a/src/connections.rs
+++ b/src/connections.rs
@@ -41,8 +41,9 @@ pub enum ConnectionError {
 }
 
 /// Manages connections for the Peer
-pub(crate) struct ConnectionManager<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static>
-{
+pub(crate) struct ConnectionManager<
+    T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static,
+> {
     /// Wraps UdpSocket with some helper methods
     pub udp: UdpWrapper,
     connections: HashMap<SocketAddr, Connection<T>>,
@@ -50,7 +51,9 @@ pub(crate) struct ConnectionManager<T: Clone + Serialize + DeserializeOwned + Se
     config: PeerConfig,
 }
 
-impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> ConnectionManager<T> {
+impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static>
+    ConnectionManager<T>
+{
     /// Creates a new ConnectionManager with the given socket and config
     pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager<T> {
         ConnectionManager {
@@ -320,38 +323,48 @@ impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Connection
             }
             Package::Messages(msgs) => {
                 if let Some(conn) = self.connections.get_mut(addr) {
-                    // Store the expected number of messages to have received
-                    // from the given connection
-                    conn.messages_expected = msgs.messages_sent.max(conn.messages_expected);
-                    conn.messages_received += 1;
+                    if conn.state == ConnectionState::Connected {
+                        // Store the expected number of messages to have received
+                        // from the given connection
+                        conn.messages_expected = msgs.messages_sent.max(conn.messages_expected);
+                        conn.messages_received += 1;
 
-                    // Clear all messages from the queue that the other peer has
-                    // already seen
-                    conn.reliable_queue.retain(|v| v.message_id > msgs.ack);
+                        // Clear all messages from the queue that the other peer has
+                        // already seen
+                        println!("{:?}", conn.reliable_queue);
+                        println!("{:?}", msgs.ack);
+                        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()));
+                        // 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(
+                                    conn.clone(),
+                                    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);
+                        // Set connection's ack value to be at least 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)
+                            .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()));
+                        // Always process all unreliable messages
+                        for unreliable_msg in &msgs.unreliable {
+                            messages.push(PeerMessage::Message(
+                                conn.clone(),
+                                unreliable_msg.message.clone(),
+                            ));
+                        }
                     }
                 }
             }
diff --git a/src/lib.rs b/src/lib.rs
index 64348d5..dd842e3 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -42,7 +42,7 @@ pub enum PeerMessage<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'st
     /// The Peer has been closed
     Closed,
     /// A single message of type T
-    Message(T),
+    Message(Connection<T>, T),
 }
 
 /// Optional configuration available for peers, mainly configuration of
@@ -102,14 +102,14 @@ impl PeerConfig {
     }
 }
 
-pub struct Peer<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
+pub struct Peer<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static> {
     connection_mgr: ConnectionManager<T>,
     closed: Arc<AtomicBool>,
     receiver: Receiver<ListenerMessage<T>>,
     messages: VecDeque<PeerMessage<T>>,
 }
 
-impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Peer<T> {
+impl<T: std::fmt::Debug + Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Peer<T> {
     /// Bind given port to listen to incoming messages. Creates a new Peer that
     /// can establish new connections.
     pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer<T>, PeerError> {