.readthedocs.yaml

@@ -8,6 +8,9 @@
 # Required
 version: 2
 
+sphinx:
+  configuration: docs/conf.py
+
 build:
   os: ubuntu-20.04
   tools:

adafruit_minimqtt/adafruit_minimqtt.py

@@ -66,7 +66,9 @@
 MQTT_PINGRESP = const(0xD0)
 MQTT_PUBLISH = const(0x30)
 MQTT_SUB = const(0x82)
+MQTT_SUBACK = const(0x90)
 MQTT_UNSUB = const(0xA2)
+MQTT_UNSUBACK = const(0xB0)
 MQTT_DISCONNECT = b"\xe0\0"
 
 MQTT_PKT_TYPE_MASK = const(0xF0)
@@ -393,12 +395,13 @@ def username_pw_set(self, username: str, password: Optional[str] = None) -> None
         if password is not None:
             self._password = password
 
-    def connect(
+    def connect(  # noqa: PLR0913, too many arguments in function definition
         self,
         clean_session: bool = True,
         host: Optional[str] = None,
         port: Optional[int] = None,
         keep_alive: Optional[int] = None,
+        session_id: Optional[str] = None,
     ) -> int:
         """Initiates connection with the MQTT Broker. Will perform exponential back-off
         on connect failures.
@@ -408,7 +411,8 @@ def connect(
         :param int port: Network port of the remote broker.
         :param int keep_alive: Maximum period allowed for communication
             within single connection attempt, in seconds.
-
+        :param str session_id: unique session ID,
+            used for multiple simultaneous connections to the same host
         """
 
         last_exception = None
@@ -430,6 +434,7 @@ def connect(
                     host=host,
                     port=port,
                     keep_alive=keep_alive,
+                    session_id=session_id,
                 )
                 self._reset_reconnect_backoff()
                 return ret
@@ -470,25 +475,32 @@ def _send_bytes(
         view = memoryview(buffer)
         while bytes_sent < bytes_to_send:
             try:
-                bytes_sent += self._sock.send(view[bytes_sent:])
+                sent_now = self._sock.send(view[bytes_sent:])
+                # Some versions of `Socket.send()` do not return the number of bytes sent.
+                if not isinstance(sent_now, int):
+                    return
+                bytes_sent += sent_now
             except OSError as exc:
-                if exc.errno == EAGAIN:
+                if exc.errno == errno.EAGAIN:
                     continue
                 raise
 
-    def _connect(  # noqa: PLR0912, PLR0915, Too many branches, Too many statements
+    def _connect(  # noqa: PLR0912, PLR0913, PLR0915, Too many branches, Too many arguments, Too many statements
         self,
         clean_session: bool = True,
         host: Optional[str] = None,
         port: Optional[int] = None,
         keep_alive: Optional[int] = None,
+        session_id: Optional[str] = None,
     ) -> int:
         """Initiates connection with the MQTT Broker.
 
         :param bool clean_session: Establishes a persistent session.
         :param str host: Hostname or IP address of the remote broker.
         :param int port: Network port of the remote broker.
         :param int keep_alive: Maximum period allowed for communication, in seconds.
+        :param str session_id: unique session ID,
+            used for multiple simultaneous connections to the same host
 
         """
         if host:
@@ -511,6 +523,7 @@ def _connect(  # noqa: PLR0912, PLR0915, Too many branches, Too many statements
             self.broker,
             self.port,
             proto="mqtt:",
+            session_id=session_id,
             timeout=self._socket_timeout,
             is_ssl=self._is_ssl,
             ssl_context=self._ssl_context,
@@ -790,7 +803,7 @@ def subscribe(  # noqa: PLR0912, PLR0915, Too many branches, Too many statements
                         f"No data received from broker for {self._recv_timeout} seconds."
                     )
             else:
-                if op == 0x90:
+                if op == MQTT_SUBACK:
                     remaining_len = self._decode_remaining_length()
                     assert remaining_len > 0
                     rc = self._sock_exact_recv(2)
@@ -868,7 +881,7 @@ def unsubscribe(  # noqa: PLR0912, Too many branches
                         f"No data received from broker for {self._recv_timeout} seconds."
                     )
             else:
-                if op == 176:
+                if op == MQTT_UNSUBACK:
                     rc = self._sock_exact_recv(3)
                     assert rc[0] == 0x02
                     # [MQTT-3.32]
@@ -878,10 +891,12 @@ def unsubscribe(  # noqa: PLR0912, Too many branches
                             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)}"
-                )
+                if op != MQTT_PUBLISH:
+                    # [3.10.4] The Server may continue to deliver existing messages buffered
+                    # for delivery to the client prior to sending the UNSUBACK Packet.
+                    raise MMQTTException(
+                        f"invalid message received as response to UNSUBSCRIBE: {hex(op)}"
+                    )
 
     def _recompute_reconnect_backoff(self) -> None:
         """
@@ -1000,7 +1015,7 @@ def _wait_for_msg(  # noqa: PLR0912, Too many branches
                     return None
                 raise MMQTTException from error
 
-        if res in [None, b"", b"\x00"]:
+        if res in [None, b""]:
             # If we get here, it means that there is nothing to be received
             return None
         pkt_type = res[0] & MQTT_PKT_TYPE_MASK

tests/test_recv_timeout.py

@@ -46,7 +46,7 @@ def test_recv_timeout_vs_keepalive(self) -> None:
                     mqtt_client.ping()
 
                 now = time.monotonic()
-                assert recv_timeout <= (now - start) <= (keep_alive + 0.2)
+                assert recv_timeout <= (now - start) <= (keep_alive + 0.5)
 
 
 if __name__ == "__main__":

tests/test_unsubscribe.py

@@ -68,6 +68,49 @@ def handle_unsubscribe(client, user_data, topic, pid):
             + [0x6F] * 257
         ),
     ),
+    # UNSUBSCRIBE responded to by PUBLISH followed by UNSUBACK
+    (
+        "foo/bar",
+        bytearray(
+            [
+                0x30,  # PUBLISH
+                0x0C,
+                0x00,
+                0x07,
+                0x66,
+                0x6F,
+                0x6F,
+                0x2F,
+                0x62,
+                0x61,
+                0x72,
+                0x66,
+                0x6F,
+                0x6F,
+                0xB0,  # UNSUBACK
+                0x02,
+                0x00,
+                0x01,
+            ]
+        ),
+        bytearray(
+            [
+                0xA2,  # fixed header
+                0x0B,  # remaining length
+                0x00,
+                0x01,  # message ID
+                0x00,
+                0x07,  # topic length
+                0x66,  # topic
+                0x6F,
+                0x6F,
+                0x2F,
+                0x62,
+                0x61,
+                0x72,
+            ]
+        ),
+    ),
     # use list of topics for more coverage. If the range was (1, 10000), that would be
     # long enough to use 3 bytes for remaining length, however that would make the test
     # run for many minutes even on modern systems, so 1000 is used instead.
@@ -95,7 +138,7 @@ def handle_unsubscribe(client, user_data, topic, pid):
 @pytest.mark.parametrize(
     "topic,to_send,exp_recv",
     testdata,
-    ids=["short_topic", "long_topic", "topic_list_long"],
+    ids=["short_topic", "long_topic", "publish_first", "topic_list_long"],
 )
 def test_unsubscribe(topic, to_send, exp_recv) -> None:
     """