teascade/teanet-rust
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix reliable messages being sent

Browse Source
This commit is contained in:
Sofia 2026-07-16 16:29:49 +03:00
parent a348dde42e
commit ad697b21b7
2 changed files with 46 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
src/connections.rs
View File

@ -41,8 +41,9 @@ pub enum ConnectionError {
} }
/// Manages connections for the Peer /// 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 /// Wraps UdpSocket with some helper methods
pub udp: UdpWrapper, pub udp: UdpWrapper,
connections: HashMap<SocketAddr, Connection<T>>, connections: HashMap<SocketAddr, Connection<T>>,
@ -50,7 +51,9 @@ pub(crate) struct ConnectionManager<T: Clone + Serialize + DeserializeOwned + Se
config: PeerConfig, 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 /// Creates a new ConnectionManager with the given socket and config
pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager<T> { pub fn new(socket: Arc<UdpSocket>, config: PeerConfig) -> ConnectionManager<T> {
ConnectionManager { ConnectionManager {
@ -320,38 +323,48 @@ impl<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'static> Connection
} }
Package::Messages(msgs) => { Package::Messages(msgs) => {
if let Some(conn) = self.connections.get_mut(addr) { if let Some(conn) = self.connections.get_mut(addr) {
// Store the expected number of messages to have received if conn.state == ConnectionState::Connected {
// from the given connection // Store the expected number of messages to have received
conn.messages_expected = msgs.messages_sent.max(conn.messages_expected); // from the given connection
conn.messages_received += 1; 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 // Clear all messages from the queue that the other peer has
// already seen // already seen
conn.reliable_queue.retain(|v| v.message_id > msgs.ack); 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 // Process all messages that have not been previously seen
// before, that is to say the connection's ack value is // before, that is to say the connection's ack value is
// lower than the message's id // lower than the message's id
for reliable_msg in &msgs.reliable { for reliable_msg in &msgs.reliable {
if reliable_msg.message_id >= conn.ack { if reliable_msg.message_id >= conn.ack {
messages.push(PeerMessage::Message(reliable_msg.message.clone())); messages.push(PeerMessage::Message(
conn.clone(),
reliable_msg.message.clone(),
));
}
} }
}
// Set connection's ack value to be at least one higher than // Set connection's ack value to be at least the highest
// the highest message_id that was received, so that we // message_id that was received, so that we inform the
// inform the remote peer not to send these messages again. // remote peer not to send these messages again.
conn.ack = msgs conn.ack = msgs
.reliable .reliable
.iter() .iter()
.map(|r| r.message_id + 1) .map(|r| r.message_id)
.max() .max()
.unwrap_or(0) .unwrap_or(0)
.max(conn.ack); .max(conn.ack);
// Always process all unreliable messages // Always process all unreliable messages
for unreliable_msg in &msgs.unreliable { for unreliable_msg in &msgs.unreliable {
messages.push(PeerMessage::Message(unreliable_msg.message.clone())); messages.push(PeerMessage::Message(
conn.clone(),
unreliable_msg.message.clone(),
));
}
} }
} }
} }

6
src/lib.rs
View File

@ -42,7 +42,7 @@ pub enum PeerMessage<T: Clone + Serialize + DeserializeOwned + Send + Sync + 'st
/// The Peer has been closed /// The Peer has been closed
Closed, Closed,
/// A single message of type T /// A single message of type T
Message(T), Message(Connection<T>, T),
} }
/// Optional configuration available for peers, mainly configuration of /// 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>, connection_mgr: ConnectionManager<T>,
closed: Arc<AtomicBool>, closed: Arc<AtomicBool>,
receiver: Receiver<ListenerMessage<T>>, receiver: Receiver<ListenerMessage<T>>,
messages: VecDeque<PeerMessage<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 /// Bind given port to listen to incoming messages. Creates a new Peer that
/// can establish new connections. /// can establish new connections.
pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer<T>, PeerError> { pub fn listen(port: Option<u16>, config: PeerConfig) -> Result<Peer<T>, PeerError> {