import { assert, describe, it } from "@effect/vitest";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect"; import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker.ts"; describe("makeKeyedCoalescingWorker", () => { it.live("waits for latest work enqueued during active processing before draining the key", () => Effect.scoped( Effect.gen(function* () { const processed: string[] = []; const firstStarted = yield* Deferred.make<void>(); const releaseFirst = yield* Deferred.make<void>(); const secondStarted = yield* Deferred.make<void>(); const releaseSecond = yield* Deferred.make<void>(); const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({ merge: (_current, next) => next, process: (key, value) => Effect.gen(function* () { processed.push(`${key}:${value}`); if (value === "first") { yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie); yield* Deferred.await(releaseFirst); } if (value === "second") { yield* Deferred.succeed(secondStarted, undefined).pipe(Effect.orDie); yield* Deferred.await(releaseSecond); } }), }); yield* worker.enqueue("terminal-1", "first"); yield* Deferred.await(firstStarted); const drained = yield* Deferred.make<void>(); yield* Effect.forkChild( worker .drainKey("terminal-1") .pipe(Effect.tap(() => Deferred.succeed(drained, undefined).pipe(Effect.orDie))), ); yield* worker.enqueue("terminal-1", "second"); yield* Deferred.succeed(releaseFirst, undefined); yield* Deferred.await(secondStarted); assert.equal(yield* Deferred.isDone(drained), false); yield* Deferred.succeed(releaseSecond, undefined); yield* Deferred.await(drained); assert.deepStrictEqual(processed, ["terminal-1:first", "terminal-1:second"]); }), ), ); it.live("requeues pending work for a key after a processor failure and keeps draining", () => Effect.scoped( Effect.gen(function* () { const processed: string[] = []; const firstStarted = yield* Deferred.make<void>(); const releaseFailure = yield* Deferred.make<void>(); const secondProcessed = yield* Deferred.make<void>(); const worker = yield* makeKeyedCoalescingWorker<string, string, string, never>({ merge: (_current, next) => next, process: (key, value) => Effect.gen(function* () { processed.push(`${key}:${value}`); if (value === "first") { yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie); yield* Deferred.await(releaseFailure); return yield* Effect.fail("boom"); } if (value === "second") { yield* Deferred.succeed(secondProcessed, undefined).pipe(Effect.orDie); } }), }); yield* worker.enqueue("terminal-1", "first"); yield* Deferred.await(firstStarted); yield* worker.enqueue("terminal-1", "second"); yield* Deferred.succeed(releaseFailure, undefined); yield* Deferred.await(secondProcessed); yield* worker.drainKey("terminal-1"); assert.deepStrictEqual(processed, ["terminal-1:first", "terminal-1:second"]); }), ), ); it.live("treats an undefined value as present queued work", () => Effect.scoped( Effect.gen(function* () { const processed: Array<string | undefined> = []; const worker = yield* makeKeyedCoalescingWorker<string, string | undefined, never, never>({ merge: (_current, next) => next, process: (_key, value) => Effect.sync(() => { processed.push(value); }), }); yield* worker.enqueue("terminal-1", undefined); yield* worker.drainKey("terminal-1"); assert.deepStrictEqual(processed, [undefined]); }), ), );});
import { assert, describe, it } from "@effect/vitest";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import { makeKeyedCoalescingWorker } from "./KeyedCoalescingWorker.ts";
describe("makeKeyedCoalescingWorker", () => {
it.live("waits for latest work enqueued during active processing before draining the key", () =>
Effect.scoped(
Effect.gen(function* () {
const processed: string[] = [];
const firstStarted = yield* Deferred.make<void>();
const releaseFirst = yield* Deferred.make<void>();
const secondStarted = yield* Deferred.make<void>();
const releaseSecond = yield* Deferred.make<void>();
const worker = yield* makeKeyedCoalescingWorker<string, string, never, never>({
merge: (_current, next) => next,
process: (key, value) =>
Effect.gen(function* () {
processed.push(`${key}:${value}`);
if (value === "first") {
yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie);
yield* Deferred.await(releaseFirst);
}
if (value === "second") {
yield* Deferred.succeed(secondStarted, undefined).pipe(Effect.orDie);
yield* Deferred.await(releaseSecond);
}
}),
});
yield* worker.enqueue("terminal-1", "first");
yield* Deferred.await(firstStarted);
const drained = yield* Deferred.make<void>();
yield* Effect.forkChild(
worker
.drainKey("terminal-1")
.pipe(Effect.tap(() => Deferred.succeed(drained, undefined).pipe(Effect.orDie))),
);
yield* worker.enqueue("terminal-1", "second");
yield* Deferred.succeed(releaseFirst, undefined);
yield* Deferred.await(secondStarted);
assert.equal(yield* Deferred.isDone(drained), false);
yield* Deferred.succeed(releaseSecond, undefined);
yield* Deferred.await(drained);
assert.deepStrictEqual(processed, ["terminal-1:first", "terminal-1:second"]);
}),
),
);
it.live("requeues pending work for a key after a processor failure and keeps draining", () =>
Effect.scoped(
Effect.gen(function* () {
const processed: string[] = [];
const firstStarted = yield* Deferred.make<void>();
const releaseFailure = yield* Deferred.make<void>();
const secondProcessed = yield* Deferred.make<void>();
const worker = yield* makeKeyedCoalescingWorker<string, string, string, never>({
merge: (_current, next) => next,
process: (key, value) =>
Effect.gen(function* () {
processed.push(`${key}:${value}`);
if (value === "first") {
yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie);
yield* Deferred.await(releaseFailure);
return yield* Effect.fail("boom");
}
if (value === "second") {
yield* Deferred.succeed(secondProcessed, undefined).pipe(Effect.orDie);
}
}),
});
yield* worker.enqueue("terminal-1", "first");
yield* Deferred.await(firstStarted);
yield* worker.enqueue("terminal-1", "second");
yield* Deferred.succeed(releaseFailure, undefined);
yield* Deferred.await(secondProcessed);
yield* worker.drainKey("terminal-1");
assert.deepStrictEqual(processed, ["terminal-1:first", "terminal-1:second"]);
}),
),
);
it.live("treats an undefined value as present queued work", () =>
Effect.scoped(
Effect.gen(function* () {
const processed: Array<string | undefined> = [];
const worker = yield* makeKeyedCoalescingWorker<string, string | undefined, never, never>({
merge: (_current, next) => next,
process: (_key, value) =>
Effect.sync(() => {
processed.push(value);
}),
});
yield* worker.enqueue("terminal-1", undefined);
yield* worker.drainKey("terminal-1");
assert.deepStrictEqual(processed, [undefined]);
}),
),
);
});