| 5 | import { Effect, Fiber, Layer, Schema, TestClock, TestServices } from "effect" |
| 6 | |
| 7 | export const suite = <E>(client: Layer.Layer<SqlClient.SqlClient, E>) => { |
| 8 | const layer = PersistedQueue.layer.pipe( |
| 9 | Layer.provide(SqlPersistedQueue.layerStore()), |
| 10 | Layer.provideMerge(client) |
| 11 | ) |
| 12 | |
| 13 | it.layer(layer, { timeout: "30 seconds" })("SqlPersistedQueue", (it) => { |
| 14 | it.effect("offer + take", () => |
| 15 | Effect.gen(function*() { |
| 16 | const queue = yield* PersistedQueue.make({ |
| 17 | name: "test-queue-a", |
| 18 | schema: Item |
| 19 | }) |
| 20 | |
| 21 | yield* queue.offer({ n: 42n }) |
| 22 | yield* queue.take(Effect.fnUntraced(function*(value) { |
| 23 | assert.strictEqual(value.n, 42n) |
| 24 | })) |
| 25 | })) |
| 26 | |
| 27 | it.effect("interrupt", () => |
| 28 | Effect.gen(function*() { |
| 29 | const queue = yield* PersistedQueue.make({ |
| 30 | name: "test-queue-b", |
| 31 | schema: Item |
| 32 | }) |
| 33 | |
| 34 | yield* queue.offer({ n: 42n }) |
| 35 | |
| 36 | const latch = Effect.unsafeMakeLatch() |
| 37 | const fiber = yield* queue.take(Effect.fnUntraced(function*(_value) { |
| 38 | yield* latch.open |
| 39 | return yield* Effect.never |
| 40 | })).pipe(Effect.fork) |
| 41 | |
| 42 | const fiber2 = yield* queue.take((val) => Effect.succeed(val)).pipe(Effect.fork) |
| 43 | |
| 44 | yield* latch.await |
| 45 | |
| 46 | // allow some real time to pass to ensure the second take is really |
| 47 | // waiting |
| 48 | yield* TestClock.adjust(1000) |
| 49 | yield* Effect.sleep(1000).pipe( |
| 50 | TestServices.provideLive |
| 51 | ) |
| 52 | assert.isNull(fiber2.unsafePoll()) |
| 53 | |
| 54 | yield* Fiber.interrupt(fiber) |
| 55 | |
| 56 | yield* TestClock.adjust(1000) |
| 57 | |
| 58 | assert.strictEqual((yield* Fiber.join(fiber2)).n, 42n) |
| 59 | })) |
| 60 | |
| 61 | it.effect("failure", () => |
| 62 | Effect.gen(function*() { |
| 63 | const queue = yield* PersistedQueue.make({ |
| 64 | name: "test-queue-c", |