# Path: aioreactive/observers.py
# class AsyncAwaitableObserver(Future[TSource], AsyncObserver[TSource], Disposable):
#     """An async awaitable observer.
# 
#     Both a future and async observer. The future resolves with the last
#     value before the observer is closed. A close without any values sent
#     is the same as cancelling the future."""
# 
#     def __init__(
#         self,
#         asend: Callable[[TSource], Awaitable[None]] = anoop,
#         athrow: Callable[[Exception], Awaitable[None]] = anoop,
#         aclose: Callable[[], Awaitable[None]] = anoop,
#     ) -> None:
#         super().__init__()
# 
#         assert iscoroutinefunction(asend)
#         self._asend = asend
# 
#         assert iscoroutinefunction(athrow)
#         self._athrow = athrow
# 
#         assert iscoroutinefunction(aclose)
#         self._aclose = aclose
# 
#         self._last_value: Optional[TSource] = None
#         self._is_stopped = False
#         self._has_value = False
# 
#     async def asend(self, value: TSource) -> None:
#         log.debug("AsyncAwaitableObserver:asend(%s)", str(value))
# 
#         if self._is_stopped:
#             log.debug("AsyncAwaitableObserver:asend(), Closed!!")
#             return
# 
#         self._last_value = value
#         self._has_value = True
#         await self._asend(value)
# 
#     async def athrow(self, error: Exception) -> None:
#         log.debug("AsyncAwaitableObserver:athrow()")
#         if self._is_stopped:
#             log.debug("AsyncAwaitableObserver:athrow(), Closed!!")
#             return
# 
#         self._is_stopped = True
# 
#         self.set_exception(error)
#         await self._athrow(error)
# 
#     async def aclose(self) -> None:
#         log.debug("AsyncAwaitableObserver:aclose")
# 
#         if self._is_stopped:
#             log.debug("AsyncAwaitableObserver:aclose(), Closed!!")
#             return
# 
#         self._is_stopped = True
# 
#         if self._has_value:
#             self.set_result(cast("TSource", self._last_value))
#         else:
#             self.cancel()
#         await self._aclose()
# 
#     def dispose(self) -> None:
#         log.debug("AsyncAwaitableObserver:dispose()")
# 
#         self._is_stopped = True
# 
# 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 asyncio
import logging
from typing import Awaitable, Callable, Optional, TypeVar
from expression.system.disposable import AsyncDisposable
from .observers import AsyncAwaitableObserver
from .types import AsyncObservable, AsyncObserver



log = logging.getLogger(__name__)

TSource = TypeVar("TSource")


async def run(
    source: AsyncObservable[TSource], observer: Optional[AsyncAwaitableObserver[TSource]] = None, timeout: int = 2

# Path: aioreactive/observers.py
# class AsyncAwaitableObserver(Future[TSource], AsyncObserver[TSource], Disposable):
#     """An async awaitable observer.
# 
#     Both a future and async observer. The future resolves with the last
#     value before the observer is closed. A close without any values sent
#     is the same as cancelling the future."""
# 
#     def __init__(
#         self,
#         asend: Callable[[TSource], Awaitable[None]] = anoop,
#         athrow: Callable[[Exception], Awaitable[None]] = anoop,
#         aclose: Callable[[], Awaitable[None]] = anoop,
#     ) -> None:
#         super().__init__()
# 
#         assert iscoroutinefunction(asend)
#         self._asend = asend
# 
#         assert iscoroutinefunction(athrow)
#         self._athrow = athrow
# 
#         assert iscoroutinefunction(aclose)
#         self._aclose = aclose
# 
#         self._last_value: Optional[TSource] = None
#         self._is_stopped = False
#         self._has_value = False
# 
#     async def asend(self, value: TSource) -> None:
#         log.debug("AsyncAwaitableObserver:asend(%s)", str(value))
# 
#         if self._is_stopped:
#             log.debug("AsyncAwaitableObserver:asend(), Closed!!")
#             return
# 
#         self._last_value = value
#         self._has_value = True
#         await self._asend(value)
# 
#     async def athrow(self, error: Exception) -> None:
#         log.debug("AsyncAwaitableObserver:athrow()")
#         if self._is_stopped:
#             log.debug("AsyncAwaitableObserver:athrow(), Closed!!")
#             return
# 
#         self._is_stopped = True
# 
#         self.set_exception(error)
#         await self._athrow(error)
# 
#     async def aclose(self) -> None:
#         log.debug("AsyncAwaitableObserver:aclose")
# 
#         if self._is_stopped:
#             log.debug("AsyncAwaitableObserver:aclose(), Closed!!")
#             return
# 
#         self._is_stopped = True
# 
#         if self._has_value:
#             self.set_result(cast("TSource", self._last_value))
#         else:
#             self.cancel()
#         await self._aclose()
# 
#     def dispose(self) -> None:
#         log.debug("AsyncAwaitableObserver:dispose()")
# 
#         self._is_stopped = True
# 
# 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/subscription.py
import asyncio
import logging
from typing import Awaitable, Callable, Optional, TypeVar
from expression.system.disposable import AsyncDisposable
from .observers import AsyncAwaitableObserver
from .types import AsyncObservable, AsyncObserver




log = logging.getLogger(__name__)

TSource = TypeVar("TSource")


async def run(
    source: AsyncObservable[TSource], observer: Optional[AsyncAwaitableObserver[TSource]] = None, timeout: int = 2

) -> TSource: