557 lines
19 KiB
TypeScript
557 lines
19 KiB
TypeScript
import { assert, describe, it } from "@effect/vitest"
|
|
import { assertExitFailure, assertFailure, assertTrue } from "@effect/vitest/utils"
|
|
import { Cause, Data, Deferred, pipe, Ref } from "effect"
|
|
import * as Channel from "effect/Channel"
|
|
import * as Chunk from "effect/Chunk"
|
|
import * as Effect from "effect/Effect"
|
|
import * as Exit from "effect/Exit"
|
|
import * as Fiber from "effect/Fiber"
|
|
import * as Filter from "effect/Filter"
|
|
import * as Latch from "effect/Latch"
|
|
import * as Queue from "effect/Queue"
|
|
import * as Result from "effect/Result"
|
|
|
|
describe("Channel", () => {
|
|
describe("constructors", () => {
|
|
it.effect("empty", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.empty.pipe(
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, [])
|
|
}))
|
|
|
|
it.effect("succeed", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.succeed(1).pipe(
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, [1])
|
|
}))
|
|
|
|
it.effect("sync", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.sync(() => 1).pipe(
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, [1])
|
|
}))
|
|
|
|
it.effect("end", () =>
|
|
Effect.gen(function*() {
|
|
let result = 0
|
|
yield* Channel.end(42).pipe(
|
|
Channel.mapDone((n) => {
|
|
result = n
|
|
}),
|
|
Channel.runDrain
|
|
)
|
|
assert.strictEqual(result, 42)
|
|
}))
|
|
|
|
it.effect("endSync", () =>
|
|
Effect.gen(function*() {
|
|
let result = 0
|
|
yield* Channel.endSync(() => 42).pipe(
|
|
Channel.mapDone((n) => {
|
|
result = n
|
|
}),
|
|
Channel.runDrain
|
|
)
|
|
assert.strictEqual(result, 42)
|
|
}))
|
|
|
|
it.effect("fromArray", () =>
|
|
Effect.gen(function*() {
|
|
const array = [0, 1, 2, 3, 4]
|
|
const result = yield* Channel.runCollect(Channel.fromArray(array))
|
|
assert.deepStrictEqual(result, array)
|
|
}))
|
|
|
|
it.effect("fromChunk", () =>
|
|
Effect.gen(function*() {
|
|
const chunk = Chunk.fromArrayUnsafe([0, 1, 2, 3, 4])
|
|
const result = yield* Channel.runCollect(Channel.fromChunk(chunk))
|
|
assert.deepStrictEqual(result, Chunk.toArray(chunk))
|
|
}))
|
|
|
|
it.effect("fromIterator", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fromIterator(() => ({
|
|
n: 0,
|
|
next(this: { n: number }) {
|
|
return this.n === 5
|
|
? { done: true, value: this.n }
|
|
: { done: false, value: this.n++ }
|
|
}
|
|
})).pipe(Channel.runCollect)
|
|
assert.deepStrictEqual(result, [0, 1, 2, 3, 4])
|
|
}))
|
|
|
|
it.effect("fromIteratorArray", () =>
|
|
Effect.gen(function*() {
|
|
function* fibonacci(): Generator<number, void, unknown> {
|
|
let a = 0, b = 1
|
|
for (let i = 0; i < 5; i++) {
|
|
yield a
|
|
;[a, b] = [b, a + b]
|
|
}
|
|
}
|
|
const result = yield* Channel.runCollect(
|
|
Channel.fromIteratorArray(() => fibonacci(), 3)
|
|
)
|
|
assert.deepStrictEqual(result, [[0, 1, 1], [2, 3]])
|
|
}))
|
|
|
|
it.effect("fromIterable", () =>
|
|
Effect.gen(function*() {
|
|
const set = new Set([1, 1, 2, 3])
|
|
const result = yield* Channel.runCollect(Channel.fromIterable(set))
|
|
assert.deepStrictEqual(result, [1, 2, 3])
|
|
}))
|
|
|
|
it.effect("fromIterableArray", () =>
|
|
Effect.gen(function*() {
|
|
const numbers = [1, 2, 3, 4, 5]
|
|
const result = yield* Channel.runCollect(Channel.fromIterableArray(numbers))
|
|
const resultChunked = yield* Channel.runCollect(Channel.fromIterableArray(numbers, 4))
|
|
assert.deepStrictEqual(result, [[1, 2, 3, 4, 5]])
|
|
assert.deepStrictEqual(resultChunked, [[1, 2, 3, 4], [5]])
|
|
}))
|
|
|
|
it.effect("acquireRelease", () =>
|
|
Effect.gen(function*() {
|
|
const acquired = yield* Ref.make(false)
|
|
const released = yield* Ref.make(false)
|
|
yield* Channel.acquireRelease(
|
|
Ref.set(acquired, true),
|
|
() => Ref.set(released, true)
|
|
).pipe(Channel.runDrain)
|
|
assert.isTrue(yield* Ref.get(acquired))
|
|
assert.isTrue(yield* Ref.get(released))
|
|
}))
|
|
})
|
|
|
|
describe("mapping", () => {
|
|
it.effect("map", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fromArray([1, 2, 3]).pipe(
|
|
Channel.map((n) => n + 1),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, [2, 3, 4])
|
|
}))
|
|
|
|
it.effect("mapEffect interrupts the running effect when the channel is interrupted", () =>
|
|
Effect.gen(function*() {
|
|
let interrupted = false
|
|
const latch = yield* Latch.make(false)
|
|
const fiber = yield* Channel.succeed(1).pipe(
|
|
Channel.mapEffect(() =>
|
|
latch.open.pipe(
|
|
Effect.andThen(Effect.never),
|
|
Effect.onInterrupt(() =>
|
|
Effect.sync(() => {
|
|
interrupted = true
|
|
})
|
|
)
|
|
), { concurrency: 2 }),
|
|
Channel.runDrain,
|
|
Effect.forkChild
|
|
)
|
|
yield* Fiber.interrupt(fiber).pipe(latch.whenOpen)
|
|
assert.isTrue(interrupted)
|
|
}))
|
|
|
|
it.effect("mapEffect - interrupts pending tasks on failure", () =>
|
|
Effect.gen(function*() {
|
|
let interrupts = 0
|
|
const latch1 = yield* Latch.make(false)
|
|
const latch2 = yield* Latch.make(false)
|
|
const result = yield* Channel.fromArray([1, 2, 3]).pipe(
|
|
Channel.mapEffect((n) => {
|
|
if (n === 1) {
|
|
return latch1.open.pipe(
|
|
Effect.andThen(Effect.never),
|
|
Effect.onInterrupt(() =>
|
|
Effect.sync(() => {
|
|
interrupts++
|
|
})
|
|
)
|
|
)
|
|
}
|
|
if (n === 2) {
|
|
return latch2.open.pipe(
|
|
Effect.andThen(Effect.never),
|
|
Effect.onInterrupt(() =>
|
|
Effect.sync(() => {
|
|
interrupts++
|
|
})
|
|
)
|
|
)
|
|
}
|
|
return Effect.fail("boom").pipe(
|
|
latch1.whenOpen,
|
|
latch2.whenOpen
|
|
)
|
|
}, { concurrency: 3 }),
|
|
Channel.runDrain,
|
|
Effect.exit
|
|
)
|
|
assert.strictEqual(interrupts, 2)
|
|
assert.deepStrictEqual(result, Exit.fail("boom"))
|
|
}))
|
|
})
|
|
|
|
describe("filtering", () => {
|
|
it.effect("filterMap with Filter", () =>
|
|
Effect.gen(function*() {
|
|
const filter = Filter.make((n: number) => n % 2 === 0 ? Result.succeed(n * 2) : Result.fail(n))
|
|
const result = yield* Channel.fromArray([1, 2, 3, 4]).pipe(
|
|
Channel.filterMap(filter),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, [4, 8])
|
|
}))
|
|
|
|
it.effect("filterMapEffect with FilterEffect", () =>
|
|
Effect.gen(function*() {
|
|
const filter = Filter.makeEffect((n: number) => Effect.succeed(n > 2 ? Result.succeed(n + 1) : Result.fail(n)))
|
|
const result = yield* Channel.fromArray([1, 2, 3, 4]).pipe(
|
|
Channel.filterMapEffect(filter),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, [4, 5])
|
|
}))
|
|
})
|
|
|
|
describe("encoding", () => {
|
|
it.effect("decodeText handles multi-byte characters split across Uint8Array boundaries", () =>
|
|
Effect.gen(function*() {
|
|
const bytes = new TextEncoder().encode("a🌍b")
|
|
const chunks: ReadonlyArray<readonly [Uint8Array, ...ReadonlyArray<Uint8Array>]> = [
|
|
[bytes.slice(0, 2)],
|
|
[bytes.slice(2, 4), bytes.slice(4)]
|
|
]
|
|
const result = yield* Channel.fromArray(chunks).pipe(
|
|
Channel.pipeTo(Channel.decodeText()),
|
|
Channel.runCollect
|
|
)
|
|
assert.strictEqual(result.flat().join(""), "a🌍b")
|
|
}))
|
|
})
|
|
|
|
describe("merging", () => {
|
|
it.effect("merge - interrupts left side if halt strategy is set to 'right'", () =>
|
|
Effect.gen(function*() {
|
|
const latch = yield* Latch.make(false)
|
|
const leftQueue = yield* Queue.make<number, Cause.Done>()
|
|
const rightQueue = yield* Queue.make<number>()
|
|
const left = Channel.fromQueue(rightQueue)
|
|
const right = Channel.fromQueue(leftQueue).pipe(
|
|
Channel.ensuring(latch.open)
|
|
)
|
|
const fiber = yield* Channel.merge(left, right, {
|
|
haltStrategy: "right"
|
|
}).pipe(Channel.runCollect, Effect.forkChild)
|
|
yield* Queue.offerAll(leftQueue, [1, 2])
|
|
yield* Queue.end(leftQueue)
|
|
yield* latch.await
|
|
yield* Queue.offerAll(rightQueue, [3, 4])
|
|
const result = yield* Fiber.join(fiber)
|
|
assert.deepStrictEqual(result, [1, 2])
|
|
}))
|
|
|
|
it.effect("merge - interrupts right side if halt strategy is set to 'left'", () =>
|
|
Effect.gen(function*() {
|
|
const latch = yield* Latch.make(false)
|
|
const leftQueue = yield* Queue.make<number, Cause.Done>()
|
|
const rightQueue = yield* Queue.make<number>()
|
|
const left = Channel.fromQueue(leftQueue).pipe(
|
|
Channel.ensuring(latch.open)
|
|
)
|
|
const right = Channel.fromQueue(rightQueue)
|
|
const fiber = yield* Channel.merge(left, right, {
|
|
haltStrategy: "left"
|
|
}).pipe(Channel.runCollect, Effect.forkChild)
|
|
yield* Queue.offerAll(leftQueue, [1, 2])
|
|
yield* Queue.end(leftQueue)
|
|
yield* latch.await
|
|
yield* Queue.offerAll(rightQueue, [3, 4])
|
|
const result = yield* Fiber.join(fiber)
|
|
assert.deepStrictEqual(result, [1, 2])
|
|
}))
|
|
|
|
it.effect("merge - interrupts losing side if halt strategy is set to 'either'", () =>
|
|
Effect.gen(function*() {
|
|
const left = Channel.fromEffect(Effect.never)
|
|
const right = Channel.succeed(1)
|
|
const result = yield* Channel.merge(left, right, {
|
|
haltStrategy: "either"
|
|
}).pipe(Channel.runCollect)
|
|
assert.deepStrictEqual(result, [1])
|
|
}))
|
|
|
|
it.effect("merge - waits for both sides if halt strategy is set to 'both'", () =>
|
|
Effect.gen(function*() {
|
|
const left = Channel.succeed(1)
|
|
const right = Channel.succeed(2)
|
|
const result = yield* Channel.merge(left, right, {
|
|
haltStrategy: "both"
|
|
}).pipe(Channel.runCollect)
|
|
assert.deepStrictEqual(result, [1, 2])
|
|
}))
|
|
|
|
it.effect("merge - prioritizes failure", () =>
|
|
Effect.gen(function*() {
|
|
const left = Channel.fromEffect(Effect.fail("boom"))
|
|
const right = Channel.fromEffect(Effect.never)
|
|
const result = yield* Channel.merge(left, right).pipe(
|
|
Channel.runCollect,
|
|
Effect.exit
|
|
)
|
|
assert.deepStrictEqual(result, Exit.fail("boom"))
|
|
}))
|
|
})
|
|
|
|
describe("switchMap", () => {
|
|
it.effect("interrupts previous channels and runs their finalizers", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fromIterable([1, 2, 3]).pipe(
|
|
Channel.switchMap((n) => n === 3 ? Channel.empty : Channel.never),
|
|
Channel.runDrain
|
|
)
|
|
assert.isUndefined(result)
|
|
}))
|
|
})
|
|
|
|
describe("interruptWhen", () => {
|
|
it.effect("interrupts the current element", () =>
|
|
Effect.gen(function*() {
|
|
const interrupted = yield* Ref.make(false)
|
|
const latch = yield* Deferred.make<void>()
|
|
const halt = yield* Deferred.make<void>()
|
|
const started = yield* Deferred.make<void>()
|
|
const channel = Deferred.succeed(started, void 0).pipe(
|
|
Effect.andThen(Deferred.await(latch)),
|
|
Effect.onInterrupt(() => Ref.set(interrupted, true)),
|
|
Channel.fromEffect,
|
|
Channel.interruptWhen(Deferred.await(halt))
|
|
)
|
|
const fiber = yield* Effect.forkChild(Channel.runDrain(channel))
|
|
yield* pipe(
|
|
Deferred.await(started),
|
|
Effect.andThen(Deferred.succeed(halt, void 0))
|
|
)
|
|
yield* Fiber.await(fiber)
|
|
const result = yield* Ref.get(interrupted)
|
|
assertTrue(result)
|
|
}))
|
|
|
|
it.effect("interruptWhen - propagates errors", () =>
|
|
Effect.gen(function*() {
|
|
const deferred = yield* Deferred.make<never, string>()
|
|
const channel = Channel.fromEffect(Effect.never).pipe(
|
|
Channel.interruptWhen(Deferred.await(deferred))
|
|
)
|
|
yield* Deferred.fail(deferred, "fail")
|
|
const result = yield* Effect.result(Channel.runDrain(channel))
|
|
assertFailure(result, "fail")
|
|
}))
|
|
})
|
|
|
|
describe("conditional catch", () => {
|
|
class HttpError extends Data.TaggedError("HttpError")<{
|
|
readonly message: string
|
|
}> {}
|
|
|
|
class ValidationError extends Data.TaggedError("ValidationError")<{
|
|
readonly field: string
|
|
}> {}
|
|
|
|
class RateLimitError extends Data.TaggedError("RateLimitError")<{
|
|
readonly retryAfter: number
|
|
}> {}
|
|
|
|
class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
|
|
readonly limit: number
|
|
}> {}
|
|
|
|
class AiError extends Data.TaggedError("AiError")<{
|
|
readonly reason: RateLimitError | QuotaExceededError
|
|
}> {}
|
|
|
|
class OtherError extends Data.TaggedError("OtherError")<{
|
|
readonly message: string
|
|
}> {}
|
|
|
|
it.effect("catchIf with refinement", () =>
|
|
Effect.gen(function*() {
|
|
const exit = yield* (Channel.fail(new ValidationError({ field: "email" })) as Channel.Channel<
|
|
never,
|
|
HttpError | ValidationError,
|
|
never
|
|
>).pipe(
|
|
Channel.catchIf(
|
|
(error): error is HttpError => error._tag === "HttpError",
|
|
() => Channel.succeed("http")
|
|
),
|
|
Channel.runCollect,
|
|
Effect.exit
|
|
)
|
|
assertExitFailure(exit, Cause.fail(new ValidationError({ field: "email" })))
|
|
}))
|
|
|
|
it.effect("catchIf with predicate", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fail("boom").pipe(
|
|
Channel.catchIf((error) => error === "boom", (error) => Channel.succeed(`recovered: ${error}`)),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, ["recovered: boom"])
|
|
}))
|
|
|
|
it.effect("catchFilter with Filter", () =>
|
|
Effect.gen(function*() {
|
|
const filter = Filter.make((error: string) =>
|
|
error === "boom"
|
|
? Result.succeed(error)
|
|
: Result.fail(error)
|
|
)
|
|
const result = yield* Channel.fail("boom").pipe(
|
|
Channel.catchFilter(filter, (error) => Channel.succeed(`recovered: ${error}`)),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, ["recovered: boom"])
|
|
}))
|
|
|
|
it.effect("catchCauseFilter with Filter", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fail("boom").pipe(
|
|
Channel.catchCauseFilter(
|
|
Cause.findError as any,
|
|
(error) => Channel.succeed(`recovered: ${error}`)
|
|
),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, ["recovered: boom"])
|
|
}))
|
|
|
|
it.effect("catchTag orElse", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.catchTag(
|
|
Channel.fail(new ValidationError({ field: "email" })) as Channel.Channel<
|
|
never,
|
|
HttpError | ValidationError,
|
|
never
|
|
>,
|
|
"HttpError",
|
|
() => Channel.succeed("http"),
|
|
() => Channel.succeed("fallback")
|
|
)
|
|
.pipe(Channel.runCollect)
|
|
assert.deepStrictEqual(result, ["fallback"])
|
|
}))
|
|
|
|
it.effect("catchReason orElse", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fail(
|
|
new AiError({ reason: new QuotaExceededError({ limit: 100 }) })
|
|
).pipe(
|
|
Channel.catchReason(
|
|
"AiError",
|
|
"RateLimitError",
|
|
(reason) => Channel.succeed(`retry: ${reason.retryAfter}`),
|
|
(reason) => Channel.succeed(`quota: ${reason.limit}`)
|
|
),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, ["quota: 100"])
|
|
}))
|
|
|
|
it.effect("catchReason ignores non-matching parent tag", () =>
|
|
Effect.gen(function*() {
|
|
const error = new OtherError({ message: "test" })
|
|
const exit = yield* (Channel.fail(error) as Channel.Channel<never, AiError | OtherError, never>).pipe(
|
|
Channel.catchReason(
|
|
"AiError",
|
|
"RateLimitError",
|
|
() => Channel.succeed("no"),
|
|
() => Channel.succeed("orElse")
|
|
),
|
|
Channel.runCollect,
|
|
Effect.exit
|
|
)
|
|
assertExitFailure(exit, Cause.fail(error))
|
|
}))
|
|
|
|
it.effect("catchReasons orElse", () =>
|
|
Effect.gen(function*() {
|
|
const result = yield* Channel.fail(
|
|
new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
|
|
).pipe(
|
|
Channel.catchReasons(
|
|
"AiError",
|
|
{
|
|
QuotaExceededError: (reason) => Channel.succeed(`quota: ${reason.limit}`)
|
|
},
|
|
(reason) => Channel.succeed(`fallback: ${reason._tag}`)
|
|
),
|
|
Channel.runCollect
|
|
)
|
|
assert.deepStrictEqual(result, ["fallback: RateLimitError"])
|
|
}))
|
|
})
|
|
|
|
describe("unwrapReason", () => {
|
|
class RateLimitError extends Data.TaggedError("RateLimitError")<{
|
|
readonly retryAfter: number
|
|
}> {}
|
|
|
|
class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
|
|
readonly limit: number
|
|
}> {}
|
|
|
|
class AiError extends Data.TaggedError("AiError")<{
|
|
readonly reason: RateLimitError | QuotaExceededError
|
|
}> {}
|
|
|
|
class OtherError extends Data.TaggedError("OtherError")<{
|
|
readonly message: string
|
|
}> {}
|
|
|
|
it.effect("extracts reason into error channel", () =>
|
|
Effect.gen(function*() {
|
|
const reason = new RateLimitError({ retryAfter: 60 })
|
|
const exit = yield* Channel.fail(new AiError({ reason })).pipe(
|
|
Channel.unwrapReason("AiError"),
|
|
Channel.runDrain,
|
|
Effect.exit
|
|
)
|
|
assertExitFailure(exit, Cause.fail(reason))
|
|
}))
|
|
|
|
it.effect("extracts second reason type", () =>
|
|
Effect.gen(function*() {
|
|
const reason = new QuotaExceededError({ limit: 100 })
|
|
const exit = yield* Channel.fail(new AiError({ reason })).pipe(
|
|
Channel.unwrapReason("AiError"),
|
|
Channel.runDrain,
|
|
Effect.exit
|
|
)
|
|
assertExitFailure(exit, Cause.fail(reason))
|
|
}))
|
|
|
|
it.effect("preserves other errors", () =>
|
|
Effect.gen(function*() {
|
|
const error = new OtherError({ message: "test" })
|
|
const exit = yield* (Channel.fail(error) as Channel.Channel<never, AiError | OtherError, never>).pipe(
|
|
Channel.unwrapReason("AiError"),
|
|
Channel.runDrain,
|
|
Effect.exit
|
|
)
|
|
assertExitFailure(exit, Cause.fail(error))
|
|
}))
|
|
})
|
|
})
|