Queue

A Queue is a lightweight in-memory queue with built-in back-pressure, enabling asynchronous, purely-functional, and type-safe handling of data.

Basic Operations

A Queue<A> stores values of type A and provides two fundamental operations:

API	Description
`Queue.offer`	Adds a value of type `A` to the queue.
`Queue.take`	Removes and returns the oldest value from the queue.

Example (Adding and Retrieving an Item)

1
import { Effect, Queue } from "effect"
2

3
const program = Effect.gen(function* () {
4
  // Creates a bounded queue with capacity 100
5
  const queue = yield* Queue.bounded<number>(100)
6
  // Adds 1 to the queue
7
  yield* Queue.offer(queue, 1)
8
  // Retrieves and removes the oldest value
9
  const value = yield* Queue.take(queue)
10
  return value
11
})
12

13
await Effect.runPromise(program) // => 1

Creating a Queue

Queues can be bounded (with a specified capacity) or unbounded (without a limit). Different types of queues handle new values differently when they reach capacity.

Bounded Queue

A bounded queue applies back-pressure when full, meaning any Queue.offer operation will suspend until there is space.

Example (Creating a Bounded Queue)

1
import { Effect, Queue } from "effect"
2

3
// Creating a bounded queue with a capacity of 100
4
const boundedQueue = Queue.bounded<number>(100)
5

6
;(await Effect.runPromise(boundedQueue)).capacity // => 100

Dropping Queue

A dropping queue discards new values if the queue is full.

Example (Creating a Dropping Queue)

1
import { Effect, Queue } from "effect"
2

3
// Creating a dropping queue with a capacity of 100
4
const droppingQueue = Queue.dropping<number>(100)
5

6
;(await Effect.runPromise(droppingQueue)).capacity // => 100

Sliding Queue

A sliding queue removes old values to make space for new ones when it reaches capacity.

Example (Creating a Sliding Queue)

1
import { Effect, Queue } from "effect"
2

3
// Creating a sliding queue with a capacity of 100
4
const slidingQueue = Queue.sliding<number>(100)
5

6
;(await Effect.runPromise(slidingQueue)).capacity // => 100

Unbounded Queue

An unbounded queue has no capacity limit, allowing unrestricted additions.

Example (Creating an Unbounded Queue)

1
import { Effect, Queue } from "effect"
2

3
// Creates an unbounded queue without a capacity limit
4
const unboundedQueue = Queue.unbounded<number>()
5

6
;(await Effect.runPromise(unboundedQueue)).capacity // => Infinity

Adding Items to a Queue

offer

Use Queue.offer to add values to the queue.

Example (Adding a Single Item)

1
import { Effect, Queue } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(100)
5
  // Adds 1 to the queue
6
  yield* Queue.offer(queue, 1)
7
  return yield* Queue.size(queue)
8
})
9

10
await Effect.runPromise(program) // => 1

When using a back-pressured queue, Queue.offer suspends if the queue is full. To avoid blocking the main fiber, you can fork the Queue.offer operation.

Example (Handling a Full Queue with Effect.forkChild)

1
import { Effect, Queue, Fiber } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(1)
5
  // Fill the queue with one item
6
  yield* Queue.offer(queue, 1)
7
  // Attempting to add a second item will suspend as the queue is full
8
  const fiber = yield* Effect.forkChild(Queue.offer(queue, 2))
9
  // Empties the queue to make space
10
  yield* Queue.take(queue)
11
  // Joins the fiber, completing the suspended offer
12
  yield* Fiber.join(fiber)
13
  // Returns the size of the queue after additions
14
  return yield* Queue.size(queue)
15
})
16

17
await Effect.runPromise(program) // => 1

offerAll

You can also add multiple items at once using Queue.offerAll.

Example (Adding Multiple Items)

1
import { Effect, Queue, Array } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(100)
5
  const items = Array.range(1, 10)
6
  // Adds all items to the queue at once
7
  yield* Queue.offerAll(queue, items)
8
  // Returns the size of the queue after additions
9
  return yield* Queue.size(queue)
10
})
11

12
await Effect.runPromise(program) // => 10

Consuming Items from a Queue

take

The Queue.take operation removes and returns the oldest item from the queue. If the queue is empty, Queue.take will suspend and only resume when an item is added. To prevent blocking, you can fork the Queue.take operation into a new fiber.

Example (Waiting for an Item in a Fiber)

1
import { Effect, Queue, Fiber } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<string>(100)
5
  // This take operation will suspend because the queue is empty
6
  const fiber = yield* Effect.forkChild(Queue.take(queue))
7
  // Adds an item to the queue
8
  yield* Queue.offer(queue, "something")
9
  // Joins the fiber to get the result of the take operation
10
  const value = yield* Fiber.join(fiber)
11
  return value
12
})
13

14
await Effect.runPromise(program) // => "something"

poll

To retrieve the queue’s first item without suspending, use Queue.poll. If the queue is empty, Queue.poll returns None; if it has an item, it wraps it in Some.

Example (Polling an Item)

1
import { Effect, Queue, Option } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(100)
5
  // Adds items to the queue
6
  yield* Queue.offer(queue, 10)
7
  yield* Queue.offer(queue, 20)
8
  // Retrieves the first item if available
9
  const head = yield* Queue.poll(queue)
10
  return head
11
})
12

13
await Effect.runPromise(program) // => Option.some(10)

takeUpTo

To retrieve multiple items, use Queue.takeBetween, which returns up to the specified number of items. If there aren’t enough items, it returns all available items without waiting for more.

This function is particularly useful for batch processing when an exact number of items is not required. It ensures the program continues working with whatever data is currently available.

If you need to wait for an exact number of items before proceeding, consider using takeN.

Example (Taking Up to N Items)

1
import { Effect, Queue } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(100)
5

6
  // Adds items to the queue
7
  yield* Queue.offer(queue, 1)
8
  yield* Queue.offer(queue, 2)
9
  yield* Queue.offer(queue, 3)
10

11
  // Retrieves up to 2 items (min must be at least 1, otherwise
12
  // takeBetween short-circuits and returns an empty array)
13
  const items = yield* Queue.takeBetween(queue, 1, 2)
14
  console.log(items)
15

16
  return "some result"
17
})
18

19
Effect.runPromise(program).then(console.log)
20
/*
21
Output:
22
[ 1, 2 ]
23
some result
24
*/

takeN

Takes a specified number of elements from a queue. If the queue does not contain enough elements, the operation suspends until the required number of elements become available.

This function is useful for scenarios where processing requires an exact number of items at a time, ensuring that the operation does not proceed until the batch is complete.

Example (Taking a Fixed Number of Items)

1
import { Effect, Queue, Fiber } from "effect"
2

3
const program = Effect.gen(function* () {
4
  // Create a queue that can hold up to 100 elements
5
  const queue = yield* Queue.bounded<number>(100)
6

7
  // Fork a fiber that attempts to take 3 items from the queue
8
  const fiber = yield* Effect.forkChild(
9
    Effect.gen(function* () {
10
      console.log("Attempting to take 3 items from the queue...")
11
      const chunk = yield* Queue.takeN(queue, 3)
12
      console.log(`Successfully took 3 items: ${chunk}`)
13
    }),
14
  )
15

16
  // Offer only 2 items initially
17
  yield* Queue.offer(queue, 1)
18
  yield* Queue.offer(queue, 2)
19
  console.log("Offered 2 items. The fiber is now waiting for the 3rd item...")
20

21
  // Simulate some delay
22
  yield* Effect.sleep("2 seconds")
23

24
  // Offer the 3rd item, which will unblock the takeN call
25
  yield* Queue.offer(queue, 3)
26
  console.log("Offered the 3rd item, which should unblock the fiber.")
27

28
  // Wait for the fiber to finish
29
  yield* Fiber.join(fiber)
30
  return "some result"
31
})
32

33
await Effect.runPromise(program) // => "some result"
34
/*
35
Output:
36
Offered 2 items. The fiber is now waiting for the 3rd item...
37
Attempting to take 3 items from the queue...
38
Offered the 3rd item, which should unblock the fiber.
39
Successfully took 3 items: 1,2,3
40
*/

takeAll

To retrieve all items from the queue at once, use Queue.takeAll. This operation completes immediately, returning an empty collection if the queue is empty.

Example (Taking All Items)

1
import { Effect, Queue } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(100)
5
  // Adds items to the queue
6
  yield* Queue.offer(queue, 10)
7
  yield* Queue.offer(queue, 20)
8
  yield* Queue.offer(queue, 30)
9
  // Retrieves all items from the queue
10
  const chunk = yield* Queue.takeAll(queue)
11
  return chunk
12
})
13

14
await Effect.runPromise(program) // => [10, 20, 30]

Shutting Down a Queue

shutdown

The Queue.shutdown operation allows you to interrupt all fibers that are currently suspended on offer* or take* operations. This action also empties the queue and makes any future offer* and take* calls terminate immediately.

Example (Interrupting Fibers on Queue Shutdown)

1
import { Effect, Queue, Fiber } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(3)
5
  // Forks a fiber that waits to take an item from the queue
6
  const fiber = yield* Effect.forkChild(Queue.take(queue))
7
  // Shuts down the queue, interrupting the fiber
8
  yield* Queue.shutdown(queue)
9
  // Joins the interrupted fiber
10
  yield* Fiber.join(fiber)
11
})
12

13
const exit = await Effect.runPromiseExit(program)
14
exit._tag // => "Failure"

await

The Queue.await operation waits until the queue reaches its Done state. Because Queue.shutdown completes the queue with an interrupt cause, an effect that is awaiting shutdown is itself interrupted, not resolved as a normal success. To run cleanup logic regardless of that outcome, use Effect.onExit instead of Effect.andThen, and use Fiber.await instead of Fiber.join so the interruption does not propagate to the fiber that joins it.

Example (Waiting for Queue Shutdown)

1
import { Effect, Queue, Fiber, Console } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(3)
5
  // Forks a fiber to await queue shutdown and log a message,
6
  // regardless of whether the wait completes or is interrupted
7
  const fiber = yield* Effect.forkChild(
8
    Queue.await(queue).pipe(Effect.onExit(() => Console.log("shutting down"))),
9
  )
10
  // Shuts down the queue, triggering (and interrupting) the await in the fiber
11
  yield* Queue.shutdown(queue)
12
  yield* Fiber.await(fiber)
13
})
14

15
await Effect.runPromise(program) // => undefined
16
// Output: shutting down

Offer-only / Take-only Queues

Sometimes, you might want certain parts of your code to only add values to a queue (Enqueue) or only retrieve values from a queue (Dequeue). Effect provides interfaces to enforce these specific capabilities.

Enqueue

All methods for adding values to a queue are defined by the Enqueue interface. This restricts the queue to only offer operations.

Example (Restricting Queue to Offer-only Operations)

1
import { Queue } from "effect"
2

3
const send = (offerOnlyQueue: Queue.Enqueue<number>, value: number) => {
4
  // This queue is restricted to offer operations only
5

6
  // Error: cannot use take on an offer-only queue
7
  // @errors: 2345
8
  Queue.take(offerOnlyQueue)
9

10
  // Valid offer operation
11
  return Queue.offer(offerOnlyQueue, value)
12
}

Dequeue

Similarly, all methods for retrieving values from a queue are defined by the Dequeue interface, which restricts the queue to only take operations.

Example (Restricting Queue to Take-only Operations)

1
import { Queue } from "effect"
2

3
const receive = (takeOnlyQueue: Queue.Dequeue<number>) => {
4
  // This queue is restricted to take operations only
5

6
  // Error: cannot use offer on a take-only queue
7
  // @errors: 2345
8
  Queue.offer(takeOnlyQueue, 1)
9

10
  // Valid take operation
11
  return Queue.take(takeOnlyQueue)
12
}

The Queue type combines both Enqueue and Dequeue, so you can easily pass it to different parts of your code, enforcing only Enqueue or Dequeue behaviors as needed.

Example (Using Offer-only and Take-only Queues Together)

1
import { Effect, Queue } from "effect"
2

3
const send = (offerOnlyQueue: Queue.Enqueue<number>, value: number) => {
4
  return Queue.offer(offerOnlyQueue, value)
5
}
6

7
const receive = (takeOnlyQueue: Queue.Dequeue<number>) => {
8
  return Queue.take(takeOnlyQueue)
9
}
10

11
const program = Effect.gen(function* () {
12
  const queue = yield* Queue.unbounded<number>()
13

14
  // Add values to the queue
15
  yield* send(queue, 1)
16
  yield* send(queue, 2)
17

18
  // Retrieve values from the queue
19
  const first = yield* receive(queue)
20
  const second = yield* receive(queue)
21
  console.log(first)
22
  console.log(second)
23
  first // => 1
24
  second // => 2
25
})
26

27
await Effect.runPromise(program) // => undefined