publish / subscribe / unsubscribe methods return a notice future that waits publish (QoS0), PubAck (QoS1) and PubRec (QoS2)

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
 
@@ -56,7 +58,7 @@ To update your code simply remove `Key::ECC()` or `Key::RSA()` from the initiali
   `rusttls-pemfile` to `2.0.0`, `async-tungstenite` to `0.24.0`, `ws_stream_tungstenite` to `0.12.0`
   and `http` to `1.0.0`. This is a breaking change as types from some of these crates are part of
   the public API.
-- `publish` / `subscribe` / `unsubscribe` methods on `AsyncClient` and `Client` now return a `PkidPromise` which resolves into the identifier value chosen by the `EventLoop` when handling the packet. 
+- `publish` / `subscribe` / `unsubscribe` methods on `AsyncClient` and `Client` now return a `NoticeFuture` which is noticed after the packet is released (sent in QoS0, ACKed in QoS1, COMPed in QoS2). 
 
 ### Deprecated
 

rumqttc/Cargo.toml

@@ -30,6 +30,7 @@ bytes = "1.5"
 log = "0.4"
 flume = { version = "0.11", default-features = false, features = ["async"] }
 thiserror = "1"
+linked-hash-map = "0.5"
 
 # Optional
 # rustls

rumqttc/examples/ack_notif.rs

@@ -1,4 +1,4 @@
-use tokio::task::{self, JoinSet};
+use tokio::{task::{self, JoinSet}, time};
 
 use rumqttc::{AsyncClient, MqttOptions, QoS};
 use std::error::Error;
@@ -35,24 +35,38 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .wait_async()
         .await
         .unwrap();
+    client
+        .subscribe("hello/world", QoS::AtLeastOnce)
+        .await
+        .unwrap()
+        .wait_async()
+        .await
+        .unwrap();
+    client
+        .subscribe("hello/world", QoS::ExactlyOnce)
+        .await
+        .unwrap()
+        .wait_async()
+        .await
+        .unwrap();
 
     // Publish and spawn wait for notification
     let mut set = JoinSet::new();
 
     let future = client
-        .publish("hello/world", QoS::AtMostOnce, false, vec![1; 1024])
+        .publish("hello/world", QoS::AtMostOnce, false, vec![1; 1])
         .await
         .unwrap();
     set.spawn(future.wait_async());
 
     let future = client
-        .publish("hello/world", QoS::AtLeastOnce, false, vec![1; 1024])
+        .publish("hello/world", QoS::AtLeastOnce, false, vec![1; 2])
         .await
         .unwrap();
     set.spawn(future.wait_async());
 
     let future = client
-        .publish("hello/world", QoS::ExactlyOnce, false, vec![1; 1024])
+        .publish("hello/world", QoS::ExactlyOnce, false, vec![1; 3])
         .await
         .unwrap();
     set.spawn(future.wait_async());
@@ -61,5 +75,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
         println!("Acknoledged = {:?}", res?);
     }
 
+    time::sleep(Duration::from_secs(6)).await;
     Ok(())
 }

rumqttc/examples/ack_notif_v5.rs

@@ -0,0 +1,80 @@
+use tokio::{task::{self, JoinSet}, time};
+
+use rumqttc::v5::{AsyncClient, MqttOptions, mqttbytes::QoS};
+use std::error::Error;
+use std::time::Duration;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    pretty_env_logger::init();
+    // color_backtrace::install();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
+    mqttoptions.set_keep_alive(Duration::from_secs(5));
+
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+    task::spawn(async move {
+        loop {
+            let event = eventloop.poll().await;
+            match &event {
+                Ok(v) => {
+                    println!("Event = {v:?}");
+                }
+                Err(e) => {
+                    println!("Error = {e:?}");
+                }
+            }
+        }
+    });
+
+    // Subscribe and wait for broker acknowledgement
+    client
+        .subscribe("hello/world", QoS::AtMostOnce)
+        .await
+        .unwrap()
+        .wait_async()
+        .await
+        .unwrap();
+    client
+        .subscribe("hello/world", QoS::AtLeastOnce)
+        .await
+        .unwrap()
+        .wait_async()
+        .await
+        .unwrap();
+    client
+        .subscribe("hello/world", QoS::ExactlyOnce)
+        .await
+        .unwrap()
+        .wait_async()
+        .await
+        .unwrap();
+
+    // Publish and spawn wait for notification
+    let mut set = JoinSet::new();
+
+    let future = client
+        .publish("hello/world", QoS::AtMostOnce, false, vec![1; 1])
+        .await
+        .unwrap();
+    set.spawn(future.wait_async());
+
+    let future = client
+        .publish("hello/world", QoS::AtLeastOnce, false, vec![1; 2])
+        .await
+        .unwrap();
+    set.spawn(future.wait_async());
+
+    let future = client
+        .publish("hello/world", QoS::ExactlyOnce, false, vec![1; 3])
+        .await
+        .unwrap();
+    set.spawn(future.wait_async());
+
+    while let Some(res) = set.join_next().await {
+        println!("Acknoledged = {:?}", res?);
+    }
+
+    time::sleep(Duration::from_secs(6)).await;
+    Ok(())
+}

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