async_rx.observable.rx_forward.rx_forward()

Complexity

Conditions

3

Size

Total Lines	30
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from typing import Any
def rx_forward(observable: Observable, except_complet: bool = False, except_error: bool = False) -> Observable:
    """Create an observable wich forward event.

    Args:
        observable (Observable): observable source
        except_complet (bool): if true then did not forward 'on_complet' (default: {False})
        except_error (bool): if true then did not forward 'on_error' (default: {False})

    Returns:
        (Observable): observable instance.

    """

    async def _dummy_on_completed():
        pass

    async def _dummy_on_error(err: Any):
        pass

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

        return await observable.subscribe(
            an_observer=rx_observer(
                on_next=an_observer.on_next,
                on_error=_dummy_on_error if except_error else an_observer.on_error,
                on_completed=_dummy_on_completed if except_complet else an_observer.on_completed,
            )
        )

    return rx_create(subscribe=_subscribe)