Non-Blocking Async Operations for AWS IOT library

Author: eman

CHANGELOG.rst

@@ -8,6 +8,16 @@ Version 0.2 (Unreleased)
 Added
 -----
 
+- **Non-Blocking MQTT Operations**: All AWS IoT SDK operations are now non-blocking
+  
+  - All MQTT operations (connect, disconnect, subscribe, unsubscribe, publish) use ``asyncio.run_in_executor()`` 
+  - Eliminates "blocking I/O detected" warnings in Home Assistant and other async applications
+  - Fully compatible with async event loops without blocking other operations
+  - No API changes required - existing code works without modification
+  - Maintains full performance and reliability of the underlying AWS IoT SDK
+  - Safe for use in Home Assistant custom integrations and other async applications
+  - Updated documentation with non-blocking implementation details
+
 - **Event Emitter Pattern (Phase 1)**: Event-driven architecture for device state changes
 
   - ``EventEmitter`` base class with multiple listeners per event

README.rst

@@ -15,6 +15,7 @@ Features
 * **Operation Mode Control**: Switch between Heat Pump, Energy Saver, High Demand, Electric, and Vacation modes
 * **Comprehensive Status Data**: Access to 70+ device status fields including compressor status, heater status, flow rates, and more
 * **MQTT Protocol Support**: Low-level MQTT communication with Navien devices
+* **Non-Blocking Async Operations**: Fully compatible with async event loops (Home Assistant safe)
 * **Automatic Reconnection**: Reconnects automatically with exponential backoff during network interruptions
 * **Command Queuing**: Commands sent while disconnected are queued and sent automatically when reconnected
 * **Data Models**: Type-safe data classes with automatic unit conversions

docs/MQTT_CLIENT.rst

@@ -12,6 +12,11 @@ enables:
 - Device control (temperature, mode, power)
 - Bidirectional communication over MQTT
 - Automatic reconnection and error handling
+- **Non-blocking async operations** (compatible with Home Assistant and other async applications)
+
+The client is designed to be fully non-blocking and integrates seamlessly
+with async event loops, avoiding the "blocking I/O detected" warnings
+commonly seen in Home Assistant and similar applications.
 
 Prerequisites
 -------------
@@ -778,6 +783,45 @@ Error Handling
 Advanced Usage
 --------------
 
+Non-Blocking Implementation
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The MQTT client is designed to be fully compatible with async event loops
+and will not block or interfere with other async operations. This makes it
+suitable for integration with Home Assistant, web servers, and other 
+async applications.
+
+**Implementation Details:**
+
+- All AWS IoT SDK operations that could block are wrapped with ``asyncio.run_in_executor()``
+- Connection, disconnection, subscription, and publishing operations are non-blocking
+- The client maintains full compatibility with the existing API
+- No additional configuration required for non-blocking behavior
+
+**Home Assistant Integration:**
+
+.. code:: python
+
+   # Safe for use in Home Assistant custom integrations
+   class MyCoordinator(DataUpdateCoordinator):
+       async def _async_update_data(self):
+           # This will not trigger "blocking I/O detected" warnings
+           await self.mqtt_client.request_device_status(self.device)
+           return self.latest_data
+
+**Concurrent Operations:**
+
+.. code:: python
+
+   # MQTT operations will not block other async tasks
+   async def main():
+       # Both tasks run concurrently without blocking
+       await asyncio.gather(
+           mqtt_client.connect(),
+           some_other_async_operation(),
+           web_server.start(),
+       )
+
 Custom Connection Configuration
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

src/nwp500/mqtt_client.py

@@ -453,21 +453,31 @@ async def connect(self) -> bool:
 
         try:
             # Build WebSocket MQTT connection with AWS credentials
-            self._connection = mqtt_connection_builder.websockets_with_default_aws_signing(
-                endpoint=self.config.endpoint,
-                region=self.config.region,
-                credentials_provider=self._create_credentials_provider(),
-                client_id=self.config.client_id,
-                clean_session=self.config.clean_session,
-                keep_alive_secs=self.config.keep_alive_secs,
-                on_connection_interrupted=self._on_connection_interrupted_internal,
-                on_connection_resumed=self._on_connection_resumed_internal,
-            )
+            # Run the connection building in a thread pool to avoid blocking I/O
+            def _build_connection():
+                return mqtt_connection_builder.websockets_with_default_aws_signing(
+                    endpoint=self.config.endpoint,
+                    region=self.config.region,
+                    credentials_provider=self._create_credentials_provider(),
+                    client_id=self.config.client_id,
+                    clean_session=self.config.clean_session,
+                    keep_alive_secs=self.config.keep_alive_secs,
+                    on_connection_interrupted=self._on_connection_interrupted_internal,
+                    on_connection_resumed=self._on_connection_resumed_internal,
+                )
+            
+            # Run connection builder in thread pool to avoid blocking I/O
+            self._connection = await self._loop.run_in_executor(None, _build_connection)
 
             # Connect
             _logger.info("Establishing MQTT connection...")
-            connect_future = self._connection.connect()
-            connect_result = connect_future.result()
+            
+            # Run the connect operation in a thread pool to avoid blocking I/O
+            def _connect():
+                connect_future = self._connection.connect()
+                return connect_future.result()
+            
+            connect_result = await self._loop.run_in_executor(None, _connect)
 
             self._connected = True
             self._reconnect_attempts = 0  # Reset on successful connection
@@ -518,8 +528,13 @@ async def disconnect(self):
         await self.stop_all_periodic_tasks()
 
         try:
-            disconnect_future = self._connection.disconnect()
-            disconnect_future.result()
+            # Run disconnect operation in thread pool to avoid blocking I/O
+            def _disconnect():
+                disconnect_future = self._connection.disconnect()
+                return disconnect_future.result()
+            
+            await self._loop.run_in_executor(None, _disconnect)
+            
             self._connected = False
             self._connection = None
             _logger.info("Disconnected successfully")
@@ -613,13 +628,16 @@ async def subscribe(
         _logger.info(f"Subscribing to topic: {topic}")
 
         try:
-            # Subscribe via MQTT connection
-            subscribe_future, packet_id = self._connection.subscribe(
-                topic=topic, qos=qos, callback=self._on_message_received
-            )
-
-            # Wait for subscription acknowledgment
-            subscribe_result = subscribe_future.result()
+            # Run subscribe operation in thread pool to avoid blocking I/O
+            def _subscribe():
+                subscribe_future, packet_id = self._connection.subscribe(
+                    topic=topic, qos=qos, callback=self._on_message_received
+                )
+                subscribe_result = subscribe_future.result()
+                return subscribe_result, packet_id
+            
+            subscribe_result, packet_id = await self._loop.run_in_executor(None, _subscribe)
+            
             _logger.info(f"Subscribed to '{topic}' with QoS {subscribe_result['qos']}")
 
             # Store subscription and handler
@@ -647,8 +665,12 @@ async def unsubscribe(self, topic: str):
         _logger.info(f"Unsubscribing from topic: {topic}")
 
         try:
-            unsubscribe_future, packet_id = self._connection.unsubscribe(topic)
-            unsubscribe_future.result()
+            # Run unsubscribe operation in thread pool to avoid blocking I/O
+            def _unsubscribe():
+                unsubscribe_future, packet_id = self._connection.unsubscribe(topic)
+                return unsubscribe_future.result()
+            
+            await self._loop.run_in_executor(None, _unsubscribe)
 
             # Remove from tracking
             self._subscriptions.pop(topic, None)
@@ -698,13 +720,16 @@ async def publish(
             # Serialize to JSON
             payload_json = json.dumps(payload)
 
-            # Publish
-            publish_future, packet_id = self._connection.publish(
-                topic=topic, payload=payload_json, qos=qos
-            )
-
-            # Wait for publish acknowledgment
-            publish_future.result()
+            # Run publish operation in thread pool to avoid blocking I/O
+            def _publish():
+                publish_future, packet_id = self._connection.publish(
+                    topic=topic, payload=payload_json, qos=qos
+                )
+                publish_future.result()
+                return packet_id
+            
+            packet_id = await self._loop.run_in_executor(None, _publish)
+            
             _logger.debug(f"Published to '{topic}' with packet_id {packet_id}")
 
             return packet_id