Hyperlinkv0.8.0-beta.28

TxQueue

TxQueue.takeconsteffect/TxQueue.ts:665
<A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>

Takes the next item from the queue, retrying the transaction while the queue is empty.

Details

If the queue is done, the effect fails with the queue's completion cause. This function mutates the original TxQueue by removing the first item. It does not return a new TxQueue reference.

Example (Taking a value)

import { Effect, TxQueue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* TxQueue.bounded<number, string>(10)
  yield* TxQueue.offer(queue, 42)

  // Take an item - blocks if empty
  const item = yield* TxQueue.take(queue)
  console.log(item) // 42

  // When queue fails, take fails with the same error
  yield* TxQueue.fail(queue, "queue error")
  const result = yield* Effect.flip(TxQueue.take(queue))
  console.log(result) // "queue error"
})
combinators
Source effect/TxQueue.ts:66530 lines
export const take = <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E> =>
  Effect.gen(function*() {
    const state = yield* TxRef.get(self.stateRef)

    // Check if queue is done - forward the cause directly
    if (state._tag === "Done") {
      return yield* Effect.failCause(state.cause)
    }

    // If no items available, retry transaction
    if (yield* isEmpty(self)) {
      return yield* Effect.txRetry
    }

    // Take item from queue
    const chunk = yield* TxChunk.get(self.items)
    const head = Chunk.head(chunk)
    if (Option.isNone(head)) {
      return yield* Effect.txRetry
    }

    yield* TxChunk.drop(self.items, 1)

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

    return head.value
  }).pipe(Effect.tx)
Referenced by 1 symbols