Function suite

packages/sql/test/SqlPersistedQueueTest.ts:7–104

(client: Layer.Layer<SqlClient.SqlClient, E>)

import { Effect, Fiber, Layer, Schema, TestClock, TestServices } from "effect"

export const suite = <E>(client: Layer.Layer<SqlClient.SqlClient, E>) => {
  const layer = PersistedQueue.layer.pipe(
    Layer.provide(SqlPersistedQueue.layerStore()),
    Layer.provideMerge(client)
  )

  it.layer(layer, { timeout: "30 seconds" })("SqlPersistedQueue", (it) => {
    it.effect("offer + take", () =>
      Effect.gen(function*() {
        const queue = yield* PersistedQueue.make({
          name: "test-queue-a",
          schema: Item
        })

        yield* queue.offer({ n: 42n })
        yield* queue.take(Effect.fnUntraced(function*(value) {
          assert.strictEqual(value.n, 42n)
        }))
      }))

    it.effect("interrupt", () =>
      Effect.gen(function*() {
        const queue = yield* PersistedQueue.make({
          name: "test-queue-b",
          schema: Item
        })

        yield* queue.offer({ n: 42n })

        const latch = Effect.unsafeMakeLatch()
        const fiber = yield* queue.take(Effect.fnUntraced(function*(_value) {
          yield* latch.open
          return yield* Effect.never
        })).pipe(Effect.fork)

        const fiber2 = yield* queue.take((val) => Effect.succeed(val)).pipe(Effect.fork)

        yield* latch.await

        // allow some real time to pass to ensure the second take is really
        // waiting
        yield* TestClock.adjust(1000)
        yield* Effect.sleep(1000).pipe(
          TestServices.provideLive
        )
        assert.isNull(fiber2.unsafePoll())

        yield* Fiber.interrupt(fiber)

        yield* TestClock.adjust(1000)

        assert.strictEqual((yield* Fiber.join(fiber2)).n, 42n)
      }))

    it.effect("failure", () =>
      Effect.gen(function*() {
        const queue = yield* PersistedQueue.make({
          name: "test-queue-c",