from __future__ import annotations

import asyncio
import logging
import signal
from typing import Any

logger = logging.getLogger(__name__)


async def wait_for_signals() -> None:
    """Wait for sigint or sigterm by setting global signal handlers"""
    signals = (signal.SIGINT, signal.SIGTERM)
    loop = asyncio.get_running_loop()
    event = asyncio.Event()

    old_handlers: dict[int, Any] = {}
    caught_signal: int | None = None

    def handle_signal(signum, frame):
        # *** Do not log or print anything in here
        # https://stackoverflow.com/questions/45680378/how-to-explain-the-reentrant-runtimeerror-caused-by-printing-in-signal-handlers
        nonlocal caught_signal
        caught_signal = signum
        # Restore old signal handler to allow for quicker exit
        # if the user sends the signal again.
        signal.signal(signum, old_handlers[signum])
        loop.call_soon_threadsafe(event.set)

    for sig in signals:
        old_handlers[sig] = signal.signal(sig, handle_signal)

    try:
        await event.wait()
        assert caught_signal
        logger.info(
            "Received signal %s (%d)", signal.Signals(caught_signal).name, caught_signal
        )
    finally:
        for sig in signals:
            signal.signal(sig, old_handlers[sig])