"""
This is an example which reduces the problem to the difference in what has
been observed to work, and what does not.

No application code is included here, and this is not a minimal reproducable example.
"""
import asyncio
import os
import signal
import sys

from foo import Application


def using_asyncio_run():
    app_instance = Application()
    exit_code = 0
    try:
        asyncio.run(app_instance.run())
    except SystemExit as exc:
        exit_code = exc.code
    except KeyboardInterrupt:
        exit_code = 1
    finally:
        sys.exit(exit_code)
    # This will crash with a RuntimeError when the application exits,
    # and the exit code will not match expecations.


# Below is manual handling which works

async def shutdown_handler():
    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for task in pending:
        task.cancel()
    await asyncio.gather(*pending, return_exceptions=True)


def manually_handling_event_loop():
    loop = asyncio.new_event_loop()
    app_instance = Application()

    exit_code = 0

    if os.name != "nt":
        signals = (signal.SIGHUP, signal.SIGTERM, signal.SIGINT)
        for sig in signals:
            loop.add_signal_handler(
                sig, lambda s: asyncio.create_task(shutdown_handler())
            )

    fut = loop.create_task(app_instance.run())
    # Normally, I'd attach a global uncaught error handler here
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        exit_code = 1
    except SystemExit as exc:
        exit_code = exc.code
    finally:
        loop.run_until_complete(shutdown_handler())
        loop.run_until_complete(loop.shutdown_asyncgens())
        if os.name == "nt":
            # Removing the below line also exhibits the crash
            # 3 seconds isn't a consistent amount
            # In all testing so far, 1 was sufficient
            loop.run_until_complete(asyncio.sleep(3))
        loop.stop()
        loop.close()
        sys.exit(exit_code)