✏️ Editing: _whiledo.py Read Only

import itertools
from asyncio import Future
from typing import Callable, TypeVar, Union

import reactivex
from reactivex import Observable
from reactivex.internal.utils import infinite
from reactivex.typing import Predicate

_T = TypeVar("_T")

def while_do_(
    condition: Predicate[Observable[_T]],
) -> Callable[[Observable[_T]], Observable[_T]]:
    def while_do(source: Union[Observable[_T], "Future[_T]"]) -> Observable[_T]:
        """Repeats source as long as condition holds emulating a while
        loop.

Args:
            source: The observable sequence that will be run if the
                condition function returns true.

Returns:
            An observable sequence which is repeated as long as the
            condition holds.
        """
        if isinstance(source, Future):
            obs = reactivex.from_future(source)
        else:
            obs = source
        it = itertools.takewhile(condition, (obs for _ in infinite()))
        return reactivex.concat_with_iterable(it)

return while_do

__all__ = ["while_do_"]

🔐 Sid Gifari File Manager Pro