async_rx.protocol.definition

Complexity

Total Complexity

29

Size/Duplication

Total Lines	343
Duplicated Lines	0 %

Importance

Changes

0

29 Methods

Rating	Name	Duplication	Size	Complexity
A	Collector.has_error()	0	3	1
A	ObservableFactory.__call__()	0	8	1
A	RefCountHandler.__call__()	0	2	1
A	ConnectableObservable.ref_count()	0	8	1
A	Observer.on_error()	0	2	1
A	Collector.is_finish()	0	3	1
A	Observable.subscribe()	0	2	1
A	_SyncAccumulatorOperator.__call__()	0	2	1
A	SubjectEventHandler.__call__()	0	2	1
A	_AsyncAccumulatorOperator.__call__()	0	2	1
A	ConnectableObservableHandler.on_connect()	0	3	1
A	_SyncPredicateOperator.__call__()	0	2	1
A	ConnectableObservableEventHandler.__call__()	0	2	1
A	Collector.error()	0	3	1
A	Subscription.__call__()	0	3	1
A	_AsyncPredicateOperator.__call__()	0	2	1
A	ConnectHandler.__call__()	0	2	1
A	Subscribe.__call__()	0	11	1
A	ConnectableObservable.connect()	0	3	1
A	CompleteHandler.__call__()	0	3	1
A	SubjectFactory.__call__()	0	2	1
A	Observer.on_completed()	0	3	1
A	ConnectableObservableHandler.on_disconnect()	0	3	1
A	ErrorHandler.__call__()	0	11	1
A	SubjectHandler.on_unsubscribe()	0	9	1
A	Observer.on_next()	0	3	1
A	Collector.result()	0	3	1
A	NextHandler.__call__()	0	3	1
A	SubjectHandler.on_subscribe()	0	9	1

"""Protocol definition."""
import sys
from typing import Any, NoReturn, Optional, TypeVar, Union

# Protocol is only available in Python 3.8+.
if sys.version_info.minor > 7:  # pragma: no cover
    from typing import Protocol
else:  # pragma: no cover
    from typing_extensions import Protocol  # type: ignore

__all__ = [
    "Subscription",
    "NextHandler",
    "CompleteHandler",
    "ErrorHandler",
    "Observable",
    "Observer",
    "Collector",
    "Subscribe",
    "Subject",
    "ConnectHandler",
    "RefCountHandler",
    "ConnectableObservable",
    "ObservableFactory",
    "SubjectEventHandler",
    "SubjectHandler",
    "ConnectableObservableEventHandler",
    "ConnectableObservableHandler",
    "PredicateOperator",
    "AccumulatorOperator",
    "SubjectFactory",
]

T = TypeVar('T')


class Subscription(Protocol):
    """Subscription Protocol.

    Subscription is a function to release resources or cancel Observable executions (act as a Disposable).
    It define something to be used and thrown away after you call it.
    """

    async def __call__(self) -> None:  # pragma: no cover
        """Release subscription."""
        pass


class NextHandler(Protocol):
    """NextHandler Protocol.

    A next handler process an item from associated observable.
    """

    async def __call__(self, item: Any) -> None:  # pragma: no cover
        """Process item."""
        pass


class CompleteHandler(Protocol):
    """CompleteHandler Protocol.

    A complete handler is call when no more item will came from the associated observable.
    """

    async def __call__(self) -> None:  # pragma: no cover
        """Signal completion of this observable."""
        pass


class ErrorHandler(Protocol):
    """ErrorHandler Protocol.

    An error handler receive a message or an exception and raise it.
    """

    async def __call__(self, err: Any) -> Optional[NoReturn]:  # pragma: no cover
        """Raise error.

        Args:
            err (Union[Any, Exception]): the error to raise

        Raises:
            (Exception): the exception

        """
        pass


class Observer(Protocol):
    """Observer Protocol.

    What is an Observer?

    An Observer is a consumer of values delivered by an Observable.

    Observers are simply a set of callbacks, one for each type of notification
    delivered by the Observable:

    - next,
    - error,
    - and complete.

    Observers are just "objects" with three callbacks, one for each type of
    notification that an Observable may deliver.
    """

    async def on_next(self, item: Any) -> None:  # pragma: no cover
        """Process item."""
        pass

    async def on_completed(self) -> None:  # pragma: no cover
        """Signal completion of this observable."""
        pass

    async def on_error(self, err: Any) -> Optional[NoReturn]:  # pragma: no cover
        pass


class Collector(Observer, Protocol):
    """Collector Observer Protocol."""

    def result(self) -> Any:  # pragma: no cover
        """Returns result."""
        pass

    def is_finish(self) -> bool:  # pragma: no cover
        """Return true if observable has completed."""
        pass

    def has_error(self) -> bool:  # pragma: no cover
        """Return true if observable has meet error."""
        pass

    def error(self) -> Any:  # pragma: no cover
        """Return error if observable has meet error."""
        pass


class Subscribe(Protocol):
    """Subscribe Protocol.

    It's a (sync/async) function wich take an observer and return a subscription.
    """

    async def __call__(self, an_observer: Observer) -> Subscription:  # pragma: no cover
        """Implement observer subscription.

        Args:
            observer (Observer): the observer instance

        Returns:
            (Subscription): subscription

        """
        pass


class Observable(Protocol):
    """Observable Protocol.

    An observable is something on which we can subscribe to listen event.
    """

    async def subscribe(self, an_observer: Observer) -> Subscription:  # pragma: no cover
        pass


class ObservableFactory(Protocol):
    """Async ObservableFactory Protocol.

    Define function which create Observable.
    """

    async def __call__(self) -> Observable:  # pragma: no cover
        """Create an Observable.

        Returns:
            (Observable): the new observable instance.

        """
        pass


class Subject(Observable, Observer, Protocol):
    """A Subject is like an Observable, but can multicast to many Observers.

    Subjects are like EventEmitters: they maintain a registry of many listeners.
    """

    pass


class SubjectEventHandler(Protocol):
    """Subject Event Handler Procotol."""

    async def __call__(self, count: int, source: Observer) -> None:  # pragma: no cover
        pass


class SubjectHandler(Protocol):
    """Subscribe Handler Protocol.

    This handler could be called on subscription/unsubscribe event.
    """

    async def on_subscribe(self, count: int, source: Observer) -> None:  # pragma: no cover
        """Notify on subscribe event.

        Args:
            count (int): current #subscribers after subscription
            source (Observer): observer source

        """
        pass

    async def on_unsubscribe(self, count: int, source: Observer) -> None:  # pragma: no cover
        """Notify on unsubscribe event.

        Args:
            count (int): current #subscribers after unsubscribe
            source (Observer): observer source

        """
        pass


class SubjectFactory(Protocol):
    def __call__(self, subject_handler: Optional[SubjectHandler] = None) -> Subject:  # pragma: no cover
        pass


class ConnectHandler(Protocol):
    """Connect Handler Protocol."""

    async def __call__(self) -> Subscription:  # pragma: no cover
        pass


class RefCountHandler(Protocol):
    """RefCount Handler Protocol."""

    async def __call__(self) -> Observable:  # pragma: no cover
        pass


class ConnectableObservable(Observable, Protocol):
    """Define a connectable observable protocol.

    We have :
        - subscribe function (it's an observable)
        - connect function: start executing
        - ref_count function: makes the Observable automatically start executing
            when the first subscriber arrives,
            and stop executing when the last subscriber leaves.
    """

    async def connect(self) -> Subscription:  # pragma: no cover
        """Connect."""
        pass

    async def ref_count(self) -> Observable:  # pragma: no cover
        """Reference counter.

        Make the multicasted Observable automatically start executing when
        the first subscriber arrives,
        and stop executing when the last subscriber leaves.
        """
        pass


class ConnectableObservableEventHandler(Protocol):
    """Connectable Observable Event Handler Protocol."""

    async def __call__(self) -> None:  # pragma: no cover
        pass


class ConnectableObservableHandler(Protocol):
    """Connectable Observable Handler Protocol.

    This handler could be called on conect/disconnect event.
    """

    async def on_connect(self) -> None:  # pragma: no cover
        """Called on connect event."""
        pass

    async def on_disconnect(self) -> None:  # pragma: no cover
        """Called on disconnect event."""
        pass


class _AsyncAccumulatorOperator(Protocol[T]):
    """Async Accumulator Operator Protocol.

    Accumulator are used in reduce operation.
    """

    async def __call__(self, buffer: T, item: T) -> T:  # pragma: no cover
        pass


class _SyncAccumulatorOperator(Protocol[T]):
    """Async Accumulator Operator Protocol.

    Accumulator are used in reduce operation.
    """

    def __call__(self, buffer: T, item: T) -> T:  # pragma: no cover
        pass


AccumulatorOperator = Union[_AsyncAccumulatorOperator, _SyncAccumulatorOperator]
"""Accumulator Operator Protocol.

Accumulator are used in reduce operation.
"""


class _AsyncPredicateOperator(Protocol):
    """Async Predicate Operator Protocol.

    Predicate are used in filter operation.
    """

    async def __call__(self, item: Any) -> bool:  # pragma: no cover
        pass


class _SyncPredicateOperator(Protocol):
    """Sync Predicate Operator Protocol.

    Predicate are used in filter operation.
    """

    def __call__(self, item: Any) -> bool:  # pragma: no cover
        pass


PredicateOperator = Union[_AsyncPredicateOperator, _SyncPredicateOperator]
"""Predicate Operator Protocol.

Predicate are used in filter operation.
"""