# Path: aioreactive/observables.py
# 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):
# class AsyncAnonymousObserver(AsyncObserver[TSource]):
# class AsyncNotificationObserver(AsyncObserver[TSource]):
# class AsyncAwaitableObserver(Future[TSource], AsyncObserver[TSource], Disposable):
#     def __init__(self, source: AsyncObservable[TSource]) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
#     async def _wait_for_pull(self) -> None:
#     async def _serialize_access(self) -> None:
#     async def wait_for_push(self) -> TSource:
#     async def dispose_async(self) -> None:
#     def __aiter__(self) -> AsyncIterator[TSource]:
#     async def __anext__(self) -> TSource:
#     def __init__(
#         self,
#         asend: Callable[[TSource], Awaitable[None]] = anoop,
#         athrow: Callable[[Exception], Awaitable[None]] = anoop,
#         aclose: Callable[[], Awaitable[None]] = anoop,
#     ) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
#     def __init__(self, fn: Callable[[Notification[TSource]], Awaitable[None]]) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
# def noop() -> AsyncObserver[Any]:
# def safe_observer(obv: AsyncObserver[TSource], disposable: AsyncDisposable) -> AsyncObserver[TSource]:
#     async def worker(inbox: MailboxProcessor[Notification[TSource]]):
#         async def message_loop(running: bool) -> None:
#     async def asend(value: TSource) -> None:
#     async def athrow(ex: Exception) -> None:
#     async def aclose() -> None:
# def auto_detach_observer(
#     obv: AsyncObserver[TSource],
# ) -> Tuple[AsyncObserver[TSource], Callable[[Awaitable[AsyncDisposable]], Awaitable[AsyncDisposable]]]:
#     async def worker(inbox: MailboxProcessor[Msg[TSource]]):
#         async def message_loop(disposables: List[AsyncDisposable]):
#     async def cancel():
#     async def auto_detach(async_disposable: Awaitable[AsyncDisposable]):
#     def __init__(
#         self,
#         asend: Callable[[TSource], Awaitable[None]] = anoop,
#         athrow: Callable[[Exception], Awaitable[None]] = anoop,
#         aclose: Callable[[], Awaitable[None]] = anoop,
#     ) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
#     def dispose(self) -> None:
# 
# Path: aioreactive/types.py
# class AsyncObservable(Generic[T_co]):
#     __slots__ = ()
# 
#     @abstractmethod
#     async def subscribe_async(self, observer: AsyncObserver[T_co]) -> AsyncDisposable:
#         raise NotImplementedError


import asyncio
import logging
from asyncio import Future
from typing import Any, AsyncIterable, Awaitable, Callable, Iterable, Optional, Tuple, TypeVar
from expression.core import TailCallResult, aiotools, tailrec_async
from expression.core.fn import TailCall
from expression.system import AsyncDisposable, CancellationToken, CancellationTokenSource
from .observables import AsyncAnonymousObservable
from .observers import AsyncObserver, safe_observer
from .types import AsyncObservable
            if task is not None:
                task.cancel()

        sub = AsyncDisposable.create(cancel)

        async def worker() -> None:
            async for value in iterable:
                try:
                    await observer.asend(value)
                except Exception as ex:
                    await observer.athrow(ex)
                    return

            await observer.aclose()

        try:
            task = asyncio.ensure_future(worker())
        except Exception as ex:
            log.debug("FromIterable:worker(), Exception: %s" % ex)
            await observer.athrow(ex)
        return sub

    return AsyncAnonymousObservable(subscribe_async)


def single(value: TSource) -> AsyncObservable[TSource]:
    """Returns an observable sequence containing the single specified
    element."""

    async def subscribe_async(aobv: AsyncObserver[TSource]) -> AsyncDisposable:

# Path: aioreactive/observables.py
# 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):
# class AsyncAnonymousObserver(AsyncObserver[TSource]):
# class AsyncNotificationObserver(AsyncObserver[TSource]):
# class AsyncAwaitableObserver(Future[TSource], AsyncObserver[TSource], Disposable):
#     def __init__(self, source: AsyncObservable[TSource]) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
#     async def _wait_for_pull(self) -> None:
#     async def _serialize_access(self) -> None:
#     async def wait_for_push(self) -> TSource:
#     async def dispose_async(self) -> None:
#     def __aiter__(self) -> AsyncIterator[TSource]:
#     async def __anext__(self) -> TSource:
#     def __init__(
#         self,
#         asend: Callable[[TSource], Awaitable[None]] = anoop,
#         athrow: Callable[[Exception], Awaitable[None]] = anoop,
#         aclose: Callable[[], Awaitable[None]] = anoop,
#     ) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
#     def __init__(self, fn: Callable[[Notification[TSource]], Awaitable[None]]) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
# def noop() -> AsyncObserver[Any]:
# def safe_observer(obv: AsyncObserver[TSource], disposable: AsyncDisposable) -> AsyncObserver[TSource]:
#     async def worker(inbox: MailboxProcessor[Notification[TSource]]):
#         async def message_loop(running: bool) -> None:
#     async def asend(value: TSource) -> None:
#     async def athrow(ex: Exception) -> None:
#     async def aclose() -> None:
# def auto_detach_observer(
#     obv: AsyncObserver[TSource],
# ) -> Tuple[AsyncObserver[TSource], Callable[[Awaitable[AsyncDisposable]], Awaitable[AsyncDisposable]]]:
#     async def worker(inbox: MailboxProcessor[Msg[TSource]]):
#         async def message_loop(disposables: List[AsyncDisposable]):
#     async def cancel():
#     async def auto_detach(async_disposable: Awaitable[AsyncDisposable]):
#     def __init__(
#         self,
#         asend: Callable[[TSource], Awaitable[None]] = anoop,
#         athrow: Callable[[Exception], Awaitable[None]] = anoop,
#         aclose: Callable[[], Awaitable[None]] = anoop,
#     ) -> None:
#     async def asend(self, value: TSource) -> None:
#     async def athrow(self, error: Exception) -> None:
#     async def aclose(self) -> None:
#     def dispose(self) -> None:
# 
# Path: aioreactive/types.py
# class AsyncObservable(Generic[T_co]):
#     __slots__ = ()
# 
#     @abstractmethod
#     async def subscribe_async(self, observer: AsyncObserver[T_co]) -> AsyncDisposable:
#         raise NotImplementedError


# Path: aioreactive/create.py
import asyncio
import logging
from asyncio import Future
from typing import Any, AsyncIterable, Awaitable, Callable, Iterable, Optional, Tuple, TypeVar
from expression.core import TailCallResult, aiotools, tailrec_async
from expression.core.fn import TailCall
from expression.system import AsyncDisposable, CancellationToken, CancellationTokenSource
from .observables import AsyncAnonymousObservable
from .observers import AsyncObserver, safe_observer
from .types import AsyncObservable

            if task is not None:
                task.cancel()

        sub = AsyncDisposable.create(cancel)

        async def worker() -> None:
            async for value in iterable:
                try:
                    await observer.asend(value)
                except Exception as ex:
                    await observer.athrow(ex)
                    return

            await observer.aclose()

        try:
            task = asyncio.ensure_future(worker())
        except Exception as ex:
            log.debug("FromIterable:worker(), Exception: %s" % ex)
            await observer.athrow(ex)
        return sub

    return AsyncAnonymousObservable(subscribe_async)


def single(value: TSource) -> AsyncObservable[TSource]:
    """Returns an observable sequence containing the single specified
    element."""

    async def subscribe_async(aobv: AsyncObserver[TSource]) -> AsyncDisposable:

        safe_obv = safe_observer(aobv, AsyncDisposable.empty())