from typing import Any, Callable, Optional, TypeVar

from reactivex import Observable, abc
from reactivex import operators as ops
from reactivex import typing
from reactivex.internal.exceptions import SequenceContainsNoElementsError

_T = TypeVar("_T")


def last_or_default_async(
    source: Observable[_T],
    has_default: bool = False,
    default_value: Optional[_T] = None,
) -> Observable[Optional[_T]]:
    def subscribe(
        observer: abc.ObserverBase[Optional[_T]],
        scheduler: Optional[abc.SchedulerBase] = None,
    ):
        value = [default_value]
        seen_value = [False]

        def on_next(x: _T) -> None:
            value[0] = x
            seen_value[0] = True

        def on_completed():
            if not seen_value[0] and not has_default:
                observer.on_error(SequenceContainsNoElementsError())
            else:
                observer.on_next(value[0])
                observer.on_completed()

        return source.subscribe(
            on_next, observer.on_error, on_completed, scheduler=scheduler
        )

    return Observable(subscribe)


def last_or_default(
    default_value: Optional[_T] = None, predicate: Optional[typing.Predicate[_T]] = None
) -> Callable[[Observable[_T]], Observable[Any]]:
    def last_or_default(source: Observable[Any]) -> Observable[Any]:
        """Return last or default element.

        Examples:
            >>> res = _last_or_default(source)

        Args:
            source: Observable sequence to get the last item from.

        Returns:
            Observable sequence containing the last element in the
            observable sequence.
        """

        if predicate:
            return source.pipe(
                ops.filter(predicate),
                ops.last_or_default(default_value),
            )

        return last_or_default_async(source, True, default_value)

    return last_or_default


__all__ = ["last_or_default", "last_or_default_async"]