###############################################################################
# Copyright 2018 The AnPyLar Team. All Rights Reserved.
# Use of this source code is governed by an MIT-style license that
# can be found in the LICENSE file at http://anpylar.com/mit-license
###############################################################################
from . import config as aconfig
from .timer import call_soon, call_delayed, call_cancel

from .utils import defaultdict

__all__ = ['Observer', 'Observable', '_MetaObservable', 'Disposable',
           'ObservableOperator', '_MetaObservableOperator',
           'ObservableSource', '_MetaObservableSource',
           'ObservableFetchError', 'ObservableStopError']


class ObservableStopError(Exception):
    pass


class ObservableFetchError(Exception):
    def __init__(self, val, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.val = val


class Disposable:
    def __init__(self, **kwargs):
        self._parents = []

        for k, v in kwargs.items():
            setattr(self, k, v)

    def _add_parent(self, parent):
        self._parents.append(parent)

    def get_val(self):
        return self.val

    def get_sid(self):
        return self.sid

    def fetch_val(self):
        return self.val


class Observer:

    def on_next(self, value):
        '''Called with each value generated by the Observable'''
        pass

    def on_completed(self):
        '''Called when the Observable can no longer produce values'''
        pass

    def on_error(self, error):
        '''Called when an error has occurred in the chain'''
        pass


class Subscription:
    _timed_op = False
    _error = None

    def __init__(self, parent, on_next=None, on_completed=None, on_error=None):
        self._parent = parent
        self._on_next = on_next
        self._on_completed = on_completed
        self._on_error = on_error

    def _suboperate(self, val, sid):
        return val

    def _operate(self, val, sid):
        return val

    def on_next(self, val, sid):
        if not self._error and self._on_next:
            self._on_next(val)

    def on_completed(self, sid):
        if not self._error and self._on_completed:
            self._on_completed()

    def on_error(self, error, sid):
        self._error = error
        if self._on_error:
            self._on_error(error)
        else:
            if aconfig.observable.log_error is True:
                print('on_error:', self, '-', error)
            elif aconfig.observable.log_error:
                aconfig.observable.log_error(error)

            if aconfig.observable.raise_error:
                raise Exception(error)


class _MetaObservable(type):
    def __call__(cls, parent, *args, **kwargs):
        self = cls.__new__(cls, *args, **kwargs)  # create

        self._parent = parent
        self._subscriptions = defaultdict(list)
        self._in_next = defaultdict(int)
        self._to_complete = defaultdict(bool)
        self._to_unsubscribe = defaultdict(bool)
        self._timers = {}

        self._next = super(cls, self).on_next

        self.__init__(*args, **kwargs)
        return self

    def __getattr__(cls, name):
        if not name.startswith('__'):
            try:
                return cls._staticmethods[name]
            except KeyError:
                pass

        return super().__getattr__(name)


class Observable(object, metaclass=_MetaObservable):
    _staticmethods = {}
    _dynamicmethods = {}

    _streamid = 1
    _error = None

    # indicator than recursive on_next timer would happen with delay_next
    _timed_op = False

    def __getattr__(self, name):
        if not name.startswith('__'):
            try:
                cls = Observable._dynamicmethods[name]
                return lambda *a, **kw: cls(self, *a, **kw)
            except KeyError:
                pass

        return super().__getattr__(name)

    def _suboperate(self, val, sid):
        val = self._operate(val, sid)
        for s in self._subscriptions[sid]:
            val = s._suboperate(val, sid)
            break  # do only for the 1st

        return val

    def _operate(self, val, sid):
        return val

    def _delay_next(self, sid, fn, tout=0):
        if self._to_complete[sid] or self._to_unsubscribe[sid]:
            return 0

        def cb():
            self._in_next[sid] -= 1  # counter the effect of the +1 below
            fn()

        self._in_next[sid] += 1
        return call_delayed(tout, cb)

    def _delay_stop(self, sid, t):
        # cancelling the +1 from above, because cb above wasn't called
        self._in_next[sid] -= 1
        call_cancel(t)

    @staticmethod
    def _get_sid():
        Observable._streamid = sid = Observable._streamid + 1
        return sid

    def _substrigger(self, sid, **kwargs):
        if self._parent is not None:
            return self._parent._subscribe(self, sid, **kwargs)
        else:
            try:
                self._subscribed(sid, **kwargs)
            except ObservableFetchError as e:
                val = self._suboperate(e.val, sid)
                return Disposable(val=val, sid=sid)

            # no meaningful value can be given in the disposable
            return Disposable(sid=sid)

    def _subscribe(self, sub, sid, **kwargs):
        self._subscriptions[sid].append(sub)
        disp = self._substrigger(sid, **kwargs)
        disp._add_parent(self)
        return disp

    def subscribe(self, on_next, on_completed=None, on_error=None,
                  observer=None, fetch=False, **kwargs):

        if isinstance(on_next, Subscription):
            sub = on_next

        elif isinstance(on_next, ObservableSource):
            # The for only using it as on_next is that we are only setting the
            # value at the source. Else, there could be circular references
            sub = Subscription(self, on_next)

        elif isinstance(on_next, Observable):
            sub = on_next

        elif observer is not None:
            sub = observer
        else:
            if not on_completed:
                on_completed = getattr(on_next, 'on_completed', None)
            if not on_error:
                on_error = getattr(on_next, 'on_error', None)

            on_next = getattr(on_next, 'on_next', on_next)
            sub = Subscription(self, on_next, on_completed, on_error)

        return self._subscribe(sub, sid=self._get_sid(), fetch=fetch, **kwargs)

    def _unsubscribe(self, sid):
        if sid is None:
            for sid in self._subscriptions:
                call_soon(lambda sid=sid: self._unsubscribe(sid))
            return

        if self._in_next[sid] > 0 or self._to_complete[sid]:
            self._to_unsubscribe[sid] = True
            return

        self._to_unsubscribe[sid] = False

        if self._parent is not None:
            self._parent._unsubscribe(sid)

        # safe pop as it could have completed meanwhile
        self._subscriptions.pop(sid, None)
        self._unsubscribed(sid)

    def on_next(self, val, sid=None):
        if sid is None:  # multicast
            for sid in self._subscriptions:
                self._delay_next(sid, lambda s=sid: self._streams_next(val, s))

        elif self._timed_op:  # gone thru delay, don't delay again
            self._streams_next(val, sid)
        else:  # yield in case someone needs something
            self._delay_next(sid, lambda: self._streams_next(val, sid))

    def _streams_next(self, val, sid):
        # The rationale to catch here the exceptions for on_error is that there
        # is no need for individual logic in the on_next implementations,
        # because if an on_next fails, its partner on_error can be hier
        # directly called to complete the chain.
        # other subscriptions at this level are not affected, because the
        self._in_next[sid] += 1

        for s in self._subscriptions[sid]:
            try:
                s.on_next(val, sid)
            except Exception as e:
                s.on_error(e, sid)

        self._in_next[sid] -= 1
        if self._to_complete[sid]:
            call_soon(lambda: self._streams_completed(sid))
        elif self._to_unsubscribe[sid]:
            call_soon(lambda: self._unsubscribe(sid))

    def on_completed(self, sid):
        if not self._error:
            if not self._timed_op:
                call_soon(lambda s=sid: self._streams_completed(s))
            else:
                self._streams_completed(sid)

    def _streams_completed(self, sid):
        if sid is None:
            for sid in self._subscriptions:
                call_soon(lambda s=sid: self._streams_completed(s))
            return

        # A completion can arrive before "next" has completed, because next may
        # be waiting on a timer. Mark it. It will be recalled at the end of
        # streams_next
        if self._in_next[sid] > 0:
            self._to_complete[sid] = True
            return

        self._to_complete[sid] = False

        # if already here, a subclassed on_completed has not generated an
        # exception and it is safe to pop the subscribtions for the given sid
        for s in self._subscriptions.pop(sid):
            try:
                s.on_completed(sid)
            except Exception as e:
                s.on_error(e, sid)

        if self._to_unsubscribe[sid]:
            call_soon(lambda: self._unsubscribe(sid))

    def on_error(self, error, sid):
        if aconfig.observable.log_error_early:
            print('on_error:', self, '-', error)

        self._error = error
        if sid is None:
            for sid in self._subscriptions:
                call_soon(lambda s=sid: self._streams_error(error, s))

        else:
            call_soon(lambda: self._streams_error(error, sid))

    def _streams_error(self, error, sid):
        # Errors from above do not come out of band like it may happen with
        # completion. If an error has happened higher in the hierarchy, on_next
        # won't have been called or will have generated an exception, there is
        # no need to wait for it then.

        # if already here, there was an error during on_next or on_completed
        # and the subscriptions will still be there (see above in
        # streams_completed)
        for s in self._subscriptions.pop(sid):
            try:
                s.on_error(error, sid)
            except Exception as e:
                pass  # cannot recover from error notification

    def _subscribed(self, sid, **kwargs):
        pass

    def _unsubscribed(self, sid):
        pass

    def __bool__(self):
        return ObsOpSingle(self, lambda x: bool(x))

    def __eq__(self, other):
        return ObsOp(self, lambda x, y: x == y, other)

    def __ne__(self, other):
        return ObsOp(self, lambda x, y: x != y, other)

    def __gt__(self, other):
        return ObsOp(self, lambda x, y: x > y, other)

    def __ge__(self, other):
        return ObsOp(self, lambda x, y: x >= y, other)

    def __lt__(self, other):
        return ObsOp(self, lambda x, y: x < y, other)

    def __le__(self, other):
        return ObsOp(self, lambda x, y: x <= y, other)


class _MetaObservableOperator(_MetaObservable):
    def __init__(cls, name, bases, dct, **kwds):
        super().__init__(name, bases, dct, **kwds)

        if name.endswith('_Operator'):
            obsname = name[:-9].lower()
            Observable._dynamicmethods[obsname] = cls


class ObservableOperator(Observable, metaclass=_MetaObservableOperator):

    def on_next(self, val, sid):
        try:
            val = self._operate(val, sid)
        except ObservableStopError:
            return

        super().on_next(val, sid)


class ObsOpSingle(ObservableOperator):

    def __init__(self, op):
        self._op = op
        self._val = None
        self._subscribing = False

    def subscribe(self, on_next, on_completed=None, on_error=None,
                  observer=None, fetch=False, **kwargs):

        self._subscribing = True
        disp1 = super().subscribe(on_next, on_completed, on_error,
                                  observer, fetch, **kwargs)

        sid = disp1.sid
        self._subscribing = False

        if fetch:
            try:
                v1 = disp1.val
            except AttributeError:  # one of the vals is not available
                pass
            else:
                self._val = disp1.val = v1

        return disp1

    def _operate(self, val, sid, ix=0):
        # during subscription we don't want to operate on the value because the
        # result will be sent as the "fetched value" and at this stage, one
        # value will still be 'NaN'
        if self._subscribing:
            return val

        self._val = val
        return self._op(val)


class ObsOp(ObservableOperator):

    def __init__(self, op, other):
        self._op = op
        self._other = other
        self._vals = defaultdict(lambda: [float('NaN'), float('NaN')])
        self._subscribing = False

    def subscribe(self, on_next, on_completed=None, on_error=None,
                  observer=None, fetch=False, **kwargs):

        self._subscribing = True
        disp1 = super().subscribe(on_next, on_completed, on_error,
                                  observer, fetch, **kwargs)

        sid = disp1.sid
        other = self._other
        if isinstance(other, Observable):
            disp2 = other.subscribe(
                lambda v, s: self._operate(v, sid, ix=1),
                fetch=fetch, **kwargs)

        else:
            disp2 = Disposable(val=other)

        self._subscribing = False

        if fetch:
            try:
                v1, v2 = disp1.val, disp2.val
            except AttributeError:  # one of the vals is not available
                pass
            else:
                self._vals[sid] = [v1, v2]
                disp1.val = self._op(v1, v2)

        return disp1

    def _operate(self, val, sid, ix=0):
        # during subscription we don't want to operate on the value because the
        # result will be sent as the "fetched value" and at this stage, one
        # value will still be 'NaN'
        if self._subscribing:
            return val

        vals = self._vals[sid]
        vals[ix] = val
        return self._op(*vals)


class _MetaObservableSource(_MetaObservable):
    def __init__(cls, name, bases, dct, **kwds):
        super().__init__(name, bases, dct, **kwds)

        if name.endswith('_Source'):
            obsname = name[:-7].lower()
            Observable._staticmethods[obsname] = (
                lambda *a, **kw: cls(*a, **kw)
            )

    def __call__(cls, *args, **kwargs):
        return super().__call__(None, *args, **kwargs)  # create


class ObservableSource(Observable, metaclass=_MetaObservableSource):
    pass
