async_rx.protocol.subscription

Complexity

Total Complexity

4

Size/Duplication

Total Lines	42
Duplicated Lines	0 %

Importance

Changes

0

2 Functions

Rating	Name	Duplication	Size	Complexity
A	disposable_subscription_on_cancel()	0	19	3
A	default_subscription()	0	10	1

"""Subscription utilities."""
from typing import Optional

import curio

from .definition import Observable, Observer, Subscription

__all__ = ["default_subscription", "disposable_subscription_on_cancel"]


async def default_subscription() -> None:
    """Default subcribe implementation method.

    Do nothing.

    Returns:
        (None) - nothing to return.

    """
    pass


async def disposable_subscription_on_cancel(an_observable: Observable, an_observer: Observer) -> Optional[Subscription]:
    """Subscribe implementation wich dispose herself on cancel.

    Arguments:
        an_observable (Observable): observable
        an_observer (Observer): observer

    Returns:
        (Optional[Subscription]): the subscription if task is not cancelled

    """
    _subscription: Optional[Subscription] = None
    try:
        _subscription = await an_observable.subscribe(an_observer=an_observer)
    except curio.CancelledError:  # pragma: no cover
        if _subscription:
            await _subscription()
            _subscription = None
    return _subscription