Hyperlinkv0.8.0-beta.28

← TxQueue

TxQueue.takeAllconsteffect/TxQueue.ts:774
<A, E>(self: TxDequeue<A, E>): Effect.Effect<Arr.NonEmptyArray<A>, E>

Takes all items from the queue. Blocks if the queue is empty.

Details

If the queue is already in a failed state, the error is propagated through the E-channel. This follows the same patterns as take and waits when there are no elements. It returns a non-empty array because it blocks until at least one item is available. This function mutates the original TxQueue by removing all items. It does not return a new TxQueue reference.

Example (Taking all queued values)

import { Array, Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])

  // Take all items atomically - returns NonEmptyArray
  const items = yield* TxQueue.takeAll(queue)
  console.log(items) // [1, 2, 3, 4, 5]
  console.log(Array.isArrayNonEmpty(items)) // true
})

// Error propagation example
const errorExample = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(5)
  yield* TxQueue.offerAll(queue, [1, 2])
  yield* TxQueue.fail(queue, "processing error")

  // takeAll() propagates the queue error through E-channel
  const result = yield* Effect.flip(TxQueue.takeAll(queue))
  console.log(result) // "processing error"
})
combinators
Source effect/TxQueue.ts:77427 lines
export const takeAll = <A, E>(self: TxDequeue<A, E>): Effect.Effect<Arr.NonEmptyArray<A>, E> =>
  Effect.gen(function*() {
    const state = yield* TxRef.get(self.stateRef)

    // Handle done queue
    if (state._tag === "Done") {
      return yield* Effect.failCause(state.cause)
    }

    // Wait if empty - same pattern as take()
    if (yield* isEmpty(self)) {
      return yield* Effect.txRetry
    }

    const chunk = yield* TxChunk.get(self.items)

    // Take all items (guaranteed non-empty due to isEmpty check above)
    const items = Chunk.toArray(chunk) as Arr.NonEmptyArray<A>
    yield* TxChunk.set(self.items, Chunk.empty())

    // Check if we need to transition Closing → Done
    if (state._tag === "Closing") {
      yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause })
    }

    return items
  }).pipe(Effect.tx)