fix: filter unexpected PubAck in event loop pending queue. (#859)

* Update eventloop.rs filter unexpected puback. * Update eventloop.rs filter unexpected puback * cargo fmt * Update CHANGELOG.md update CHANGELOG.md * Update rumqttc/src/eventloop.rs apply suggestion from rumqtt team Co-authored-by: Devdutt Shenoi <[email protected]> * Update rumqttc/src/v5/eventloop.rs apply suggestion from rumqtt team Co-authored-by: Devdutt Shenoi <[email protected]> --------- Co-authored-by: Devdutt Shenoi <[email protected]> Co-authored-by: Devdutt Shenoi <[email protected]>
bytebeamio · May 22, 2024 · 51224e1 · 51224e1
1 parent 83d8f77
commit 51224e1
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 2 deletions.
diff --git a/rumqttc/CHANGELOG.md b/rumqttc/CHANGELOG.md
@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Validate filters while creating subscription requests.
 * Make v4::Connect::write return correct value
 * Ordering of `State.events` related to `QoS > 0` publishes
+* Filter PUBACK in pending save requests to fix unexpected PUBACK sent to reconnected broker.
 * Resume session only if broker sends `CONNACK` with `session_present == 1`.
 
 ### Security

diff --git a/rumqttc/src/eventloop.rs b/rumqttc/src/eventloop.rs
@@ -130,7 +130,15 @@ impl EventLoop {
         self.pending.extend(self.state.clean());
 
         // drain requests from channel which weren't yet received
-        let requests_in_channel = self.requests_rx.drain();
+        let mut requests_in_channel: Vec<_> = self.requests_rx.drain().collect();
+
+        requests_in_channel.retain(|request| {
+            match request {
+                Request::PubAck(_) => false,  // Wait for publish retransmission, else the broker could be confused by an unexpected ack
+                _ => true,
+            }
+        });
+
         self.pending.extend(requests_in_channel);
     }
 

diff --git a/rumqttc/src/v5/eventloop.rs b/rumqttc/src/v5/eventloop.rs
@@ -126,7 +126,15 @@ impl EventLoop {
         self.pending.extend(self.state.clean());
 
         // drain requests from channel which weren't yet received
-        let requests_in_channel = self.requests_rx.drain();
+        let mut requests_in_channel: Vec<_> = self.requests_rx.drain().collect();
+
+        requests_in_channel.retain(|request| {
+            match request {
+                Request::PubAck(_) => false,  // Wait for publish retransmission, else the broker could be confused by an unexpected ack
+                _ => true,
+            }
+        });
+
         self.pending.extend(requests_in_channel);
     }