from typing import Any, Callable, Optional, TypeVar

from reactivex import Observable, abc
from reactivex.scheduler import CurrentThreadScheduler

_T = TypeVar("_T")


def return_value_(
    value: _T, scheduler: Optional[abc.SchedulerBase] = None
) -> Observable[_T]:
    """Returns an observable sequence that contains a single element,
    using the specified scheduler to send out observer messages.
    There is an alias called 'just'.

    Examples:
        >>> res = return(42)
        >>> res = return(42, rx.Scheduler.timeout)

    Args:
        value: Single element in the resulting observable sequence.

    Returns:
        An observable sequence containing the single specified
        element.
    """

    def subscribe(
        observer: abc.ObserverBase[_T], scheduler_: Optional[abc.SchedulerBase] = None
    ) -> abc.DisposableBase:
        _scheduler = scheduler or scheduler_ or CurrentThreadScheduler.singleton()

        def action(scheduler: abc.SchedulerBase, state: Any = None) -> None:
            observer.on_next(value)
            observer.on_completed()

        return _scheduler.schedule(action)

    return Observable(subscribe)


def from_callable_(
    supplier: Callable[[], _T], scheduler: Optional[abc.SchedulerBase] = None
) -> Observable[_T]:
    def subscribe(
        observer: abc.ObserverBase[_T], scheduler_: Optional[abc.SchedulerBase] = None
    ) -> abc.DisposableBase:
        _scheduler = scheduler or scheduler_ or CurrentThreadScheduler.singleton()

        def action(_: abc.SchedulerBase, __: Any = None) -> None:
            nonlocal observer

            try:
                observer.on_next(supplier())
                observer.on_completed()
            except Exception as e:  # pylint: disable=broad-except
                observer.on_error(e)

        return _scheduler.schedule(action)

    return Observable(subscribe)


__all__ = ["return_value_", "from_callable_"]