fix(notifier): ignore queued messages after stopping

can/notifier.py

@@ -196,6 +196,9 @@ def stop(self, timeout: float = 5.0) -> None:
         """Stop notifying Listeners when new :class:`~can.Message` objects arrive
         and call :meth:`~can.Listener.stop` on each Listener.
 
+        Queued message delivery callbacks are ignored after stopping. Already
+        scheduled coroutine tasks are not cancelled.
+
         :param timeout:
             Max time in seconds to wait for receive threads to finish.
             Should be longer than timeout given at instantiation.
@@ -252,6 +255,8 @@ def _on_message_available(self, bus: BusABC) -> None:
 
     def _on_message_received(self, msg: Message) -> None:
         for callback in self.listeners:
+            if self._stopped:
+                return
             res = callback(msg)
             if res and self._loop and asyncio.iscoroutine(res):
                 # Schedule coroutine and keep a reference to the task

doc/changelog.d/1712.fixed.rst

@@ -0,0 +1,2 @@
+Ignore queued message deliveries after ``Notifier.stop()`` so they do not call
+listeners that have already been stopped or write to closed log files.

test/notifier_test.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 
 import asyncio
+import io
 import time
 import unittest
 
@@ -74,6 +75,83 @@ def test_registry(self):
 
 
 class AsyncNotifierTest(unittest.TestCase):
+    def test_queued_message_is_not_delivered_after_stop(self):
+        async def run_it():
+            loop = asyncio.get_running_loop()
+            errors = []
+            loop.set_exception_handler(lambda loop, context: errors.append(context))
+            stream = io.StringIO()
+            writer = can.CanutilsLogWriter(stream)
+            reader = can.BufferedReader()
+            notifier = can.Notifier([], [writer, reader], loop=loop)
+            delivered = can.Message(
+                arbitration_id=0x123, is_extended_id=False, data=[0x12]
+            )
+            queued = can.Message(
+                arbitration_id=0x456, is_extended_id=False, data=[0x34]
+            )
+            try:
+                loop.call_soon_threadsafe(notifier._on_message_received, delivered)
+                await asyncio.sleep(0)
+                self.assertIn("123#12", stream.getvalue())
+
+                loop.call_soon_threadsafe(notifier._on_message_received, queued)
+                notifier.stop()
+                await asyncio.sleep(0)
+
+                self.assertTrue(stream.closed)
+                self.assertIs(reader.get_message(timeout=0), delivered)
+                self.assertIsNone(reader.get_message(timeout=0))
+                self.assertEqual(errors, [])
+            finally:
+                notifier.stop()
+
+        asyncio.run(run_it())
+
+    def test_listener_can_stop_delivery_to_remaining_listeners(self):
+        async def run_it():
+            delivered = []
+
+            def stop_and_record(msg):
+                delivered.append(msg)
+                notifier.stop()
+
+            notifier = can.Notifier(
+                [],
+                [stop_and_record, lambda msg: stop_and_record(msg)],
+                loop=asyncio.get_running_loop(),
+            )
+            message = can.Message()
+            try:
+                notifier._on_message_received(message)
+                self.assertEqual(delivered, [message])
+                self.assertTrue(notifier.stopped)
+            finally:
+                notifier.stop()
+
+        asyncio.run(run_it())
+
+    def test_active_listener_error_is_reported_to_event_loop(self):
+        async def run_it():
+            loop = asyncio.get_running_loop()
+            errors = []
+            loop.set_exception_handler(lambda loop, context: errors.append(context))
+
+            def raising_listener(msg):
+                raise ValueError("active listener failed")
+
+            notifier = can.Notifier([], [raising_listener], loop=loop)
+            try:
+                loop.call_soon_threadsafe(notifier._on_message_received, can.Message())
+                await asyncio.sleep(0)
+                self.assertEqual(len(errors), 1)
+                self.assertIsInstance(errors[0]["exception"], ValueError)
+                self.assertEqual(str(errors[0]["exception"]), "active listener failed")
+            finally:
+                notifier.stop()
+
+        asyncio.run(run_it())
+
     def test_asyncio_notifier(self):
         async def run_it():
             with can.Bus("test", interface="virtual", receive_own_messages=True) as bus: