✏️ Editing: fromfuture.py Read Only

import asyncio
from asyncio import Future
from typing import Any, Optional, TypeVar, cast

from reactivex import Observable, abc
from reactivex.disposable import Disposable

_T = TypeVar("_T")

def from_future_(future: "Future[_T]") -> Observable[_T]:
    """Converts a Future to an Observable sequence

Args:
        future -- A Python 3 compatible future.
            https://docs.python.org/3/library/asyncio-task.html#future

Returns:
        An Observable sequence which wraps the existing future success
        and failure.
    """

def subscribe(
        observer: abc.ObserverBase[Any], scheduler: Optional[abc.SchedulerBase] = None
    ) -> abc.DisposableBase:
        def done(future: "Future[_T]") -> None:
            try:
                value: Any = future.result()
            except Exception as ex:
                observer.on_error(ex)
            except asyncio.CancelledError as ex:  # pylint: disable=broad-except
                # asyncio.CancelledError is a BaseException, so need to cast
                observer.on_error(cast(Exception, ex))
            else:
                observer.on_next(value)
                observer.on_completed()

future.add_done_callback(done)

def dispose() -> None:
            if future:
                future.cancel()

return Disposable(dispose)

return Observable(subscribe)

__all__ = ["from_future_"]

🔐 Sid Gifari File Manager Pro