42 - Queue: Asynchronous Producer / Consumer Buffering

Previous · Next

A Queue<A> is an asynchronous FIFO buffer. offer and take are Effects that suspend the fiber when the queue is full or empty, waking it up automatically when conditions change.

The capacity strategy decides what happens on overflow.

What you’ll work through

index.test.ts contains three beats: the four capacity strategies (bounded, unbounded, sliding, dropping) and how each handles overflow, basic offer / take semantics including back-pressure, and a producer-consumer pipeline assembled from two concurrent fibers.

By the end you should be able to choose between Queue (point-to-point buffer), PubSub (broadcast), and Semaphore (capacity gate) without hesitation.

Concept

1
Queue<A>   asynchronous FIFO
2

3
  Producer ──offer(item)──▶  [a, b, c]  ──take──▶ Consumer
4

5
Capacity strategies:
6
  bounded   ──▶ offer suspends when full   (default for back-pressure)
7
  dropping  ──▶ offer drops the new item   (shed load)
8
  sliding   ──▶ offer evicts the oldest    (keep the newest)
9
  unbounded ──▶ no cap                     (memory risk; use sparingly)

take always suspends when the queue is empty, there is no polling. takeAll returns whatever is available right now and never suspends.

Constructors

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

3
//      ┌─── Effect<Queue.Queue<string>>
4
//      ▼
5
const bounded   = Queue.bounded<string>(100)
6
const dropping  = Queue.dropping<string>(100)
7
const sliding   = Queue.sliding<string>(100)
8
const unbounded = Queue.unbounded<string>()

When to Use

bounded, the safe default. Back-pressure protects the consumer.
dropping, when missing a message is preferable to slowing the producer.
sliding, telemetry, last-N readings, “latest wins” semantics.
unbounded, only when both sides are clearly rate-matched.

`Queue.offer` and `Queue.take`

1
//      ┌─── Effect<ReadonlyArray<string>>
2
//      ▼
3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<string>(100)
5
  yield* Queue.offer(queue, "hello")
6
  yield* Queue.offer(queue, "world")
7
  const a = yield* Queue.take(queue)
8
  const b = yield* Queue.take(queue)
9
  return [a, b] as const
10
})
11
// ["hello", "world"]

offer returns a boolean indicating whether the item was accepted (false for a dropping queue that discarded it). take returns A once one is available.

`Queue.offerAll` and `Queue.takeAll`

Bulk variants. takeAll returns a Chunk<A> of the items available right now, it does not wait.

1
yield* Queue.offerAll(queue, ["a", "b", "c"])
2
const drained = yield* Queue.takeAll(queue) // Chunk<string>

`Queue.shutdown`

Signals that no more items will be offered. Pending and future take calls fail with an interruption, which is how consumers learn the producer is done.

1
yield* Queue.shutdown(queue)

Composite Example: Producer / Consumer

1
//      ┌─── Effect<ReadonlyArray<number>>
2
//      ▼
3
const program = Effect.gen(function* () {
4
  const queue = yield* Queue.bounded<number>(10)
5
  const processed: Array<number> = []
6

7
  const producer = Effect.gen(function* () {
8
    for (let i = 1; i <= 5; i++) {
9
      yield* Queue.offer(queue, i)
10
    }
11
  })
12

13
  const consumer = Effect.gen(function* () {
14
    for (let i = 0; i < 5; i++) {
15
      const item = yield* Queue.take(queue)
16
      processed.push(item * 2)
17
    }
18
  })
19

20
  yield* Effect.all([producer, consumer], { concurrency: "unbounded" })
21
  return processed
22
})

The producer and consumer run as separate fibers. Back-pressure handles the case where the producer is faster than the consumer.

Build a producer/consumer from scratch

The producer/consumer pattern is two fibers talking through a bounded Queue. The producer offers items; the consumer takes them.

The Queue takes care of back-pressure: when full it suspends the producer; when empty it suspends the consumer.

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

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

6
  const producer = Effect.gen(function* () {
7
    for (let i = 0; i < 10; i++) yield* queue.offer(i)
8
  })
9

10
  const consumer = Effect.gen(function* () {
11
    const out: Array<number> = []
12
    for (let i = 0; i < 10; i++) out.push(yield* queue.take)
13
    return out
14
  })
15

16
  // Run both concurrently — Queue, Effect.all, and concurrency fit as one pattern.
17
  const [, result] = yield* Effect.all([producer, consumer], {
18
    concurrency: "unbounded",
19
  })
20
  return result
21
})

Building this from scratch shows that there’s nothing magical about “producer/consumer” — it’s Queue + two concurrent Effects. Once the shape is in your head, every variation (multiple producers, multiple consumers, sliding overflow) is a one-line change.

Your turn wire a `producer` and a `consumer` against a single bounded `Queue`, run them concurrently with `Effect.all`, and assert the consumer collected every item the producer offered.

Common Pitfalls

1, Calling takeAll when you mean take. takeAll returns immediately with whatever is available, possibly nothing. take waits.

2, Reaching for Queue.unbounded to avoid back-pressure. Without a cap, a fast producer is a memory leak waiting to happen. Pick a finite bound and a strategy.

3, Treating Queue.shutdown as graceful flush. It is not. Shutdown interrupts pending takers.

To flush, drain explicitly with Queue.takeAll before shutting down.

4, Forgetting that Queue.offer on a dropping queue returns false on drop. Inspect the boolean if drop must be observed.

overflow. - [ ] Exercise `offer` / `take` and confirm back-pressure when the queue is full or empty. - [ ] Build a producer-consumer pipeline as two concurrent fibers connected by the queue.

The full test suite

The index.test.ts for this lesson, verbatim from steps/42-queue/.

1
import { describe, it, expect } from "vite-plus/test"
2
import { Effect, Fiber, Queue } from "effect"
3

4
// =============================================================================
5
// 39 - Queue: Asynchronous Producer / Consumer Buffering
6
// =============================================================================
7
//
8
// Queue.bounded / dropping / sliding / unbounded
9
// Queue.offer / offerAll / take / takeAll / size / shutdown
10

11
describe("39 - Queue", () => {
12
  // @beat
13
  //   id: offer-and-take
14
  //   shape: one-todo
15
  //   title: Offer and take preserve FIFO order
16
  //   teaches: Queue.offer enqueues, Queue.take dequeues; items come back out in the order they went in.
17
  //   hint: Queue.bounded<number>(10), then Queue.offer(queue, n) and yield* Queue.take(queue)
18
  describe("offer and take", () => {
19
    it("round-trips a single item", async () => {
20
      const program = /* @stub */ Effect.gen(function* () {
21
        const queue = yield* Queue.bounded<number>(10)
22
        yield* Queue.offer(queue, 42)
23
        return yield* Queue.take(queue)
24
      }) /* @end */
25

26
      expect(await Effect.runPromise(program)).toBe(42)
27
    })
28

29
    it("preserves FIFO order", async () => {
30
      const program = /* @stub */ Effect.gen(function* () {
31
        const queue = yield* Queue.bounded<number>(10)
32
        yield* Queue.offer(queue, 1)
33
        yield* Queue.offer(queue, 2)
34
        yield* Queue.offer(queue, 3)
35
        return [
36
          yield* Queue.take(queue),
37
          yield* Queue.take(queue),
38
          yield* Queue.take(queue),
39
        ]
40
      }) /* @end */
41

42
      expect(await Effect.runPromise(program)).toEqual([1, 2, 3])
43
    })
44
  })
45
  // @end-beat
46

47
  describe("offerAll and takeAll", () => {
48
    it("bulk-offers and drains available items", async () => {
49
      const program = Effect.gen(function* () {
50
        const queue = yield* Queue.unbounded<string>()
51
        yield* Queue.offerAll(queue, ["a", "b", "c"])
52
        return yield* Queue.takeAll(queue)
53
      })
54

55
      expect([...(await Effect.runPromise(program))]).toEqual(["a", "b", "c"])
56
    })
57
  })
58

59
  // @beat
60
  //   id: capacity-strategies
61
  //   shape: multi-todo
62
  //   title: Pick a capacity strategy
63
  //   teaches: bounded back-pressures, dropping discards new items, sliding evicts the oldest — match the strategy to the overflow policy you want.
64
  //   hint: Queue.bounded blocks the producer; Queue.dropping discards the new value; Queue.sliding evicts the head
65
  describe("Capacity strategies", () => {
66
    it("bounded queue applies back-pressure when full", async () => {
67
      const program = Effect.gen(function* () {
68
        const queue = yield* Queue.bounded<number>(2)
69
        yield* Queue.offer(queue, 1)
70
        yield* Queue.offer(queue, 2)
71

72
        // Forked offer suspends until a take frees a slot.
73
        const producer = yield* Effect.forkChild(Queue.offer(queue, 3))
74

75
        const taken = yield* Queue.take(queue)
76
        yield* Fiber.join(producer)
77
        const size = yield* Queue.size(queue)
78
        return { taken, size }
79
      })
80

81
      const result = await Effect.runPromise(program)
82
      expect(result.taken).toBe(1)
83
      expect(result.size).toBe(2) // 2 and 3 remain
84
    })
85

86
    it("dropping queue discards new items when full", async () => {
87
      const program = Effect.gen(function* () {
88
        const queue = yield* Queue.dropping<number>(2)
89
        yield* Queue.offer(queue, 1)
90
        yield* Queue.offer(queue, 2)
91
        yield* Queue.offer(queue, 3) // dropped
92
        return yield* Queue.takeAll(queue)
93
      })
94

95
      expect([...(await Effect.runPromise(program))]).toEqual([1, 2])
96
    })
97

98
    it("sliding queue evicts the oldest item when full", async () => {
99
      const program = /* @stub */ Effect.gen(function* () {
100
        const queue = yield* Queue.sliding<number>(2)
101
        yield* Queue.offer(queue, 1)
102
        yield* Queue.offer(queue, 2)
103
        yield* Queue.offer(queue, 3) // 1 evicted
104
        return yield* Queue.takeAll(queue)
105
      }) /* @end */
106

107
      expect([...(await Effect.runPromise(program))]).toEqual([2, 3])
108
    })
109
  })
110
  // @end-beat
111

112
  // @beat
113
  //   id: producer-consumer
114
  //   shape: one-todo
115
  //   title: Run producer and consumer fibers concurrently
116
  //   teaches: Queues suspend on offer when full and on take when empty — perfect for handing work between fibers.
117
  //   hint: the producer offers 1..5 in order; the consumer doubles each — predict the final array
118
  describe("Producer / consumer", () => {
119
    it("runs producer and consumer fibers concurrently", async () => {
120
      const program = Effect.gen(function* () {
121
        const queue = yield* Queue.bounded<number>(10)
122
        const processed: Array<number> = []
123

124
        const producer = Effect.gen(function* () {
125
          for (let i = 1; i <= 5; i++) {
126
            yield* Queue.offer(queue, i)
127
          }
128
        })
129

130
        const consumer = Effect.gen(function* () {
131
          for (let i = 0; i < 5; i++) {
132
            processed.push((yield* Queue.take(queue)) * 2)
133
          }
134
        })
135

136
        yield* Effect.all([producer, consumer], { concurrency: "unbounded" })
137
        return processed
138
      })
139

140
      // Predict the consumed-and-doubled array in FIFO order
141
      expect(await Effect.runPromise(program)).toEqual(/* @stub */ [2, 4, 6, 8, 10] /* @end */)
142
    })
143
  })
144
  // @end-beat
145

146
  // @beat
147
  //   id: producer-consumer-build
148
  //   shape: from-scratch
149
  //   title: Build the producer-consumer pipeline yourself
150
  //   teaches: A Queue connects two concurrent fibers — the producer offers, the consumer takes. Build this from scratch to see how Queue, Effect.all, and concurrency fit together as one pattern.
151
  //   hint: Queue.bounded<number>(10), then two generators — one offers 1..5, one takes 5 times; run both with Effect.all concurrency unbounded
152

153
  describe("capstone: build the pipeline", () => {
154
    it("producer offers 1..5, consumer doubles each value — build both fibers", async () => {
155
      const processed: Array<number> = []
156

157
      //      ┌─── Effect<void, never, never>
158
      //      ▼
159
      const program = /* @stub */ Effect.gen(function* () {
160
        const queue = yield* Queue.bounded<number>(10)
161

162
        const producer = Effect.gen(function* () {
163
          for (let i = 1; i <= 5; i++) {
164
            yield* Queue.offer(queue, i)
165
          }
166
        })
167

168
        const consumer = Effect.gen(function* () {
169
          for (let i = 0; i < 5; i++) {
170
            const n = yield* Queue.take(queue)
171
            processed.push(n * 2)
172
          }
173
        })
174

175
        yield* Effect.all([producer, consumer], { concurrency: "unbounded" })
176
      }) /* @end */
177

178
      await Effect.runPromise(program)
179
      expect(processed).toEqual([2, 4, 6, 8, 10])
180
    })
181
  })
182
  // @end-beat
183
})

07·01Queue

42 - Queue: Asynchronous Producer / Consumer Buffering

What you’ll work through

Concept

Constructors

`Queue.offer` and `Queue.take`

`Queue.offerAll` and `Queue.takeAll`

`Queue.shutdown`

Composite Example: Producer / Consumer

Build a producer/consumer from scratch

Common Pitfalls

See Also

The full test suite

07·01Queue

42 - Queue: Asynchronous Producer / Consumer Buffering

What you’ll work through

Concept

Constructors

Queue.offer and Queue.take

Queue.offerAll and Queue.takeAll

Queue.shutdown

Composite Example: Producer / Consumer

Build a producer/consumer from scratch

Common Pitfalls

See Also

The full test suite

`Queue.offer` and `Queue.take`

`Queue.offerAll` and `Queue.takeAll`

`Queue.shutdown`