async_rx.observable.rx_last.rx_last()

Complexity

Conditions

3

Size

Total Lines	37
Code Lines	16

Duplication

Lines	37
Ratio	100 %

Importance

Changes

0

from collections import deque
def rx_last(observable: Observable, count: int = 1) -> Observable:

This code seems to be duplicated in your project.

    """Create an observale which only take #count (or less) last events and complete.

    Args:
        observable (Observable): observable source
        count (int): number of event to get (default 1)

    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:
        # local buffer of #count
        _q: Deque = deque(maxlen=count)

        async def _on_next(item: Any):
            nonlocal _q

            _q.append(item)

        async def _on_completed():
            nonlocal _q

            for item in _q:
                await an_observer.on_next(item)
            _q.clear()
            await an_observer.on_completed()

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

    return rx_create(subscribe=_subscribe)