async_rx.observable.rx_range.rx_range()

Complexity

Conditions

2

Size

Total Lines	21
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from ..protocol import Observable, Observer, Subscription, default_subscription
def rx_range(start: int, stop: int, step: int = 1) -> Observable:
    """Create an observable sequence of range.

    Args:
        start (int): initiale value
        stop (int): last value
        step (int): default increment (default: {1})

    Returns:
        (Observable): observable instance.

    """

    async def _subscribe(an_observer: Observer) -> Subscription:
        for i in range(start, stop, step):
            await an_observer.on_next(i)
        await an_observer.on_completed()

        return default_subscription

    return rx_create(subscribe=_subscribe)