import asyncio
import errno
import signal
import socket
import sys
import typing
import warnings

from ._log import logger


def _sighandler_noop(signum, frame):
    """Dummy signal handler."""
    pass


class SignalMixin:
    def __init__(self):
        self._signal_handlers = {}
        self._ssock = self._csock = None

        self._make_self_pipe()

    def close(self):
        if self._signal_handlers:
            if sys.is_finalizing():
                warnings.warn(
                    f"Closing the loop {self!r} "
                    f"on interpreter shutdown "
                    f"stage, skipping signal handlers removal",
                    ResourceWarning,
                    source=self,
                )
                self._signal_handlers.clear()

            else:
                for signum in list(self._signal_handlers):
                    self.remove_signal_handler(signum)

        self._remove_reader(self._ssock.fileno())
        self._ssock.close()
        self._ssock = None

        self._csock.close()
        self._csock = None

    def _make_self_pipe(self):
        self._ssock, self._csock = socket.socketpair()
        self._ssock.setblocking(False)
        self._csock.setblocking(False)

        self.add_reader(self._ssock.fileno(), self._read_from_self)

    def _read_from_self(self):
        # The loop below would be clearer as a "while True:"
        # loop with "break" statements, but that confuses
        # coverage.py (likely due to the generated byte code).
        running = True
        while running:
            try:
                if self._ssock is None:
                    running = False

                else:
                    data = self._ssock.recv(128)
                    if not data:
                        running = False

                    else:
                        self._process_self_data(data)
            except InterruptedError:
                continue
            except BlockingIOError:
                running = False

    def _process_self_data(self, data):
        for signum in data:
            handle = self._signal_handlers.get(signum)
            if handle is not None:
                if handle._cancelled:
                    self.remove_signal_handler(signum)
                else:
                    self._add_callback(handle)

    def add_signal_handler(
        self, sig: int, callback: typing.Callable[..., typing.Any], *args: typing.Any
    ):
        """Add a handler for a signal.  UNIX only.

        Raise ValueError if the signal number is invalid or uncatchable.
        Raise RuntimeError if there is a problem setting up the handler.
        """
        if asyncio.iscoroutine(callback) or asyncio.iscoroutinefunction(callback):
            raise TypeError("coroutines cannot be used " "with add_signal_handler()")
        if not callable(callback):
            raise TypeError(f"{callback!r} is not callable")

        self._check_signal(sig)
        self._check_closed()
        try:
            # set_wakeup_fd() raises ValueError if this is not the
            # main thread.  By calling it early we ensure that an
            # event loop running in another thread cannot add a signal
            # handler.
            signal.set_wakeup_fd(self._csock.fileno())
        except (ValueError, OSError) as exc:
            raise RuntimeError(str(exc))

        handle = asyncio.Handle(callback, args, self, None)
        self._signal_handlers[sig] = handle

        try:
            # Register a dummy signal handler to ask Python to write the signal
            # number in the wakup file descriptor. _process_self_data() will
            # read signal numbers from this file descriptor to handle signals.
            signal.signal(sig, _sighandler_noop)

            # Set SA_RESTART to limit EINTR occurrences.
            signal.siginterrupt(sig, False)
        except OSError as exc:
            del self._signal_handlers[sig]
            if not self._signal_handlers:
                try:
                    signal.set_wakeup_fd(-1)
                except (ValueError, OSError) as nexc:
                    logger.info("set_wakeup_fd(-1) failed: %s", nexc)

            if exc.errno == errno.EINVAL:
                raise RuntimeError(f"sig {sig} cannot be caught")
            else:  # pragma: nocover
                raise

    def remove_signal_handler(self, sig: int):
        """Remove a handler for a signal.  UNIX only.

        Return True if a signal handler was removed, False if not.
        """
        self._check_signal(sig)
        try:
            del self._signal_handlers[sig]
        except KeyError:
            return False

        if sig == signal.SIGINT:
            handler = signal.default_int_handler
        else:
            handler = signal.SIG_DFL

        try:
            signal.signal(sig, handler)
        except OSError as exc:
            if exc.errno == errno.EINVAL:
                raise RuntimeError(f"sig {sig} cannot be caught")
            else:  # pragma: nocover
                raise

        if not self._signal_handlers:
            try:
                signal.set_wakeup_fd(-1)
            except (ValueError, OSError) as exc:
                logger.info("set_wakeup_fd(-1) failed: %s", exc)

        return True

    if sys.version_info[:2] >= (3, 8):

        def _check_signal(self, sig: int):
            """Internal helper to validate a signal.

            Raise ValueError if the signal number is invalid or uncatchable.
            Raise RuntimeError if there is a problem setting up the handler.
            """
            if not isinstance(sig, int):
                raise TypeError(f"sig must be an int, not {sig!r}")

            if sig not in signal.valid_signals():
                raise ValueError(f"invalid signal number {sig}")

    else:  # pragma: nocover

        def _check_signal(self, sig: int):
            """Internal helper to validate a signal.

            Raise ValueError if the signal number is invalid or uncatchable.
            Raise RuntimeError if there is a problem setting up the handler.
            """
            if not isinstance(sig, int):
                raise TypeError(f"sig must be an int, not {sig!r}")

            if not (1 <= sig < signal.NSIG):
                raise ValueError(f"sig {sig} out of range(1, {signal.NSIG})")
