handle ack packet

Author: xiazhvera

AWSIoTPythonSDK/core/protocol/mqtt_core.py

@@ -29,6 +29,7 @@
 from AWSIoTPythonSDK.core.protocol.internal.defaults import ALPN_PROTCOLS
 from AWSIoTPythonSDK.core.protocol.internal.events import FixedEventMids
 from AWSIoTPythonSDK.core.protocol.paho.client import MQTT_ERR_SUCCESS
+from AWSIoTPythonSDK.core.protocol.paho.client import MQTT_ERR_SUBACK_ERROR
 from AWSIoTPythonSDK.exception.AWSIoTExceptions import connectError
 from AWSIoTPythonSDK.exception.AWSIoTExceptions import connectTimeoutException
 from AWSIoTPythonSDK.exception.AWSIoTExceptions import disconnectError
@@ -58,6 +59,12 @@
     from queue import Queue
 
 
+class AckPacket(object):
+    def __init__(self):
+        self.event = Event()
+        self.data = None
+
+
 class MqttCore(object):
 
     _logger = logging.getLogger(__name__)
@@ -298,12 +305,15 @@ def subscribe(self, topic, qos, message_callback=None):
         if ClientStatus.STABLE != self._client_status.get_status():
             self._handle_offline_request(RequestTypes.SUBSCRIBE, (topic, qos, message_callback, None))
         else:
-            event = Event()
-            rc, mid = self._subscribe_async(topic, qos, self._create_blocking_ack_callback(event), message_callback)
-            if not event.wait(self._operation_timeout_sec):
+            ack = AckPacket()
+            rc, mid = self._subscribe_async(topic, qos, self._create_blocking_ack_callback_ret(ack), message_callback)
+            if not ack.event.wait(self._operation_timeout_sec):
                 self._internal_async_client.remove_event_callback(mid)
                 self._logger.error("Subscribe timed out")
                 raise subscribeTimeoutException()
+            if ack.data[0] == MQTT_ERR_SUBACK_ERROR:
+                self._logger.error(f"Subscribe error: {ack.data}")
+                raise subscribeError(ack.data)
             ret = True
         return ret
 
@@ -361,6 +371,12 @@ def ack_callback(mid, data=None):
             event.set()
         return ack_callback
 
+    def _create_blocking_ack_callback_ret(self, ack: AckPacket):
+        def ack_callback(mid, data=None):
+            ack.data = data
+            ack.event.set()
+        return ack_callback
+
     def _handle_offline_request(self, type, data):
         self._logger.info("Offline request detected!")
         offline_request = QueueableRequest(type, data)

AWSIoTPythonSDK/core/protocol/paho/client.py

@@ -137,6 +137,9 @@
 MSG_QUEUEING_DROP_OLDEST = 0
 MSG_QUEUEING_DROP_NEWEST = 1
 
+# Packet Error Codes
+MQTT_ERR_SUBACK_ERROR = 0x80
+
 if sys.version_info[0] < 3:
     sockpair_data = "0"
 else: