async_rx.observable.rx_sample

Complexity

Total Complexity

8

Size/Duplication

Total Lines	89
Duplicated Lines	86.52 %

Importance

Changes

0

1 Function

Rating	Name	Duplication	Size	Complexity
C	rx_sample()	77	77	8

from datetime import timedelta
from typing import Any, Optional

import curio

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

__all__ = ["rx_sample"]


def rx_sample(observable: Observable, duration: timedelta) -> Observable:

This code seems to be duplicated in your project.

    """Sample operator used to rate-limit the sequence.

    Sample filter out elements based on the timing.
    Sample will emit the LATEST value on a set interval or emit nothing if no new value arrived during the last interval.


    Args:
        observable (Observable): an observable instance
        duration (timedelta): timedelta of interval (the duration)

    Returns:
        (Observable): observable instance

    Raise:
        (RuntimeError): if no observable or duration are provided

    """
    if not observable or not duration:
        raise RuntimeError("observable and duration are mandatory")

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

        _receive_value = False
        _lastest_value = None
        _consumer_task = None
        _subscription: Optional[Subscription] = None
        _duration = duration.total_seconds()

        async def consumer():
            nonlocal _duration, _lastest_value, _receive_value

The variable _receive_value does not seem to be defined.

            try:
                while True:
                    await curio.sleep(_duration)  # add duration delay before process a new one
                    if _receive_value:
                        await an_observer.on_next(_lastest_value)
                        _receive_value = False

            except curio.TaskCancelled:
                # it's time to finish
                pass

        async def _on_next(item: Any):
            nonlocal _lastest_value, _receive_value

The variable _lastest_value does not seem to be defined.

The variable _receive_value does not seem to be defined.

            _lastest_value = item
            _receive_value = True

        async def _cancel_consumer():
            nonlocal _consumer_task

The variable _consumer_task does not seem to be defined.

            if _consumer_task:
                await _consumer_task.cancel()
                _consumer_task = None

        async def _on_completed():
            nonlocal _consumer_task
            await _cancel_consumer()
            await an_observer.on_completed()

        async def _on_error(err: Any):
            nonlocal _consumer_task
            await _cancel_consumer()
            await an_observer.on_error(err=err)

        async def _subscribe():
            nonlocal _consumer_task, _subscription

The variable _subscription does not seem to be defined.

            await _cancel_consumer()
            if _subscription:
                await _subscription()
                _subscription = None

        _consumer_task = await curio.spawn(consumer())

        _subscription = await observable.subscribe(rx_observer(on_next=_on_next, on_error=_on_error, on_completed=_on_completed))

        return _subscribe

    return rx_create(subscribe=_subscribe, max_observer=1)