FiberHandle

__init__() -> None

def __init__(self) -> None:
    self._state: _Open[A, E] | None = _Open()  # None once the scope has closed
    # Completed by the first reported failure, or when the scope closes.
    self.deferred: Deferred[None, E] = Deferred()

make() -> Effect[FiberHandle[A, E], Never, Scope]

@classmethod
def make(cls) -> Effect[FiberHandle[A, E], Never, Scope]:
    """A handle whose fibers are interrupted when the current scope
    closes."""
    return acquire_release(_Sync(cls), lambda h, _: h._close())

make_runtime(*tags) -> Effect[FiberHandleRuntime[A, E, R], Never, Scope | R]

@classmethod
def make_runtime[R = Never](
    cls, *tags: TagLike[Any, R]
) -> Effect[FiberHandleRuntime[A, E, R], Never, Scope | R]:
    """``make`` and ``runtime`` in one."""
    return cls.make().flat_map(lambda h: h.runtime(*tags))

make_runtime_async(*tags) -> Effect[FiberHandleAsyncRuntime[A, E, R], Never, Scope | R]

@classmethod
def make_runtime_async[R = Never](
    cls, *tags: TagLike[Any, R]
) -> Effect[FiberHandleAsyncRuntime[A, E, R], Never, Scope | R]:
    """``make`` and ``runtime_async`` in one."""
    return cls.make().flat_map(lambda h: h.runtime_async(*tags))

unsafe_set(fiber, *, only_if_missing=False, propagate_interruption=False) -> None

def unsafe_set(
    self,
    fiber: Fiber[A, E],
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
) -> None:
    """Make a running fiber the current one, interrupting the one there
    (or, with ``only_if_missing``, interrupting ``fiber`` instead). Into
    a closed handle the fiber is interrupted. Call on the scheduler
    thread."""
    state = self._state
    if state is None:
        fiber.unsafe_interrupt_as(INTERNAL_FIBER_ID)
        return
    previous = state.current
    if previous is fiber:
        return
    if previous is not None and only_if_missing:
        fiber.unsafe_interrupt_as(INTERNAL_FIBER_ID)
        return
    # Install the replacement before the interrupt can re-enter the
    # handle through a finalizer.
    state.current = fiber
    state.owned[fiber] = None
    if previous is not None:
        previous.unsafe_interrupt_as(INTERNAL_FIBER_ID)

    def observer(ex: Exit[A, E]) -> None:
        state = self._state
        if state is not None:
            state.owned.pop(fiber, None)
            if state.current is fiber:
                state.current = None
        if reports_failure(ex, propagate_interruption):
            self.deferred.unsafe_done(ex)

    fiber.unsafe_add_observer(observer)

set(fiber, *, only_if_missing=False, propagate_interruption=False) -> Effect[None]

def set(
    self,
    fiber: Fiber[A, E],
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
) -> Effect[None]:
    return _Sync(
        lambda: self.unsafe_set(
            fiber,
            only_if_missing=only_if_missing,
            propagate_interruption=propagate_interruption,
        )
    )

unsafe_get() -> Fiber[A, E] | None

def unsafe_get(self) -> Fiber[A, E] | None:
    return None if self._state is None else self._state.current

get() -> Effect[Fiber[A, E] | None]

def get(self) -> Effect[Fiber[A, E] | None]:
    return _Sync(self.unsafe_get)

clear() -> Effect[None]

def clear(self) -> Effect[None]:
    """Interrupt the current fiber, if any, and wait for it. A fiber set
    meanwhile stays current."""

    def body(restore: Restore) -> Effect[None]:
        fiber = self.unsafe_get()
        if fiber is None:
            return _Succeed(None)

        def forget() -> None:
            if self._state is not None and self._state.current is fiber:
                self._state.current = None

        return restore(fiber.interrupt_as(INTERNAL_FIBER_ID)).and_then(_Sync(forget))

    return uninterruptible_mask(body)

run(eff, *, only_if_missing=False, propagate_interruption=False, start_immediately=True) -> Effect[Fiber[A, E], Never, R]

def run[R = Never](
    self,
    eff: Effect[A, E, R],
    *,
    only_if_missing: bool = False,
    propagate_interruption: bool = False,
    start_immediately: bool = True,
) -> Effect[Fiber[A, E], Never, R]:
    """Fork ``eff`` as the current fiber. It starts at once unless
    ``start_immediately=False``. With ``only_if_missing`` and a current
    fiber, gives back an interrupted fiber that never runs ``eff``.
    Running into a closed handle interrupts the caller."""
    require_effect(eff, "eff")

    def body(parent: Fiber[Any, Any]) -> Effect[Fiber[A, E]]:
        state = self._state
        if state is None:
            return interrupt()
        if only_if_missing and state.current is not None:
            return _Succeed(interrupted_fiber(parent._run_scheduler, parent._context))
        fiber: Fiber[A, E] = parent._fork(eff, daemon=True, start_immediately=start_immediately)
        self.unsafe_set(
            fiber,
            only_if_missing=only_if_missing,
            propagate_interruption=propagate_interruption,
        )
        return _AfterStarted(fiber)

    return _WithFiber(body)

runtime(*tags) -> Effect[FiberHandleRuntime[A, E, R], Never, R]

def runtime[R = Never](
    self, *tags: TagLike[Any, R]
) -> Effect[FiberHandleRuntime[A, E, R], Never, R]:
    """A runner that forks effects needing ``tags`` into the handle,
    with the services of the fiber calling this."""
    return _WithFiber(lambda fiber: _Succeed(FiberHandleRuntime(self, CapturedRuntime(fiber))))

runtime_async(*tags) -> Effect[FiberHandleAsyncRuntime[A, E, R], Never, R]

def runtime_async[R = Never](
    self, *tags: TagLike[Any, R]
) -> Effect[FiberHandleAsyncRuntime[A, E, R], Never, R]:
    """``runtime``, but the runner returns a future of the value (needs
    a running asyncio loop when called)."""
    return self.runtime(*tags).map(FiberHandleAsyncRuntime)

join() -> Effect[None, E]

def join(self) -> Effect[None, E]:
    """Wait for the first reported failure and fail with it; succeed
    when the scope closes."""
    return self.deferred.wait()

await_empty() -> Effect[None]

def await_empty(self) -> Effect[None]:
    """Wait for the current fiber, if any."""

    def body() -> Effect[None]:
        fiber = self.unsafe_get()
        return _Succeed(None) if fiber is None else fiber.await_exit().as_void()

    return _Suspend(body)