Fix to ensure all event listeners are stopped when disconnected from SD websocket server

Author: strohganoff

streamdeck/event_listener.py

@@ -31,6 +31,10 @@ def is_sentinal(cls, event: str | bytes | type[_SENTINAL]) -> TypeIs[type[_SENTI
         return event is cls
 
 
+class StopStreaming(Exception):  # noqa: N818
+    """Raised by an EventListener implementation to signal that the entire EventManagerListener should stop streaming events."""
+
+
 class EventListenerManager:
     """Manages event listeners and provides a shared event queue for them to push events into.
 
@@ -65,8 +69,12 @@ def _listener_wrapper(self, listener: EventListener) -> None:
                 if not self.running:
                     break
 
+        except StopStreaming:
+            logger.debug("Event listener requested to stop streaming.")
+            self.event_queue.put(_SENTINAL)
+
         except Exception:
-            logger.exception("Error in wrapped listener.")
+            logger.exception("Unexpected error in wrapped listener %s. Stopping just this listener.", listener)
 
     def stop(self) -> None:
         """Stops the event generation loop and waits for all threads to finish.
@@ -79,10 +87,12 @@ def stop(self) -> None:
         self.event_queue.put(_SENTINAL)
 
         for thread in self.listeners_lookup_by_thread:
-            self.listeners_lookup_by_thread[thread].stop()
-            thread.join()
+            logger.debug("Stopping listener %s.")
+            if thread.is_alive():
+                self.listeners_lookup_by_thread[thread].stop()
+                thread.join()
 
-        logger.info("All listeners have been stopped.")
+            logger.info("All listeners have been stopped.")
 
     def event_stream(self) -> Generator[str | bytes, None, None]:
         """Starts all registered listeners, sets the running flag to True, and yields events from the shared queue."""
@@ -96,6 +106,7 @@ def event_stream(self) -> Generator[str | bytes, None, None]:
             while True:
                 event = self.event_queue.get()
                 if _SENTINAL.is_sentinal(event):
+                    logger.debug("Sentinal received, stopping event stream.")
                     break  # Exit loop immediately if the sentinal is received
                 yield event
         finally:

streamdeck/manager.py

@@ -166,4 +166,4 @@ def run(self) -> None:
 
                     processed_handler(data)
 
-
+            logger.info("PluginManager has stopped processing events.")

streamdeck/websocket.py

@@ -8,10 +8,11 @@
 from logging import getLogger
 from typing import TYPE_CHECKING
 
+from websockets import ConnectionClosedError, WebSocketException
 from websockets.exceptions import ConnectionClosed, ConnectionClosedOK
 from websockets.sync.client import ClientConnection, connect
 
-from streamdeck.event_listener import EventListener
+from streamdeck.event_listener import EventListener, StopStreaming
 from streamdeck.models import events
 
 
@@ -72,14 +73,21 @@ def listen(self) -> Generator[str | bytes, Any, None]:
                 message: str | bytes = self._client.recv()
                 yield message
 
-        except ConnectionClosedOK:
-            logger.debug("Connection was closed normally, stopping the client.")
+        except WebSocketException as exc:
+            if isinstance(exc, ConnectionClosedOK):
+                logger.debug("Connection was closed normally, stopping the client.")
+            elif isinstance(exc, ConnectionClosedError):
+                logger.exception("Connection was terminated with an error.")
+            elif isinstance(exc, ConnectionClosed):
+                logger.exception("Connection was already closed.")
+            else:
+                logger.exception("Connection is closed due to an unexpected WebSocket error.")
 
-        except ConnectionClosed:
-            logger.exception("Connection was closed with an error.")
+            raise StopStreaming from None
 
         except Exception:
             logger.exception("Failed to receive messages from websocket server due to unexpected error.")
+            raise
 
     def start(self) -> None:
         """Start the connection to the websocket server."""