geronimo-iia / async-rx

__all__ = ["rx_distinct"]

def rx_distinct(observable: Observable, frame_size: int) -> Observable:

    """Create an observable which send distinct event inside a windows of size #frame_size.

    Args:

        observable (Observable): observable source

        frame_size (int): windows size

    Returns:

        (Observable): observable instance

    Raise:

        (RuntimeError): if frame_size <= 0

    """

    if frame_size <= 0:

        raise RuntimeError('framesize must be greather than zero')

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

        # our frame buffer

        _q: Deque = deque(maxlen=frame_size)

        async def _on_next(item: Any):

            nonlocal _q

            if item not in _q:  # distinct value

                _q.append(item)

                await an_observer.on_next(item)

        async def _on_completed():

            nonlocal _q

            _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)

__all__ = ["rx_last"]

def rx_last(observable: Observable, count: int = 1) -> Observable:

    """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)