# Path: aioreactive/observables.py
# T1 = TypeVar("T1")
# T2 = TypeVar("T2")
# class AsyncAnonymousObservable(AsyncObservable[TSource]):
# class AsyncIterableObservable(AsyncIterable[TSource], AsyncObservable[TSource]):
#     def __init__(self, subscribe: Callable[[AsyncObserver[TSource]], Awaitable[AsyncDisposable]]) -> None:
#     async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
#     def __init__(self, source: AsyncObservable[TSource]) -> None:
#     async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
#     def __aiter__(self) -> AsyncIterator[TSource]:
# 
# Path: aioreactive/types.py
# 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 asyncio import Future
from typing import List, Optional, TypeVar
from expression.system import AsyncDisposable, ObjectDisposedException
from .observables import AsyncObservable
from .types import AsyncObserver
        log.debug("AsyncSingleStream:aclose()")

        while self._observer is None:
            log.debug("AsyncSingleStream:aclose:awaiting start")
            await self._wait

        self.check_disposed()
        if not self._is_stopped:
            log.debug("AsyncSingleStream:subscription disposed")
            await self._observer.aclose()
            self._is_stopped = True

    async def dispose_async(self):
        log.debug("AsyncSingleStream:dispose_async()")

        self._observer = None
        self._is_disposed = True

    async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
        """Start streaming."""

        self.check_disposed()
        self._observer = observer

        if not self._wait.done():
            self._wait.set_result(True)

        return AsyncDisposable.create(self.dispose_async)



# Path: aioreactive/observables.py
# T1 = TypeVar("T1")
# T2 = TypeVar("T2")
# class AsyncAnonymousObservable(AsyncObservable[TSource]):
# class AsyncIterableObservable(AsyncIterable[TSource], AsyncObservable[TSource]):
#     def __init__(self, subscribe: Callable[[AsyncObserver[TSource]], Awaitable[AsyncDisposable]]) -> None:
#     async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
#     def __init__(self, source: AsyncObservable[TSource]) -> None:
#     async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
#     def __aiter__(self) -> AsyncIterator[TSource]:
# 
# Path: aioreactive/types.py
# 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/subject.py
import logging
from asyncio import Future
from typing import List, Optional, TypeVar
from expression.system import AsyncDisposable, ObjectDisposedException
from .observables import AsyncObservable
from .types import AsyncObserver

        log.debug("AsyncSingleStream:aclose()")

        while self._observer is None:
            log.debug("AsyncSingleStream:aclose:awaiting start")
            await self._wait

        self.check_disposed()
        if not self._is_stopped:
            log.debug("AsyncSingleStream:subscription disposed")
            await self._observer.aclose()
            self._is_stopped = True

    async def dispose_async(self):
        log.debug("AsyncSingleStream:dispose_async()")

        self._observer = None
        self._is_disposed = True

    async def subscribe_async(self, observer: AsyncObserver[TSource]) -> AsyncDisposable:
        """Start streaming."""

        self.check_disposed()
        self._observer = observer

        if not self._wait.done():
            self._wait.set_result(True)

        return AsyncDisposable.create(self.dispose_async)



class AsyncMultiSubject(AsyncObserver[TSource], AsyncObservable[TSource], AsyncDisposable):