revert the higher level timeout changes

Author: vladak

adafruit_minimqtt/adafruit_minimqtt.py

@@ -627,6 +627,7 @@ def _connect(
             self._send_str(self._username)
             self._send_str(self._password)
         self.logger.debug("Receiving CONNACK packet from broker")
+        stamp = time.monotonic()
         while True:
             op = self._wait_for_msg()
             if op == 32:
@@ -641,6 +642,12 @@ def _connect(
 
                 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) -> None:
         """Disconnects the MiniMQTT client from the MQTT broker."""
         self._connected()
@@ -752,6 +759,7 @@ def publish(
         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:
@@ -764,6 +772,12 @@ def publish(
                             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: str, qos: int = 0) -> None:
         """Subscribes to a topic on the MQTT Broker.
         This method can subscribe to one topics or multiple topics.
@@ -811,26 +825,34 @@ def subscribe(self, topic: str, qos: int = 0) -> None:
         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:
-                rc = self._sock_exact_recv(3)
-                # Check packet identifier.
-                assert rc[1] == packet[2] and rc[2] == packet[3]
-                remaining_len = rc[0] - 2
-                assert remaining_len > 0
-                rc = self._sock_exact_recv(remaining_len)
-                for i in range(0, remaining_len):
-                    if rc[i] not in [0, 1, 2]:
-                        raise MMQTTException(
-                            f"SUBACK Failure for topic {topics[i][0]}: {hex(rc[i])}"
-                        )
-
-                for t, q in topics:
-                    if self.on_subscribe is not None:
-                        self.on_subscribe(self, self._user_data, t, q)
-                    self._subscribed_topics.append(t)
+            if op is None:
+                if time.monotonic() - stamp > self._recv_timeout:
+                    raise MMQTTException(
+                        f"No data received from broker for {self._recv_timeout} seconds."
+                    )
             else:
+                if op == 0x90:
+                    rc = self._sock_exact_recv(3)
+                    # Check packet identifier.
+                    assert rc[1] == packet[2] and rc[2] == packet[3]
+                    remaining_len = rc[0] - 2
+                    assert remaining_len > 0
+                    rc = self._sock_exact_recv(remaining_len)
+                    for i in range(0, remaining_len):
+                        if rc[i] not in [0, 1, 2]:
+                            raise MMQTTException(
+                                f"SUBACK Failure for topic {topics[i][0]}: {hex(rc[i])}"
+                            )
+
+                    for t, q in topics:
+                        if self.on_subscribe is not None:
+                            self.on_subscribe(self, self._user_data, t, q)
+                        self._subscribed_topics.append(t)
+                    return
+
                 raise MMQTTException(
                     f"invalid message received as response to SUBSCRIBE: {hex(op)}"
                 )
@@ -870,20 +892,27 @@ def unsubscribe(self, topic: str) -> None:
         self._sock.send(packet)
         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)
-                assert rc[0] == 0x02
-                # [MQTT-3.32]
-                assert rc[1] == packet_id_bytes[0] and rc[2] == packet_id_bytes[1]
-                for t in topics:
-                    if self.on_unsubscribe is not None:
-                        self.on_unsubscribe(self, self._user_data, t, self._pid)
-                    self._subscribed_topics.remove(t)
+            if op is None:
+                if time.monotonic() - stamp > self._recv_timeout:
+                    raise MMQTTException(
+                        f"No data received from broker for {self._recv_timeout} seconds."
+                    )
             else:
-                raise MMQTTException(
-                    f"invalid message received as response to UNSUBSCRIBE: {hex(op)}"
-                )
+                if op == 176:
+                    rc = self._sock_exact_recv(3)
+                    assert rc[0] == 0x02
+                    # [MQTT-3.32]
+                    assert rc[1] == packet_id_bytes[0] and rc[2] == packet_id_bytes[1]
+                    for t in topics:
+                        if self.on_unsubscribe is not None:
+                            self.on_unsubscribe(self, self._user_data, t, self._pid)
+                        self._subscribed_topics.remove(t)
+                else:
+                    raise MMQTTException(
+                        f"invalid message received as response to UNSUBSCRIBE: {hex(op)}"
+                    )
 
     def _recompute_reconnect_backoff(self) -> None:
         """