| 774 | * @since 2.0.0 |
| 775 | */ |
| 776 | export const takeAll = <A, E>(self: TxDequeue<A, E>): Effect.Effect<Arr.NonEmptyArray<A>, E> => |
| 777 | Effect.gen(function*() { |
| 778 | const state = yield* TxRef.get(self.stateRef) |
| 779 | |
| 780 | // Handle done queue |
| 781 | if (state._tag === "Done") { |
| 782 | return yield* Effect.failCause(state.cause) |
| 783 | } |
| 784 | |
| 785 | // Wait if empty - same pattern as take() |
| 786 | if (yield* isEmpty(self)) { |
| 787 | return yield* Effect.txRetry |
| 788 | } |
| 789 | |
| 790 | const chunk = yield* TxChunk.get(self.items) |
| 791 | |
| 792 | // Take all items (guaranteed non-empty due to isEmpty check above) |
| 793 | const items = Chunk.toArray(chunk) as Arr.NonEmptyArray<A> |
| 794 | yield* TxChunk.set(self.items, Chunk.empty()) |
| 795 | |
| 796 | // Check if we need to transition Closing → Done |
| 797 | if (state._tag === "Closing") { |
| 798 | yield* TxRef.set(self.stateRef, { _tag: "Done", cause: state.cause }) |
| 799 | } |
| 800 | |
| 801 | return items |
| 802 | }).pipe(Effect.tx) |
| 803 | |
| 804 | /** |
| 805 | * Takes up to `n` items from the queue in a single transaction. |