async_rx.subject.rx_subject_from.rx_subject_from()

Complexity

Conditions

5

Size

Total Lines	25
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

from typing import Optional
def rx_subject_from(
    a_subject: Subject,
    subscribe: Optional[Subscribe] = None,
    on_next: Optional[NextHandler] = None,
    on_error: Optional[ErrorHandler] = None,
    on_completed: Optional[CompleteHandler] = None,
) -> Subject:
    """Build a subject from another one by override some function.

    Args:
        a_subject (Subject): the source subject
        subscribe (Optional[Subscribe]): override subscribe if set
        on_next (Optional[NextHandler]): override on_next if set
        on_error (Optional[ErrorHandler]): override on_error if set
        on_completed (Optional[CompleteHandler]): override on_completed if set

    Returns;
        (Subject): a new subject

    """
    return subject(
        subscribe=subscribe if subscribe else a_subject.subscribe,
        on_next=on_next if on_next else a_subject.on_next,
        on_error=on_error if on_error else a_subject.on_error,
        on_completed=on_completed if on_completed else a_subject.on_completed,
    )