Queue

__init__(capacity=None, strategy='suspend') -> None

def __init__(self, capacity: int | None = None, strategy: Strategy = "suspend") -> None:
    if capacity is not None and capacity < 0:
        raise ValueError(f"capacity must be >= 0, got {capacity}")
    self._capacity = capacity
    self._strategy: Strategy = strategy
    self._items: deque[A] = deque()
    self._state: st.State[A, E] = st.Open()

make(capacity=None, *, strategy='suspend') -> Effect[Queue[A, E]]

@classmethod
def make(
    cls, capacity: int | None = None, *, strategy: Strategy = "suspend"
) -> Effect[Queue[A, E]]:
    """A queue of ``capacity`` items (``None``: unbounded) that handles a
    full queue by ``strategy``."""
    return _Sync(lambda: cls(capacity, strategy))

unbounded() -> Effect[Queue[A, E]]

@classmethod
def unbounded(cls) -> Effect[Queue[A, E]]:
    return _Sync(cls)

bounded(capacity) -> Effect[Queue[A, E]]

@classmethod
def bounded(cls, capacity: int) -> Effect[Queue[A, E]]:
    """``offer`` waits while the queue holds ``capacity`` items."""
    return _Sync(lambda: cls(capacity, "suspend"))

dropping(capacity) -> Effect[Queue[A, E]]

@classmethod
def dropping(cls, capacity: int) -> Effect[Queue[A, E]]:
    return _Sync(lambda: cls(capacity, "dropping"))

sliding(capacity) -> Effect[Queue[A, E]]

@classmethod
def sliding(cls, capacity: int) -> Effect[Queue[A, E]]:
    return _Sync(lambda: cls(capacity, "sliding"))

offer(item) -> Effect[bool]

def offer(self, item: A) -> Effect[bool]:
    """Add an item. ``False`` if the queue is finished, or if it is full
    and drops new items."""

    def register(resume: Callable[[Effect[bool]], None]) -> Effect[None] | None:
        if self.unsafe_offer(item):
            resume(_Succeed(True))
        elif isinstance(self._state, st.Open) and self._strategy == "suspend":
            return self._park(self._state.offers, st.OfferOne(item, resume))
        else:
            resume(_Succeed(False))
        return None

    return _Callback(register)

unsafe_offer(item) -> bool

def unsafe_offer(self, item: A) -> bool:
    """``offer`` without waiting, from outside an effect: ``False`` if it
    would have to wait. Call on the queue's fibers' scheduler thread."""
    match self._state:
        case st.Open(takers=takers):
            if self._free(takers) > 0 or self._strategy == "sliding":
                # As Effect's ``offerUnsafe``: a full sliding queue
                # swaps its oldest item for this one, even at capacity 0.
                self._push([item], takers, keep_at_least=1)
                return True
            return False
        case st.Closing() | st.Done():
            return False

offer_all(items) -> Effect[list[T]]

def offer_all[T](self: Queue[T, Any], items: Iterable[T]) -> Effect[list[T]]:
    """Add items in order; succeeds with those not added. What fits is
    added at once. A suspending queue waits to add the rest, and returns
    what is left if the queue is finished meanwhile."""

    def register(resume: Callable[[Effect[list[T]]], None]) -> Effect[None] | None:
        rest = self.unsafe_offer_all(items)
        if rest and isinstance(self._state, st.Open) and self._strategy == "suspend":
            return self._park(self._state.offers, st.OfferMany(rest, resume))
        resume(_Succeed(rest))
        return None

    return _Callback(register)

unsafe_offer_all(items) -> list[T]

def unsafe_offer_all[T](self: Queue[T, Any], items: Iterable[T]) -> list[T]:
    """``offer_all`` without waiting: the items that did not fit."""
    batch = list(items)
    match self._state:
        case st.Open(takers=takers):
            fits = len(batch) if self._strategy == "sliding" else self._free(takers)
            self._push(batch[:fits], takers)
            return batch[fits:]
        case st.Closing() | st.Done():
            return batch

end() -> Effect[bool]

def end(self: Enqueue[Any, Done]) -> Effect[bool]:
    """No more items. Takers fail with ``Done()`` once the queue drains.
    Only available on queues whose error type includes ``Done``."""
    return self.fail_cause(fail_cause(Done()))

unsafe_end() -> bool

def unsafe_end(self: Queue[Any, Done]) -> bool:
    return self.unsafe_fail_cause(fail_cause(Done()))

fail(error) -> Effect[bool]

def fail(self, error: E) -> Effect[bool]:
    """Takers fail with ``error`` once the queue drains."""
    return self.fail_cause(fail_cause(error))

interrupt() -> Effect[bool]

def interrupt(self) -> Effect[bool]:
    """Takers are interrupted (by the calling fiber) once the queue
    drains."""
    return _WithFiber(lambda fiber: self.fail_cause(interrupt_cause(fiber.id)))

fail_cause(cause) -> Effect[bool]

def fail_cause(self, cause: Cause[E]) -> Effect[bool]:
    """``False`` if the queue was already ended, failed or shut down."""
    return _Sync(lambda: self.unsafe_fail_cause(cause))

unsafe_fail_cause(cause) -> bool

def unsafe_fail_cause(self, cause: Cause[E]) -> bool:
    match self._state:
        case st.Open(takers=takers, offers=offers, awaiters=awaiters):
            self._state = st.Closing(takers, offers, awaiters, cause)
            # A batch take waiting for its minimum can now drain the rest.
            takers.notify_ready()
            self._finish_if_drained()
            return True
        case st.Closing() | st.Done():
            return False

shutdown() -> Effect[bool]

def shutdown(self) -> Effect[bool]:
    """Drop all items and waiting offers; takers are interrupted (or see
    the failure of a queue that was closing). ``False`` if the queue was
    already done."""
    return _Sync(self.unsafe_shutdown)

unsafe_shutdown() -> bool

def unsafe_shutdown(self) -> bool:
    match self._state:
        case st.Done():
            return False
        case st.Open(offers=offers):
            self._shut(interrupt_cause(), offers)
        case st.Closing(offers=offers, cause=cause):
            self._shut(cause, offers)
    return True

into(eff) -> Effect[bool, Never, R]

def into[R, EE](
    self: Queue[Any, EE | Done], eff: Effect[object, EE | Done, R]
) -> Effect[bool, Never, R]:
    """Run ``eff``, then end the queue, or fail it with ``eff``'s cause.
    The queue is finished even if the running fiber is interrupted."""
    require_effect(eff, "eff")
    return uninterruptible_mask(
        lambda restore: restore(eff).match_cause_effect(
            on_failure=self.fail_cause, on_success=lambda _: self.end()
        )
    )

take() -> Effect[A, E]

def take(self) -> Effect[A, E]:
    """Wait for an item."""

    def attempt() -> Effect[A, E]:
        ex = self.unsafe_take()
        return self._wait_then(1, attempt) if ex is None else _FromExit(ex)

    return _Suspend(attempt)

unsafe_take() -> Exit[A, E] | None

def unsafe_take(self) -> Exit[A, E] | None:
    """An item, the failure of a finished queue, or ``None`` if ``take``
    would wait."""
    match self._state:
        case st.Done(cause=cause):
            return Failure(cause)
        case st.Open(offers=offers) | st.Closing(offers=offers):
            if not self._items and not self._takes_from(offers):
                return None
            item = self._take_one(offers)
            self._release()
            return Success(item)

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

def take_between(self, minimum: int, maximum: int) -> Effect[list[A], E]:
    """Wait until ``minimum`` items are there (or the queue is finishing),
    then take up to ``maximum``. A bounded queue never holds more than
    its capacity, so it waits for at most that many. A count of 0 or less
    takes nothing and does not wait (but still fails on a finished
    queue)."""
    return self._take_between(minimum, maximum)

take_n(n) -> Effect[list[A], E]

def take_n(self, n: int) -> Effect[list[A], E]:
    """``take_between(n, n)``."""
    return self._take_between(n, n)

take_all() -> Effect[list[A], E]

def take_all(self) -> Effect[list[A], E]:
    """Wait for at least one item, then take everything."""
    return self._take_between(1, None)

poll() -> Effect[Option[A]]

def poll(self) -> Effect[Option[A]]:
    """An item if one is there now; ``None`` if not, or if the queue is
    finished."""

    def attempt() -> Effect[Option[A]]:
        ex = self.unsafe_take()
        return _Succeed(Some(ex.value) if isinstance(ex, Success) else None)

    return _Suspend(attempt)

peek() -> Effect[A, E]

def peek(self) -> Effect[A, E]:
    """Wait for an item and succeed with it, leaving it in the queue.
    Offers waiting on a rendezvous queue are not visible to ``peek``."""

    def attempt() -> Effect[A, E]:
        match self._state:
            case st.Done(cause=cause):
                return _FailCause(cause)
            case st.Open(takers=takers) | st.Closing(takers=takers):
                if not self._items:
                    ready = self._finished_or(lambda: bool(self._items))
                    return takers.wait(ready).and_then(_Suspend(attempt))
                return _Succeed(self._items[0])

    return _Suspend(attempt)

clear() -> Effect[list[Any], Any]

def clear(self: Queue[Any, Any]) -> Effect[list[Any], Any]:
    """Take every item there is now, without waiting. On a queue that
    was ended, succeeds with ``[]``; on one that failed, fails with that
    failure (without ``Done``)."""

    def attempt() -> Effect[list[Any], Any]:
        match self._state:
            case st.Done(cause=cause):
                return _catch_done(cause, [])
            case st.Open(offers=offers) | st.Closing(offers=offers):
                if self._items:
                    items = self._take_buffered(len(self._items))
                elif self._takes_from(offers):
                    items = [st.take_offer(offers)]
                else:
                    return _Succeed([])
                self._release()
                return _Succeed(items)

    return _Suspend(attempt)

collect() -> Effect[list[Any], Any]

def collect(self: Queue[Any, Any]) -> Effect[list[Any], Any]:
    """Take items until the queue is ended, and succeed with all of them.
    If the queue fails instead, fail with that failure."""

    def body() -> Generator[Effect[Any, Any, Any], Any, list[Any]]:
        out: list[Any] = []
        while True:
            taken: Exit[list[Any], Any] = yield self.take_all().exit()
            match taken:
                case Success(value=items):
                    out.extend(items)
                case Failure(cause=cause):
                    return (yield _catch_done(cause, out))

    return _Gen(body)

await_done() -> Effect[None, Any]

def await_done(self: Queue[Any, Any]) -> Effect[None, Any]:
    """Wait until the queue is finished: ended, failed or shut down, and
    drained. Succeeds if it was ended; otherwise fails as its takers do
    (Effect's ``Queue.await``)."""

    def attempt() -> Effect[None, Any]:
        match self._state:
            case st.Done(cause=cause):
                return _catch_done(cause, None)
            case st.Open(awaiters=awaiters) | st.Closing(awaiters=awaiters):
                return awaiters.wait(self._finished_or(lambda: False)).and_then(
                    _Suspend(attempt)
                )

    return _Suspend(attempt)

size() -> Effect[int]

def size(self) -> Effect[int]:
    """How many items the queue holds (0 once it is finished)."""
    return _Sync(self.unsafe_size)

unsafe_size() -> int

def unsafe_size(self) -> int:
    return 0 if isinstance(self._state, st.Done) else len(self._items)

is_full() -> Effect[bool]

def is_full(self) -> Effect[bool]:
    """Whether the queue holds ``capacity`` items."""
    return _Sync(self.unsafe_is_full)

unsafe_is_full() -> bool

def unsafe_is_full(self) -> bool:
    return self.unsafe_size() == self._capacity