async_rx.observable.rx_list

Complexity

Total Complexity

25

Size/Duplication

Total Lines	131
Duplicated Lines	22.9 %

Importance

Changes

0

18 Methods

Rating	Name	Duplication	Size	Complexity
A	_RxList.copy()	0	2	1
A	_RxList.append()	0	3	1
B	_RxList.subscribe()	30	30	5
A	_RxList.sort()	0	3	1
A	_RxList.remove()	0	3	1
A	_RxList.__iadd__()	0	4	1
A	_RxList._set_event()	0	3	2
A	_RxList.__imul__()	0	4	1
A	_RxList.__add__()	0	4	1
A	_RxList.clear()	0	3	1
A	_RxList.pop()	0	3	1
A	_RxList.extend()	0	3	1
A	_RxList.__init__()	0	4	2
A	_RxList.reverse()	0	3	1
A	_RxList.__delitem__()	0	3	1
A	_RxList.__mul__()	0	4	1
A	_RxList.__setitem__()	0	3	1
A	_RxList.insert()	0	3	1

from collections import UserList
from typing import List, Optional, Union

import curio

from ..protocol import Observable, Observer, Subscription

__all__ = ["rx_list"]


class _RxList(UserList):
    def __init__(self, initlist: Optional[Union[List, "_RxList"]] = None):
        self._event = curio.UniversalEvent()
        self._subscribers = 0
        super().__init__(initlist=initlist if initlist else [])

    async def subscribe(self, an_observer: Observer) -> Subscription:

This code seems to be duplicated in your project.

        if self._subscribers > 0:
            raise RuntimeError("Only one subscription is supported")

        self._subscribers += 1

        _consumer_task = None

        async def consumer():
            try:
                while True:
                    await self._event.wait()
                    await an_observer.on_next(list(self.data))
                    self._event.clear()
            except curio.TaskCancelled:
                # it's time to finish
                pass

        async def _subscription():
            nonlocal _consumer_task

The variable _consumer_task does not seem to be defined.

            if _consumer_task:
                await _consumer_task.cancel()
                _consumer_task = None
            self._subscribers -= 1

        _consumer_task = await curio.spawn(consumer())

        await an_observer.on_next(list(self.data))

        return _subscription

    def _set_event(self):
        if not self._event.is_set():
            self._event.set()

    def __setitem__(self, i, item):
        super().__setitem__(i, item)
        self._set_event()

    def __delitem__(self, i):
        super().__delitem__(i)
        self._set_event()

    def __add__(self, other):
        result = super().__add__(other)
        self._set_event()
        return _RxList(result)

    def __iadd__(self, other):
        super().__iadd__(other)
        self._set_event()
        return self

    def __mul__(self, n):
        result = super().__mul__(n)
        self._set_event()
        return _RxList(result)

    def __imul__(self, n):
        super().__imul__(n)
        self._set_event()
        return self

    def append(self, item):
        super().append(item)
        self._set_event()

    def insert(self, i, item):
        super().insert(i, item)
        self._set_event()

    def pop(self, i=-1):
        super().pop(i)
        self._set_event()

    def remove(self, item):
        super().remove(item)
        self._set_event()

    def clear(self):
        super().clear()
        self._set_event()

    def copy(self):
        return _RxList(super().copy())

    def reverse(self):
        super().reverse()
        self._set_event()

    def sort(self, *args, **kwds):
        super().sort(*args, **kwds)
        self._set_event()

    def extend(self, other):
        super().extend(other)
        self._set_event()


def rx_list(initial_value: Optional[List] = None) -> Observable:
    """Create an observable on list.

    The observer receive the current value of list on subscribe
    and when an item change (added, updated or deleted, ...).
    This observable implements a UserList, so you can use it as a classic list.

    Args:
        initial_value (Optional[List]): intial value (default: [])

    Returns:
        (Observable): observable instance

    """
    return _RxList(initlist=initial_value)