Unallocated pkid passthrough

rumqttc/src/client.rs

@@ -41,7 +41,7 @@ impl From<TrySendError<Request>> for ClientError {
 /// from the broker, i.e. move ahead.
 #[derive(Clone, Debug)]
 pub struct AsyncClient {
-    request_tx: Sender<Request>,
+    pub request_tx: Sender<Request>,
 }
 
 impl AsyncClient {

rumqttc/src/eventloop.rs

@@ -161,6 +161,7 @@ impl EventLoop {
         match self.select().await {
             Ok(v) => Ok(v),
             Err(e) => {
+                log::error!("connection error : {e}");
                 self.clean();
                 Err(e)
             }

rumqttc/src/state.rs

@@ -221,6 +221,10 @@ impl MqttState {
     }
 
     fn handle_incoming_puback(&mut self, puback: &PubAck) -> Result<Option<Packet>, StateError> {
+        if puback.pkid > self.max_inflight {
+            return Ok(None);
+        }
+
         let publish = self
             .outgoing_pub
             .get_mut(puback.pkid as usize)
@@ -319,7 +323,7 @@ impl MqttState {
     /// Adds next packet identifier to QoS 1 and 2 publish packets and returns
     /// it buy wrapping publish in packet
     fn outgoing_publish(&mut self, mut publish: Publish) -> Result<Option<Packet>, StateError> {
-        if publish.qos != QoS::AtMostOnce {
+        if publish.qos != QoS::AtMostOnce && publish.pkid <= self.max_inflight {
             if publish.pkid == 0 {
                 publish.pkid = self.next_pkid();
             }

rumqttc/src/v5/state.rs

@@ -65,7 +65,7 @@ pub enum StateError {
     #[error("Connection failed with reason '{reason:?}' ")]
     ConnFail { reason: ConnectReturnCode },
     #[error("Connection closed by peer abruptly")]
-    ConnectionAborted
+    ConnectionAborted,
 }
 
 impl From<mqttbytes::Error> for StateError {