RcMap

__init__(lookup, context, scope, idle_time_to_live, capacity) -> None

def __init__(
    self,
    lookup: Callable[[K], Effect[A, E, Any]],
    context: Context,
    scope: CloseableScope,
    idle_time_to_live: Callable[[K], Seconds],
    capacity: int | None,
) -> None:
    self._lookup = lookup
    self._context = context
    self._scope = scope
    self._idle_time_to_live = idle_time_to_live
    self._capacity = capacity
    self._entries: dict[K, _Entry[A, E]] | None = {}  # None once closed

make(lookup, *, idle_time_to_live=None, capacity=None) -> Effect[RcMap[Any, Any, Any], Never, Any]

@staticmethod
def make(
    lookup: Callable[[Any], Effect[Any, Any, Any]],
    *,
    idle_time_to_live: IdleTimeToLive[Any] | None = None,
    capacity: int | None = None,
) -> Effect[RcMap[Any, Any, Any], Never, Any]:
    """A map that looks keys up with ``lookup`` (with the services in
    scope now and of the caller, and a scope per key), released when
    the current scope closes. ``idle_time_to_live`` defaults to 0:
    release as soon as the last reference goes."""
    ttl = _idle_time_to_live(idle_time_to_live)
    bound = None if capacity is None else max(capacity, 0)

    def body(fiber: Fiber[Any, Any]) -> Effect[RcMap[Any, Any, Any]]:
        context = fiber._context
        scope = context.unsafe_get(Scope)
        rc_map = RcMap(lookup, context, scope, ttl, bound)
        return scope.add_finalizer(lambda _: rc_map._close()).as_(rc_map)

    return _WithFiber(body)

get(key) -> Effect[A, E, Scope]

def get(self, key: K) -> Effect[A, E, Scope]:
    """The resource for ``key``, looked up if needed and held until the
    current scope closes. Interrupted once the map is closed."""

    def body(restore: Restore) -> Effect[A, E, Scope]:
        return _WithFiber(lambda parent: self._get(parent, key, restore))

    return uninterruptible_mask(body)

get_option(key) -> Effect[Option[A], E, Scope]

def get_option(self, key: K) -> Effect[Option[A], E, Scope]:
    """``get`` for a key the map already holds (``Some``), without a
    lookup: ``None`` for a missing key or a closed map. An entry still
    being looked up is waited for; a failed one fails the same way."""

    def body(restore: Restore) -> Effect[Option[A], E, Scope]:
        def run(fiber: Fiber[Any, Any]) -> Effect[Option[A], E]:
            entry = None if self._entries is None else self._entries.get(key)
            if entry is None:
                return _Succeed(None)
            return self._retain(fiber, entry, restore(entry.deferred.wait().map(Some)))

        return _WithFiber(run)

    return uninterruptible_mask(body)

keys() -> Effect[list[K]]

def keys(self) -> Effect[list[K]]:
    """The keys held now. Interrupted once the map is closed."""
    return _Suspend(
        lambda: interrupt() if self._entries is None else _Succeed(list(self._entries))
    )

has(key) -> Effect[bool]

def has(self, key: K) -> Effect[bool]:
    """Whether the map holds ``key``; ``False`` once it is closed."""
    return _Sync(lambda: self._entries is not None and key in self._entries)

invalidate(key) -> Effect[None]

def invalidate(self, key: K) -> Effect[None]:
    """Drop ``key``, so the next ``get`` looks it up afresh. Its resource
    is released now if nobody holds it, else when the last holder lets
    go."""

    def body() -> Effect[None]:
        entries = self._entries
        entry = None if entries is None else entries.pop(key, None)
        if entry is None or entry.ref_count > 0:
            return _VOID
        return _close_entry(entry)

    return _Suspend(body).uninterruptible()

touch(key) -> Effect[None]

def touch(self, key: K) -> Effect[None]:
    """Restart ``key``'s idle time to live."""

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        entry = None if self._entries is None else self._entries.get(key)
        if entry is not None and entry.idle_time_to_live != 0:
            clock: ClockService = fiber._context.lookup(Clock)[1]
            entry.expires_at = clock.unsafe_monotonic() + entry.idle_time_to_live
        return _VOID

    return _WithFiber(body)