test thread sync

grimmpp · May 14, 2024 · 8c671e0 · 8c671e0
1 parent 6cb66fd
commit 8c671e0
Showing 1 changed file with 12 additions and 8 deletions.
diff --git a/custom_components/eltako/gateway.py b/custom_components/eltako/gateway.py
@@ -93,9 +93,9 @@ def set_connection_state_changed_handler(self, handler):
 
     def _fire_connection_state_changed_event(self, status):
         if self._connection_state_handler:
-            asyncio.ensure_future(
+            asyncio.run_coroutine_threadsafe(
                 self._connection_state_handler(status),
-                loop= self._loop
+                loop = self._loop
             )
 
 
@@ -105,9 +105,9 @@ def set_last_message_received_handler(self, handler):
 
     def _fire_last_message_received_event(self):
         if self._last_message_received_handler:
-            asyncio.ensure_future(
-                self._last_message_received_handler( datetime.utcnow().replace(tzinfo=pytz.utc) ),
-                loop= self._loop
+            asyncio.run_coroutine_threadsafe(
+                self._last_message_received_handler( datetime.now(datetime.UTC).replace(tzinfo=pytz.UTC) ),
+                loop = self._loop
             )
 
 
@@ -118,9 +118,9 @@ def set_received_message_count_handler(self, handler):
     def _fire_received_message_count_event(self):
         self._received_message_count += 1
         if self._received_message_count_handler:
-            asyncio.ensure_future(
+            asyncio.run_coroutine_threadsafe(
                 self._received_message_count_handler( self._received_message_count ),
-                loop= self._loop
+                loop = self._loop
             )
 
     def process_messages(self, data=None):
@@ -293,7 +293,11 @@ def _callback_send_message_to_serial_bus(self, msg):
                 LOGGER.debug("[Gateway] [Id: %d] Send message: %s - Serialized: %s", self.dev_id, msg, msg.serialize().hex())
 
                 # put message on serial bus
-                asyncio.ensure_future(self._bus.send(msg), loop=self._loop)
+                # asyncio.ensure_future(self._bus.send(msg), loop=self._loop)
+                asyncio.run_coroutine_threadsafe(
+                    self._bus.send(msg), 
+                    loop = self._loop
+                )
         else:
             LOGGER.warn("[Gateway] [Id: %d] Serial port %s is not available!!! message (%s) was not sent.", self.dev_id, self.serial_path, msg)