diff --git a/can/notifier.py b/can/notifier.py index fd21a0662..2b30fbff5 100644 --- a/can/notifier.py +++ b/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 diff --git a/doc/changelog.d/1712.fixed.rst b/doc/changelog.d/1712.fixed.rst new file mode 100644 index 000000000..6223154b6 --- /dev/null +++ b/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. diff --git a/test/notifier_test.py b/test/notifier_test.py index d8512a00b..75b112040 100644 --- a/test/notifier_test.py +++ b/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: