# 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
) -> TSource:
    """Run the source with the given observer.

    Similar to `subscribe_async()` but also awaits until the stream
    closes and returns the final value received.

    Args:
        timeout -- Seconds before timing out in case source never closes.

    Returns:
        The last event sent through the stream. If any values have been
        sent through the stream it will return the last value. If the
        stream is closed without any previous values it will throw
        `StopAsyncIteration`. For any other errors it will throw the
        exception.
    """

    # For run we need a noopobserver if no observer is specified to avoid
    # blocking the last single stream in the chain.

# 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:
    """Run the source with the given observer.

    Similar to `subscribe_async()` but also awaits until the stream
    closes and returns the final value received.

    Args:
        timeout -- Seconds before timing out in case source never closes.

    Returns:
        The last event sent through the stream. If any values have been
        sent through the stream it will return the last value. If the
        stream is closed without any previous values it will throw
        `StopAsyncIteration`. For any other errors it will throw the
        exception.
    """

    # For run we need a noopobserver if no observer is specified to avoid
    # blocking the last single stream in the chain.

    obv: AsyncObserver[TSource] = observer or AsyncAwaitableObserver()