# Path: aioreactive/msg.py
# class Msg(SupportsMatch[TSource], ABC):
# class SourceMsg(Msg[Notification[TSource]], SupportsMatch[TSource]):
# class OtherMsg(Msg[Notification[TOther]], SupportsMatch[TOther]):
# class DisposableMsg(Msg[AsyncDisposable], SupportsMatch[AsyncDisposable]):
# class InnerObservableMsg(Msg[AsyncObservable[TSource]], SupportsMatch[AsyncObservable[TSource]]):
# class InnerCompletedMsg(Msg[TSource]):
# class CompletedMsg_(Msg[Any]):
# class DisposeMsg_(Msg[None]):
#     def __match__(self, pattern: Any) -> Iterable[Notification[TSource]]:
#     def __match__(self, pattern: Any) -> Iterable[Notification[TOther]]:
#     def __match__(self, pattern: Any) -> Iterable[AsyncDisposable]:
#     def __match__(self, pattern: Any) -> Iterable[AsyncObservable[TSource]]:
#     def __match__(self, pattern: Any) -> Iterable[Key]:
#     def __match__(self, pattern: Any) -> Iterable[bool]:
#     def __match__(self, pattern: Any) -> Iterable[bool]:
# 
# Path: aioreactive/notification.py
# class MsgKind(Enum):
# class Notification(Generic[TSource], ABC):
# class OnNext(Notification[TSource], SupportsMatch[TSource]):
# class OnError(Notification[TSource], SupportsMatch[Exception]):
# class _OnCompleted(Notification[TSource], SupportsMatch[bool]):
#     ON_NEXT = 1
#     ON_ERROR = 2
#     ON_COMPLETED = 3
#     def __init__(self, kind: MsgKind):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ) -> None:
#     async def accept_observer(self, obv: AsyncObserver[TSource]) -> None:
#     def __repr__(self) -> str:
#     def __init__(self, value: TSource):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ) -> None:
#     async def accept_observer(self, obv: AsyncObserver[TSource]) -> None:
#     def __match__(self, pattern: Any) -> Iterable[TSource]:
#     def __eq__(self, other: Any) -> bool:
#     def __str__(self) -> str:
#     def __init__(self, exception: Exception):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ):
#     async def accept_observer(self, obv: AsyncObserver[TSource]):
#     def __match__(self, pattern: Any) -> Iterable[Exception]:
#     def __eq__(self, other: Any) -> bool:
#     def __str__(self) -> str:
#     def __init__(self):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ):
#     async def accept_observer(self, obv: AsyncObserver[TSource]):
#     def __match__(self, pattern: Any) -> Iterable[bool]:
#     def __eq__(self, other: Any) -> bool:
#     def __str__(self) -> str:
# 
# Path: aioreactive/types.py
# class AsyncObservable(Generic[T_co]):
#     __slots__ = ()
# 
#     @abstractmethod
#     async def subscribe_async(self, observer: AsyncObserver[T_co]) -> AsyncDisposable:
#         raise NotImplementedError
# 
# class AsyncObserver(Generic[T_contra]):
#     """An asynchronous observable."""
# 
#     __slots__ = ()
# 
#     @abstractmethod
#     async def asend(self, value: T_contra) -> None:
#         raise NotImplementedError
# 
#     @abstractmethod
#     async def athrow(self, error: Exception) -> None:
#         raise NotImplementedError
# 
#     @abstractmethod
#     async def aclose(self) -> None:
#         raise NotImplementedError
# 
# Path: aioreactive/utils.py
# async def anoop(value: Optional[Any] = None):
#     """Async no operation. Returns nothing"""
#     pass


import logging
from asyncio import Future, iscoroutinefunction
from typing import Any, AsyncIterable, AsyncIterator, Awaitable, Callable, List, Optional, Tuple, TypeVar, cast
from expression.core import MailboxProcessor, TailCall, tailrec_async
from expression.system import AsyncDisposable, CancellationTokenSource, Disposable
from .msg import DisposableMsg, DisposeMsg, Msg
from .notification import MsgKind, Notification, OnCompleted, OnError, OnNext
from .types import AsyncObservable, AsyncObserver
from .utils import anoop

    Args:
        obv: Observer to serialize access to
        disposable: Disposable to dispose when the observer closes.
    """

    async def worker(inbox: MailboxProcessor[Notification[TSource]]):
        async def message_loop(running: bool) -> None:
            while running:
                msg = await inbox.receive()
                if msg.kind == MsgKind.ON_NEXT:
                    try:
                        await msg.accept_observer(obv)
                    except Exception as ex:
                        await obv.athrow(ex)
                        running = False
                elif msg.kind == MsgKind.ON_ERROR:
                    await disposable.dispose_async()
                    await msg.accept_observer(obv)
                    running = False
                else:
                    await disposable.dispose_async()
                    await obv.aclose()
                    running = False

        await message_loop(running=True)

    agent = MailboxProcessor.start(worker)

    async def asend(value: TSource) -> None:

# Path: aioreactive/msg.py
# class Msg(SupportsMatch[TSource], ABC):
# class SourceMsg(Msg[Notification[TSource]], SupportsMatch[TSource]):
# class OtherMsg(Msg[Notification[TOther]], SupportsMatch[TOther]):
# class DisposableMsg(Msg[AsyncDisposable], SupportsMatch[AsyncDisposable]):
# class InnerObservableMsg(Msg[AsyncObservable[TSource]], SupportsMatch[AsyncObservable[TSource]]):
# class InnerCompletedMsg(Msg[TSource]):
# class CompletedMsg_(Msg[Any]):
# class DisposeMsg_(Msg[None]):
#     def __match__(self, pattern: Any) -> Iterable[Notification[TSource]]:
#     def __match__(self, pattern: Any) -> Iterable[Notification[TOther]]:
#     def __match__(self, pattern: Any) -> Iterable[AsyncDisposable]:
#     def __match__(self, pattern: Any) -> Iterable[AsyncObservable[TSource]]:
#     def __match__(self, pattern: Any) -> Iterable[Key]:
#     def __match__(self, pattern: Any) -> Iterable[bool]:
#     def __match__(self, pattern: Any) -> Iterable[bool]:
# 
# Path: aioreactive/notification.py
# class MsgKind(Enum):
# class Notification(Generic[TSource], ABC):
# class OnNext(Notification[TSource], SupportsMatch[TSource]):
# class OnError(Notification[TSource], SupportsMatch[Exception]):
# class _OnCompleted(Notification[TSource], SupportsMatch[bool]):
#     ON_NEXT = 1
#     ON_ERROR = 2
#     ON_COMPLETED = 3
#     def __init__(self, kind: MsgKind):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ) -> None:
#     async def accept_observer(self, obv: AsyncObserver[TSource]) -> None:
#     def __repr__(self) -> str:
#     def __init__(self, value: TSource):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ) -> None:
#     async def accept_observer(self, obv: AsyncObserver[TSource]) -> None:
#     def __match__(self, pattern: Any) -> Iterable[TSource]:
#     def __eq__(self, other: Any) -> bool:
#     def __str__(self) -> str:
#     def __init__(self, exception: Exception):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ):
#     async def accept_observer(self, obv: AsyncObserver[TSource]):
#     def __match__(self, pattern: Any) -> Iterable[Exception]:
#     def __eq__(self, other: Any) -> bool:
#     def __str__(self) -> str:
#     def __init__(self):
#     async def accept(
#         self,
#         asend: Callable[[TSource], Awaitable[None]],
#         athrow: Callable[[Exception], Awaitable[None]],
#         aclose: Callable[[], Awaitable[None]],
#     ):
#     async def accept_observer(self, obv: AsyncObserver[TSource]):
#     def __match__(self, pattern: Any) -> Iterable[bool]:
#     def __eq__(self, other: Any) -> bool:
#     def __str__(self) -> str:
# 
# Path: aioreactive/types.py
# class AsyncObservable(Generic[T_co]):
#     __slots__ = ()
# 
#     @abstractmethod
#     async def subscribe_async(self, observer: AsyncObserver[T_co]) -> AsyncDisposable:
#         raise NotImplementedError
# 
# class AsyncObserver(Generic[T_contra]):
#     """An asynchronous observable."""
# 
#     __slots__ = ()
# 
#     @abstractmethod
#     async def asend(self, value: T_contra) -> None:
#         raise NotImplementedError
# 
#     @abstractmethod
#     async def athrow(self, error: Exception) -> None:
#         raise NotImplementedError
# 
#     @abstractmethod
#     async def aclose(self) -> None:
#         raise NotImplementedError
# 
# Path: aioreactive/utils.py
# async def anoop(value: Optional[Any] = None):
#     """Async no operation. Returns nothing"""
#     pass


# Path: aioreactive/observers.py
import logging
from asyncio import Future, iscoroutinefunction
from typing import Any, AsyncIterable, AsyncIterator, Awaitable, Callable, List, Optional, Tuple, TypeVar, cast
from expression.core import MailboxProcessor, TailCall, tailrec_async
from expression.system import AsyncDisposable, CancellationTokenSource, Disposable
from .msg import DisposableMsg, DisposeMsg, Msg
from .notification import MsgKind, Notification, OnCompleted, OnError, OnNext
from .types import AsyncObservable, AsyncObserver
from .utils import anoop


    Args:
        obv: Observer to serialize access to
        disposable: Disposable to dispose when the observer closes.
    """

    async def worker(inbox: MailboxProcessor[Notification[TSource]]):
        async def message_loop(running: bool) -> None:
            while running:
                msg = await inbox.receive()
                if msg.kind == MsgKind.ON_NEXT:
                    try:
                        await msg.accept_observer(obv)
                    except Exception as ex:
                        await obv.athrow(ex)
                        running = False
                elif msg.kind == MsgKind.ON_ERROR:
                    await disposable.dispose_async()
                    await msg.accept_observer(obv)
                    running = False
                else:
                    await disposable.dispose_async()
                    await obv.aclose()
                    running = False

        await message_loop(running=True)

    agent = MailboxProcessor.start(worker)

    async def asend(value: TSource) -> None:

        agent.post(OnNext(value))