GitHub

FiberHandleRuntime

API reference for FiberHandleRuntime.

Forks effects into a FiberHandle from plain code; see FiberHandle.runtime.

Functions

__init__

__init__(handle, runtime) -> None
ParameterTypeDescription
handleFiberHandle[A, E]
runtimeCapturedRuntime
def __init__(self, handle: FiberHandle[A, E], runtime: CapturedRuntime) -> None:
    self._handle = handle
    self._runtime = runtime

__call__

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

Fork eff on a root fiber and make it current. A closed handle, or a current fiber with only_if_missing, gives back an interrupted fiber that never runs eff.

ParameterTypeDescription
effEffect[A, E, R]
only_if_missingbool = False
propagate_interruptionbool = False
uninterruptiblebool = False
schedulerScheduler | None = None

Returns Fiber[A, E]

def __call__(
    self,
    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 and make it current. A closed
    handle, or a current fiber with ``only_if_missing``, gives back an
    interrupted fiber that never runs ``eff``."""
    require_effect(eff, "eff")
    handle = self._handle
    if handle._state is None or (only_if_missing and handle._state.current is not None):
        return self._runtime.interrupted()
    fiber = self._runtime.fork(eff, uninterruptible=uninterruptible, scheduler=scheduler)
    handle.unsafe_set(
        fiber, only_if_missing=only_if_missing, propagate_interruption=propagate_interruption
    )
    return fiber