1import { describe, expect, it } from "@effect/vitest";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as Option from "effect/Option";
6import * as PubSub from "effect/PubSub";
7import * as Ref from "effect/Ref";
8import * as Semaphore from "effect/Semaphore";
9import * as Stream from "effect/Stream";
10
11import {
12 subscribeBeforeSnapshot,
13 subscribeBeforeSnapshotWithoutMutex,
14} from "./subscribeBeforeSnapshot.ts";
15
16describe("subscribeBeforeSnapshot", () => {
17 it.effect("atomically reads the initial snapshot before receiving later changes", () =>
18 Effect.gen(function* () {
19 const changes = yield* PubSub.sliding<number>(1);
20 const latest = yield* Ref.make(1);
21 const mutex = yield* Semaphore.make(1);
22 const snapshotStarted = yield* Deferred.make<void>();
23 const finishSnapshot = yield* Deferred.make<void>();
24 const subscriptionFiber = yield* subscribeBeforeSnapshot(
25 changes,
26 Deferred.succeed(snapshotStarted, undefined).pipe(
27 Effect.andThen(Deferred.await(finishSnapshot)),
28 Effect.andThen(Ref.get(latest)),
29 ),
30 mutex,
31 ).pipe(Effect.forkChild);
32
33 yield* Deferred.await(snapshotStarted);
34 const publishFiber = yield* mutex
35 .withPermits(1)(Ref.set(latest, 2).pipe(Effect.andThen(PubSub.publish(changes, 2))))
36 .pipe(Effect.forkChild);
37 yield* Deferred.succeed(finishSnapshot, undefined);
38
39 const subscription = yield* Fiber.join(subscriptionFiber);
40 yield* Fiber.join(publishFiber);
41 const firstChange = yield* subscription.changes.pipe(
42 Stream.runHead,
43 Effect.timeout("1 second"),
44 );
45
46 expect(subscription.latest).toBe(1);
47 expect(firstChange).toEqual(Option.some(2));
48 }),
49 );
50
51 it.effect("subscribes before an uncoordinated snapshot can change", () =>
52 Effect.gen(function* () {
53 const changes = yield* PubSub.sliding<number>(1);
54 const latest = yield* Ref.make(1);
55 const snapshotStarted = yield* Deferred.make<void>();
56 const finishSnapshot = yield* Deferred.make<void>();
57 const subscriptionFiber = yield* subscribeBeforeSnapshotWithoutMutex(
58 changes,
59 Deferred.succeed(snapshotStarted, undefined).pipe(
60 Effect.andThen(Deferred.await(finishSnapshot)),
61 Effect.andThen(Ref.get(latest)),
62 ),
63 ).pipe(Effect.forkChild);
64
65 yield* Deferred.await(snapshotStarted);
66 yield* Ref.set(latest, 2);
67 yield* PubSub.publish(changes, 2);
68 yield* Deferred.succeed(finishSnapshot, undefined);
69
70 const subscription = yield* Fiber.join(subscriptionFiber);
71 const firstChange = yield* subscription.changes.pipe(
72 Stream.runHead,
73 Effect.timeout("1 second"),
74 );
75
76 expect(subscription.latest).toBe(2);
77 expect(firstChange).toEqual(Option.some(2));
78 }),
79 );
80});