Rewrote program loop handling with asyncio.run function

Author: agramgurka

lumimqtt/__main__.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import os
+import signal
 from uuid import getnode as get_mac
 
 from .__version__ import version
@@ -29,9 +30,12 @@ def read_mac():
     return mac
 
 
-def main():
+def signal_handler():
+    raise KeyboardInterrupt()
+
+
+async def amain():
     logging.basicConfig(level='INFO')
-    loop = aio.new_event_loop()
 
     os.environ.setdefault('LUMIMQTT_CONFIG', '/etc/lumimqtt.json')
     config = {}
@@ -62,7 +66,6 @@ def main():
         replace('{MAC}', device_id)  # support old configs
     server = LumiMqtt(
         reconnection_interval=10,
-        loop=loop,
         device_id=config['device_id'],
         topic_root=topic_root,
         host=config['mqtt_host'],
@@ -83,17 +86,23 @@ def main():
     ):
         server.register(device)
 
+    loop = aio.get_running_loop()
+    for _signal in (signal.SIGTERM, signal.SIGQUIT, signal.SIGHUP):
+        loop.add_signal_handler(_signal, signal_handler)
+
     try:
         logger.info(f'Start lumimqtt {version}')
-        loop.run_until_complete(server.start())
+        await server.start()
+    finally:
+        await server.close()
+
+
+def main():
+    try:
+        aio.run(amain())
     except KeyboardInterrupt:
         pass
 
-    finally:
-        loop.run_until_complete(server.close())
-        loop.run_until_complete(loop.shutdown_asyncgens())
-        loop.close()
-
 
 if __name__ == '__main__':
     main()

lumimqtt/lumimqtt.py

@@ -47,7 +47,6 @@ def __init__(
             sensor_debounce_period: int,
             light_transition_period: float,
             light_notification_period: float,
-            loop: ty.Optional[aio.AbstractEventLoop] = None,
     ) -> None:
         self.dev_id = device_id
         self._topic_root = topic_root
@@ -71,11 +70,9 @@ def __init__(
         self._light_transition_period = light_transition_period
         self._light_notification_period = light_notification_period
         self._light_last_sent = None
-
         self._reconnection_interval = reconnection_interval
-        self._loop = loop or aio.get_event_loop()
         self._client = aio_mqtt.Client(
-            loop=self._loop,
+            loop=aio.get_running_loop(),
             client_id_prefix='lumimqtt_',
         )
         self._tasks: ty.List[aio.Future] = []
@@ -89,33 +86,19 @@ def __init__(
 
     async def start(self):
         self._tasks = [
-            self._loop.create_task(self._connect_forever()),
-            self._loop.create_task(self._handle_messages()),
-            self._loop.create_task(self._periodic_publish()),
-            self._loop.create_task(self._handle_buttons()),
+            aio.create_task(self._connect_forever()),
+            aio.create_task(self._handle_messages()),
+            aio.create_task(self._periodic_publish()),
+            aio.create_task(self._handle_buttons()),
         ]
-        finished, unfinished = await aio.wait(
+        finished, _ = await aio.wait(
             self._tasks,
             return_when=aio.FIRST_COMPLETED,
         )
-        for t in unfinished:
-            t.cancel()
-            try:
-                await t
-            except aio.CancelledError:
-                pass
         for t in finished:
             t.result()
 
     async def close(self) -> None:
-        for task in self._tasks:
-            if task.done():
-                continue
-            task.cancel()
-            try:
-                await task
-            except aio.CancelledError:
-                pass
         if self._client.is_connected():
             await self._client.disconnect()
 
@@ -208,7 +191,7 @@ async def _handle_messages(self) -> None:
                     except ValueError as e:
                         logger.exception(str(e))
                         continue
-                    new_light_task = self._loop.create_task(self._light_handler(light, value))
+                    new_light_task = aio.create_task(self._light_handler(light, value))
                     running_message_tasks.append(new_light_task)
                     continue
                 command: ty.Optional[Command] = None
@@ -222,7 +205,7 @@ async def _handle_messages(self) -> None:
                         value = json.loads(message.payload)
                     except ValueError:
                         value = message.payload.decode()
-                    new_command_task = self._loop.create_task(self._command_handler(command, value))
+                    new_command_task = aio.create_task(self._command_handler(command, value))
                     running_message_tasks.append(new_command_task)
         except aio.CancelledError:
             for task in running_message_tasks: