async_rx.observable.rx_skip.rx_skip()

Complexity

Conditions

3

Size

Total Lines	32
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from typing import Any
def rx_skip(observable: Observable, count: int) -> Observable:
    """Create an obervable wich skip #count event on source.

    Args:
        observable (Observable): observable source
        count (int): number of event to skip

    Returns:
        (Observable): observable instance

    Raise:
        (RuntimeError): if count <= 0

    """
    if count <= 0:
        raise RuntimeError('count must be greather than zero')

    async def _subscribe(an_observer: Observer) -> Subscription:

        _count: int = 0

        async def _on_next(item: Any):
            nonlocal _count

The variable _count does not seem to be defined.

            if _count < count:
                _count += 1
            else:
                await an_observer.on_next(item)

        return await observable.subscribe(an_observer=rx_observer_from(observer=an_observer, on_next=_on_next))

    return rx_create(subscribe=_subscribe)