Repository: dbrattli/aioreactive (language: python)

Issue:
split.py example fails with InvalidStateError
1. Create a new Python 3.9 environment.
2. pip install aioreactive
3. Download and run split.py from examples/streams
Expected result: see values being printed out.
Actual result:
```
Exception in callback MailboxProcessor.__process_events()
handle: <Handle MailboxProcessor.__process_events()>
Traceback (most recent call last):
  File "C:\Users\Ian\.conda\envs\aioreactive\lib\asyncio\events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\Ian\.conda\envs\aioreactive\lib\site-packages\expression\core\mailbox.py", line 152, in __process_events
    cont(msg)
  File "C:\Users\Ian\.conda\envs\aioreactive\lib\site-packages\expression\core\aiotools.py", line 45, in done
    future.set_result(value)
asyncio.exceptions.InvalidStateError: invalid state
Exception in callback MailboxProcessor.__process_events()
handle: <Handle MailboxProcessor.__process_events()>
Traceback (most recent call last):
  File "C:\Users\Ian\.conda\envs\aioreactive\lib\asyncio\events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\Ian\.conda\envs\aioreactive\lib\site-packages\expression\core\mailbox.py", line 152, in __process_events
    cont(msg)
  File "C:\Users\Ian\.conda\envs\aioreactive\lib\site-packages\expression\core\aiotools.py", line 45, in done
    future.set_result(value)
asyncio.exceptions.InvalidStateError: invalid state
```


Discussion / hints:
The asyncio event loop is closing immediately and cancelling the futures that do the work resulting in this error.  If you stick
```
await asyncio.sleep(5)
```
at the end of the main function the error goes away.

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index f04bb60..1410945 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -6,7 +6,7 @@ repos:
       - id: ruff-format
         args: [--check]
     repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.1.11
+    rev: v0.1.13
   - hooks:
       - id: pyright
         name: pyright
diff --git a/aioreactive/__init__.py b/aioreactive/__init__.py
index cec59b4..a905776 100644
--- a/aioreactive/__init__.py
+++ b/aioreactive/__init__.py
@@ -314,16 +314,14 @@ def merge(self, other: AsyncObservable[_TSource]) -> AsyncRx[_TSource]:
             AsyncRx.create,
         )
 
-    def reduce(
-        self, accumulator: Callable[[_TResult, _TSource], _TResult], initial: _TResult
-    ) -> "AsyncRx[_TResult]":
+    def reduce(self, accumulator: Callable[[_TResult, _TSource], _TResult], initial: _TResult) -> AsyncRx[_TResult]:
         return pipe(self, reduce(accumulator, initial), AsyncRx[_TResult])
 
     def reduce_async(
         self,
         accumulator: Callable[[_TResult, _TSource], Awaitable[_TResult]],
         initial: _TResult,
-    ) -> "AsyncRx[_TResult]":
+    ) -> AsyncRx[_TResult]:
         return pipe(self, reduce_async(accumulator, initial), AsyncRx[_TResult])
 
     def skip(self, count: int) -> AsyncObservable[_TSource]:
diff --git a/aioreactive/notification.py b/aioreactive/notification.py
index 16d3f04..6ed793b 100644
--- a/aioreactive/notification.py
+++ b/aioreactive/notification.py
@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable, Callable, Iterable
+from collections.abc import Awaitable, Callable
 from enum import Enum
-from typing import Any, Generic, TypeVar, get_origin
+from typing import Any, Generic, TypeVar
 
 from .types import AsyncObserver
 
@@ -59,15 +59,6 @@ async def accept(
     async def accept_observer(self, obv: AsyncObserver[_TSource]) -> None:
         await obv.asend(self.value)
 
-    def __match__(self, pattern: Any) -> Iterable[_TSource]:
-        origin: Any = get_origin(pattern)
-        try:
-            if isinstance(self, origin or pattern):
-                return [self.value]
-        except TypeError:
-            pass
-        return []
-
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, OnNext):
             return self.value == other.value  # type: ignore
@@ -98,15 +89,6 @@ async def accept(
     async def accept_observer(self, obv: AsyncObserver[_TSource]) -> None:
         await obv.athrow(self.exception)
 
-    def __match__(self, pattern: Any) -> Iterable[Exception]:
-        origin: Any = get_origin(pattern)
-        try:
-            if isinstance(self, origin or pattern):
-                return [self.exception]
-        except TypeError:
-            pass
-        return []
-
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, OnError):
             return self.exception == other.exception
@@ -137,18 +119,6 @@ async def accept(
     async def accept_observer(self, obv: AsyncObserver[_TSource]) -> None:
         await obv.aclose()
 
-    def __match__(self, pattern: Any) -> Iterable[bool]:
-        if self is pattern:
-            return [True]
-
-        origin: Any = get_origin(pattern)
-        try:
-            if isinstance(self, origin or pattern):
-                return [True]
-        except TypeError:
-            pass
-        return []
-
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, OnCompleted):
             return True
diff --git a/examples/streams/split.py b/examples/streams/split.py
index f193205..d4d9633 100644
--- a/examples/streams/split.py
+++ b/examples/streams/split.py
@@ -19,6 +19,9 @@ async def mysink(value: int):
     await odds.subscribe_async(rx.AsyncAnonymousObserver(mysink))
     await evens.subscribe_async(rx.AsyncAnonymousObserver(mysink))
 
+    # Wait to avoid the program exiting before the streams are finished.
+    await asyncio.sleep(1)
+
 
 if __name__ == "__main__":
     asyncio.run(main())
diff --git a/poetry.lock b/poetry.lock
index 71c6cea..afdd969 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -264,13 +264,13 @@ packaging = ">=20.9"
 
 [[package]]
 name = "expression"
-version = "5.0.1"
-description = "Practical functional programming for Python 3.9+"
+version = "5.0.2"
+description = "Practical functional programming for Python 3.11+"
 optional = false
 python-versions = ">=3.11,<4"
 files = [
-    {file = "expression-5.0.1-py3-none-any.whl", hash = "sha256:b5fe712a2e8c7515e4a688cff885f2e24b597f522099b3256e83e1edd6c99965"},
-    {file = "expression-5.0.1.tar.gz", hash = "sha256:3755d9115242d1b6c51f2fc6facc769b9b5ce30077abe861e836639c1e935906"},
+    {file = "expression-5.0.2-py3-none-any.whl", hash = "sha256:2b54eeccb6247d86f598a3ce1a4e1efc56e4c7ca901f622d8203919d4106e147"},
+    {file = "expression-5.0.2.tar.gz", hash = "sha256:37019851b0794820c1167b185d6be4a22111604f3bab221f4136e429e5d9b57f"},
 ]
 
 [package.dependencies]


Test patch:
diff --git a/tests/test_reduce.py b/tests/test_reduce.py
index 068eb73..398227a 100644
--- a/tests/test_reduce.py
+++ b/tests/test_reduce.py
@@ -40,7 +40,7 @@ async def test_reduce():
     values = list(map(lambda t: t[1], observer.values))
     assert values == [
         OnNext(10),
-        OnCompleted,
+        OnCompleted(),
     ]
 
 
@@ -55,5 +55,5 @@ async def test_reduce_async():
     values = list(map(lambda t: t[1], observer.values))
     assert values == [
         OnNext(10),
-        OnCompleted,
+        OnCompleted(),
     ]