feat: return pkid of ack

bytebeamio · Sep 30, 2024 · 8ad2223 · 8ad2223
1 parent 2632ab1
commit 8ad2223
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 14 deletions.
diff --git a/rumqttc/src/lib.rs b/rumqttc/src/lib.rs
@@ -223,11 +223,12 @@ impl From<Unsubscribe> for Request {
     }
 }
 
-pub type AckPromise = oneshot::Receiver<()>;
+pub type Pkid = u16;
+pub type AckPromise = oneshot::Receiver<Pkid>;
 
 #[derive(Debug)]
 pub struct PromiseTx {
-    inner: oneshot::Sender<()>,
+    inner: oneshot::Sender<Pkid>,
 }
 
 impl PromiseTx {
@@ -237,8 +238,8 @@ impl PromiseTx {
         (PromiseTx { inner }, promise)
     }
 
-    fn resolve(self) {
-        if self.inner.send(()).is_err() {
+    fn resolve(self, pkid: Pkid) {
+        if self.inner.send(pkid).is_err() {
             trace!("Promise was drpped")
         }
     }

diff --git a/rumqttc/src/state.rs b/rumqttc/src/state.rs
@@ -206,7 +206,7 @@ impl MqttState {
             .any(|x| matches!(x, SubscribeReasonCode::Success(_)))
         {
             if let Some(tx) = self.ack_waiter[suback.pkid as usize].take() {
-                tx.resolve();
+                tx.resolve(suback.pkid);
             }
         }
 
@@ -223,7 +223,7 @@ impl MqttState {
         }
 
         if let Some(tx) = self.ack_waiter[unsuback.pkid as usize].take() {
-            tx.resolve();
+            tx.resolve(unsuback.pkid);
         }
 
         Ok(None)
@@ -271,7 +271,7 @@ impl MqttState {
 
         if let Some(tx) = self.ack_waiter[puback.pkid as usize].take() {
             // Resolve promise for QoS 1
-            tx.resolve();
+            tx.resolve(puback.pkid);
         }
 
         self.inflight -= 1;
@@ -333,7 +333,7 @@ impl MqttState {
 
         if let Some(tx) = self.ack_waiter[pubcomp.pkid as usize].take() {
             // Resolve promise for QoS 2
-            tx.resolve();
+            tx.resolve(pubcomp.pkid);
         }
 
         self.outgoing_rel.set(pubcomp.pkid as usize, false);
@@ -398,7 +398,7 @@ impl MqttState {
         let event = Event::Outgoing(Outgoing::Publish(publish.pkid));
         self.events.push_back(event);
         match (publish.qos, tx) {
-            (QoS::AtMostOnce, Some(tx)) => tx.resolve(),
+            (QoS::AtMostOnce, Some(tx)) => tx.resolve(publish.pkid),
             (_, tx) => self.ack_waiter[publish.pkid as usize] = tx,
         }
 

diff --git a/rumqttc/src/v5/state.rs b/rumqttc/src/v5/state.rs
@@ -269,7 +269,7 @@ impl MqttState {
             .any(|x| matches!(x, SubscribeReasonCode::Success(_)))
         {
             if let Some(tx) = self.ack_waiter[suback.pkid as usize].take() {
-                tx.resolve();
+                tx.resolve(suback.pkid);
             }
         }
 
@@ -293,7 +293,7 @@ impl MqttState {
 
         if unsuback.reasons.contains(&UnsubAckReason::Success) {
             if let Some(tx) = self.ack_waiter[unsuback.pkid as usize].take() {
-                tx.resolve();
+                tx.resolve(unsuback.pkid);
             }
         }
 
@@ -401,7 +401,7 @@ impl MqttState {
 
         if let Some(tx) = self.ack_waiter[puback.pkid as usize].take() {
             // Resolve promise for QoS 1
-            tx.resolve();
+            tx.resolve(puback.pkid);
         }
 
         self.inflight -= 1;
@@ -490,7 +490,7 @@ impl MqttState {
 
         if let Some(tx) = self.ack_waiter[pubcomp.pkid as usize].take() {
             // Resolve promise for QoS 2
-            tx.resolve();
+            tx.resolve(pubcomp.pkid);
         }
 
         self.outgoing_rel.set(pubcomp.pkid as usize, false);
@@ -577,7 +577,7 @@ impl MqttState {
         let event = Event::Outgoing(Outgoing::Publish(pkid));
         self.events.push_back(event);
         match (publish.qos, tx) {
-            (QoS::AtMostOnce, Some(tx)) => tx.resolve(),
+            (QoS::AtMostOnce, Some(tx)) => tx.resolve(0),
             (_, tx) => self.ack_waiter[publish.pkid as usize] = tx,
         }