Atomic

__init__(capacity, replay=None) -> None

def __init__(self, capacity: int | None, replay: int | None = None) -> None:
    if capacity is not None and capacity <= 0:
        raise ValueError(f"Cannot construct PubSub with capacity of {capacity}")
    self._capacity = capacity
    self._replay = ReplayBuffer[A](replay) if replay is not None and replay > 0 else None
    self._slots: deque[_Slot[A]] = deque()
    # The publish index of the oldest stored message.
    self._start = 0
    self._subscribers = 0

bounded(capacity, *, replay=None) -> Atomic[A]

@classmethod
def bounded(cls, capacity: int | None, *, replay: int | None = None) -> Atomic[A]:
    """A store of at most ``capacity`` messages (``None``: unbounded),
    replaying the last ``replay`` to new subscribers (Effect's
    ``makeAtomicBounded``). A capacity below 1 is a ``ValueError``."""
    return cls(capacity, replay)

unbounded(*, replay=None) -> Atomic[A]

@classmethod
def unbounded(cls, *, replay: int | None = None) -> Atomic[A]:
    """Effect's ``makeAtomicUnbounded``."""
    return cls(None, replay)

size() -> int

def size(self) -> int:
    """How many messages are stored (each once, however many readers)."""
    return len(self._slots)

is_empty() -> bool

def is_empty(self) -> bool:
    return not self._slots

is_full() -> bool

def is_full(self) -> bool:
    return len(self._slots) == self._capacity

publish(value) -> bool

def publish(self, value: A) -> bool:
    """Store ``value`` for the current subscribers; ``False`` if full."""
    if self.is_full():
        return False
    replay_index = 0 if self._replay is None else self._replay.offer(value)
    if self._subscribers:
        self._slots.append(_Slot(value, self._subscribers, replay_index))
    return True

publish_all(items) -> list[A]

def publish_all(self, items: Iterable[A]) -> list[A]:
    """Store what fits, in order; the items that did not."""
    if not self._subscribers:
        if self._replay is not None:
            for item in items:
                self._replay.offer(item)
        return []
    batch = list(items)
    fits = len(batch) if self._capacity is None else self._capacity - len(self._slots)
    for item in batch[:fits]:
        self.publish(item)
    return batch[fits:]

slide() -> None

def slide(self) -> None:
    """Drop the oldest stored message for every subscriber."""
    if self._slots:
        slot = self._slots.popleft()
        self._start += 1
        if self._replay is not None:
            self._replay.slide(slot.value, slot.replay_index)

subscribe() -> BackingSubscription[A]

def subscribe(self) -> BackingSubscription[A]:
    """A reader of the messages published from now on."""
    self._subscribers += 1
    return BackingSubscription(self)

replay_window() -> ReplayWindow[A]

def replay_window(self) -> ReplayWindow[A]:
    """The messages a subscriber that joins now replays."""
    return ReplayWindow(self._replay)