Move request for unknown check

Author: allenporter

roborock/cloud_api.py

@@ -76,29 +76,24 @@ def __init__(self, user_data: UserData, device_info: DeviceData) -> None:
 
     def _mqtt_on_connect(self, *args, **kwargs):
         _, __, ___, rc, ____ = args
-        connection_queue = self._waiting_queue.safe_pop(RequestKey(CONNECT_REQUEST_ID))
+        if not (connection_queue := self._waiting_queue.safe_pop(RequestKey(CONNECT_REQUEST_ID), "connect")):
+            self._logger.info("Received unexpected connect event")
+            return
         if rc != mqtt.MQTT_ERR_SUCCESS:
             message = f"Failed to connect ({mqtt.error_string(rc)})"
             self._logger.error(message)
-            if connection_queue:
-                connection_queue.set_exception(VacuumError(message))
-            else:
-                self._logger.debug("Failed to notify connect future, not in queue")
+            connection_queue.set_exception(VacuumError(message))
             return
         self._logger.info(f"Connected to mqtt {self._mqtt_host}:{self._mqtt_port}")
         topic = f"rr/m/o/{self._mqtt_user}/{self._hashed_user}/{self.device_info.device.duid}"
         (result, mid) = self._mqtt_client.subscribe(topic)
         if result != 0:
             message = f"Failed to subscribe ({mqtt.error_string(rc)})"
             self._logger.error(message)
-            if connection_queue:
-                connection_queue.set_exception(VacuumError(message))
+            connection_queue.set_exception(VacuumError(message))
             return
         self._logger.info(f"Subscribed to topic {topic}")
-        if connection_queue:
-            connection_queue.set_result(True)
-        else:
-            self._logger.debug("Connected but no connect future")
+        connection_queue.set_result(True)
 
     def _mqtt_on_message(self, *args, **kwargs):
         client, __, msg = args
@@ -113,11 +108,8 @@ def _mqtt_on_disconnect(self, *args, **kwargs):
         try:
             exc = RoborockException(mqtt.error_string(rc)) if rc != mqtt.MQTT_ERR_SUCCESS else None
             super().on_connection_lost(exc)
-            connection_queue = self._waiting_queue.safe_pop(RequestKey(DISCONNECT_REQUEST_ID))
-            if connection_queue:
+            if connection_queue := self._waiting_queue.safe_pop(RequestKey(DISCONNECT_REQUEST_ID), "disconnect"):
                 connection_queue.set_result(True)
-            else:
-                self._logger.debug("Disconnected but no disconnect future")
         except Exception as ex:
             self._logger.exception(ex)
 

roborock/roborock_future.py

@@ -12,7 +12,6 @@
 from .roborock_message import RoborockMessageProtocol
 
 _LOGGER = logging.getLogger(__name__)
-_TRIES = 3
 
 
 @dataclass(frozen=True)
@@ -43,11 +42,17 @@ def put(self, request_key: RequestKey, future: RoborockFuture) -> None:
                 raise ValueError(f"Request key {request_key} already exists in the queue")
             self._queue[request_key] = future
 
-    def safe_pop(self, request_key: RequestKey) -> RoborockFuture | None:
-        """Get the future from the queue if it has not yet been popped, otherwise ignore."""
-        _LOGGER.debug("Popping request key %s from the queue", request_key)
+    def safe_pop(self, request_key: RequestKey, label: str | None = None) -> RoborockFuture | None:
+        """Get the future from the queue if it has not yet been popped, otherwise ignore.
+
+        The label is used for logging when the request key is not found in the queue.
+        """
+        _LOGGER.debug("Popping request key %s (%s) from the queue", request_key, label)
         with self._lock:
-            return self._queue.pop(request_key, None)
+            future = self._queue.pop(request_key, None)
+            if future is None and label is not None:
+                _LOGGER.warning("Received message for key %s (%s) not found in the queue", request_key, label)
+            return future
 
 
 class RoborockFuture:

roborock/version_1_apis/roborock_client_v1.py

@@ -393,8 +393,7 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None:
                             data_point_response = json.loads(data_point)
                             request_id = data_point_response.get("id")
                             request_key = RequestKey(request_id, protocol)
-                            queue = self._waiting_queue.safe_pop(request_key)
-                            if queue:
+                            if queue := self._waiting_queue.safe_pop(request_key, "v1_rpc"):
                                 error = data_point_response.get("error")
                                 if error:
                                     queue.set_exception(
@@ -408,8 +407,6 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None:
                                     if isinstance(result, list) and len(result) == 1:
                                         result = result[0]
                                     queue.set_result(result)
-                            else:
-                                self._logger.debug("Received response for unknown request id %s", request_key)
                         else:
                             try:
                                 data_protocol = RoborockDataProtocol(int(data_point_number))
@@ -470,20 +467,14 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None:
                             raise RoborockException(f"Failed to decode {data.payload!r} for {data.protocol}") from err
                         decompressed = Utils.decompress(decrypted)
                         request_key = RequestKey(request_id, protocol)
-                        queue = self._waiting_queue.safe_pop(request_key)
-                        if queue:
+                        if queue := self._waiting_queue.safe_pop(request_key, "v1_map"):
                             if isinstance(decompressed, list):
                                 decompressed = decompressed[0]
                             queue.set_result(decompressed)
-                        else:
-                            self._logger.debug("Received response for unknown request id %s", request_key)
                 else:
                     request_key = RequestKey(data.seq, protocol)
-                    queue = self._waiting_queue.safe_pop(request_key)
-                    if queue:
+                    if queue := self._waiting_queue.safe_pop(request_key, "v1_other"):
                         queue.set_result(data.payload)
-                    else:
-                        self._logger.debug("Received response for unknown request id %s", request_key)
         except Exception as ex:
             self._logger.exception(ex)
 

roborock/version_a01_apis/roborock_client_a01.py

@@ -144,11 +144,8 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None:
                         # Auto convert into data struct we want.
                         converted_response = entries[data_point_protocol].post_process_fn(data_point)
                         request_key = RequestKey(int(data_point_number), protocol)
-                        future = self._waiting_queue.safe_pop(request_key)
-                        if future is not None:
+                        if future := self._waiting_queue.safe_pop(request_key, "a01"):
                             future.set_result(converted_response)
-                        else:
-                            self._logger.debug(f"Got response for {request_key} but no future found")
 
     @abstractmethod
     async def update_values(