1import { assert, describe, it } from "@effect/vitest"
2import { assertExitFailure, assertFailure, assertTrue } from "@effect/vitest/utils"
3import { Cause, Data, Deferred, pipe, Ref } from "effect"
4import * as Arr from "effect/Array"
5import * as Channel from "effect/Channel"
6import * as Chunk from "effect/Chunk"
7import * as Effect from "effect/Effect"
8import * as Exit from "effect/Exit"
9import * as Fiber from "effect/Fiber"
10import * as Filter from "effect/Filter"
11import * as Latch from "effect/Latch"
12import * as Queue from "effect/Queue"
13import * as Result from "effect/Result"
14import * as Schedule from "effect/Schedule"
15import * as Scheduler from "effect/Scheduler"
16import * as Stream from "effect/Stream"
17
18describe("Channel", () => {
19 describe("repetition", () => {
20 for (const kind of ["repeat", "forever"] as const) {
21 const repeated = (source: Channel.Channel<Arr.NonEmptyArray<number>>) =>
22 kind === "repeat" ? Channel.repeat(source, Schedule.forever) : Channel.forever(source)
23
24 it.effect(kind + " work per repetition does not grow", () =>
25 Effect.gen(function*() {
26 const stream = Stream.fromChannel(repeated(Channel.succeed(Arr.make(1))))
27 const ops = (n: number) => countOps(stream.pipe(Stream.take(n), Stream.runDrain))
28 const small = yield* ops(1_000)
29 const large = yield* ops(2_000)
30 assert.isBelow(large / small, 2.5)
31 }))
32 }
33 })
34
35 describe("constructors", () => {
36 it.effect("empty", () =>
37 Effect.gen(function*() {
38 const result = yield* Channel.empty.pipe(
39 Channel.runCollect
40 )
41 assert.deepStrictEqual(result, [])
42 }))
43
44 it.effect("succeed", () =>
45 Effect.gen(function*() {
46 const result = yield* Channel.succeed(1).pipe(
47 Channel.runCollect
48 )
49 assert.deepStrictEqual(result, [1])
50 }))
51
52 it.effect("sync", () =>
53 Effect.gen(function*() {
54 const result = yield* Channel.sync(() => 1).pipe(
55 Channel.runCollect
56 )
57 assert.deepStrictEqual(result, [1])
58 }))
59
60 it.effect("end", () =>
61 Effect.gen(function*() {
62 let result = 0
63 yield* Channel.end(42).pipe(
64 Channel.mapDone((n) => {
65 result = n
66 }),
67 Channel.runDrain
68 )
69 assert.strictEqual(result, 42)
70 }))
71
72 it.effect("endSync", () =>
73 Effect.gen(function*() {
74 let result = 0
75 yield* Channel.endSync(() => 42).pipe(
76 Channel.mapDone((n) => {
77 result = n
78 }),
79 Channel.runDrain
80 )
81 assert.strictEqual(result, 42)
82 }))
83
84 it.effect("fromArray", () =>
85 Effect.gen(function*() {
86 const array = [0, 1, 2, 3, 4]
87 const result = yield* Channel.runCollect(Channel.fromArray(array))
88 assert.deepStrictEqual(result, array)
89 }))
90
91 it.effect("fromChunk", () =>
92 Effect.gen(function*() {
93 const chunk = Chunk.fromArrayUnsafe([0, 1, 2, 3, 4])
94 const result = yield* Channel.runCollect(Channel.fromChunk(chunk))
95 assert.deepStrictEqual(result, Chunk.toArray(chunk))
96 }))
97
98 it.effect("fromIterator", () =>
99 Effect.gen(function*() {
100 const result = yield* Channel.fromIterator(() => ({
101 n: 0,
102 next(this: { n: number }) {
103 return this.n === 5
104 ? { done: true, value: this.n }
105 : { done: false, value: this.n++ }
106 }
107 })).pipe(Channel.runCollect)
108 assert.deepStrictEqual(result, [0, 1, 2, 3, 4])
109 }))
110
111 it.effect("fromIteratorArray", () =>
112 Effect.gen(function*() {
113 function* fibonacci(): Generator<number, void, unknown> {
114 let a = 0, b = 1
115 for (let i = 0; i < 5; i++) {
116 yield a
117 ;[a, b] = [b, a + b]
118 }
119 }
120 const result = yield* Channel.runCollect(
121 Channel.fromIteratorArray(() => fibonacci(), 3)
122 )
123 assert.deepStrictEqual(result, [[0, 1, 1], [2, 3]])
124 }))
125
126 it.effect("fromIteratorArray - normalizes the chunk size", () =>
127 Effect.gen(function*() {
128 const results = yield* Effect.forEach([Number.NaN, -1, 0.5, 1.9], (chunkSize) =>
129 Channel.fromIteratorArray(() =>
130 [1, 2, 3][Symbol.iterator](), chunkSize).pipe(
131 Channel.runCollect
132 ))
133 assert.deepStrictEqual(results, [
134 [[1], [2], [3]],
135 [[1], [2], [3]],
136 [[1], [2], [3]],
137 [[1], [2], [3]]
138 ])
139 }))
140
141 it.effect("fromIterable", () =>
142 Effect.gen(function*() {
143 const set = new Set([1, 1, 2, 3])
144 const result = yield* Channel.runCollect(Channel.fromIterable(set))
145 assert.deepStrictEqual(result, [1, 2, 3])
146 }))
147
148 it.effect("fromIterableArray", () =>
149 Effect.gen(function*() {
150 const numbers = [1, 2, 3, 4, 5]
151 const result = yield* Channel.runCollect(Channel.fromIterableArray(numbers))
152 const resultChunked = yield* Channel.runCollect(Channel.fromIterableArray(numbers, 4))
153 assert.deepStrictEqual(result, [[1, 2, 3, 4, 5]])
154 assert.deepStrictEqual(resultChunked, [[1, 2, 3, 4], [5]])
155 }))
156
157 it.effect("fromIterableArray - normalizes the chunk size", () =>
158 Effect.gen(function*() {
159 const results = yield* Effect.forEach([Number.NaN, 0, 2.9], (chunkSize) =>
160 Channel.runCollect(Channel.fromIterableArray([1, 2, 3], chunkSize)))
161 assert.deepStrictEqual(results, [
162 [[1], [2], [3]],
163 [[1], [2], [3]],
164 [[1, 2], [3]]
165 ])
166 }))
167
168 it.effect("fromReadableStream", () =>
169 Effect.gen(function*() {
170 const result = yield* Channel.fromReadableStream({
171 evaluate: () =>
172 new ReadableStream<number>({
173 start(controller) {
174 controller.enqueue(1)
175 controller.enqueue(2)
176 controller.close()
177 }
178 }),
179 onError: (error) => error
180 }).pipe(Channel.runCollect)
181
182 assert.deepStrictEqual(result, [[1], [2]])
183 }))
184
185 it.effect("fromTransformStream - surfaces write-side errors through the read side", () =>
186 Effect.gen(function*() {
187 const error = new Error("write failed")
188 const channel = Channel.fromTransformStream<never, number, number, Error>({
189 evaluate: () =>
190 new TransformStream<number, number>({
191 transform() {
192 throw error
193 }
194 }),
195 onError: (cause) => cause as Error
196 })
197 const exit = yield* Channel.fromArray([[1] as [number]]).pipe(
198 Channel.pipeTo(channel),
199 Channel.runDrain,
200 Effect.exit
201 )
202
203 assertExitFailure(exit, Cause.fail(error))
204 }))
205
206 it.effect("acquireRelease", () =>
207 Effect.gen(function*() {
208 const acquired = yield* Ref.make(false)
209 const released = yield* Ref.make(false)
210 yield* Channel.acquireRelease(
211 Ref.set(acquired, true),
212 () => Ref.set(released, true)
213 ).pipe(Channel.runDrain)
214 assert.isTrue(yield* Ref.get(acquired))
215 assert.isTrue(yield* Ref.get(released))
216 }))
217
218 it.effect("acquireUseRelease combines usage and release failures", () =>
219 Effect.gen(function*() {
220 const result = yield* Channel.acquireUseRelease(
221 Effect.void,
222 () => Channel.fail("usage failure"),
223 () => Effect.die("release failure")
224 ).pipe(Channel.runDrain, Effect.exit)
225 assert.deepStrictEqual(
226 result,
227 Exit.failCause(Cause.combine(Cause.fail("usage failure"), Cause.die("release failure")))
228 )
229 }))
230
231 it.effect("acquireUseRelease surfaces release failure after successful usage", () =>
232 Effect.gen(function*() {
233 const result = yield* Channel.acquireUseRelease(
234 Effect.void,
235 () => Channel.succeed(1),
236 () => Effect.die("release failure")
237 ).pipe(Channel.runDrain, Effect.exit)
238 assert.deepStrictEqual(result, Exit.die("release failure"))
239 }))
240 })
241
242 describe("destructors", () => {
243 it.effect("runDrain returns the done value after emitted elements", () =>
244 Effect.gen(function*() {
245 const result = yield* Channel.fromArray([1]).pipe(
246 Channel.concat(Channel.end("done")),
247 Channel.runDrain
248 )
249
250 assert.strictEqual(result, "done")
251 }))
252
253 it.effect("mkUint8Array", () =>
254 Effect.gen(function*() {
255 const bytes = yield* Channel.fromArray(
256 [
257 [new Uint8Array([1, 2])],
258 [new Uint8Array([3]), new Uint8Array([4, 5])]
259 ] as const
260 ).pipe(Channel.mkUint8Array)
261
262 assert.deepStrictEqual(bytes, new Uint8Array([1, 2, 3, 4, 5]))
263 }))
264 })
265
266 describe("mapping", () => {
267 it.effect("map", () =>
268 Effect.gen(function*() {
269 const result = yield* Channel.fromArray([1, 2, 3]).pipe(
270 Channel.map((n) => n + 1),
271 Channel.runCollect
272 )
273 assert.deepStrictEqual(result, [2, 3, 4])
274 }))
275
276 it.effect("mapEffect interrupts the running effect when the channel is interrupted", () =>
277 Effect.gen(function*() {
278 let interrupted = false
279 const latch = yield* Latch.make(false)
280 const fiber = yield* Channel.succeed(1).pipe(
281 Channel.mapEffect(() =>
282 latch.open.pipe(
283 Effect.andThen(Effect.never),
284 Effect.onInterrupt(() =>
285 Effect.sync(() => {
286 interrupted = true
287 })
288 )
289 ), { concurrency: 2 }),
290 Channel.runDrain,
291 Effect.forkChild
292 )
293 yield* Fiber.interrupt(fiber).pipe(latch.whenOpen)
294 assert.isTrue(interrupted)
295 }))
296
297 it.effect("mapEffect - interrupts pending tasks on failure", () =>
298 Effect.gen(function*() {
299 let interrupts = 0
300 const latch1 = yield* Latch.make(false)
301 const latch2 = yield* Latch.make(false)
302 const result = yield* Channel.fromArray([1, 2, 3]).pipe(
303 Channel.mapEffect((n) => {
304 if (n === 1) {
305 return latch1.open.pipe(
306 Effect.andThen(Effect.never),
307 Effect.onInterrupt(() =>
308 Effect.sync(() => {
309 interrupts++
310 })
311 )
312 )
313 }
314 if (n === 2) {
315 return latch2.open.pipe(
316 Effect.andThen(Effect.never),
317 Effect.onInterrupt(() =>
318 Effect.sync(() => {
319 interrupts++
320 })
321 )
322 )
323 }
324 return Effect.fail("boom").pipe(
325 latch1.whenOpen,
326 latch2.whenOpen
327 )
328 }, { concurrency: 3 }),
329 Channel.runDrain,
330 Effect.exit
331 )
332 assert.strictEqual(interrupts, 2)
333 assert.deepStrictEqual(result, Exit.fail("boom"))
334 }))
335 })
336
337 describe("filtering", () => {
338 it.effect("filterMap with Filter", () =>
339 Effect.gen(function*() {
340 const filter = Filter.make((n: number) => n % 2 === 0 ? Result.succeed(n * 2) : Result.fail(n))
341 const result = yield* Channel.fromArray([1, 2, 3, 4]).pipe(
342 Channel.filterMap(filter),
343 Channel.runCollect
344 )
345 assert.deepStrictEqual(result, [4, 8])
346 }))
347
348 it.effect("filterMapEffect with FilterEffect", () =>
349 Effect.gen(function*() {
350 const filter = Filter.makeEffect((n: number) => Effect.succeed(n > 2 ? Result.succeed(n + 1) : Result.fail(n)))
351 const result = yield* Channel.fromArray([1, 2, 3, 4]).pipe(
352 Channel.filterMapEffect(filter),
353 Channel.runCollect
354 )
355 assert.deepStrictEqual(result, [4, 5])
356 }))
357 })
358
359 describe("encoding", () => {
360 it.effect("decodeText handles multi-byte characters split across Uint8Array boundaries", () =>
361 Effect.gen(function*() {
362 const bytes = new TextEncoder().encode("a🌍b")
363 const chunks: ReadonlyArray<readonly [Uint8Array, ...ReadonlyArray<Uint8Array>]> = [
364 [bytes.slice(0, 2)],
365 [bytes.slice(2, 4), bytes.slice(4)]
366 ]
367 const result = yield* Channel.fromArray(chunks).pipe(
368 Channel.pipeTo(Channel.decodeText()),
369 Channel.runCollect
370 )
371 assert.strictEqual(result.flat().join(""), "a🌍b")
372 }))
373 })
374
375 describe("merging", () => {
376 it.effect("merge - interrupts left side if halt strategy is set to 'right'", () =>
377 Effect.gen(function*() {
378 const latch = yield* Latch.make(false)
379 const leftQueue = yield* Queue.make<number, Cause.Done>()
380 const rightQueue = yield* Queue.make<number>()
381 const left = Channel.fromQueue(rightQueue)
382 const right = Channel.fromQueue(leftQueue).pipe(
383 Channel.ensuring(latch.open)
384 )
385 const fiber = yield* Channel.merge(left, right, {
386 haltStrategy: "right"
387 }).pipe(Channel.runCollect, Effect.forkChild)
388 yield* Queue.offerAll(leftQueue, [1, 2])
389 yield* Queue.end(leftQueue)
390 yield* latch.await
391 yield* Queue.offerAll(rightQueue, [3, 4])
392 const result = yield* Fiber.join(fiber)
393 assert.deepStrictEqual(result, [1, 2])
394 }))
395
396 it.effect("merge - interrupts right side if halt strategy is set to 'left'", () =>
397 Effect.gen(function*() {
398 const latch = yield* Latch.make(false)
399 const leftQueue = yield* Queue.make<number, Cause.Done>()
400 const rightQueue = yield* Queue.make<number>()
401 const left = Channel.fromQueue(leftQueue).pipe(
402 Channel.ensuring(latch.open)
403 )
404 const right = Channel.fromQueue(rightQueue)
405 const fiber = yield* Channel.merge(left, right, {
406 haltStrategy: "left"
407 }).pipe(Channel.runCollect, Effect.forkChild)
408 yield* Queue.offerAll(leftQueue, [1, 2])
409 yield* Queue.end(leftQueue)
410 yield* latch.await
411 yield* Queue.offerAll(rightQueue, [3, 4])
412 const result = yield* Fiber.join(fiber)
413 assert.deepStrictEqual(result, [1, 2])
414 }))
415
416 it.effect("merge - interrupts losing side if halt strategy is set to 'either'", () =>
417 Effect.gen(function*() {
418 const left = Channel.fromEffect(Effect.never)
419 const right = Channel.succeed(1)
420 const result = yield* Channel.merge(left, right, {
421 haltStrategy: "either"
422 }).pipe(Channel.runCollect)
423 assert.deepStrictEqual(result, [1])
424 }))
425
426 it.effect("merge - waits for both sides if halt strategy is set to 'both'", () =>
427 Effect.gen(function*() {
428 const left = Channel.succeed(1)
429 const right = Channel.succeed(2)
430 const result = yield* Channel.merge(left, right, {
431 haltStrategy: "both"
432 }).pipe(Channel.runCollect)
433 assert.deepStrictEqual(result, [1, 2])
434 }))
435
436 it.effect("merge - prioritizes failure", () =>
437 Effect.gen(function*() {
438 const left = Channel.fromEffect(Effect.fail("boom"))
439 const right = Channel.fromEffect(Effect.never)
440 const result = yield* Channel.merge(left, right).pipe(
441 Channel.runCollect,
442 Effect.exit
443 )
444 assert.deepStrictEqual(result, Exit.fail("boom"))
445 }))
446 })
447
448 describe("switchMap", () => {
449 it.effect("interrupts previous channels and runs their finalizers", () =>
450 Effect.gen(function*() {
451 const result = yield* Channel.fromIterable([1, 2, 3]).pipe(
452 Channel.switchMap((n) => n === 3 ? Channel.empty : Channel.never),
453 Channel.runDrain
454 )
455 assert.isUndefined(result)
456 }))
457 })
458
459 describe("interruptWhen", () => {
460 it.effect("interrupts the current element", () =>
461 Effect.gen(function*() {
462 const interrupted = yield* Ref.make(false)
463 const latch = yield* Deferred.make<void>()
464 const halt = yield* Deferred.make<void>()
465 const started = yield* Deferred.make<void>()
466 const channel = Deferred.succeed(started, void 0).pipe(
467 Effect.andThen(Deferred.await(latch)),
468 Effect.onInterrupt(() => Ref.set(interrupted, true)),
469 Channel.fromEffect,
470 Channel.interruptWhen(Deferred.await(halt))
471 )
472 const fiber = yield* Effect.forkChild(Channel.runDrain(channel))
473 yield* pipe(
474 Deferred.await(started),
475 Effect.andThen(Deferred.succeed(halt, void 0))
476 )
477 yield* Fiber.await(fiber)
478 const result = yield* Ref.get(interrupted)
479 assertTrue(result)
480 }))
481
482 it.effect("interruptWhen - propagates errors", () =>
483 Effect.gen(function*() {
484 const deferred = yield* Deferred.make<never, string>()
485 const channel = Channel.fromEffect(Effect.never).pipe(
486 Channel.interruptWhen(Deferred.await(deferred))
487 )
488 yield* Deferred.fail(deferred, "fail")
489 const result = yield* Effect.result(Channel.runDrain(channel))
490 assertFailure(result, "fail")
491 }))
492 })
493
494 describe("conditional catch", () => {
495 class HttpError extends Data.TaggedError("HttpError")<{
496 readonly message: string
497 }> {}
498
499 class ValidationError extends Data.TaggedError("ValidationError")<{
500 readonly field: string
501 }> {}
502
503 class RateLimitError extends Data.TaggedError("RateLimitError")<{
504 readonly retryAfter: number
505 }> {}
506
507 class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
508 readonly limit: number
509 }> {}
510
511 class AiError extends Data.TaggedError("AiError")<{
512 readonly reason: RateLimitError | QuotaExceededError
513 }> {}
514
515 class OtherError extends Data.TaggedError("OtherError")<{
516 readonly message: string
517 }> {}
518
519 it.effect("catchDefect", () =>
520 Effect.gen(function*() {
521 const defect = new Error("boom")
522 const recovered = yield* Channel.fromEffect(Effect.die(defect)).pipe(
523 Channel.catchDefect((caught) => Channel.succeed(caught)),
524 Channel.runCollect
525 )
526 const failed = yield* Channel.fail("failure").pipe(
527 Channel.catchDefect(() => Channel.succeed("recovered")),
528 Channel.runCollect,
529 Effect.exit
530 )
531 assert.deepStrictEqual(recovered, [defect])
532 assertExitFailure(failed, Cause.fail("failure"))
533 }))
534
535 it.effect("catchIf with refinement", () =>
536 Effect.gen(function*() {
537 const exit = yield* (Channel.fail(new ValidationError({ field: "email" })) as Channel.Channel<
538 never,
539 HttpError | ValidationError,
540 never
541 >).pipe(
542 Channel.catchIf(
543 (error): error is HttpError => error._tag === "HttpError",
544 () => Channel.succeed("http")
545 ),
546 Channel.runCollect,
547 Effect.exit
548 )
549 assertExitFailure(exit, Cause.fail(new ValidationError({ field: "email" })))
550 }))
551
552 it.effect("catchIf with predicate", () =>
553 Effect.gen(function*() {
554 const result = yield* Channel.fail("boom").pipe(
555 Channel.catchIf((error) => error === "boom", (error) => Channel.succeed(`recovered: ${error}`)),
556 Channel.runCollect
557 )
558 assert.deepStrictEqual(result, ["recovered: boom"])
559 }))
560
561 it.effect("catchFilter with Filter", () =>
562 Effect.gen(function*() {
563 const filter = Filter.make((error: string) =>
564 error === "boom"
565 ? Result.succeed(error)
566 : Result.fail(error)
567 )
568 const result = yield* Channel.fail("boom").pipe(
569 Channel.catchFilter(filter, (error) => Channel.succeed(`recovered: ${error}`)),
570 Channel.runCollect
571 )
572 assert.deepStrictEqual(result, ["recovered: boom"])
573 }))
574
575 it.effect("catchCauseFilter with Filter", () =>
576 Effect.gen(function*() {
577 const result = yield* Channel.fail("boom").pipe(
578 Channel.catchCauseFilter(
579 Cause.findError as any,
580 (error) => Channel.succeed(`recovered: ${error}`)
581 ),
582 Channel.runCollect
583 )
584 assert.deepStrictEqual(result, ["recovered: boom"])
585 }))
586
587 it.effect("catchTag orElse", () =>
588 Effect.gen(function*() {
589 const result = yield* Channel.catchTag(
590 Channel.fail(new ValidationError({ field: "email" })) as Channel.Channel<
591 never,
592 HttpError | ValidationError,
593 never
594 >,
595 "HttpError",
596 () => Channel.succeed("http"),
597 () => Channel.succeed("fallback")
598 )
599 .pipe(Channel.runCollect)
600 assert.deepStrictEqual(result, ["fallback"])
601 }))
602
603 it.effect("catchReason orElse", () =>
604 Effect.gen(function*() {
605 const result = yield* Channel.fail(
606 new AiError({ reason: new QuotaExceededError({ limit: 100 }) })
607 ).pipe(
608 Channel.catchReason(
609 "AiError",
610 "RateLimitError",
611 (reason) => Channel.succeed(`retry: ${reason.retryAfter}`),
612 (reason) => Channel.succeed(`quota: ${reason.limit}`)
613 ),
614 Channel.runCollect
615 )
616 assert.deepStrictEqual(result, ["quota: 100"])
617 }))
618
619 it.effect("catchReason ignores non-matching parent tag", () =>
620 Effect.gen(function*() {
621 const error = new OtherError({ message: "test" })
622 const exit = yield* (Channel.fail(error) as Channel.Channel<never, AiError | OtherError, never>).pipe(
623 Channel.catchReason(
624 "AiError",
625 "RateLimitError",
626 () => Channel.succeed("no"),
627 () => Channel.succeed("orElse")
628 ),
629 Channel.runCollect,
630 Effect.exit
631 )
632 assertExitFailure(exit, Cause.fail(error))
633 }))
634
635 it.effect("catchReasons orElse", () =>
636 Effect.gen(function*() {
637 const result = yield* Channel.fail(
638 new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
639 ).pipe(
640 Channel.catchReasons(
641 "AiError",
642 {
643 QuotaExceededError: (reason) => Channel.succeed(`quota: ${reason.limit}`)
644 },
645 (reason) => Channel.succeed(`fallback: ${reason._tag}`)
646 ),
647 Channel.runCollect
648 )
649 assert.deepStrictEqual(result, ["fallback: RateLimitError"])
650 }))
651 })
652
653 describe("unwrapReason", () => {
654 class RateLimitError extends Data.TaggedError("RateLimitError")<{
655 readonly retryAfter: number
656 }> {}
657
658 class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
659 readonly limit: number
660 }> {}
661
662 class AiError extends Data.TaggedError("AiError")<{
663 readonly reason: RateLimitError | QuotaExceededError
664 }> {}
665
666 class OtherError extends Data.TaggedError("OtherError")<{
667 readonly message: string
668 }> {}
669
670 it.effect("extracts reason into error channel", () =>
671 Effect.gen(function*() {
672 const reason = new RateLimitError({ retryAfter: 60 })
673 const exit = yield* Channel.fail(new AiError({ reason })).pipe(
674 Channel.unwrapReason("AiError"),
675 Channel.runDrain,
676 Effect.exit
677 )
678 assertExitFailure(exit, Cause.fail(reason))
679 }))
680
681 it.effect("extracts second reason type", () =>
682 Effect.gen(function*() {
683 const reason = new QuotaExceededError({ limit: 100 })
684 const exit = yield* Channel.fail(new AiError({ reason })).pipe(
685 Channel.unwrapReason("AiError"),
686 Channel.runDrain,
687 Effect.exit
688 )
689 assertExitFailure(exit, Cause.fail(reason))
690 }))
691
692 it.effect("preserves other errors", () =>
693 Effect.gen(function*() {
694 const error = new OtherError({ message: "test" })
695 const exit = yield* (Channel.fail(error) as Channel.Channel<never, AiError | OtherError, never>).pipe(
696 Channel.unwrapReason("AiError"),
697 Channel.runDrain,
698 Effect.exit
699 )
700 assertExitFailure(exit, Cause.fail(error))
701 }))
702 })
703})
704
705// The scheduler is asked whether to yield once for each fiber run-loop operation.
706const countOps = <A, E>(effect: Effect.Effect<A, E>): Effect.Effect<number, E> =>
707 Effect.suspend(() => {
708 let ops = 0
709 const scheduler: Scheduler.Scheduler = {
710 executionMode: "sync",
711 makeDispatcher: () => new Scheduler.MixedScheduler("sync").makeDispatcher(),
712 shouldYield: () => {
713 ops++
714 return false
715 }
716 }
717 return Effect.map(Effect.provideService(effect, Scheduler.Scheduler, scheduler), () => ops)
718 })