feat: oneshot channel returns Pkid

rumqttc/CHANGELOG.md

@@ -29,6 +29,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. 
 
 ### Deprecated
 

rumqttc/Cargo.toml

@@ -24,7 +24,7 @@ proxy = ["dep:async-http-proxy"]
 
 [dependencies]
 futures-util = { version = "0.3", default_features = false, features = ["std"] }
-tokio = { version = "1.36", features = ["rt", "macros", "io-util", "net", "time"] }
+tokio = { version = "1.36", features = ["rt", "macros", "io-util", "net", "time", "sync"] }
 bytes = "1.5"
 log = "0.4"
 flume = { version = "0.11", default-features = false, features = ["async"] }
@@ -55,6 +55,7 @@ matches = "0.1"
 pretty_assertions = "1"
 pretty_env_logger = "0.5"
 serde = { version = "1", features = ["derive"] }
+tokio-util = { version = "0.7", features = ["time"] }
 
 [[example]]
 name = "tls"

rumqttc/examples/pkid_promise.rs

@@ -0,0 +1,69 @@
+use futures_util::stream::StreamExt;
+use tokio::{
+    select,
+    task::{self, JoinSet},
+};
+use tokio_util::time::DelayQueue;
+
+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>> {
+    pretty_env_logger::init();
+    // color_backtrace::install();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "broker.emqx.io", 1883);
+    mqttoptions.set_keep_alive(Duration::from_secs(5));
+
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+    task::spawn(async move {
+        requests(client).await;
+    });
+
+    loop {
+        let event = eventloop.poll().await;
+        match &event {
+            Ok(v) => {
+                println!("Event = {v:?}");
+            }
+            Err(e) => {
+                println!("Error = {e:?}");
+                return Ok(());
+            }
+        }
+    }
+}
+
+async fn requests(client: AsyncClient) {
+    let mut joins = JoinSet::new();
+    joins.spawn(
+        client
+            .subscribe("hello/world", QoS::AtMostOnce)
+            .await
+            .unwrap(),
+    );
+
+    let mut queue = DelayQueue::new();
+    for i in 1..=10 {
+        queue.insert(i as usize, Duration::from_secs(i));
+    }
+
+    loop {
+        select! {
+            Some(i) = queue.next() => {
+                joins.spawn(
+                    client
+                        .publish("hello/world", QoS::ExactlyOnce, false, vec![1; i.into_inner()])
+                        .await
+                        .unwrap(),
+                );
+            }
+            Some(Ok(Ok(pkid))) = joins.join_next() => {
+                println!("Pkid: {:?}", pkid);
+            }
+            else => break,
+        }
+    }
+}

rumqttc/examples/pkid_promise_v5.rs

@@ -0,0 +1,69 @@
+use futures_util::stream::StreamExt;
+use tokio::{
+    select,
+    task::{self, JoinSet},
+};
+use tokio_util::time::DelayQueue;
+
+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>> {
+    pretty_env_logger::init();
+    // color_backtrace::install();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "broker.emqx.io", 1883);
+    mqttoptions.set_keep_alive(Duration::from_secs(5));
+
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+    task::spawn(async move {
+        requests(client).await;
+    });
+
+    loop {
+        let event = eventloop.poll().await;
+        match &event {
+            Ok(v) => {
+                println!("Event = {v:?}");
+            }
+            Err(e) => {
+                println!("Error = {e:?}");
+                return Ok(());
+            }
+        }
+    }
+}
+
+async fn requests(client: AsyncClient) {
+    let mut joins = JoinSet::new();
+    joins.spawn(
+        client
+            .subscribe("hello/world", QoS::AtMostOnce)
+            .await
+            .unwrap(),
+    );
+
+    let mut queue = DelayQueue::new();
+    for i in 1..=10 {
+        queue.insert(i as usize, Duration::from_secs(i));
+    }
+
+    loop {
+        select! {
+            Some(i) = queue.next() => {
+                joins.spawn(
+                    client
+                        .publish("hello/world", QoS::ExactlyOnce, false, vec![1; i.into_inner()])
+                        .await
+                        .unwrap(),
+                );
+            }
+            Some(Ok(Ok(pkid))) = joins.join_next() => {
+                println!("Pkid: {:?}", pkid);
+            }
+            else => break,
+        }
+    }
+}