refactor: side effect runners always run the side effect in the event loop provided to them regardless of the return value of the side effect being a coroutine or not, this is because even if the side effect is not a coroutine, it might still use async features internally

Author: sassanh

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Upcoming
 
 - refactor: provide correct signature for the autorun instance based on the function it decorates
+- refactor: side effect runners always run the side effect in the event loop provided to them regardless of the return value of the side effect being a coroutine or not, this is because even if the side effect is not a coroutine, it might still use async features internally
 
 ## Version 0.18.3
 

redux/main.py

@@ -93,7 +93,10 @@ def __init__(
             tuple[EventHandler[Event], Event] | None
         ]()
         self._workers = [
-            SideEffectRunnerThread(task_queue=self._event_handlers_queue)
+            SideEffectRunnerThread(
+                task_queue=self._event_handlers_queue,
+                create_task=self._create_task,
+            )
             for _ in range(self.store_options.threads)
         ]
         for worker in self._workers:

redux/side_effect_runner.py

@@ -4,14 +4,14 @@
 
 import asyncio
 import contextlib
+import inspect
 import threading
 import weakref
 from asyncio import Handle, iscoroutine
 from collections.abc import Callable
-from inspect import signature
 from typing import TYPE_CHECKING, Any, Generic, cast
 
-from redux.basic_types import Event, EventHandler
+from redux.basic_types import Event, EventHandler, TaskCreator
 
 if TYPE_CHECKING:
     import queue
@@ -24,15 +24,14 @@ def __init__(
         self: SideEffectRunnerThread,
         *,
         task_queue: queue.Queue[tuple[EventHandler[Event], Event] | None],
+        create_task: TaskCreator | None,
     ) -> None:
         """Initialize the side effect runner thread."""
         super().__init__()
         self.task_queue = task_queue
         self.loop = asyncio.get_event_loop()
         self._handles: set[Handle] = set()
-        self.create_task = lambda coro: self._handles.add(
-            self.loop.call_soon_threadsafe(self.loop.create_task, coro),
-        )
+        self.create_task = create_task
 
     def run(self: SideEffectRunnerThread[Event]) -> None:
         """Run the side effect runner thread."""
@@ -51,12 +50,27 @@ def run(self: SideEffectRunnerThread[Event]) -> None:
                     event_handler = event_handler_
                 parameters = 1
                 with contextlib.suppress(Exception):
-                    parameters = len(signature(event_handler).parameters)
-                if parameters == 1:
-                    result = cast(Callable[[Event], Any], event_handler)(event)
-                else:
-                    result = cast(Callable[[], Any], event_handler)()
-                if iscoroutine(result):
-                    self.create_task(result)
+                    parameters = len(inspect.signature(event_handler).parameters)
+
+                if self.create_task:
+
+                    async def _(
+                        event_handler: EventHandler[Event],
+                        event: Event,
+                        parameters: int,
+                    ) -> None:
+                        if parameters == 1:
+                            result = cast(Callable[[Event], Any], event_handler)(event)
+                        else:
+                            result = cast(Callable[[], Any], event_handler)()
+                        if iscoroutine(result):
+                            await result
+
+                    self.create_task(_(event_handler, event, parameters))
+                else:  # noqa: PLR5501
+                    if parameters == 1:
+                        cast(Callable[[Event], Any], event_handler)(event)
+                    else:
+                        cast(Callable[[], Any], event_handler)()
             finally:
                 self.task_queue.task_done()

redux_pytest/fixtures/event_loop.py

@@ -8,24 +8,37 @@
 import pytest
 
 if TYPE_CHECKING:
-    from collections.abc import Coroutine
+    from collections.abc import Callable, Coroutine
+
+    from redux.basic_types import TaskCreatorCallback
 
 
 class LoopThread(threading.Thread):
     def __init__(self: LoopThread) -> None:
         super().__init__()
         self.loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(self.loop)
 
     def run(self: LoopThread) -> None:
         self.loop.run_forever()
 
     def stop(self: LoopThread) -> None:
-        asyncio.set_event_loop(None)
         self.loop.call_soon_threadsafe(self.loop.stop)
 
-    def create_task(self: LoopThread, coro: Coroutine) -> None:
-        self.loop.call_soon_threadsafe(self.loop.create_task, coro)
+    def create_task(
+        self: LoopThread,
+        coro: Coroutine,
+        *,
+        callback: TaskCreatorCallback | None = None,
+    ) -> None:
+        def _(
+            coro: Coroutine,
+            callback: Callable[[asyncio.Task], None] | None = None,
+        ) -> None:
+            task = self.loop.create_task(coro)
+            if callback:
+                task.add_done_callback(callback)
+
+        self.loop.call_soon_threadsafe(_, coro, callback)
 
 
 @pytest.fixture