use recv_timeout instead of keep_alive

fixes #189

Author: vladak

adafruit_minimqtt/adafruit_minimqtt.py

@@ -627,7 +627,6 @@ 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:
@@ -642,12 +641,6 @@ 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()
@@ -759,7 +752,6 @@ 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:
@@ -772,12 +764,6 @@ 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.
@@ -825,34 +811,26 @@ 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 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
+            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)
+            else:
                 raise MMQTTException(
                     f"invalid message received as response to SUBSCRIBE: {hex(op)}"
                 )
@@ -892,14 +870,8 @@ 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 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:
                 if op == 176:
                     rc = self._sock_exact_recv(3)
                     assert rc[0] == 0x02
@@ -910,10 +882,10 @@ def unsubscribe(self, topic: str) -> None:
                             self.on_unsubscribe(self, self._user_data, t, self._pid)
                         self._subscribed_topics.remove(t)
                     return
-
-                raise MMQTTException(
-                    f"invalid message received as response to UNSUBSCRIBE: {hex(op)}"
-                )
+                else:
+                    raise MMQTTException(
+                        f"invalid message received as response to UNSUBSCRIBE: {hex(op)}"
+                    )
 
     def _recompute_reconnect_backoff(self) -> None:
         """
@@ -1115,7 +1087,7 @@ def _sock_exact_recv(self, bufsize: int) -> bytearray:
             to_read = bufsize - recv_len
             if to_read < 0:
                 raise MMQTTException(f"negative number of bytes to read: {to_read}")
-            read_timeout = self.keep_alive
+            read_timeout = self._recv_timeout
             mv = mv[recv_len:]
             while to_read > 0:
                 recv_len = self._sock.recv_into(mv, to_read)
@@ -1136,7 +1108,7 @@ def _sock_exact_recv(self, bufsize: int) -> bytearray:
             # or raise exception if wait longer than read_timeout
             to_read = bufsize - len(rc)
             assert to_read >= 0
-            read_timeout = self.keep_alive
+            read_timeout = self._recv_timeout
             while to_read > 0:
                 recv = self._sock.recv(to_read)
                 to_read -= len(recv)