Subscription

__init__(hub, reader) -> None

def __init__(self, hub: Hub[Any], reader: Reader[Any]) -> None:
    self._hub = hub
    self._reader = reader

take() -> Effect[A]

def take(self) -> Effect[A]:
    """Wait for the next message."""
    reader = self._reader
    hub = self._hub

    def attempt() -> Effect[A]:
        if reader.unsubscribed:
            return _interrupt()
        replayed = reader.replay.take()
        if replayed is not None:
            return _Succeed(replayed.value)
        polled = reader.read_up_to(1)
        if polled:
            hub.deliver()
            return _Succeed(polled[0])
        if hub.ended is not None:
            return _Succeed(hub.ended.value)
        return self._park(_identity, attempt)

    return _Suspend(attempt)

take_all() -> Effect[Sequence[A]]

def take_all(self) -> Effect[Sequence[A]]:
    """Wait for at least one message, then take every one there is."""
    reader = self._reader
    hub = self._hub

    def more(first: A) -> list[A]:
        rest = reader.read_up_to(reader.size())
        hub.deliver()
        return [first, *rest]

    def attempt() -> Effect[list[A]]:
        if reader.unsubscribed:
            return _interrupt()
        polled = reader.read_up_to(reader.size())
        hub.deliver()
        if reader.replay.remaining:
            return _Succeed(reader.replay.take_all() + polled)
        if polled:
            return _Succeed(polled)
        if hub.ended is not None:
            return _Succeed([hub.ended.value])
        return self._park(more, attempt)

    return _Suspend(attempt)

take_up_to(n) -> Effect[Sequence[A]]

def take_up_to(self, n: int) -> Effect[Sequence[A]]:
    """Take up to ``n`` messages without waiting (none if ``n`` is 0 or
    less, or while other takes wait)."""
    reader = self._reader

    def attempt() -> Effect[list[A]]:
        if reader.unsubscribed:
            return _interrupt()
        limit = max(n, 0)
        if reader.replay.remaining >= limit:
            return _Succeed(reader.replay.take_n(limit))
        replayed = reader.replay.take_all()
        polled = reader.read_up_to(limit - len(replayed))
        self._hub.deliver()
        return _Succeed(replayed + polled)

    return _Suspend(attempt)

take_between(minimum, maximum) -> Effect[Sequence[A]]

def take_between(self, minimum: int, maximum: int) -> Effect[Sequence[A]]:
    """Take at least ``minimum`` and at most ``maximum`` messages, waiting
    for the minimum. Nothing, at once, if ``maximum < minimum``."""

    def body() -> Generator[Effect[Any], Any, list[A]]:
        low, high = max(minimum, 0), max(maximum, 0)
        out: list[A] = []
        while high >= low:
            taken: Sequence[A] = yield self.take_up_to(high)
            out.extend(taken)
            missing = low - len(taken)
            if missing <= 0:
                break
            out.append((yield self.take()))
            low, high = missing - 1, high - len(taken) - 1
            if low == 0:
                break
        return out

    return _Gen(body)

remaining() -> Effect[int]

def remaining(self) -> Effect[int]:
    """How many messages there are to take now."""

    def attempt() -> Effect[int]:
        count = self.unsafe_remaining()
        return _interrupt() if count is None else _Succeed(count)

    return _Suspend(attempt)

unsafe_remaining() -> int | None

def unsafe_remaining(self) -> int | None:
    """``remaining`` now; ``None`` once unsubscribed."""
    reader = self._reader
    if reader.unsubscribed:
        return None
    return reader.size() + reader.replay.remaining