feat: return token when request is made, resolve token when request handling completes

rumqttc/CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * `set_session_expiry_interval` and `session_expiry_interval` methods on `MqttOptions`.
 * `Auth` packet as per MQTT5 standards
 * Allow configuring  the `nodelay` property of underlying TCP client with the `tcp_nodelay` field in `NetworkOptions`
+* `publish` / `subscribe` / `unsubscribe` methods on `AsyncClient` and `Client` now return an `AckPromise` which resolves when the packet(except for QoS 0 publishes, which resolve as soon as handled) is acknowledged by the broker.
 
 ### Changed
 

rumqttc/examples/ack_promise.rs

@@ -0,0 +1,81 @@
+use tokio::task::{self, JoinSet};
+
+use rumqttc::{AsyncClient, MqttOptions, QoS};
+use std::error::Error;
+use std::time::Duration;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    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
+    match client
+        .subscribe("hello/world", QoS::AtMostOnce)
+        .await
+        .unwrap()
+        .await
+    {
+        Ok(pkid) => println!("Acknowledged Sub({pkid:?})"),
+        Err(e) => println!("Subscription failed: {e:?}"),
+    }
+
+    // Publish at all QoS levels and wait for broker acknowledgement
+    for (i, qos) in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce]
+        .into_iter()
+        .enumerate()
+    {
+        match client
+            .publish("hello/world", qos, false, vec![1; i])
+            .await
+            .unwrap()
+            .await
+        {
+            Ok(ack) => println!("Acknowledged Pub({ack:?})"),
+            Err(e) => println!("Publish failed: {e:?}"),
+        }
+    }
+
+    // Publish with different QoS levels and spawn wait for notification
+    let mut set = JoinSet::new();
+    for (i, qos) in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce]
+        .into_iter()
+        .enumerate()
+    {
+        let token = client
+            .publish("hello/world", qos, false, vec![1; i])
+            .await
+            .unwrap();
+        set.spawn(token);
+    }
+
+    while let Some(Ok(res)) = set.join_next().await {
+        match res {
+            Ok(ack) => println!("Acknowledged Pub({ack:?})"),
+            Err(e) => println!("Publish failed: {e:?}"),
+        }
+    }
+
+    // Unsubscribe and wait for broker acknowledgement
+    match client.unsubscribe("hello/world").await.unwrap().await {
+        Ok(ack) => println!("Acknowledged Unsub({ack:?})"),
+        Err(e) => println!("Unsubscription failed: {e:?}"),
+    }
+
+    Ok(())
+}

rumqttc/examples/ack_promise_sync.rs

@@ -0,0 +1,98 @@
+use flume::bounded;
+use rumqttc::{Client, MqttOptions, QoS, TokenError};
+use std::error::Error;
+use std::thread::{self, sleep};
+use std::time::Duration;
+
+fn main() -> Result<(), Box<dyn Error>> {
+    let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
+    mqttoptions.set_keep_alive(Duration::from_secs(5));
+
+    let (client, mut conn) = Client::new(mqttoptions, 10);
+    thread::spawn(move || {
+        for event in conn.iter() {
+            match &event {
+                Ok(v) => {
+                    println!("Event = {v:?}");
+                }
+                Err(e) => {
+                    println!("Error = {e:?}");
+                }
+            }
+        }
+    });
+
+    // Subscribe and wait for broker acknowledgement
+    match client
+        .subscribe("hello/world", QoS::AtMostOnce)
+        .unwrap()
+        .wait()
+    {
+        Ok(pkid) => println!("Acknowledged Sub({pkid:?})"),
+        Err(e) => println!("Subscription failed: {e:?}"),
+    }
+
+    // Publish at all QoS levels and wait for broker acknowledgement
+    for (i, qos) in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce]
+        .into_iter()
+        .enumerate()
+    {
+        match client
+            .publish("hello/world", qos, false, vec![1; i])
+            .unwrap()
+            .wait()
+        {
+            Ok(ack) => println!("Acknowledged Pub({ack:?})"),
+            Err(e) => println!("Publish failed: {e:?}"),
+        }
+    }
+
+    // Spawn threads for each publish, use channel to notify result
+    let (tx, rx) = bounded(1);
+
+    for (i, qos) in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce]
+        .into_iter()
+        .enumerate()
+    {
+        let token = client
+            .publish("hello/world", qos, false, vec![1; i])
+            .unwrap();
+        let tx = tx.clone();
+        thread::spawn(move || {
+            let res = token.wait();
+            tx.send(res).unwrap()
+        });
+    }
+
+    // Try resolving a promise, if it is waiting to resolve, try again after a sleep of 1s
+    let mut token = client
+        .publish("hello/world", QoS::AtMostOnce, false, vec![1; 4])
+        .unwrap();
+    thread::spawn(move || loop {
+        match token.check() {
+            Err(TokenError::Waiting) => {
+                println!("Promise yet to resolve, retrying");
+                sleep(Duration::from_secs(1));
+            }
+            res => {
+                tx.send(res).unwrap();
+                break;
+            }
+        }
+    });
+
+    while let Ok(res) = rx.recv() {
+        match res {
+            Ok(ack) => println!("Acknowledged Pub({ack:?})"),
+            Err(e) => println!("Publish failed: {e:?}"),
+        }
+    }
+
+    // Unsubscribe and wait for broker acknowledgement
+    match client.unsubscribe("hello/world").unwrap().wait() {
+        Ok(ack) => println!("Acknowledged Unsub({ack:?})"),
+        Err(e) => println!("Unsubscription failed: {e:?}"),
+    }
+
+    Ok(())
+}

rumqttc/examples/ack_promise_v5.rs

@@ -0,0 +1,81 @@
+use tokio::task::{self, JoinSet};
+
+use rumqttc::v5::{mqttbytes::QoS, AsyncClient, MqttOptions};
+use std::error::Error;
+use std::time::Duration;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    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
+    match client
+        .subscribe("hello/world", QoS::AtMostOnce)
+        .await
+        .unwrap()
+        .await
+    {
+        Ok(pkid) => println!("Acknowledged Sub({pkid:?})"),
+        Err(e) => println!("Subscription failed: {e:?}"),
+    }
+
+    // Publish at all QoS levels and wait for broker acknowledgement
+    for (i, qos) in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce]
+        .into_iter()
+        .enumerate()
+    {
+        match client
+            .publish("hello/world", qos, false, vec![1; i])
+            .await
+            .unwrap()
+            .await
+        {
+            Ok(pkid) => println!("Acknowledged Pub({pkid:?})"),
+            Err(e) => println!("Publish failed: {e:?}"),
+        }
+    }
+
+    // Publish with different QoS levels and spawn wait for notification
+    let mut set = JoinSet::new();
+    for (i, qos) in [QoS::AtMostOnce, QoS::AtLeastOnce, QoS::ExactlyOnce]
+        .into_iter()
+        .enumerate()
+    {
+        let token = client
+            .publish("hello/world", qos, false, vec![1; i])
+            .await
+            .unwrap();
+        set.spawn(token);
+    }
+
+    while let Some(Ok(res)) = set.join_next().await {
+        match res {
+            Ok(pkid) => println!("Acknowledged Pub({pkid:?})"),
+            Err(e) => println!("Publish failed: {e:?}"),
+        }
+    }
+
+    // Unsubscribe and wait for broker acknowledgement
+    match client.unsubscribe("hello/world").await.unwrap().await {
+        Ok(pkid) => println!("Acknowledged Unsub({pkid:?})"),
+        Err(e) => println!("Unsubscription failed: {e:?}"),
+    }
+
+    Ok(())
+}