GitHub

LiveClock

PreviousNext

API reference for LiveClock.

The system clock. A sleep of zero or less yields once; an infinite sleep never wakes and holds no timer (as Effect's).

Functions

unsafe_now

unsafe_now() -> float

Returns float

def unsafe_now(self) -> float:
    return time.time_ns() / _NANOS

unsafe_monotonic

unsafe_monotonic() -> float

Returns float

def unsafe_monotonic(self) -> float:
    return time.monotonic_ns() / _NANOS

unsafe_now_nanos

unsafe_now_nanos() -> int

Returns int

def unsafe_now_nanos(self) -> int:
    return time.time_ns()

unsafe_monotonic_nanos

unsafe_monotonic_nanos() -> int

Returns int

def unsafe_monotonic_nanos(self) -> int:
    return time.monotonic_ns()

sleep

sleep(seconds) -> Effect[None]
ParameterTypeDescription
secondsSeconds

Returns Effect[None]

def sleep(self, seconds: Seconds) -> Effect[None]:
    if seconds <= 0:
        return _YieldNow()
    if math.isinf(seconds):
        return never

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        def register(resume: Callable[[Effect[None]], None]) -> Effect[None]:
            cancel = fiber._scheduler.call_later(seconds, lambda: resume(_Succeed(None)))
            return _Sync(cancel)

        return _Callback(register)

    return _WithFiber(body)