Refactor streaming support

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [5.2.0] - 2025-05-08
+
+* Refactor stream support in publish builder
+
 ## [5.1.0] - 2025-05-05
 
 * Fix .read_all() method return type

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-mqtt"
-version = "5.1.0"
+version = "5.2.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Client and Server framework for MQTT v5 and v3.1.1 protocols"
 documentation = "https://docs.rs/ntex-mqtt"

examples/mqtt-ws-client.rs

@@ -63,7 +63,7 @@ async fn main() -> std::io::Result<()> {
     });
     ntex::rt::spawn(router.start_default());
 
-    sink.publish("test-topic", Bytes::from_static(b"data")).send_at_least_once().await.unwrap();
+    sink.publish("test-topic").send_at_least_once(Bytes::from_static(b"data")).await.unwrap();
     println!("ack is received");
 
     sleep(Millis(10_000)).await;

examples/openssl-client.rs

@@ -45,7 +45,7 @@ async fn main() -> std::io::Result<()> {
     let router = client.resource("response", publish);
     ntex::rt::spawn(router.start_default());
 
-    sink.publish("test-topic", "Publish data".into()).send_at_least_once().await.unwrap();
+    sink.publish("test-topic").send_at_least_once("Publish data".into()).await.unwrap();
 
     sleep(Millis(10_000)).await;
 

examples/subs.rs

@@ -60,8 +60,8 @@ async fn publish(
         let payload = publish.read_all().await.unwrap();
         session
             .sink
-            .publish(publish.packet().topic.clone(), payload)
-            .send_at_least_once()
+            .publish(publish.packet().topic.clone())
+            .send_at_least_once(payload)
             .await
             .unwrap();
     }

examples/subs_client.rs

@@ -83,7 +83,7 @@ async fn main() -> std::io::Result<()> {
         .unwrap();
 
     log::info!("sending client publish");
-    let ack = sink.publish("topic1", "Hello world!".into()).send_at_least_once().await.unwrap();
+    let ack = sink.publish("topic1").send_at_least_once("Hello world!".into()).await.unwrap();
     log::info!("ack received: {:?}", ack);
 
     sleep(Millis(1_000)).await;