async_rx.observable.rx_last

Complexity

Total Complexity

3

Size/Duplication

Total Lines	47
Duplicated Lines	78.72 %

Importance

Changes

0

1 Function

Rating	Name	Duplication	Size	Complexity
A	rx_last()	37	37	3

from collections import deque
from typing import Any, Deque

from ..protocol import Observable, Observer, Subscription, rx_observer_from
from .rx_create import rx_create

__all__ = ["rx_last"]


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)