async_rx.observable.rx_avg.rx_avg()

Complexity

Conditions

2

Size

Total Lines	32
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from typing import Any
def rx_avg(observable: Observable) -> Observable:
    """Create an observable wich return the average items in the source when completes.

    Args:
        observable (observable): the observable source

    Returns:
        (Observable): observable instance

    """

    _count = 0

    async def accumulator(current, item):
        nonlocal _count

The variable _count does not seem to be defined.

        _count += 1
        return current + item

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

        reducer = rx_reduce(observable=observable, accumulator=accumulator, seed=0)

        async def _on_next(item: Any):
            nonlocal _count
            if _count == 0:
                await an_observer.on_error('No value emitted')
            else:
                await an_observer.on_next(item / _count)

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

    return rx_create(subscribe=_subscribe)