1import { assert, describe, it } from "@effect/vitest"
2import { Array, Cause, Effect, Exit, Fiber, Latch, Number, Option, Random, Stream, SubscriptionRef } from "effect"
3
4describe("SubscriptionRef", () => {
5 it.effect("isSubscriptionRef", () =>
6 Effect.gen(function*() {
7 const ref = yield* SubscriptionRef.make(0)
8 assert.isTrue(SubscriptionRef.isSubscriptionRef(ref))
9 assert.isFalse(SubscriptionRef.isSubscriptionRef([0]))
10 }))
11
12 it.effect("getAndUpdateEffect", () =>
13 Effect.gen(function*() {
14 const ref = yield* SubscriptionRef.make(1)
15 const previous = yield* SubscriptionRef.getAndUpdateEffect(ref, (value) => Effect.succeed(value + 1))
16 assert.strictEqual(previous, 1)
17 assert.strictEqual(yield* SubscriptionRef.get(ref), 2)
18 }))
19
20 it.effect("getAndUpdateSome returns the current value when no update is selected", () =>
21 Effect.gen(function*() {
22 const ref = yield* SubscriptionRef.make(1)
23 const previous = yield* SubscriptionRef.getAndUpdateSome(ref, () => Option.none())
24
25 assert.strictEqual(previous, 1)
26 }))
27
28 it.effect("multiple subscribers can receive changes", () =>
29 Effect.gen(function*() {
30 const ref = yield* SubscriptionRef.make(0)
31 const latch1 = yield* Latch.make()
32 const latch2 = yield* Latch.make()
33 const fiber1 = yield* SubscriptionRef.changes(ref).pipe(
34 Stream.tap(() => latch1.open),
35 Stream.take(3),
36 Stream.runCollect,
37 Effect.forkScoped
38 )
39 yield* latch1.await
40 yield* SubscriptionRef.update(ref, Number.increment)
41 const fiber2 = yield* SubscriptionRef.changes(ref).pipe(
42 Stream.tap(() => latch2.open),
43 Stream.take(2),
44 Stream.runCollect,
45 Effect.forkScoped
46 )
47 yield* latch2.await
48 yield* SubscriptionRef.update(ref, Number.increment)
49 const result1 = yield* Fiber.join(fiber1)
50 const result2 = yield* Fiber.join(fiber2)
51 assert.deepStrictEqual(result1, [0, 1, 2])
52 assert.deepStrictEqual(result2, [1, 2])
53 }))
54
55 it.effect("subscriptions are interruptible", () =>
56 Effect.gen(function*() {
57 const ref = yield* SubscriptionRef.make(0)
58 const latch1 = yield* Latch.make()
59 const latch2 = yield* Latch.make()
60 const fiber1 = yield* SubscriptionRef.changes(ref).pipe(
61 Stream.tap(() => latch1.open),
62 Stream.take(5),
63 Stream.runCollect,
64 Effect.forkScoped
65 )
66 yield* latch1.await
67 yield* SubscriptionRef.update(ref, Number.increment)
68 const fiber2 = yield* SubscriptionRef.changes(ref).pipe(
69 Stream.tap(() => latch2.open),
70 Stream.take(2),
71 Stream.runCollect,
72 Effect.forkScoped
73 )
74 yield* latch2.await
75 yield* SubscriptionRef.update(ref, Number.increment)
76 yield* Fiber.interrupt(fiber1)
77 const result1 = yield* Fiber.await(fiber1)
78 const result2 = yield* Fiber.join(fiber2)
79 assert.isTrue(Exit.isFailure(result1) && Cause.hasInterruptsOnly(result1.cause))
80 assert.deepStrictEqual(result2, [1, 2])
81 }))
82
83 it.effect("concurrent subscribes and unsubscribes are handled correctly", () =>
84 Effect.gen(function*() {
85 const ref = yield* SubscriptionRef.make(0)
86 const producer = yield* SubscriptionRef.update(ref, Number.increment).pipe(
87 Effect.forever,
88 Effect.forkScoped
89 )
90 const [result1, result2] = yield* Effect.all([
91 makeConsumer(ref),
92 makeConsumer(ref)
93 ], { concurrency: 2 })
94 yield* Fiber.interrupt(producer)
95 assert.deepStrictEqual(result1, Array.sort(Number.Order)(result1))
96 assert.deepStrictEqual(result2, Array.sort(Number.Order)(result2))
97 }))
98})
99
100const makeConsumer = Effect.fnUntraced(
101 function*(ref: SubscriptionRef.SubscriptionRef<number>) {
102 const n = yield* Random.nextIntBetween(0, 200)
103 const changes = SubscriptionRef.changes(ref)
104 return yield* changes.pipe(Stream.take(n), Stream.runCollect)
105 }
106)