FiberMapRuntime

__init__(fiber_map, runtime) -> None

def __init__(self, fiber_map: FiberMap[K, A, E], runtime: CapturedRuntime) -> None:
    self._map = fiber_map
    self._runtime = runtime

__call__(key, eff, /, *, only_if_missing=False, propagate_interruption=False, uninterruptible=False, scheduler=None) -> Fiber[A, E]

def __call__(
    self,
    key: K,
    eff: Effect[A, E, R],
    /,
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
    uninterruptible: bool = False,
    scheduler: Scheduler | None = None,
) -> Fiber[A, E]:
    """Fork ``eff`` on a root fiber under ``key``. A closed map, or a
    taken key with ``only_if_missing``, gives back an interrupted fiber
    that never runs ``eff``."""
    require_effect(eff, "eff")
    fiber_map = self._map
    if fiber_map._state is None or (only_if_missing and fiber_map.unsafe_has(key)):
        return self._runtime.interrupted()
    fiber = self._runtime.fork(eff, uninterruptible=uninterruptible, scheduler=scheduler)
    fiber_map.unsafe_set(
        key,
        fiber,
        only_if_missing=only_if_missing,
        propagate_interruption=propagate_interruption,
    )
    return fiber