Queue

A Queue passes values between fibers, delivering each offered value to one consumer. Use it to separate producers from workers and to apply backpressure when workers cannot keep up.

Choosing an overflow strategy

function chooseQueues() {
  const jobs = Queue.bounded<string>(32)
  const optionalEvents = Queue.dropping<string>(32)
  const recentPositions = Queue.sliding<number>(32)
  const unrestricted = Queue.unbounded<string>()
  return { jobs, optionalEvents, recentPositions, unrestricted }
}

A bounded queue suspends offers when full until consumers free space. A dropping queue rejects new values when full, while a sliding queue discards the oldest buffered values to accept new ones. An unbounded queue imposes no capacity limit, so memory use grows if producers stay ahead of consumers.

Offering values and batches

function offerWithBackpressure() {
  const queue = Queue.bounded<number>(1)
  queue.offer(10)
  const secondOffer = fork queue.offer(20)
  const first = queue.take()
  join secondOffer
  const second = queue.take()
  return [first, second] // [10, 20]
}

function rejectedBatch() {
  const queue = Queue.dropping<number>(2)
  return queue.offerAll([10, 20, 30]) // [30]
}

offer returns whether the value was accepted. offerAll returns the values it could not add, so an empty array means the whole batch was accepted. Fork a producer when the current fiber must consume values to free the space that producer needs; offering into a full bounded queue and only then taking from it would wait forever.

Taking, polling, and draining values

function consumeBatches() {
  const queue = Queue.bounded<number>(8)
  queue.offerAll([1, 2, 3, 4, 5, 6])
  const first = queue.take()           // 1
  const optional = queue.poll()       // Some(2)
  const pair = queue.takeN(2)        // [3, 4]
  const batch = queue.takeBetween(1, 4) // [5, 6]
  const empty = queue.clear()         // []
  return { first, optional, pair, batch, empty }
}

take waits for one value, takeN waits for an exact count, and takeBetween waits for its minimum then takes up to its maximum. In this version, takeAll waits for at least one value and returns the currently available nonempty batch. Use poll for an optional single value without waiting, or clear to drain buffered values without waiting for new ones.

Querying queue state

function inspectQueue() {
  const queue = Queue.bounded<string>(2)
  queue.offer("task")
  return {
    capacity: queue.capacity,
    size: queue.size(),
    full: queue.isFull()
  }
}
// Returns { capacity: 2, size: 1, full: false }.

Size counts buffered messages, and capacity is the configured limit. These values are observations, not reservations; another fiber can change the queue immediately after a check. Call offer or take directly to coordinate instead of checking state first.

Shutting down and awaiting closure

function finishNormally() {
  const queue = Queue.bounded<number, Cause.Done>(4)
  queue.offerAll([7, 8])
  queue.end()
  const values = queue.collect()
  Queue.await(queue)
  return values // [7, 8]
}

function stopImmediately() {
  const queue = Queue.bounded<number>(4)
  const waiter = fork {
    try {
      Queue.await(queue)
    } finally {
      logInfo("queue closed")
    }
  }
  queue.shutdown()
  return Fiber.await(waiter)
}

end stops new offers and lets buffered values drain before normal completion; collect consumes until that completion. shutdown immediately discards buffered values and interrupts waiting operations. Queue.await waits for the terminal state and preserves its outcome, so awaiting a shut-down queue is interrupted; inspect that child with Fiber.await rather than propagating it through join.

Queues do not automatically close at ordinary block exit. Have their owner call shutdown in finally when consumers and producers must stop with it.

Restricting producer and consumer capabilities

function send(outbox: Queue.Enqueue<string>, message: string) {
  return outbox.offer(message)
}

function receive(inbox: Queue.Dequeue<string>) {
  return inbox.take()
}

function handoff() {
  const queue = Queue.bounded<string>(2)
  send(queue.asEnqueue(), "ready")
  return receive(queue.asDequeue())
}

Enqueue gives a producer access to offering operations, while Dequeue gives a consumer access to taking operations. The adapters return the same underlying queue with narrower types. This restriction documents ownership and catches accidental use of the wrong side at compile time.