"""Event loop implementation."""

from __future__ import annotations

import errno
import os
import socket
import threading
from contextlib import contextmanager, suppress
from copy import copy
from queue import Empty
from time import sleep
from types import GeneratorType as generator

from vine import Thenable, promise

from kombu.log import get_logger
from kombu.utils.compat import fileno
from kombu.utils.eventio import ERR, READ, WRITE, poll
from kombu.utils.objects import cached_property

from .timer import Timer

__all__ = ('Hub', 'get_event_loop', 'set_event_loop')
logger = get_logger(__name__)

_current_loop: Hub | None = None

W_UNKNOWN_EVENT = """\
Received unknown event %r for fd %r, please contact support!\
"""


class Stop(BaseException):
    """Stops the event loop."""


def _raise_stop_error():
    raise Stop()


@contextmanager
def _dummy_context(*args, **kwargs):
    yield


class _Waker:
    """Socket pair that lets another thread interrupt the poller.

    Sockets rather than a pipe, as ``select()`` on Windows only accepts
    sockets.
    """

    def __init__(self):
        self.reader, self.writer = socket.socketpair()
        self.reader.setblocking(False)
        self.writer.setblocking(False)
        self.pid = os.getpid()

    def fileno(self):
        return self.reader.fileno()

    def wake(self):
        # The hub can be closed already.
        with suppress(OSError):
            self.writer.send(b'\0')

    def drain(self):
        # One read is enough: call_soon() only sends while no wakeup
        # is pending.
        with suppress(OSError):
            self.reader.recv(4096)

    def close(self):
        self.reader.close()
        self.writer.close()


def get_event_loop() -> Hub | None:
    """Get current event loop object."""
    return _current_loop


def set_event_loop(loop: Hub | None) -> Hub | None:
    """Set the current event loop object."""
    global _current_loop
    _current_loop = loop
    return loop


class Hub:
    """Event loop object.

    Arguments:
    ---------
        timer (kombu.asynchronous.Timer): Specify custom timer instance.
    """

    #: Flag set if reading from an fd will not block.
    READ = READ

    #: Flag set if writing to an fd will not block.
    WRITE = WRITE

    #: Flag set on error, and the fd should be read from asap.
    ERR = ERR

    #: List of callbacks to be called when the loop is exiting,
    #: applied with the hub instance as sole argument.
    on_close = None

    def __init__(self, timer=None):
        self.timer = timer if timer is not None else Timer()

        self.readers = {}
        self.writers = {}
        self.on_tick = set()
        self.on_close = set()
        self._ready = set()
        self._ready_lock = threading.Lock()

        # Lets call_soon() from another thread interrupt the poller,
        # see _create_waker().
        self._waker = None
        self._wake_pending = False
        self._loop_thread = None

        self._running = False
        self._loop = None

        # The eventloop (in celery.worker.loops)
        # will merge fds in this set and then instead of calling
        # the callback for each ready fd it will call the
        # :attr:`consolidate_callback` with the list of ready_fds
        # as an argument.  This API is internal and is only
        # used by the multiprocessing pool to find inqueues
        # that are ready to write.
        self.consolidate = set()
        self.consolidate_callback = None

        self.propagate_errors = ()

        self._create_poller()

    @property
    def poller(self):
        if not self._poller:
            self._create_poller()
        return self._poller

    @poller.setter
    def poller(self, value):
        self._poller = value

    def reset(self):
        # Unlike close(), keeps the waker: closing it could race with
        # another thread's write.
        self._close()
        self._create_poller()

    def _create_poller(self):
        self._poller = poll()
        self._register_fd = self._poller.register
        self._unregister_fd = self._poller.unregister
        if self._waker is not None:
            self._register_fd(self._waker.fileno(), READ)

    def _create_waker(self):
        # Created by the loop, so that a hub that never runs holds no
        # sockets.  A forked child must not share the parent's.
        if self._waker is not None:
            self._unregister(self._waker.fileno())
            self._waker.close()
        self._waker = _Waker()
        with self._ready_lock:
            self._wake_pending = False
        self.poller.register(self._waker.fileno(), READ)
        return self._waker

    def _close_poller(self):
        if self._poller is not None:
            self._poller.close()
            self._poller = None
            self._register_fd = None
            self._unregister_fd = None

    def stop(self):
        self.call_soon(_raise_stop_error)

    def __repr__(self):
        return '<Hub@{:#x}: R:{} W:{}>'.format(
            id(self), len(self.readers), len(self.writers),
        )

    def fire_timers(self, min_delay=1, max_delay=10, max_timers=10,
                    propagate=()):
        timer = self.timer
        delay = None
        if timer and timer._queue:
            for i in range(max_timers):
                delay, entry = next(self.scheduler)
                if entry is None:
                    break
                try:
                    entry()
                except propagate:
                    raise
                except (MemoryError, AssertionError):
                    raise
                except OSError as exc:
                    if exc.errno == errno.ENOMEM:
                        raise
                    logger.error('Error in timer: %r', exc, exc_info=1)
                except Exception as exc:
                    logger.error('Error in timer: %r', exc, exc_info=1)
        return min(delay or min_delay, max_delay)

    def _remove_from_loop(self, fd):
        try:
            self._unregister(fd)
        finally:
            self._discard(fd)

    def add(self, fd, callback, flags, args=(), consolidate=False):
        fd = fileno(fd)
        try:
            self.poller.register(fd, flags)
        except ValueError:
            self._remove_from_loop(fd)
            raise
        else:
            dest = self.readers if flags & READ else self.writers
            if consolidate:
                self.consolidate.add(fd)
                dest[fd] = None
            else:
                dest[fd] = callback, args

    def remove(self, fd):
        fd = fileno(fd)
        self._remove_from_loop(fd)

    def run_forever(self):
        self._running = True
        try:
            while 1:
                try:
                    self.run_once()
                except Stop:
                    break
        finally:
            self._running = False

    def run_once(self):
        try:
            next(self.loop)
        except StopIteration:
            self._loop = None

    def call_soon(self, callback, *args):
        if not isinstance(callback, Thenable):
            callback = promise(callback, args)
        with self._ready_lock:
            self._ready.add(callback)
            # The loop thread checks _ready before polling again, and one
            # pending wakeup is enough for any number of callbacks.
            wake = (not self._wake_pending and
                    threading.get_ident() != self._loop_thread)
            if wake:
                self._wake_pending = True
        if wake:
            waker = self._waker
            if waker is not None:
                waker.wake()
        return callback

    def call_later(self, delay, callback, *args):
        return self.timer.call_after(delay, callback, args)

    def call_at(self, when, callback, *args):
        return self.timer.call_at(when, callback, args)

    def call_repeatedly(self, delay, callback, *args):
        return self.timer.call_repeatedly(delay, callback, args)

    def add_reader(self, fds, callback, *args):
        return self.add(fds, callback, READ | ERR, args)

    def add_writer(self, fds, callback, *args):
        return self.add(fds, callback, WRITE, args)

    def remove_reader(self, fd):
        writable = fd in self.writers
        on_write = self.writers.get(fd)
        try:
            self._remove_from_loop(fd)
        finally:
            if writable:
                cb, args = on_write
                self.add(fd, cb, WRITE, args)

    def remove_writer(self, fd):
        readable = fd in self.readers
        on_read = self.readers.get(fd)
        try:
            self._remove_from_loop(fd)
        finally:
            if readable:
                cb, args = on_read
                self.add(fd, cb, READ | ERR, args)

    def _unregister(self, fd):
        try:
            self.poller.unregister(fd)
        except (AttributeError, KeyError, OSError):
            pass

    def _pop_ready(self):
        with self._ready_lock:
            ready = self._ready
            self._ready = set()
            return ready

    def _on_wakeup(self, waker):
        # Drain first: clearing the flag first could let a wakeup sent
        # in between be drained here, leaving the flag set for good.
        waker.drain()
        with self._ready_lock:
            self._wake_pending = False

    def close(self, *args):
        self._close()
        if self._waker is not None:
            self._waker.close()
            self._waker = None

    def _close(self):
        [self._unregister(fd) for fd in self.readers]
        self.readers.clear()
        [self._unregister(fd) for fd in self.writers]
        self.writers.clear()
        self.consolidate.clear()
        self._close_poller()
        # A callback may remove itself from on_close, so iterate a copy.
        for callback in list(self.on_close):
            callback(self)

        # Complete remaining todo before Hub close
        # Eg: Acknowledge message
        # To avoid infinite loop where one of the callables adds items
        # to self._ready (via call_soon or otherwise).
        # we create new list with current self._ready
        todos = self._pop_ready()
        for item in todos:
            item()

        # Clear global event loop variable if this hub is the current loop
        if _current_loop is self:
            set_event_loop(None)

    def _discard(self, fd):
        fd = fileno(fd)
        self.readers.pop(fd, None)
        self.writers.pop(fd, None)
        self.consolidate.discard(fd)

    def on_callback_error(self, callback, exc):
        logger.error(
            'Callback %r raised exception: %r', callback, exc, exc_info=1,
        )

    def create_loop(self,
                    generator=generator, sleep=sleep, min=min, next=next,
                    Empty=Empty, StopIteration=StopIteration,
                    KeyError=KeyError, READ=READ, WRITE=WRITE, ERR=ERR):
        readers, writers = self.readers, self.writers
        poll = self.poller.poll
        waker = self._waker
        if waker is None or waker.pid != os.getpid():
            waker = self._create_waker()
        waker_fd = waker.fileno()
        fire_timers = self.fire_timers
        hub_remove = self.remove
        scheduled = self.timer._queue
        consolidate = self.consolidate
        consolidate_callback = self.consolidate_callback
        propagate = self.propagate_errors
        self._loop_thread = threading.get_ident()

        while 1:
            todo = self._pop_ready()

            for item in todo:
                if item:
                    item()

            poll_timeout = fire_timers(propagate=propagate) if scheduled else 1

            for tick_callback in copy(self.on_tick):
                tick_callback()

            #  print('[[[HUB]]]: %s' % (self.repr_active(),))
            if readers or writers:
                to_consolidate = []
                try:
                    events = poll(poll_timeout)
                    #  print('[EVENTS]: %s' % (self.repr_events(events),))
                except ValueError:  # Issue celery/#882
                    return

                for fd, event in events or ():
                    if fd == waker_fd:
                        self._on_wakeup(waker)
                        continue
                    general_error = False
                    if fd in consolidate and \
                            writers.get(fd) is None:
                        to_consolidate.append(fd)
                        continue
                    cb = cbargs = None

                    if event & READ:
                        try:
                            cb, cbargs = readers[fd]
                        except KeyError:
                            self.remove_reader(fd)
                            continue
                    elif event & WRITE:
                        try:
                            cb, cbargs = writers[fd]
                        except KeyError:
                            self.remove_writer(fd)
                            continue
                    elif event & ERR:
                        general_error = True
                    else:
                        logger.info(W_UNKNOWN_EVENT, event, fd)
                        general_error = True

                    if general_error:
                        try:
                            cb, cbargs = (readers.get(fd) or
                                          writers.get(fd))
                        except TypeError:
                            pass

                    if cb is None:
                        self.remove(fd)
                        continue

                    if isinstance(cb, generator):
                        try:
                            next(cb)
                        except OSError as exc:
                            if exc.errno != errno.EBADF:
                                raise
                            hub_remove(fd)
                        except StopIteration:
                            pass
                        except Exception:
                            hub_remove(fd)
                            raise
                    else:
                        try:
                            cb(*cbargs)
                        except Empty:
                            pass
                if to_consolidate:
                    consolidate_callback(to_consolidate)
            else:
                # no sockets yet, startup is probably not done.
                sleep(min(poll_timeout, 0.1))
            yield

    def repr_active(self):
        from .debug import repr_active
        return repr_active(self)

    def repr_events(self, events):
        from .debug import repr_events
        return repr_events(self, events or [])

    @cached_property
    def scheduler(self):
        return iter(self.timer)

    @property
    def loop(self):
        if self._loop is None:
            self._loop = self.create_loop()
        return self._loop