refactor: Store no longer aggregates changes, it now calls listeners with every change

refactor: `SideEffectRunnerThread` now runs async side effects in the event loop of the thread in which it was instantiated in (it used to create its own event loop)
refactor(test): `event_loop` fixture now sets the global event loop on setup and restores it on teardown

Author: sassanh

.github/workflows/integration_delivery.yml

@@ -6,7 +6,7 @@ on:
   workflow_dispatch:
 
 env:
-  PYTHON_VERSION: "3.11"
+  PYTHON_VERSION: '3.11'
 
 jobs:
   dependencies:
@@ -267,7 +267,7 @@ jobs:
           path: artifacts
 
       - name: Release
-        uses: softprops/action-gh-release@v1
+        uses: softprops/action-gh-release@v2
         with:
           files: artifacts/*
           tag_name: v${{ needs.build.outputs.version }}

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## Version 0.14.0
+
+- refactor: `Store` no longer aggregates changes, it now calls listeners with every
+  change
+- refactor: `SideEffectRunnerThread` now runs async side effects in the event loop
+  of the thread in which it was instantiated in (it used to create its own event
+  loop)
+- refactor(test): `event_loop` fixture now sets the global event loop on setup and
+  restores it on teardown
+
 ## Version 0.13.2
 
 - fix: initial snapshot cleanup which used to mistakenly remove files with store:...

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python-redux"
-version = "0.13.2"
+version = "0.14.0"
 description = "Redux implementation for Python"
 authors = ["Sassan Haradji <sassanh@gmail.com>"]
 license = "Apache-2.0"
@@ -60,6 +60,7 @@ profile = "black"
 
 [tool.pyright]
 exclude = ['typings']
+filterwarnings = 'error'
 
 [tool.pytest.ini_options]
 log_cli = 1

pyproject.toml

@@ -171,8 +171,7 @@ def _check_and_call(
                 create_task = self._store._create_task  # noqa: SLF001
                 if iscoroutine(self._latest_value) and create_task:
                     create_task(self._latest_value, callback=self._task_callback)
-                else:
-                    self.inform_subscribers()
+                self.inform_subscribers()
             else:
                 self.unsubscribe()
 

redux/autorun.py

@@ -50,6 +50,7 @@ def __init__(
         options: CreateStoreOptions[Action, Event] | None = None,
     ) -> None:
         """Create a new store."""
+        self.finished = False
         self.store_options = options or CreateStoreOptions()
         self.reducer = reducer
         self._create_task = self.store_options.task_creator
@@ -95,35 +96,37 @@ def __init__(
         if self.store_options.scheduler:
             self.store_options.scheduler(self.run, interval=True)
 
+    def _call_listeners(self: Store[State, Action, Event], state: State) -> None:
+        for listener_ in self._listeners.copy():
+            if isinstance(listener_, weakref.ref):
+                listener = listener_()
+                if listener is None:
+                    self._listeners.discard(listener_)
+                    continue
+            else:
+                listener = listener_
+            result = listener(state)
+            if asyncio.iscoroutine(result) and self._create_task:
+                self._create_task(result)
+
     def _run_actions(self: Store[State, Action, Event]) -> None:
         action = self._actions.pop(0)
         result = self.reducer(self._state, action)
         if is_complete_reducer_result(result):
             self._state = result.state
+            self._call_listeners(self._state)
             self.dispatch([*(result.actions or []), *(result.events or [])])
         elif is_state_reducer_result(result):
             self._state = result
+            self._call_listeners(self._state)
 
         if isinstance(action, FinishAction):
             self.dispatch(cast(Event, FinishEvent()))
 
-        if len(self._actions) == 0 and self._state:
-            for listener_ in self._listeners.copy():
-                if isinstance(listener_, weakref.ref):
-                    listener = listener_()
-                    if listener is None:
-                        self._listeners.discard(listener_)
-                        continue
-                else:
-                    listener = listener_
-                result = listener(self._state)
-                if asyncio.iscoroutine(result) and self._create_task:
-                    self._create_task(result)
-
     def _run_event_handlers(self: Store[State, Action, Event]) -> None:
         event = self._events.pop(0)
         for event_handler_ in self._event_handlers[type(event)].copy():
-            self._event_handlers_queue.put((event_handler_, event))
+            self._event_handlers_queue.put_nowait((event_handler_, event))
 
     def run(self: Store[State, Action, Event]) -> None:
         """Run the store."""
@@ -134,7 +137,12 @@ def run(self: Store[State, Action, Event]) -> None:
 
                 if len(self._events) > 0:
                     self._run_event_handlers()
-        if not any(worker.is_alive() for worker in self._workers):
+        if (
+            self.finished
+            and self._actions == []
+            and self._events == []
+            and not any(worker.is_alive() for worker in self._workers)
+        ):
             self.clean_up()
 
     def clean_up(self: Store[State, Action, Event]) -> None:
@@ -219,7 +227,8 @@ def unsubscribe() -> None:
 
     def _handle_finish_event(self: Store[State, Action, Event]) -> None:
         for _ in range(self.store_options.threads):
-            self._event_handlers_queue.put(None)
+            self._event_handlers_queue.put_nowait(None)
+        self.finished = True
 
     def autorun(
         self: Store[State, Action, Event],

redux/main.py

@@ -6,7 +6,7 @@
 import contextlib
 import threading
 import weakref
-from asyncio import Task, iscoroutine
+from asyncio import Handle, iscoroutine
 from inspect import signature
 from typing import TYPE_CHECKING, Any, Callable, Generic, cast
 
@@ -27,16 +27,14 @@ def __init__(
         """Initialize the side effect runner thread."""
         super().__init__()
         self.task_queue = task_queue
-        self._tasks: set[Task] = set()
+        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),
+        )
 
     def run(self: SideEffectRunnerThread[Event]) -> None:
         """Run the side effect runner thread."""
-        self.loop = asyncio.new_event_loop()
-        self.create_task = lambda coro: self._tasks.add(self.loop.create_task(coro))
-        self.loop.run_until_complete(self.work())
-
-    async def work(self: SideEffectRunnerThread[Event]) -> None:
-        """Run the side effects."""
         while True:
             task = self.task_queue.get()
             if task is None:
@@ -61,17 +59,3 @@ async def work(self: SideEffectRunnerThread[Event]) -> None:
                     self.create_task(result)
             finally:
                 self.task_queue.task_done()
-        await self.clean_up()
-
-    async def clean_up(self: SideEffectRunnerThread[Event]) -> None:
-        """Clean up the side effect runner thread."""
-        while True:
-            tasks = [
-                task
-                for task in asyncio.all_tasks(self.loop)
-                if task is not asyncio.current_task(self.loop)
-            ]
-            if not tasks:
-                break
-            for task in tasks:
-                await task

redux/side_effect_runner.py

@@ -12,15 +12,17 @@ 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(lambda: self.loop.create_task(coro))
+        self.loop.call_soon_threadsafe(self.loop.create_task, coro)
 
 
 @pytest.fixture()

redux_pytest/fixtures/event_loop.py

@@ -51,7 +51,12 @@ def __init__(
     def json_snapshot(self: StoreSnapshot) -> str:
         """Return the snapshot of the current state of the store."""
         return (
-            json.dumps(self.store.snapshot, indent=2, sort_keys=True)
+            json.dumps(
+                self.store.snapshot,
+                indent=2,
+                sort_keys=True,
+                ensure_ascii=False,
+            )
             if self.store._state  # noqa: SLF001
             else ''
         )

redux_pytest/fixtures/snapshot.py

@@ -2,12 +2,12 @@
 {
   "_id": "e3e70682c2094cac629f6fbed82c07cd",
   "base10": {
-    "count": 12
+    "count": 11
   },
   "inverse": {
-    "count": -1
+    "count": 0
   },
   "straight": {
-    "count": 2
+    "count": 1
   }
 }