fix shutdown mqtt server correctly

bumper/__init__.py

@@ -140,6 +140,8 @@ async def maintenance():
 async def shutdown():
     try:
         bumperlog.info("Shutting down")
+        global shutting_down
+        shutting_down = True
 
         await mqtt_helperbot.disconnect()
         await web_server.shutdown()
@@ -151,18 +153,11 @@ async def shutdown():
             if xmpp_server.server._serving:
                 xmpp_server.server.close()
             await xmpp_server.server.wait_closed()
-        global shutting_down
-        shutting_down = True
 
+        bumperlog.info("Shutdown complete")
     except asyncio.CancelledError:
         bumperlog.info("Coroutine canceled")
 
-    except Exception as e:
-        bumperlog.info(f"Exception: {e}")
-
-    finally:
-        bumperlog.info("Shutdown complete")
-
 
 def main(argv=None):
     import argparse

bumper/mqtt/server.py

@@ -105,6 +105,12 @@ class MQTTServer:
 
     async def shutdown(self) -> None:
         """Shutdown server."""
+        # stop session handler manually otherwise connection will not be closed correctly
+        for (
+            _,
+            handler,
+        ) in self._broker._sessions.values():  # pylint: disable=protected-access
+            await handler.stop()
         await self._broker.shutdown()
 
 

bumper/web/server.py

@@ -243,20 +243,10 @@ class WebServer:
         asyncio.create_task(bumper.mqtt_helperbot.start())
 
     async def _restart_mqtt_server(self) -> None:
-        loop = asyncio.get_event_loop()
-
         if bumper.mqtt_server.state not in ["stopped", "not_started"]:
-            # close session writers - this was required so bots would reconnect properly after restarting
-            for sess in list(bumper.mqtt_server.broker._sessions):
-                sessobj = bumper.mqtt_server.broker._sessions[sess][1]
-                if sessobj.session.transitions.state == "connected":
-                    await sessobj.writer.close()
+            await bumper.mqtt_server.shutdown()
 
-            loop.call_later(
-                0.1, lambda: asyncio.create_task(bumper.mqtt_server.shutdown())
-            )
-
-        loop.call_later(1.5, lambda: asyncio.create_task(bumper.mqtt_server.start()))
+        asyncio.create_task(bumper.mqtt_server.start())
 
     async def _handle_restart_service(self, request: Request) -> Response:
         try:

tests/test_mqttserver.py

@@ -354,8 +354,7 @@ async def test_nofileauth_mqttserver():
 
         mqtt_server = MQTTServer(HOST, MQTT_PORT, password_file="tests/passwd-notfound")
         await mqtt_server.start()
-        await mqtt_server.shutdown()
-
+        try:
             l.check_present(
                 (
                     "amqtt.broker.plugins.bumper",
@@ -364,3 +363,5 @@ async def test_nofileauth_mqttserver():
                 ),
                 order_matters=False,
             )
+        finally:
+            await mqtt_server.shutdown()