Latch

__init__(open=False) -> None

def __init__(self, open: bool = False) -> None:
    self._is_open = open
    self._waiters: list[_Resume] = []
    # Waiters a scheduled flush will resume; None when no flush is pending.
    self._scheduled: list[_Resume] | None = None

make(open=False) -> Effect[Latch]

@classmethod
def make(cls, open: bool = False) -> Effect[Latch]:
    return _Sync(lambda: cls(open))

is_open() -> bool

def is_open(self) -> bool:
    return self._is_open

open() -> Effect[bool]

def open(self) -> Effect[bool]:
    """Open the latch, letting every waiter through. ``False`` if it was
    already open."""

    def run(fiber: Fiber[Any, Any]) -> Effect[bool]:
        if self._is_open:
            return _FALSE
        self._is_open = True
        return self._schedule_flush(fiber)

    return _WithFiber(run)

release() -> Effect[bool]

def release(self) -> Effect[bool]:
    """Let the current waiters through, leaving the latch closed.
    ``False`` if it is open."""
    return _WithFiber(lambda fiber: _FALSE if self._is_open else self._schedule_flush(fiber))

close() -> Effect[bool]

def close(self) -> Effect[bool]:
    """Close the latch. ``False`` if it was already closed."""
    return _Sync(self.unsafe_close)

wait() -> Effect[None]

def wait(self) -> Effect[None]:
    """Wait until the latch is open or released (Effect's ``await``)."""
    return _Callback(self._register)

when_open(eff) -> Effect[A, E, R]

def when_open[A, E, R](self, eff: Effect[A, E, R]) -> Effect[A, E, R]:
    """Run ``eff`` once the latch lets this fiber through."""
    require_effect(eff, "eff")
    return self.wait().and_then(eff)

unsafe_open() -> bool

def unsafe_open(self) -> bool:
    """Open the latch from outside an effect, resuming waiters at once.
    Call on the fibers' scheduler thread."""
    if self._is_open:
        return False
    self._is_open = True
    # Take both lists before resuming anyone: a waiter that closes the
    # latch again and registers new waiters must not be drained here.
    waiters, self._waiters = self._waiters, []
    self._flush_scheduled()
    for resume in waiters:
        resume(_VOID)
    return True

unsafe_close() -> bool

def unsafe_close(self) -> bool:
    if not self._is_open:
        return False
    self._is_open = False
    return True