Skip to content

tapmap.lifecycle

tapmap.lifecycle

Application lifecycle coordination: shutdown signaling and the server thread.

request_shutdown() is the single entry point every shutdown trigger (signal, UI action, or the server thread exiting unexpectedly) calls; only the main thread, via wait_for_shutdown() returning, ever acts on it.

logger = logging.getLogger(__name__) module-attribute

_WAIT_POLL_S = 0.5 module-attribute

LifecycleCoordinator

Coordinate shutdown requests across threads.

Source code in src/tapmap/lifecycle.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class LifecycleCoordinator:
    """Coordinate shutdown requests across threads."""

    def __init__(self) -> None:
        self._shutdown_event = threading.Event()
        self._icon: Icon | None = None

    def set_tray_icon(self, icon: Icon) -> None:
        """Register the running tray icon so request_shutdown() can also stop it."""
        self._icon = icon

    def request_shutdown(self) -> None:
        """Signal that the application should shut down. Safe from any thread, idempotent."""
        self._shutdown_event.set()
        if self._icon is not None:
            self._icon.stop()

    def run_tray(self, icon: Icon, *, on_ready: Callable[[], None] | None = None) -> None:
        """Run icon's blocking loop, honoring a shutdown already requested before it started.

        pystray's Icon.stop() has no effect until the icon is actually
        running, so a shutdown requested between set_tray_icon() and this
        call (e.g. the server thread dying immediately) would otherwise be
        lost, and icon.run() would then block forever. pystray only invokes
        the setup callback once it has marked the icon running, so
        re-checking and re-stopping there closes that window.

        If provided, on_ready runs from pystray's setup callback after the
        icon has been marked ready.
        """

        def _setup(icon: Icon) -> None:
            icon.visible = True
            if self._shutdown_event.is_set():
                icon.stop()
                return
            if on_ready is not None:
                try:
                    on_ready()
                except Exception:
                    logger.exception("Unable to run the tray-ready callback.")

        timer_id = _start_windows_message_loop_nudge()
        try:
            icon.run(setup=_setup)
        finally:
            _stop_windows_message_loop_nudge(timer_id)

    def wait_for_shutdown(self) -> None:
        """Block the calling thread until shutdown has been requested."""
        # Poll instead of blocking indefinitely: on Windows, an unbounded wait
        # never returns to the interpreter loop, so a pending Ctrl+C handler
        # never runs.
        while not self._shutdown_event.wait(timeout=_WAIT_POLL_S):
            pass

    def install_signal_handlers(self) -> None:
        """Register SIGINT/SIGTERM handlers that request shutdown."""

        def _handle_signal(signum: int, frame: FrameType | None) -> None:
            self.request_shutdown()

        signal.signal(signal.SIGINT, _handle_signal)
        signal.signal(signal.SIGTERM, _handle_signal)

set_tray_icon(icon)

Register the running tray icon so request_shutdown() can also stop it.

Source code in src/tapmap/lifecycle.py 
36
37
38
def set_tray_icon(self, icon: Icon) -> None:
    """Register the running tray icon so request_shutdown() can also stop it."""
    self._icon = icon

request_shutdown()

Signal that the application should shut down. Safe from any thread, idempotent.

Source code in src/tapmap/lifecycle.py 
40
41
42
43
44
def request_shutdown(self) -> None:
    """Signal that the application should shut down. Safe from any thread, idempotent."""
    self._shutdown_event.set()
    if self._icon is not None:
        self._icon.stop()

run_tray(icon, *, on_ready=None)

Run icon's blocking loop, honoring a shutdown already requested before it started.

pystray's Icon.stop() has no effect until the icon is actually running, so a shutdown requested between set_tray_icon() and this call (e.g. the server thread dying immediately) would otherwise be lost, and icon.run() would then block forever. pystray only invokes the setup callback once it has marked the icon running, so re-checking and re-stopping there closes that window.

If provided, on_ready runs from pystray's setup callback after the icon has been marked ready.

Source code in src/tapmap/lifecycle.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def run_tray(self, icon: Icon, *, on_ready: Callable[[], None] | None = None) -> None:
    """Run icon's blocking loop, honoring a shutdown already requested before it started.

    pystray's Icon.stop() has no effect until the icon is actually
    running, so a shutdown requested between set_tray_icon() and this
    call (e.g. the server thread dying immediately) would otherwise be
    lost, and icon.run() would then block forever. pystray only invokes
    the setup callback once it has marked the icon running, so
    re-checking and re-stopping there closes that window.

    If provided, on_ready runs from pystray's setup callback after the
    icon has been marked ready.
    """

    def _setup(icon: Icon) -> None:
        icon.visible = True
        if self._shutdown_event.is_set():
            icon.stop()
            return
        if on_ready is not None:
            try:
                on_ready()
            except Exception:
                logger.exception("Unable to run the tray-ready callback.")

    timer_id = _start_windows_message_loop_nudge()
    try:
        icon.run(setup=_setup)
    finally:
        _stop_windows_message_loop_nudge(timer_id)

wait_for_shutdown()

Block the calling thread until shutdown has been requested.

Source code in src/tapmap/lifecycle.py 
77
78
79
80
81
82
83
def wait_for_shutdown(self) -> None:
    """Block the calling thread until shutdown has been requested."""
    # Poll instead of blocking indefinitely: on Windows, an unbounded wait
    # never returns to the interpreter loop, so a pending Ctrl+C handler
    # never runs.
    while not self._shutdown_event.wait(timeout=_WAIT_POLL_S):
        pass

install_signal_handlers()

Register SIGINT/SIGTERM handlers that request shutdown.

Source code in src/tapmap/lifecycle.py 
85
86
87
88
89
90
91
92
def install_signal_handlers(self) -> None:
    """Register SIGINT/SIGTERM handlers that request shutdown."""

    def _handle_signal(signum: int, frame: FrameType | None) -> None:
        self.request_shutdown()

    signal.signal(signal.SIGINT, _handle_signal)
    signal.signal(signal.SIGTERM, _handle_signal)

_start_windows_message_loop_nudge()

Start a periodic Windows timer that wakes pystray's blocking message loop, or None.

Source code in src/tapmap/lifecycle.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def _start_windows_message_loop_nudge() -> int | None:
    """Start a periodic Windows timer that wakes pystray's blocking message loop, or None."""
    # pystray's Windows backend blocks in a plain GetMessage() call with no
    # timeout, which only returns when a real window message arrives - so a
    # pending Ctrl+C/SIGTERM sits undispatched until something unrelated
    # happens to wake it. A NULL-window timer posts WM_TIMER straight to the
    # calling (main) thread's own queue at a fixed interval, which
    # GetMessage() picks up like any other message - no pystray internals needed.
    if platform.system() != "Windows":
        return None

    timer_id = ctypes.windll.user32.SetTimer(0, 0, int(_WAIT_POLL_S * 1000), None)
    if timer_id == 0:
        logger.warning("Unable to create the Windows Ctrl+C responsiveness timer.")
        return None
    return timer_id

_stop_windows_message_loop_nudge(timer_id)

Remove the timer started by _start_windows_message_loop_nudge(), if one was created.

Source code in src/tapmap/lifecycle.py 
113
114
115
116
def _stop_windows_message_loop_nudge(timer_id: int | None) -> None:
    """Remove the timer started by _start_windows_message_loop_nudge(), if one was created."""
    if timer_id is not None:
        ctypes.windll.user32.KillTimer(0, timer_id)

start_server_thread(server, coordinator)

Run server.serve_forever() on a background thread; request shutdown when it exits.

Source code in src/tapmap/lifecycle.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def start_server_thread(
    server: BaseWSGIServer, coordinator: LifecycleCoordinator
) -> threading.Thread:
    """Run server.serve_forever() on a background thread; request shutdown when it exits."""

    def _serve() -> None:
        try:
            server.serve_forever()
        finally:
            coordinator.request_shutdown()

    # daemon=True is only a backstop for an unplanned main-thread failure -
    # the normal shutdown path always joins this thread before proceeding.
    thread = threading.Thread(target=_serve, daemon=True, name="tapmap-server")
    thread.start()
    return thread