Squashed commit of the following:

commit 518773d Author: CQ Xiao <[email protected]> Date: Tue May 21 14:47:32 2024 +0800 rumqttc: resume session only if CONNACK with session present 1 (bytebeamio#864) * Check if session present to restore pending publishes. * Modify changelog. * remove changes that don't seem to be related * refactor: improve readability * feat: apply changes to v4 * Remove session_expiry_interval related code. * test: set clean session * test: broker saved session * test: fix resume reconnect --------- Co-authored-by: Devdutt Shenoi <[email protected]> Co-authored-by: Devdutt Shenoi <[email protected]> commit 67d9ca7 Author: CQ Xiao <[email protected]> Date: Thu May 16 23:05:10 2024 +0800 feat(rumqttc): set `session_expiry_interval` in v5 (bytebeamio#854) Co-authored-by: Devdutt Shenoi <[email protected]> commit 98997d1 Author: CQ Xiao <[email protected]> Date: Wed Jul 3 18:13:54 2024 +0800 HashMap -> VecDeque, cleanup
xiaocq2001 · Jul 4, 2024 · 68fcb73 · 68fcb73
1 parent fb290fb
commit 68fcb73
Show file tree

Hide file tree

Showing 8 changed files with 173 additions and 90 deletions.
diff --git a/rumqttc/CHANGELOG.md b/rumqttc/CHANGELOG.md
@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * `size()` method on `Packet` calculates size once serialized.
 * `read()` and `write()` methods on `Packet`.
 * `ConnectionAborted` variant on `StateError` type to denote abrupt end to a connection
+* `set_session_expiry_interval` and `session_expiry_interval` methods on `MqttOptions`.
 
 ### Changed
 
@@ -27,6 +28,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
+* Resume session only if broker sends `CONNACK` with `session_present == 1`.
 
 ### Security
 

diff --git a/rumqttc/examples/async_manual_acks_v5.rs b/rumqttc/examples/async_manual_acks_v5.rs
@@ -10,6 +10,7 @@ fn create_conn() -> (AsyncClient, EventLoop) {
     let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1884);
     mqttoptions
         .set_keep_alive(Duration::from_secs(5))
+        .set_session_expiry_interval(u32::MAX.into())
         .set_manual_acks(true)
         .set_clean_start(false);
 

diff --git a/rumqttc/src/eventloop.rs b/rumqttc/src/eventloop.rs
@@ -172,6 +172,8 @@ impl EventLoop {
         match self.select().await {
             Ok(v) => Ok(v),
             Err(e) => {
+                // MQTT requires that packets pending acknowledgement should be republished on session resume.
+                // Move pending messages from state to eventloop.
                 self.clean();
                 Err(e)
             }