Forks effects into a FiberHandle from plain code; see
FiberHandle.runtime.
Functions#
__init__#
__init__(handle, runtime) -> None| Parameter | Type | Description |
|---|---|---|
handle | FiberHandle[A, E] | |
runtime | CapturedRuntime |
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.
| Parameter | Type | Description |
|---|---|---|
eff | Effect[A, E, R] | |
only_if_missing | bool = False | |
propagate_interruption | bool = False | |
uninterruptible | bool = False | |
scheduler | Scheduler | 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