Keep a reference to tasks

docs/userguide/testing.rst

@@ -42,7 +42,7 @@ Create a ``tests`` directory at the root of the project directory and create a m
                 client = ClientComponent("Hello!")
                 await client.start(ctx)
 
-        event_loop.create_task(run())
+        task = event_loop.create_task(run())
         with pytest.raises(SystemExit) as exc:
             event_loop.run_forever()
 

examples/tutorial1/tests/test_client_server.py

@@ -21,7 +21,7 @@ async def run() -> None:
             client = ClientComponent("Hello!")
             await client.start(ctx)
 
-    event_loop.create_task(run())
+    task = event_loop.create_task(run())  # noqa: F841
     with pytest.raises(SystemExit) as exc:
         event_loop.run_forever()
 

src/asphalt/core/event.py

@@ -5,7 +5,7 @@
 import logging
 import sys
 import weakref
-from asyncio import Queue, create_task, get_running_loop, iscoroutine, wait
+from asyncio import Queue, Task, create_task, get_running_loop, iscoroutine, wait
 from datetime import datetime, timezone
 from inspect import getmembers, isawaitable
 from time import time as stdlib_time
@@ -19,6 +19,7 @@
     MutableMapping,
     Optional,
     Sequence,
+    Set,
     Type,
     TypeVar,
     cast,
@@ -89,7 +90,14 @@ class Signal(Generic[T_Event]):
     :param event_class: an event class
     """
 
-    __slots__ = "event_class", "topic", "source", "listeners", "bound_signals"
+    __slots__ = (
+        "event_class",
+        "topic",
+        "source",
+        "listeners",
+        "bound_signals",
+        "_background_tasks",
+    )
 
     def __init__(
         self,
@@ -109,6 +117,7 @@ def __init__(
                 event_class, Event
             ), "event_class must be a subclass of Event"
             self.bound_signals: MutableMapping[Any, Signal] = WeakKeyDictionary()
+        self._background_tasks: Set[Task] = set()
 
     def __get__(self, instance, owner) -> Signal:
         if instance is None:
@@ -225,7 +234,9 @@ async def do_dispatch() -> None:
         future = loop.create_future()
         if self.listeners:
             listeners = list(self.listeners)
-            loop.create_task(do_dispatch())
+            task = loop.create_task(do_dispatch())
+            self._background_tasks.add(task)
+            task.add_done_callback(self._background_tasks.discard)
         else:
             future.set_result(True)
 

tests/test_event.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import gc
-from asyncio import AbstractEventLoop, Queue, all_tasks, current_task
+from asyncio import AbstractEventLoop, Queue, all_tasks, current_task, sleep
 from datetime import datetime, timedelta, timezone
 from typing import NoReturn
 
@@ -93,6 +93,21 @@ async def callback(event: Event) -> None:
         assert events[0].args == ("x", "y")
         assert events[0].kwargs == {"a": 1, "b": 2}
 
+    @pytest.mark.asyncio
+    async def test_dispatch_event_coroutine_complete(
+        self, source: DummySource, capfd
+    ) -> None:
+        """Test that a coroutine function listening to an event runs until complete."""
+
+        async def callback(event: Event) -> None:
+            await sleep(0.1)
+            print("callback done")
+
+        source.event_a.connect(callback)
+        assert await source.event_a.dispatch()
+        out, err = capfd.readouterr()
+        assert out == "callback done\n"
+
     @pytest.mark.asyncio
     async def test_dispatch_raw(self, source: DummySource) -> None:
         """Test that dispatch_raw() correctly dispatches the given event."""