feat(rumqttc): set session_expiry_interval in v5 (#854)

Co-authored-by: Devdutt Shenoi <[email protected]>

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
 

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);
 

rumqttc/src/v5/eventloop.rs

@@ -412,6 +412,11 @@ async fn mqtt_connect(
                     options.keep_alive = Duration::from_secs(keep_alive as u64);
                 }
                 network.set_max_outgoing_size(props.max_packet_size);
+
+                // Override local session_expiry_interval value if set by server.
+                if props.session_expiry_interval.is_some() {
+                    options.set_session_expiry_interval(props.session_expiry_interval);
+                }
             }
             Ok(Packet::ConnAck(connack))
         }

rumqttc/src/v5/mod.rs

@@ -312,6 +312,27 @@ impl MqttOptions {
         self.connect_properties.clone()
     }
 
+    /// set session expiry interval on connection properties
+    pub fn set_session_expiry_interval(&mut self, interval: Option<u32>) -> &mut Self {
+        if let Some(conn_props) = &mut self.connect_properties {
+            conn_props.session_expiry_interval = interval;
+            self
+        } else {
+            let mut conn_props = ConnectProperties::new();
+            conn_props.session_expiry_interval = interval;
+            self.set_connect_properties(conn_props)
+        }
+    }
+
+    /// get session expiry interval on connection properties
+    pub fn session_expiry_interval(&self) -> Option<u32> {
+        if let Some(conn_props) = &self.connect_properties {
+            conn_props.session_expiry_interval
+        } else {
+            None
+        }
+    }
+
     /// set receive maximum on connection properties
     pub fn set_receive_maximum(&mut self, recv_max: Option<u16>) -> &mut Self {
         if let Some(conn_props) = &mut self.connect_properties {