1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Effect, Exit, Fiber, Option, Queue, Stream } from "effect"
3import * as Scheduler from "effect/Scheduler"
4
5describe("Queue", () => {
6 describe("waiter registration after a yield", () => {
7 // Budget 3 yields after the first queue check but before the old waiter registration.
8 const withYield = <A, E>(effect: Effect.Effect<A, E>) =>
9 Effect.provideService(effect, Scheduler.MaxOpsBeforeYield, 3)
10
11 const assertCompletes = (fiber: Fiber.Fiber<unknown, unknown>) =>
12 Effect.gen(function*() {
13 for (let i = 0; i < 200 && fiber.pollUnsafe() === undefined; i++) yield* Effect.yieldNow
14 const done = fiber.pollUnsafe() !== undefined
15 yield* Fiber.interrupt(fiber)
16 assert.isTrue(done, "lost wake-up")
17 })
18
19 it.effect("offerAll does not park beside available capacity", () =>
20 Effect.gen(function*() {
21 const queue = yield* Queue.bounded<number>(1)
22 yield* Queue.offer(queue, 0)
23 const fiber = yield* Effect.forkChild(withYield(Queue.offerAll(queue, [1])), { startImmediately: true })
24 yield* Queue.take(queue)
25 yield* assertCompletes(fiber)
26 }))
27
28 it.effect("zero-capacity offer and take can rendezvous", () =>
29 Effect.gen(function*() {
30 const queue = yield* Queue.bounded<number>(0)
31 const offerer = yield* Effect.forkChild(withYield(Queue.offer(queue, 1)), { startImmediately: true })
32 const taker = yield* Effect.forkChild(Queue.take(queue), { startImmediately: true })
33 yield* assertCompletes(offerer)
34 yield* assertCompletes(taker)
35 }))
36 })
37
38 it.effect("isEnqueue type guard", () =>
39 Effect.gen(function*() {
40 const queue = yield* Queue.bounded<number>(10)
41
42 assert.isTrue(Queue.isEnqueue(queue))
43 assert.isFalse(Queue.isEnqueue({}))
44 assert.isFalse(Queue.isEnqueue(null))
45 }))
46
47 it.effect("isDequeue type guard", () =>
48 Effect.gen(function*() {
49 const queue = yield* Queue.bounded<number>(10)
50
51 assert.isTrue(Queue.isDequeue(queue))
52 assert.isFalse(Queue.isDequeue({}))
53 assert.isFalse(Queue.isDequeue(null))
54 }))
55
56 it.effect("asEnqueue converts to write-only interface", () =>
57 Effect.gen(function*() {
58 const queue = yield* Queue.bounded<number>(10)
59 const enqueue: Queue.Enqueue<number> = Queue.asEnqueue(queue)
60
61 // Verify it's recognized as an enqueue
62 assert.isTrue(Queue.isEnqueue(enqueue))
63
64 // Verify queue operations still work through enqueue reference
65 assert.isTrue(Queue.isQueue(enqueue))
66 }))
67
68 it.effect("asDequeue converts to read-only interface", () =>
69 Effect.gen(function*() {
70 const queue = yield* Queue.bounded<number>(10)
71 yield* Queue.offer(queue, 42)
72
73 const dequeue = Queue.asDequeue(queue)
74
75 // Can use dequeue operations
76 const item = yield* Queue.take(dequeue)
77 assert.strictEqual(item, 42)
78
79 // Verify it's recognized as a dequeue
80 assert.isTrue(Queue.isDequeue(dequeue))
81 }))
82
83 it.effect("bounded offerAll waits until capacity is released", () =>
84 Effect.gen(function*() {
85 const queue = yield* Queue.bounded<number>(2)
86 const fiber = yield* Queue.offerAll(queue, [1, 2, 3, 4]).pipe(
87 Effect.forkChild
88 )
89 yield* Effect.yieldNow
90 assert.isUndefined(fiber.pollUnsafe())
91
92 let result = yield* Queue.takeAll(queue)
93 assert.deepStrictEqual(result, [1, 2])
94
95 yield* Effect.yieldNow
96 assert.isDefined(fiber.pollUnsafe())
97
98 result = yield* Queue.takeAll(queue)
99 assert.deepStrictEqual(result, [3, 4])
100
101 yield* Effect.yieldNow
102 assert.deepStrictEqual(fiber.pollUnsafe(), Exit.succeed([]))
103 }))
104
105 it.effect("interrupting a suspended offer after end withdraws its message", () =>
106 Effect.gen(function*() {
107 const queue = yield* Queue.bounded<number, Cause.Done>(1)
108 yield* Queue.offer(queue, 1)
109 const producer = yield* Effect.forkChild(Queue.offer(queue, 2), { startImmediately: true })
110 assert.isUndefined(producer.pollUnsafe(), "offer must be suspended before end")
111
112 yield* Queue.end(queue)
113 yield* Fiber.interrupt(producer)
114
115 assert.deepStrictEqual(yield* Queue.collect(queue), [1])
116 }))
117
118 it.effect("interrupting a suspended offerAll after end completes an empty queue", () =>
119 Effect.gen(function*() {
120 const queue = yield* Queue.bounded<number, Cause.Done>(0)
121 const producer = yield* Effect.forkChild(Queue.offerAll(queue, [1, 2]), { startImmediately: true })
122 assert.isUndefined(producer.pollUnsafe(), "offerAll must be suspended before end")
123
124 yield* Queue.end(queue)
125 const awaiter = yield* Effect.forkChild(Queue.await(queue), { startImmediately: true })
126 assert.isUndefined(awaiter.pollUnsafe(), "await must wait for the pending offer")
127
128 yield* Fiber.interrupt(producer)
129 yield* Fiber.join(awaiter)
130 assert.deepStrictEqual(yield* Effect.exit(Queue.take(queue)), Exit.fail(Cause.Done()))
131 }))
132
133 it.effect("resuming a blocked producer does not enqueue its message twice", () =>
134 Effect.gen(function*() {
135 const queue = yield* Queue.bounded<number>(1)
136 yield* Queue.offer(queue, 0)
137 const producer = yield* Effect.forkChild(
138 Effect.gen(function*() {
139 yield* Queue.offer(queue, 1)
140 return yield* Queue.take(queue)
141 }),
142 { startImmediately: true }
143 )
144
145 const first = yield* Queue.take(queue)
146 const second = yield* Fiber.join(producer)
147 const remaining = yield* Queue.clear(queue)
148 yield* Queue.shutdown(queue)
149
150 assert.deepStrictEqual([first, second], [0, 1])
151 assert.deepStrictEqual(remaining, [])
152 }))
153
154 for (const method of ["offer", "offerAll"]) {
155 for (const take of ["take", "takeAll", "clear"]) {
156 it.effect(`zero-capacity ${take} reserves the message before resuming ${method}`, () =>
157 Effect.gen(function*() {
158 const queue = yield* Queue.bounded<number>(0)
159 const producer = yield* Effect.forkChild(
160 Effect.gen(function*() {
161 if (method === "offer") yield* Queue.offer(queue, 1)
162 else yield* Queue.offerAll(queue, [1])
163 return yield* Queue.take(queue)
164 }),
165 { startImmediately: true }
166 )
167
168 const first = take === "take"
169 ? [yield* Queue.take(queue)]
170 : take === "takeAll"
171 ? yield* Queue.takeAll(queue)
172 : yield* Queue.clear(queue)
173 const producerExit = producer.pollUnsafe()
174 yield* Queue.shutdown(queue)
175
176 assert.deepStrictEqual({ first, producerExit }, { first: [1], producerExit: undefined })
177 }))
178 }
179
180 it.effect(`resuming ${method} can shut down the queue without defecting the consumer`, () =>
181 Effect.gen(function*() {
182 const queue = yield* Queue.bounded<number>(1)
183 yield* Queue.offer(queue, 0)
184 const producer = yield* Effect.forkChild(
185 Effect.gen(function*() {
186 if (method === "offer") yield* Queue.offer(queue, 1)
187 else yield* Queue.offerAll(queue, [1])
188 yield* Queue.shutdown(queue)
189 }),
190 { startImmediately: true }
191 )
192
193 const exit = yield* Effect.exit(Queue.take(queue))
194 yield* Fiber.join(producer)
195
196 assert.deepStrictEqual(exit, Exit.succeed(0))
197 }))
198
199 it.effect(`resuming ${method} does not overfill the queue with reentrant offers`, () =>
200 Effect.gen(function*() {
201 const queue = yield* Queue.bounded<number>(2)
202 yield* Queue.offerAll(queue, [0, 1])
203 yield* Effect.forkChild(
204 Effect.gen(function*() {
205 if (method === "offer") yield* Queue.offer(queue, 2)
206 else yield* Queue.offerAll(queue, [2])
207 yield* Queue.offer(queue, 3)
208 }),
209 { startImmediately: true }
210 )
211 yield* Effect.forkChild(Queue.offer(queue, 4), { startImmediately: true })
212
213 const first = yield* Queue.takeAll(queue)
214 const size = Queue.sizeUnsafe(queue)
215 yield* Queue.shutdown(queue)
216
217 assert.deepStrictEqual(first, [0, 1])
218 assert.isAtMost(size, 2)
219 }))
220 }
221
222 it.effect("takeN", () =>
223 Effect.gen(function*() {
224 const queue = yield* Queue.unbounded<number>()
225 yield* Queue.offerAll(queue, [1, 2, 3, 4]).pipe(Effect.forkChild)
226 const a = yield* Queue.takeN(queue, 2)
227 const b = yield* Queue.takeN(queue, 2)
228 assert.deepEqual(a, [1, 2])
229 assert.deepEqual(b, [3, 4])
230 }))
231
232 it("takeN on a partial batch does not wake its taker", () => {
233 // Drive the queue's scheduler by hand so every taker release runs
234 // synchronously inside the test instead of on a host timer.
235 const tasks: Array<() => void> = []
236 const scheduler: Scheduler.Scheduler = {
237 executionMode: "async",
238 shouldYield: () => false,
239 makeDispatcher: () => ({
240 scheduleTask: (task) => {
241 tasks.push(task)
242 },
243 flush() {}
244 })
245 }
246 const queue = Effect.runSync(
247 Queue.unbounded<number>().pipe(Effect.provideService(Scheduler.Scheduler, scheduler))
248 )
249 const taker = Effect.runFork(Queue.takeN(queue, 2), { scheduler })
250 assert(queue.state._tag === "Open")
251 const takers = queue.state.takers
252 assert.strictEqual(takers.size, 1)
253 // Releasing a taker deletes it first, so the old wake/retry loop trips
254 // this on its first turn instead of spinning forever.
255 takers.delete = () => {
256 throw new Error("partial batch woke its taker")
257 }
258
259 Queue.offerUnsafe(queue, 1)
260 assert.strictEqual(tasks.length, 1)
261 tasks.shift()!()
262
263 assert.isUndefined(taker.pollUnsafe())
264 assert.strictEqual(takers.size, 1)
265 assert.isEmpty(tasks)
266 })
267
268 const lostWakeupCases: ReadonlyArray<readonly [string, (queue: Queue.Queue<number>) => Effect.Effect<unknown>]> = [
269 ["take", Queue.take],
270 ["takeN", (queue) => Queue.takeN(queue, 1)],
271 ["takeAll", Queue.takeAll],
272 ["takeBetween", (queue) => Queue.takeBetween(queue, 1, 5)],
273 ["peek", Queue.peek]
274 ]
275 for (const [name, receive] of lostWakeupCases) {
276 it.effect(`${name} does not miss an offer during the check-to-registration yield`, () =>
277 Effect.gen(function*() {
278 const queue = yield* Queue.unbounded<number>()
279 // At a budget of 8, this padding exhausts the taker budget after
280 // checking the empty queue but before registering in awaitTake.
281 let pad: Effect.Effect<void> = Effect.void
282 for (let i = 0; i < 5; i++) pad = Effect.andThen(pad, Effect.void)
283 const taker = yield* Effect.forkDetach(Effect.andThen(pad, receive(queue)))
284 const offerer = yield* Effect.forkDetach(Queue.offer(queue, 1))
285 for (let i = 0; i < 20; i++) yield* Effect.yieldNow
286
287 const exit = taker.pollUnsafe()
288 const size = yield* Queue.size(queue)
289 yield* Queue.shutdown(queue)
290 assert.deepStrictEqual(offerer.pollUnsafe(), Exit.succeed(true))
291 // Without the fix the taker stays suspended with no future wakeup, and
292 // the offered message remains in the queue.
293 assert.deepStrictEqual(exit, Exit.succeed(name === "take" || name === "peek" ? 1 : [1]))
294 assert.strictEqual(size, name === "peek" ? 1 : 0)
295 }).pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, 8)))
296 }
297
298 it.effect("takeN and takeBetween keep waiting on a partial batch", () =>
299 Effect.gen(function*() {
300 const takeNQueue = yield* Queue.unbounded<number>()
301 yield* Queue.offer(takeNQueue, 1)
302 const takeN = yield* Effect.forkDetach(Queue.takeN(takeNQueue, 3))
303
304 const takeBetweenQueue = yield* Queue.unbounded<number>()
305 yield* Queue.offer(takeBetweenQueue, 1)
306 const takeBetween = yield* Effect.forkDetach(Queue.takeBetween(takeBetweenQueue, 2, 5))
307
308 for (let i = 0; i < 20; i++) yield* Effect.yieldNow
309 // A buffered message below the minimum must not resume the taker.
310 assert.strictEqual(takeN.pollUnsafe(), undefined)
311 assert.strictEqual(takeBetween.pollUnsafe(), undefined)
312
313 yield* Queue.offerAll(takeNQueue, [2, 3])
314 yield* Queue.offerAll(takeBetweenQueue, [2, 3])
315 assert.deepStrictEqual(yield* Fiber.await(takeN), Exit.succeed([1, 2, 3]))
316 assert.deepStrictEqual(yield* Fiber.await(takeBetween), Exit.succeed([1, 2, 3]))
317 }))
318
319 it.effect("takeN drains an insufficient batch after end", () =>
320 Effect.gen(function*() {
321 const queue = yield* Queue.unbounded<number, Cause.Done>()
322 yield* Queue.offerAll(queue, [1, 2])
323 yield* Queue.end(queue)
324
325 assert.deepStrictEqual(yield* Queue.takeN(queue, 5), [1, 2])
326 assert.strictEqual(queue.state._tag, "Done")
327 assert.deepStrictEqual(yield* Effect.exit(Queue.takeN(queue, 5)), Exit.fail(Cause.Done()))
328 }))
329
330 it.effect("takeBetween drains a waiting batch on failure", () =>
331 Effect.gen(function*() {
332 const queue = yield* Queue.unbounded<number, string>()
333 yield* Queue.offerAll(queue, [1, 2])
334 const taker = yield* Effect.forkChild(Queue.takeBetween(queue, 5, 8), { startImmediately: true })
335 assert.strictEqual(taker.pollUnsafe(), undefined)
336
337 yield* Queue.fail(queue, "boom")
338 assert.deepStrictEqual(yield* Fiber.join(taker), [1, 2])
339 assert.strictEqual(queue.state._tag, "Done")
340 assert.deepStrictEqual(yield* Effect.exit(Queue.takeBetween(queue, 5, 8)), Exit.fail("boom"))
341 }))
342
343 it.effect("takeN drains on failure between the batch check and waiter registration", () =>
344 Effect.gen(function*() {
345 const queue = yield* Queue.unbounded<number, string>()
346 yield* Queue.offerAll(queue, [1, 2])
347 // At a budget of 8, this padding exhausts the taker budget after the
348 // first batch check but before registering in awaitTake.
349 const pad = Effect.andThen(Effect.void, Effect.void)
350 const taker = yield* Effect.forkDetach(Effect.andThen(pad, Effect.exit(Queue.takeN(queue, 5))))
351 let takersAtFail = -1
352 yield* Effect.forkDetach(Effect.suspend(() => {
353 takersAtFail = queue.state._tag === "Done" ? -1 : queue.state.takers.size
354 return Queue.fail(queue, "boom")
355 }))
356
357 assert.deepStrictEqual(yield* Fiber.join(taker), Exit.succeed([1, 2]))
358 // The taker yielded after its check and before registering, so the
359 // termination could not wake it.
360 assert.strictEqual(takersAtFail, 0)
361 assert.strictEqual(queue.state._tag, "Done")
362 assert.deepStrictEqual(yield* Effect.exit(Queue.takeN(queue, 5)), Exit.fail("boom"))
363 }).pipe(Effect.provideService(Scheduler.MaxOpsBeforeYield, 8)))
364
365 it.effect("takeN ending at an offerAll boundary keeps the next message", () =>
366 Effect.gen(function*() {
367 const queue = yield* Queue.unbounded<number>()
368 yield* Queue.offerAll(queue, [1, 2])
369 yield* Queue.offerAll(queue, [3])
370 assert.deepStrictEqual(yield* Queue.takeN(queue, 2), [1, 2])
371 assert.strictEqual(yield* Queue.size(queue), 1)
372 assert.strictEqual(yield* Queue.peek(queue), 3)
373 assert.strictEqual(yield* Queue.take(queue), 3)
374 yield* Queue.offer(queue, 99)
375 assert.strictEqual(yield* Queue.take(queue), 99)
376 assert.strictEqual(yield* Queue.size(queue), 0)
377 }))
378
379 it.effect("takeN and takeBetween normalize element counts before waiting", () =>
380 Effect.gen(function*() {
381 const takeNQueue = yield* Queue.unbounded<number>()
382 yield* Queue.offer(takeNQueue, 1)
383 assert.deepStrictEqual(yield* Queue.takeN(takeNQueue, 1.9), [1])
384
385 const takeBetweenQueue = yield* Queue.unbounded<number>()
386 yield* Queue.offer(takeBetweenQueue, 1)
387 assert.deepStrictEqual(yield* Queue.takeBetween(takeBetweenQueue, 1.9, 2.9), [1])
388
389 const invalidQueue = yield* Queue.unbounded<number>()
390 yield* Queue.offer(invalidQueue, 1)
391 assert.deepStrictEqual(yield* Queue.takeN(invalidQueue, Number.NaN), [])
392 assert.deepStrictEqual(yield* Queue.takeAll(invalidQueue), [1])
393 }))
394
395 it.effect("collect does not duplicate messages", () =>
396 Effect.gen(function*() {
397 const queue = yield* Queue.bounded<{ id: number }, Cause.Done>(10)
398 yield* Queue.offer(queue, { id: 0 })
399 yield* Queue.end(queue)
400
401 const result = yield* Queue.collect(queue)
402 assert.deepStrictEqual(result, [{ id: 0 }])
403 }))
404
405 it.effect("offer dropping", () =>
406 Effect.gen(function*() {
407 const queue = yield* Queue.make<number>({ capacity: 2, strategy: "dropping" })
408 const remaining = yield* Queue.offerAll(queue, [1, 2, 3, 4])
409 assert.deepStrictEqual(remaining, [3, 4])
410 const result = yield* Queue.offer(queue, 5)
411 assert.isFalse(result)
412 assert.deepStrictEqual(yield* Queue.takeAll(queue), [1, 2])
413 }))
414
415 it.effect("offer sliding", () =>
416 Effect.gen(function*() {
417 const queue = yield* Queue.make<number>({ capacity: 2, strategy: "sliding" })
418 const remaining = yield* Queue.offerAll(queue, [1, 2, 3, 4])
419 assert.deepStrictEqual(remaining, [])
420 const result = yield* Queue.offer(queue, 5)
421 assert.isTrue(result)
422 assert.deepStrictEqual(yield* Queue.takeAll(queue), [4, 5])
423 }))
424
425 it.effect("offerAll can be interrupted", () =>
426 Effect.gen(function*() {
427 const queue = yield* Queue.bounded<number>(2)
428 const fiber = yield* Queue.offerAll(queue, [1, 2, 3, 4]).pipe(
429 Effect.forkChild
430 )
431
432 yield* Effect.yieldNow
433 yield* Fiber.interrupt(fiber)
434 yield* Effect.yieldNow
435
436 let result = yield* Queue.takeAll(queue)
437 assert.deepStrictEqual(result, [1, 2])
438
439 yield* Queue.offer(queue, 5)
440 yield* Effect.yieldNow
441
442 result = yield* Queue.takeAll(queue)
443 assert.deepStrictEqual(result, [5])
444 }))
445
446 it.effect("take can be interrupted without losing offers", () =>
447 Effect.gen(function*() {
448 const queue = yield* Queue.unbounded<number>()
449 const interruptedFiber = yield* Queue.take(queue).pipe(Effect.forkChild)
450
451 yield* Effect.yieldNow
452 yield* Fiber.interrupt(interruptedFiber)
453 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(interruptedFiber)))
454
455 const liveFiber = yield* Queue.take(queue).pipe(Effect.forkChild)
456 yield* Effect.yieldNow
457 yield* Queue.offer(queue, 1)
458
459 assert.strictEqual(yield* Fiber.join(liveFiber), 1)
460 }))
461
462 it.effect("done completes takes", () =>
463 Effect.gen(function*() {
464 const queue = yield* Queue.bounded<number, Cause.Done>(2)
465 const fiber = yield* Queue.takeAll(queue).pipe(
466 Effect.forkChild
467 )
468 yield* Effect.yieldNow
469 yield* Queue.end(queue)
470 assert.deepStrictEqual(yield* Fiber.await(fiber), Exit.fail(Cause.Done()))
471 }))
472
473 it.effect("end", () =>
474 Effect.gen(function*() {
475 const queue = yield* Queue.bounded<number, Cause.Done>(2)
476 yield* Effect.forkChild(Queue.offerAll(queue, [1, 2, 3, 4]))
477 yield* Effect.forkChild(Queue.offerAll(queue, [5, 6, 7, 8]))
478 yield* Effect.forkChild(Queue.offer(queue, 9))
479 yield* Effect.forkChild(Queue.end(queue))
480 const items = yield* Stream.runCollect(Stream.fromQueue(queue))
481 assert.deepStrictEqual(items, [1, 2, 3, 4, 5, 6, 7, 8, 9])
482 assert.strictEqual(yield* Queue.await(queue), void 0)
483 assert.strictEqual(yield* Queue.offer(queue, 10), false)
484 }))
485
486 it.effect("end with take", () =>
487 Effect.gen(function*() {
488 const queue = yield* Queue.bounded<number, Cause.Done>(2)
489 yield* Effect.forkChild(Queue.offerAll(queue, [1, 2]))
490 yield* Effect.forkChild(Queue.offer(queue, 3))
491 yield* Effect.forkChild(Queue.end(queue))
492 assert.strictEqual(yield* Queue.take(queue), 1)
493 assert.strictEqual(yield* Queue.take(queue), 2)
494 assert.strictEqual(yield* Queue.take(queue), 3)
495 assert.strictEqual(Cause.isDone(yield* Queue.take(queue).pipe(Effect.flip)), true)
496 assert.strictEqual(yield* Queue.await(queue), void 0)
497 assert.strictEqual(yield* Queue.offer(queue, 10), false)
498 }))
499
500 it.effect("interrupt allows draining", () =>
501 Effect.gen(function*() {
502 const queue = yield* Queue.bounded<number>(10)
503 yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
504
505 // Interrupt gracefully
506 const interrupted = yield* Queue.interrupt(queue)
507 assert.strictEqual(interrupted, true)
508
509 // No more offers accepted
510 const offerResult = yield* Queue.offer(queue, 6)
511 assert.strictEqual(offerResult, false)
512
513 // But can still drain existing messages
514 assert.strictEqual(yield* Queue.take(queue), 1)
515 assert.strictEqual(yield* Queue.take(queue), 2)
516 assert.strictEqual(yield* Queue.take(queue), 3)
517 assert.strictEqual(yield* Queue.take(queue), 4)
518 assert.strictEqual(yield* Queue.take(queue), 5)
519
520 // Now queue is done and take fails with interrupt
521 const exit = yield* Queue.take(queue).pipe(Effect.exit)
522 assert.isTrue(Exit.hasInterrupts(exit))
523 }))
524
525 it.effect("poll returns Option for non-blocking take", () =>
526 Effect.gen(function*() {
527 const queue = yield* Queue.bounded<number>(10)
528
529 // Poll returns Option.none when empty
530 const empty = yield* Queue.poll(queue)
531 assert.isTrue(Option.isNone(empty))
532
533 // Add an item
534 yield* Queue.offer(queue, 42)
535
536 // Poll returns Option.some with the item
537 const item = yield* Queue.poll(queue)
538 assert.isTrue(Option.isSome(item))
539 assert.strictEqual(Option.getOrNull(item), 42)
540
541 // Queue is now empty again
542 const empty2 = yield* Queue.poll(queue)
543 assert.isTrue(Option.isNone(empty2))
544 }))
545
546 it.effect("peek views item without removing it", () =>
547 Effect.gen(function*() {
548 const queue = yield* Queue.bounded<number>(10)
549 yield* Queue.offer(queue, 42)
550
551 // Peek at the item
552 const item = yield* Queue.peek(queue)
553 assert.strictEqual(item, 42)
554
555 // Item is still in the queue
556 const size = yield* Queue.size(queue)
557 assert.strictEqual(size, 1)
558
559 // Peek again - same item
560 const item2 = yield* Queue.peek(queue)
561 assert.strictEqual(item2, 42)
562
563 // Now take it
564 const taken = yield* Queue.take(queue)
565 assert.strictEqual(taken, 42)
566
567 // Queue is now empty
568 const newSize = yield* Queue.size(queue)
569 assert.strictEqual(newSize, 0)
570 }))
571
572 it.effect("fail drains buffered values before failing takers", () =>
573 Effect.gen(function*() {
574 const queue = yield* Queue.bounded<number, string>(2)
575 yield* Effect.forkChild(Queue.offerAll(queue, [1, 2, 3, 4]))
576 yield* Effect.forkChild(Queue.offer(queue, 5))
577 yield* Effect.forkChild(Queue.fail(queue, "boom"))
578 const takeArr = Queue.takeAll(queue)
579 assert.deepStrictEqual(yield* takeArr, [1, 2])
580 assert.deepStrictEqual(yield* takeArr, [3, 4])
581 const items = yield* Queue.takeAll(queue)
582 assert.deepStrictEqual(items, [5])
583 const error = yield* Queue.takeAll(queue).pipe(Effect.flip)
584 assert.deepStrictEqual(error, "boom")
585 assert.strictEqual(yield* Queue.await(queue).pipe(Effect.flip), "boom")
586 assert.strictEqual(yield* Queue.offer(queue, 6), false)
587 }))
588
589 it.effect("shutdown", () =>
590 Effect.gen(function*() {
591 const queue = yield* Queue.bounded<number>(2)
592 yield* Effect.forkChild(Queue.offerAll(queue, [1, 2, 3, 4]))
593 yield* Effect.forkChild(Queue.offerAll(queue, [5, 6, 7, 8]))
594 yield* Effect.forkChild(Queue.shutdown(queue))
595 const exit = yield* Stream.runCollect(Stream.fromQueue(queue)).pipe(
596 Effect.exit
597 )
598 assert.isTrue(Exit.hasInterrupts(exit))
599 assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(Queue.await(queue))))
600 assert.strictEqual(yield* Queue.offer(queue, 10), false)
601 }))
602
603 it.effect("shutdownUnsafe interrupts waiting takers synchronously", () =>
604 Effect.gen(function*() {
605 const queue = yield* Queue.unbounded<number>()
606 const taker = yield* Queue.take(queue).pipe(Effect.forkChild({ startImmediately: true }))
607
608 Queue.shutdownUnsafe(queue)
609
610 assert.isTrue(Exit.hasInterrupts(taker.pollUnsafe()!))
611 }))
612
613 it.effect("shutdownUnsafe clears buffered messages and settles pending offers synchronously", () =>
614 Effect.gen(function*() {
615 const queue = yield* Queue.bounded<number>(1)
616 Queue.offerUnsafe(queue, 1)
617 const offerer = yield* Queue.offer(queue, 2).pipe(Effect.forkChild({ startImmediately: true }))
618
619 Queue.shutdownUnsafe(queue)
620
621 assert.strictEqual(queue.messages.length, 0)
622 assert.deepStrictEqual(offerer.pollUnsafe(), Exit.succeed(false))
623 }))
624
625 it.effect("shutdownUnsafe resumes batch offers with their unaccepted suffix", () =>
626 Effect.gen(function*() {
627 const queue = yield* Queue.bounded<number>(1)
628 const producer = yield* Queue.offerAll(queue, [1, 2, 3]).pipe(
629 Effect.forkChild({ startImmediately: true })
630 )
631 assert.strictEqual(yield* Queue.take(queue), 1)
632
633 Queue.shutdownUnsafe(queue)
634
635 assert.deepStrictEqual(producer.pollUnsafe(), Exit.succeed([3]))
636 }))
637
638 it.effect("shutdownUnsafe returns true and preserves an existing failure", () =>
639 Effect.gen(function*() {
640 const queue = yield* Queue.unbounded<number, string>()
641 Queue.offerUnsafe(queue, 1)
642 Queue.failCauseUnsafe(queue, Cause.fail("boom"))
643
644 assert.isTrue(Queue.shutdownUnsafe(queue))
645
646 assert.strictEqual(yield* Queue.take(queue).pipe(Effect.flip), "boom")
647 }))
648
649 it.effect("shutdownUnsafe returns false when already shut down", () =>
650 Effect.gen(function*() {
651 const queue = yield* Queue.unbounded<number>()
652
653 assert.isTrue(Queue.shutdownUnsafe(queue))
654 assert.isFalse(Queue.shutdownUnsafe(queue))
655 assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(Queue.await(queue))))
656 }))
657
658 it.effect("shutdown returns false and preserves a completed queue's failure", () =>
659 Effect.gen(function*() {
660 const queue = yield* Queue.unbounded<number, string>()
661 Queue.failCauseUnsafe(queue, Cause.fail("boom"))
662
663 assert.isFalse(yield* Queue.shutdown(queue))
664 assert.strictEqual(yield* Queue.take(queue).pipe(Effect.flip), "boom")
665 }))
666
667 it.effect("fail doesnt drop items", () =>
668 Effect.gen(function*() {
669 const queue = yield* Queue.bounded<number, string>(2)
670 yield* Effect.forkChild(Queue.offerAll(queue, [1, 2, 3, 4]))
671 yield* Effect.forkChild(Queue.offer(queue, 5))
672 yield* Effect.forkChild(Queue.fail(queue, "boom"))
673 const items: Array<number> = []
674 const error = yield* Stream.fromQueue(queue).pipe(
675 Stream.runForEach((item) => Effect.sync(() => items.push(item))),
676 Effect.flip
677 )
678 assert.deepStrictEqual(items, [1, 2, 3, 4, 5])
679 assert.strictEqual(error, "boom")
680 }))
681
682 it.effect("await waits for no items", () =>
683 Effect.gen(function*() {
684 const queue = yield* Queue.unbounded<number, Cause.Done>()
685 const fiber = yield* Queue.await(queue).pipe(Effect.forkChild)
686 yield* Effect.yieldNow
687 yield* Queue.offer(queue, 1)
688 yield* Queue.end(queue)
689
690 yield* Effect.yieldNow
691 assert.isUndefined(fiber.pollUnsafe())
692 const result = yield* Queue.takeAll(queue)
693 assert.deepStrictEqual(result, [1])
694 yield* Effect.flip(Queue.takeAll(queue))
695 yield* Effect.yieldNow
696 assert.isNotNull(fiber.pollUnsafe())
697 }))
698
699 it.effect("end preserves Done for take and excludes it from await", () =>
700 Effect.gen(function*() {
701 const queue = yield* Queue.unbounded<never, Cause.Done>()
702 const takeFiber = yield* Queue.take(queue).pipe(Effect.forkChild)
703 const awaitFiber = yield* Queue.await(queue).pipe(Effect.forkChild)
704 yield* Effect.yieldNow
705 yield* Queue.end(queue)
706
707 assert.deepStrictEqual(yield* Fiber.await(takeFiber), Exit.fail(Cause.Done()))
708 assert.strictEqual(yield* Fiber.join(awaitFiber), void 0)
709 }))
710
711 it.effect("await preserves non-Done failures", () =>
712 Effect.gen(function*() {
713 const queue = yield* Queue.unbounded<number, string>()
714 const fiber = yield* Queue.await(queue).pipe(Effect.exit, Effect.forkChild)
715 yield* Effect.yieldNow
716 yield* Queue.fail(queue, "boom")
717
718 assert.deepStrictEqual(yield* Fiber.join(fiber), Exit.fail("boom"))
719 }))
720
721 it.effect("flushUnsafe releases a waiting taker after offerUnsafe without completing the queue", () =>
722 Effect.gen(function*() {
723 const queue = yield* Queue.unbounded<number>()
724 const taker = yield* Queue.take(queue).pipe(Effect.forkChild)
725 const awaiter = yield* Queue.await(queue).pipe(Effect.forkChild)
726 yield* Effect.yieldNow
727
728 Queue.offerUnsafe(queue, 1)
729 assert.isUndefined(taker.pollUnsafe())
730
731 Queue.flushUnsafe(queue)
732
733 assert.deepStrictEqual(taker.pollUnsafe(), Exit.succeed(1))
734 assert.isUndefined(awaiter.pollUnsafe())
735 yield* Queue.shutdown(queue)
736 }))
737
738 it.effect("flush releases a waiting taker after offerUnsafe without completing the queue", () =>
739 Effect.gen(function*() {
740 const queue = yield* Queue.unbounded<number>()
741 const taker = yield* Queue.take(queue).pipe(Effect.forkChild)
742 const awaiter = yield* Queue.await(queue).pipe(Effect.forkChild)
743 yield* Effect.yieldNow
744
745 Queue.offerUnsafe(queue, 1)
746 assert.isUndefined(taker.pollUnsafe())
747
748 yield* Queue.flush(queue)
749
750 assert.deepStrictEqual(taker.pollUnsafe(), Exit.succeed(1))
751 assert.isUndefined(awaiter.pollUnsafe())
752 yield* Queue.shutdown(queue)
753 }))
754
755 it.effect("flushUnsafe coalesces scheduled releases across repeated offers", () =>
756 Effect.gen(function*() {
757 const tasks: Array<() => void> = []
758 const scheduler: Scheduler.Scheduler = {
759 executionMode: "async",
760 makeDispatcher() {
761 return {
762 scheduleTask(task, _priority) {
763 tasks.push(task)
764 },
765 flush() {
766 }
767 }
768 },
769 shouldYield: () => false
770 }
771 const queue = yield* Queue.unbounded<number>().pipe(
772 Effect.provideService(Scheduler.Scheduler, scheduler)
773 )
774 for (let i = 0; i < 10; i++) {
775 yield* Queue.take(queue).pipe(Effect.forkChild)
776 }
777 yield* Effect.yieldNow
778
779 for (let i = 0; i < 8; i++) {
780 Queue.offerUnsafe(queue, i)
781 Queue.flushUnsafe(queue)
782 }
783
784 assert.lengthOf(tasks, 1)
785 yield* Queue.shutdown(queue)
786 }))
787
788 it.effect("flush is a no-op after shutdown", () =>
789 Effect.gen(function*() {
790 const queue = yield* Queue.unbounded<number>()
791 yield* Queue.shutdown(queue)
792
793 Queue.flushUnsafe(queue)
794 yield* Queue.flush(queue)
795
796 assert.isFalse(yield* Queue.offer(queue, 1))
797 }))
798
799 it.effect("flush keeps a buffered message when no taker is waiting", () =>
800 Effect.gen(function*() {
801 const queue = yield* Queue.unbounded<number>()
802 Queue.offerUnsafe(queue, 1)
803
804 yield* Queue.flush(queue)
805
806 assert.strictEqual(yield* Queue.take(queue), 1)
807 }))
808
809 it.effect("flushUnsafe preserves later offers when the queue is empty", () =>
810 Effect.gen(function*() {
811 const queue = yield* Queue.unbounded<number>()
812 const first = yield* Queue.take(queue).pipe(Effect.forkChild)
813 const second = yield* Queue.take(queue).pipe(Effect.forkChild)
814 yield* Effect.yieldNow
815
816 Queue.flushUnsafe(queue)
817 yield* Queue.offerAll(queue, [1, 2])
818
819 const values = yield* Effect.all([Fiber.join(first), Fiber.join(second)])
820 assert.deepStrictEqual(new Set(values), new Set([1, 2]))
821 yield* Queue.shutdown(queue)
822 }))
823
824 it.effect("bounded 0 capacity", () =>
825 Effect.gen(function*() {
826 const queue = yield* Queue.bounded<number>(0)
827 yield* Queue.offer(queue, 1).pipe(Effect.forkChild)
828 let result = yield* Queue.take(queue)
829 assert.strictEqual(result, 1)
830 const fiber = yield* Queue.take(queue).pipe(Effect.forkChild)
831 yield* Queue.offer(queue, 2)
832 result = yield* Fiber.join(fiber)
833 assert.strictEqual(result, 2)
834 }))
835
836 it.effect("zero-capacity takeAll consumes buffered messages before pending offers", () =>
837 Effect.gen(function*() {
838 const queue = yield* Queue.bounded<number>(0)
839 const waiting = yield* Queue.take(queue).pipe(Effect.forkChild({ startImmediately: true }))
840 const producer = yield* Queue.offerAll(queue, [1, 2]).pipe(Effect.forkChild({ startImmediately: true }))
841
842 const batch = yield* Queue.takeAll(queue)
843 const next = yield* Fiber.join(waiting)
844 yield* Fiber.join(producer)
845
846 assert.deepStrictEqual(batch, [1])
847 assert.strictEqual(next, 2)
848 }))
849
850 it.effect("zero-capacity offerAll drains in order and completes a closing queue", () =>
851 Effect.gen(function*() {
852 const queue = yield* Queue.bounded<number, Cause.Done>(0)
853 const producer = yield* Queue.offerAll(queue, [1, 2, 3]).pipe(Effect.forkChild({ startImmediately: true }))
854
855 const first = yield* Queue.takeAll(queue)
856 yield* Queue.end(queue)
857 const second = yield* Queue.clear(queue)
858 const third = yield* Queue.poll(queue)
859 const producerExit = producer.pollUnsafe()
860 const done = yield* Effect.exit(Queue.await(queue))
861
862 assert.deepStrictEqual(first, [1])
863 assert.deepStrictEqual(second, [2])
864 assert.deepStrictEqual(third, Option.some(3))
865 assert.deepStrictEqual(producerExit, Exit.succeed([]))
866 assert.deepStrictEqual(done, Exit.void)
867 }))
868})