BackingSubscription

__init__(atomic) -> None

def __init__(self, atomic: Atomic[A]) -> None:
    self._atomic = atomic
    self._index = atomic._end()
    self._unsubscribed = False

size() -> int

def size(self) -> int:
    """How many messages are there to read."""
    if self._unsubscribed:
        return 0
    atomic = self._atomic
    return atomic._end() - max(self._index, atomic._start)

is_empty() -> bool

def is_empty(self) -> bool:
    return self.size() == 0

poll() -> Option[A]

def poll(self) -> Option[A]:
    """The next message, if there is one."""
    items = self.poll_up_to(1)
    return Some(items[0]) if items else None

poll_up_to(n) -> list[A]

def poll_up_to(self, n: int) -> list[A]:
    """Up to ``n`` messages (none if ``n`` is 0 or less)."""
    count = min(n, self.size())
    if count <= 0:
        return []
    atomic = self._atomic
    start = max(self._index, atomic._start)
    self._index = start + count
    return [atomic._read(i) for i in range(start, start + count)]

poll_all() -> list[A]

def poll_all(self) -> list[A]:
    return self.poll_up_to(self.size())

unsubscribe() -> None

def unsubscribe(self) -> None:
    """Stop reading; messages only this subscriber owed are dropped."""
    if not self._unsubscribed:
        self._unsubscribed = True
        self._atomic._leave(self._index)