ReplayWindow

__init__(buffer) -> None

def __init__(self, buffer: ReplayBuffer[A] | None) -> None:
    self._buffer = buffer
    self._values: deque[A] = deque()
    self._newest = -1
    self._slides_seen = 0
    if buffer is not None:
        for index, value in buffer.items:
            self._values.append(value)
            self._newest = index
        self._slides_seen = buffer.slide_count

take() -> Option[A]

def take(self) -> Option[A]:
    """The next replayed message; ``None`` once drained."""
    if not self._values:
        return None
    self._sync()
    value = self._values.popleft()
    if not self._values:
        self.close()
    return Some(value)

take_n(n) -> list[A]

def take_n(self, n: int) -> list[A]:
    """Up to ``n`` replayed messages (none if ``n`` is 0 or less)."""
    if not self._values or n <= 0:
        return []
    self._sync()
    values = [self._values.popleft() for _ in range(min(n, len(self._values)))]
    if not self._values:
        self.close()
    return values

take_all() -> list[A]

def take_all(self) -> list[A]:
    return self.take_n(len(self._values))

close() -> None

def close(self) -> None:
    """Drop what is left; a drained window keeps nothing alive."""
    self._values.clear()
    self._buffer = None