RcRef

__init__(acquire, context, scope, idle_time_to_live) -> None

def __init__(
    self,
    acquire: Effect[A_co, E_co, Any],
    context: Context,
    scope: CloseableScope,
    idle_time_to_live: Seconds | None,
) -> None:
    self._acquire = acquire
    self._context = context
    self._scope = scope
    self._idle_time_to_live = idle_time_to_live
    self._state: _State[A_co] = "empty"

make(acquire, *, idle_time_to_live=None) -> Effect[RcRef[T, X], Never, Scope | R]

@staticmethod
def make[T, X = Never, R = Never](
    acquire: Effect[T, X, R], *, idle_time_to_live: Duration | None = None
) -> Effect[RcRef[T, X], Never, Scope | R]:
    """A ref that runs ``acquire`` (with the services in scope now, and
    a scope of its own) when first used. Without ``idle_time_to_live``
    the resource is released as soon as the last reference goes."""
    require_effect(acquire, "acquire")
    ttl = None if idle_time_to_live is None else to_seconds(idle_time_to_live)

    def body(fiber: Fiber[Any, Any]) -> Effect[RcRef[T, X]]:
        context = fiber._context
        scope = context.unsafe_get(Scope)
        ref = RcRef(acquire, context, scope, ttl)
        return scope.add_finalizer(lambda _: ref._close()).as_(ref)

    return _WithFiber(body)

get() -> Effect[A_co, E_co, Scope]

def get(self) -> Effect[A_co, E_co, Scope]:
    """The resource, held until the current scope closes."""

    def body(restore: Restore) -> Effect[A_co, E_co, Scope]:
        def with_fiber(fiber: Fiber[Any, Any]) -> Effect[A_co, E_co]:
            scope = fiber._context.unsafe_get(Scope)
            return self._get_state(fiber, restore).flat_map(
                lambda state: scope.add_finalizer(lambda _: self._release(state)).as_(
                    state.value
                )
            )

        return _WithFiber(with_fiber)

    return uninterruptible_mask(body)

invalidate() -> Effect[None]

def invalidate(self) -> Effect[None]:
    """Forget the resource, so the next ``get`` acquires afresh. It is
    released now if nobody holds it, else when the last holder lets go."""

    def body() -> Effect[None]:
        state = self._state
        if not isinstance(state, _Acquired):
            return _VOID
        self._state = "empty"
        state.invalidated = True
        if state.ref_count > 0:
            return _VOID
        if state.fiber is not None:
            state.fiber.unsafe_interrupt_as(None)
        return state.scope.close(_DONE)

    return _Suspend(body).uninterruptible()