avoid endless loop when waiting for data from broker

fixes #115

Author: vladak

adafruit_minimqtt/adafruit_minimqtt.py

@@ -6,6 +6,8 @@
 # Modified Work Copyright (c) 2019 Bradley Beach, esp32spi_mqtt
 # Modified Work Copyright (c) 2012-2019 Roger Light and others, Paho MQTT Python
 
+# pylint: disable=too-many-lines
+
 """
 `adafruit_minimqtt`
 ================================================================================
@@ -126,6 +128,7 @@ class MQTT:
     :param str client_id: Optional client identifier, defaults to a unique, generated string.
     :param bool is_ssl: Sets a secure or insecure connection with the broker.
     :param int keep_alive: KeepAlive interval between the broker and the MiniMQTT client.
+    :param int recv_timeout: receive timeout, in seconds.
     :param socket socket_pool: A pool of socket resources available for the given radio.
     :param ssl_context: SSL context for long-lived SSL connections.
     :param bool use_binary_mode: Messages are passed as bytearray instead of string to callbacks.
@@ -142,6 +145,7 @@ def __init__(
         client_id=None,
         is_ssl=True,
         keep_alive=60,
+        recv_timeout=10,
         socket_pool=None,
         ssl_context=None,
         use_binary_mode=False,
@@ -154,6 +158,7 @@ def __init__(
         self._use_binary_mode = use_binary_mode
 
         self.keep_alive = keep_alive
+        self._recv_timeout = recv_timeout
         self._user_data = None
         self._is_connected = False
         self._msg_size_lim = MQTT_MSG_SZ_LIM
@@ -514,6 +519,7 @@ def connect(self, clean_session=True, host=None, port=None, keep_alive=None):
             self._send_str(self._password)
         if self.logger is not None:
             self.logger.debug("Receiving CONNACK packet from broker")
+        stamp = time.monotonic()
         while True:
             op = self._wait_for_msg()
             if op == 32:
@@ -527,6 +533,12 @@ def connect(self, clean_session=True, host=None, port=None, keep_alive=None):
                     self.on_connect(self, self._user_data, result, rc[2])
                 return result
 
+            if op is None:
+                if time.monotonic() - stamp > self._recv_timeout:
+                    raise MMQTTException(
+                        f"No data received from broker for {self._recv_timeout} seconds."
+                    )
+
     def disconnect(self):
         """Disconnects the MiniMQTT client from the MQTT broker."""
         self.is_connected()
@@ -637,6 +649,7 @@ def publish(self, topic, msg, retain=False, qos=0):
         if qos == 0 and self.on_publish is not None:
             self.on_publish(self, self._user_data, topic, self._pid)
         if qos == 1:
+            stamp = time.monotonic()
             while True:
                 op = self._wait_for_msg()
                 if op == 0x40:
@@ -649,6 +662,12 @@ def publish(self, topic, msg, retain=False, qos=0):
                             self.on_publish(self, self._user_data, topic, rcv_pid)
                         return
 
+                if op is None:
+                    if time.monotonic() - stamp > self._recv_timeout:
+                        raise MMQTTException(
+                            f"No data received from broker for {self._recv_timeout} seconds."
+                        )
+
     def subscribe(self, topic, qos=0):
         """Subscribes to a topic on the MQTT Broker.
         This method can subscribe to one topics or multiple topics.
@@ -697,6 +716,7 @@ def subscribe(self, topic, qos=0):
             for t, q in topics:
                 self.logger.debug("SUBSCRIBING to topic %s with QoS %d", t, q)
         self._sock.send(packet)
+        stamp = time.monotonic()
         while True:
             op = self._wait_for_msg()
             if op == 0x90:
@@ -710,6 +730,12 @@ def subscribe(self, topic, qos=0):
                     self._subscribed_topics.append(t)
                 return
 
+            if op is None:
+                if time.monotonic() - stamp > self._recv_timeout:
+                    raise MMQTTException(
+                        f"No data received from broker for {self._recv_timeout} seconds."
+                    )
+
     def unsubscribe(self, topic):
         """Unsubscribes from a MQTT topic.
 
@@ -747,6 +773,7 @@ def unsubscribe(self, topic):
         if self.logger is not None:
             self.logger.debug("Waiting for UNSUBACK...")
         while True:
+            stamp = time.monotonic()
             op = self._wait_for_msg()
             if op == 176:
                 rc = self._sock_exact_recv(3)
@@ -759,6 +786,12 @@ def unsubscribe(self, topic):
                     self._subscribed_topics.remove(t)
                 return
 
+            if op is None:
+                if time.monotonic() - stamp > self._recv_timeout:
+                    raise MMQTTException(
+                        f"No data received from broker for {self._recv_timeout} seconds."
+                    )
+
     def reconnect(self, resub_topics=True):
         """Attempts to reconnect to the MQTT broker.