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.scoped(
19 Effect.gen(function* () {
20 const changes = yield* PubSub.sliding<number>(1);
21 const latest = yield* Ref.make(1);
22 const mutex = yield* Semaphore.make(1);
23 const snapshotStarted = yield* Deferred.make<void>();
24 const finishSnapshot = yield* Deferred.make<void>();
25 const subscriptionFiber = yield* subscribeBeforeSnapshot(
26 changes,
27 Deferred.succeed(snapshotStarted, undefined).pipe(
28 Effect.andThen(Deferred.await(finishSnapshot)),
29 Effect.andThen(Ref.get(latest)),
30 ),
31 mutex,
32 ).pipe(Effect.forkChild);
33
34 yield* Deferred.await(snapshotStarted);
35 const publishFiber = yield* mutex
36 .withPermits(1)(Ref.set(latest, 2).pipe(Effect.andThen(PubSub.publish(changes, 2))))
37 .pipe(Effect.forkChild);
38 yield* Deferred.succeed(finishSnapshot, undefined);
39
40 const subscription = yield* Fiber.join(subscriptionFiber);
41 yield* Fiber.join(publishFiber);
42 const firstChange = yield* subscription.changes.pipe(
43 Stream.runHead,
44 Effect.timeout("1 second"),
45 );
46
47 expect(subscription.latest).toBe(1);
48 expect(firstChange).toEqual(Option.some(2));
49 }),
50 ),
51 );
52
53 it.effect("subscribes before an uncoordinated snapshot can change", () =>
54 Effect.scoped(
55 Effect.gen(function* () {
56 const changes = yield* PubSub.sliding<number>(1);
57 const latest = yield* Ref.make(1);
58 const snapshotStarted = yield* Deferred.make<void>();
59 const finishSnapshot = yield* Deferred.make<void>();
60 const subscriptionFiber = yield* subscribeBeforeSnapshotWithoutMutex(
61 changes,
62 Deferred.succeed(snapshotStarted, undefined).pipe(
63 Effect.andThen(Deferred.await(finishSnapshot)),
64 Effect.andThen(Ref.get(latest)),
65 ),
66 ).pipe(Effect.forkChild);
67
68 yield* Deferred.await(snapshotStarted);
69 yield* Ref.set(latest, 2);
70 yield* PubSub.publish(changes, 2);
71 yield* Deferred.succeed(finishSnapshot, undefined);
72
73 const subscription = yield* Fiber.join(subscriptionFiber);
74 const firstChange = yield* subscription.changes.pipe(
75 Stream.runHead,
76 Effect.timeout("1 second"),
77 );
78
79 expect(subscription.latest).toBe(2);
80 expect(firstChange).toEqual(Option.some(2));
81 }),
82 ),
83 );
84});