# Path: aioreactive/observers.py
# class AsyncIteratorObserver(AsyncObserver[TSource], AsyncIterable[TSource], AsyncDisposable):
#     """An async observer that might be iterated asynchronously."""
# 
#     def __init__(self, source: AsyncObservable[TSource]) -> None:
#         super().__init__()
# 
#         self._push: Future[TSource] = Future()
#         self._pull: Future[bool] = Future()
# 
#         self._awaiters: List[Future[bool]] = []
#         self._subscription: Optional[AsyncDisposable] = None
#         self._source = source
#         self._busy = False
# 
#     async def asend(self, value: TSource) -> None:
#         log.debug("AsyncIteratorObserver:asend(%s)", value)
# 
#         await self._serialize_access()
# 
#         self._push.set_result(value)
#         await self._wait_for_pull()
# 
#     async def athrow(self, error: Exception) -> None:
#         log.debug("AsyncIteratorObserver:athrow()", error)
#         await self._serialize_access()
# 
#         self._push.set_exception(error)
#         await self._wait_for_pull()
# 
#     async def aclose(self) -> None:
#         await self._serialize_access()
# 
#         self._push.set_exception(StopAsyncIteration)
#         await self._wait_for_pull()
# 
#     async def _wait_for_pull(self) -> None:
#         await self._pull
#         self._pull = Future()
#         self._busy = False
# 
#     async def _serialize_access(self) -> None:
#         # Serialize producer event to the iterator
#         while self._busy:
#             fut: Future[bool] = Future()
#             self._awaiters.append(fut)
#             await fut
#             self._awaiters.remove(fut)
# 
#         self._busy = True
# 
#     async def wait_for_push(self) -> TSource:
#         if self._subscription is None:
#             self._subscription = await self._source.subscribe_async(self)
# 
#         value = await self._push
#         self._push = Future()
#         self._pull.set_result(True)
# 
#         # Wake up any awaiters
#         for awaiter in self._awaiters[:1]:
#             awaiter.set_result(True)
#         return value
# 
#     async def dispose_async(self) -> None:
#         if self._subscription is not None:
#             await self._subscription.dispose_async()
#         self._subscription = None
# 
#     def __aiter__(self) -> AsyncIterator[TSource]:
#         log.debug("AsyncIteratorObserver:__aiter__")
#         return self
# 
#     async def __anext__(self) -> TSource:
#         log.debug("AsyncIteratorObserver:__anext__()")
#         return await self.wait_for_push()
# 
# 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


import logging
from typing import AsyncIterable, AsyncIterator, Awaitable, Callable, TypeVar
from expression.system import AsyncDisposable
from .observers import AsyncIteratorObserver
from .types import AsyncObservable, AsyncObserver



TSource = TypeVar("TSource")
TResult = TypeVar("TResult")
TOther = TypeVar("TOther")
TError = TypeVar("TError")
T1 = TypeVar("T1")
T2 = TypeVar("T2")

log = logging.getLogger(__name__)


class AsyncAnonymousObservable(AsyncObservable[TSource]):

    """An anonymous AsyncObservable.

    Uses a custom subscribe method.
    """

    def __init__(self, subscribe: Callable[[AsyncObserver[TSource]], Awaitable[AsyncDisposable]]) -> None:
        self._subscribe = subscribe

    async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
        log.debug("AsyncAnonymousObservable:subscribe_async(%s)", self._subscribe)
        return await self._subscribe(observer)



# Path: aioreactive/observers.py
# class AsyncIteratorObserver(AsyncObserver[TSource], AsyncIterable[TSource], AsyncDisposable):
#     """An async observer that might be iterated asynchronously."""
# 
#     def __init__(self, source: AsyncObservable[TSource]) -> None:
#         super().__init__()
# 
#         self._push: Future[TSource] = Future()
#         self._pull: Future[bool] = Future()
# 
#         self._awaiters: List[Future[bool]] = []
#         self._subscription: Optional[AsyncDisposable] = None
#         self._source = source
#         self._busy = False
# 
#     async def asend(self, value: TSource) -> None:
#         log.debug("AsyncIteratorObserver:asend(%s)", value)
# 
#         await self._serialize_access()
# 
#         self._push.set_result(value)
#         await self._wait_for_pull()
# 
#     async def athrow(self, error: Exception) -> None:
#         log.debug("AsyncIteratorObserver:athrow()", error)
#         await self._serialize_access()
# 
#         self._push.set_exception(error)
#         await self._wait_for_pull()
# 
#     async def aclose(self) -> None:
#         await self._serialize_access()
# 
#         self._push.set_exception(StopAsyncIteration)
#         await self._wait_for_pull()
# 
#     async def _wait_for_pull(self) -> None:
#         await self._pull
#         self._pull = Future()
#         self._busy = False
# 
#     async def _serialize_access(self) -> None:
#         # Serialize producer event to the iterator
#         while self._busy:
#             fut: Future[bool] = Future()
#             self._awaiters.append(fut)
#             await fut
#             self._awaiters.remove(fut)
# 
#         self._busy = True
# 
#     async def wait_for_push(self) -> TSource:
#         if self._subscription is None:
#             self._subscription = await self._source.subscribe_async(self)
# 
#         value = await self._push
#         self._push = Future()
#         self._pull.set_result(True)
# 
#         # Wake up any awaiters
#         for awaiter in self._awaiters[:1]:
#             awaiter.set_result(True)
#         return value
# 
#     async def dispose_async(self) -> None:
#         if self._subscription is not None:
#             await self._subscription.dispose_async()
#         self._subscription = None
# 
#     def __aiter__(self) -> AsyncIterator[TSource]:
#         log.debug("AsyncIteratorObserver:__aiter__")
#         return self
# 
#     async def __anext__(self) -> TSource:
#         log.debug("AsyncIteratorObserver:__anext__()")
#         return await self.wait_for_push()
# 
# 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/observables.py
import logging
from typing import AsyncIterable, AsyncIterator, Awaitable, Callable, TypeVar
from expression.system import AsyncDisposable
from .observers import AsyncIteratorObserver
from .types import AsyncObservable, AsyncObserver




TSource = TypeVar("TSource")
TResult = TypeVar("TResult")
TOther = TypeVar("TOther")
TError = TypeVar("TError")
T1 = TypeVar("T1")
T2 = TypeVar("T2")

log = logging.getLogger(__name__)


class AsyncAnonymousObservable(AsyncObservable[TSource]):

    """An anonymous AsyncObservable.

    Uses a custom subscribe method.
    """

    def __init__(self, subscribe: Callable[[AsyncObserver[TSource]], Awaitable[AsyncDisposable]]) -> None:
        self._subscribe = subscribe

    async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
        log.debug("AsyncAnonymousObservable:subscribe_async(%s)", self._subscribe)
        return await self._subscribe(observer)



class AsyncIterableObservable(AsyncIterable[TSource], AsyncObservable[TSource]):