Source code for trader.daemon.sleeper

"""Sleeping, as an injectable seam.

The daemon spends almost all of its life asleep, which would make its tests
either slow or untrustworthy. Taking `sleep` and `monotonic` as arguments makes
them instant and exact instead.
"""

import time
from collections.abc import Callable
from typing import Protocol

from trader.daemon.signals import ShutdownFlag

__all__ = ["RealSleeper", "Sleeper"]




class Sleeper(Protocol):
    """Waits, but gives up early when shutdown is requested."""



    def sleep(self, seconds: float, shutdown: ShutdownFlag) -> None:
        """Sleep up to `seconds`, returning early if `shutdown` is set."""
        ...






class RealSleeper:
    """Sleeps in short slices so a signal is noticed promptly.

    One `time.sleep(3600)` would be interrupted by a signal on POSIX, but the
    handler only sets a flag — the sleep resumes for the remaining hour unless
    it is broken up. Slicing bounds the shutdown delay at `slice_seconds`
    regardless of how long the requested wait was.
    """

    def __init__(
        self,
        *,
        slice_seconds: float = 1.0,
        sleep: Callable[[float], None] = time.sleep,
        monotonic: Callable[[], float] = time.monotonic,
    ) -> None:
        self._slice_seconds = slice_seconds
        self._sleep = sleep
        self._monotonic = monotonic



    def sleep(self, seconds: float, shutdown: ShutdownFlag) -> None:
        """Sleep up to `seconds`, in slices, stopping early on shutdown."""
        if seconds <= 0:
            return

        deadline = self._monotonic() + seconds
        while not shutdown.is_set():
            remaining = deadline - self._monotonic()
            if remaining <= 0:
                return
            self._sleep(min(self._slice_seconds, remaining))