1/**
2 * Stores transactional state and publishes committed changes.
3 *
4 * A `TxSubscriptionRef<A>` combines a `TxRef<A>` for the current value with a
5 * transactional pub/sub channel for updates. Subscribers first receive the
6 * current value and then every later value that is published by committed
7 * updates. This module includes constructors, reads, writes, update and modify
8 * helpers, transactional-queue subscriptions, stream subscriptions, and a guard.
9 *
10 * @since 4.0.0
11 */
12import * as Effect from "./Effect.ts"
13import { dual } from "./Function.ts"
14import type { Inspectable } from "./Inspectable.ts"
15import { NodeInspectSymbol, toJson } from "./Inspectable.ts"
16import type { Pipeable } from "./Pipeable.ts"
17import { pipeArguments } from "./Pipeable.ts"
18import { hasProperty } from "./Predicate.ts"
19import type * as Scope from "./Scope.ts"
20import * as Stream from "./Stream.ts"
21import * as TxPubSub from "./TxPubSub.ts"
22import * as TxQueue from "./TxQueue.ts"
23import * as TxRef from "./TxRef.ts"
24
25const TypeId = "~effect/TxSubscriptionRef"
26
27/**
28 * A TxSubscriptionRef is a transactional reference that allows subscribing to all
29 * committed changes. Subscribers receive the current value followed by every subsequent
30 * update via a transactional dequeue.
31 *
32 * **When to use**
33 *
34 * Use to store transactional state whose committed changes must be observable by
35 * subscribers.
36 *
37 * **Example** (Subscribing to transactional changes)
38 *
39 * ```ts import.meta.vitest
40 * import { Effect, TxQueue, TxSubscriptionRef } from "effect"
41 *
42 * const program = Effect.gen(function*() {
43 * const ref = yield* TxSubscriptionRef.make(0)
44 *
45 * return yield* Effect.scoped(
46 * Effect.gen(function*() {
47 * const sub = yield* TxSubscriptionRef.changes(ref)
48 * const initial = yield* TxQueue.take(sub)
49 *
50 * yield* TxSubscriptionRef.set(ref, 1)
51 * const next = yield* TxQueue.take(sub)
52 * return [initial, next]
53 * })
54 * )
55 * })
56 *
57 * await Effect.runPromise(program) // => [0, 1]
58 * ```
59 *
60 * @see {@link make} for creating a transactional subscription reference
61 * @see {@link changes} for subscribing through a transactional queue
62 * @see {@link changesStream} for subscribing through a `Stream`
63 *
64 * @category models
65 * @since 4.0.0
66 */
67export interface TxSubscriptionRef<in out A> extends Inspectable, Pipeable {
68 readonly [TypeId]: typeof TypeId
69 /** @internal */
70 readonly ref: TxRef.TxRef<A>
71 /** @internal */
72 readonly pubsub: TxPubSub.TxPubSub<A>
73}
74
75const TxSubscriptionRefProto: Omit<TxSubscriptionRef<any>, typeof TypeId | "ref" | "pubsub"> = {
76 [NodeInspectSymbol](this: TxSubscriptionRef<unknown>) {
77 return toJson(this)
78 },
79 toJSON(this: TxSubscriptionRef<unknown>) {
80 return { _id: "TxSubscriptionRef" }
81 },
82 toString() {
83 return "TxSubscriptionRef"
84 },
85 pipe() {
86 return pipeArguments(this, arguments)
87 }
88}
89
90// =============================================================================
91// Constructors
92// =============================================================================
93
94/**
95 * Creates a new TxSubscriptionRef with the specified initial value.
96 *
97 * **When to use**
98 *
99 * Use to create a `TxSubscriptionRef` that publishes every committed update to
100 * subscribers.
101 *
102 * **Example** (Creating a transactional subscription reference)
103 *
104 * ```ts import.meta.vitest
105 * import { Effect, TxSubscriptionRef } from "effect"
106 *
107 * const program = Effect.gen(function*() {
108 * const ref = yield* TxSubscriptionRef.make(42)
109 * return yield* TxSubscriptionRef.get(ref)
110 * })
111 *
112 * await Effect.runPromise(program) // => 42
113 * ```
114 *
115 * @see {@link changes} for subscribing to the created reference
116 *
117 * @category constructors
118 * @since 3.10.0
119 */
120export const make = <A>(value: A): Effect.Effect<TxSubscriptionRef<A>> =>
121 Effect.gen(function*() {
122 const ref = yield* TxRef.make(value)
123 const pubsub = yield* TxPubSub.unbounded<A>()
124 const self = Object.create(TxSubscriptionRefProto)
125 self[TypeId] = TypeId
126 self.ref = ref
127 self.pubsub = pubsub
128 return self
129 }).pipe(Effect.tx)
130
131// =============================================================================
132// Getters
133// =============================================================================
134
135/**
136 * Reads the current value of the TxSubscriptionRef.
137 *
138 * **When to use**
139 *
140 * Use to read the current `TxSubscriptionRef` value without subscribing to
141 * future changes.
142 *
143 * **Example** (Reading the current value)
144 *
145 * ```ts import.meta.vitest
146 * import { Effect, TxSubscriptionRef } from "effect"
147 *
148 * const program = Effect.gen(function*() {
149 * const ref = yield* TxSubscriptionRef.make("hello")
150 * return yield* TxSubscriptionRef.get(ref)
151 * })
152 *
153 * await Effect.runPromise(program) // => "hello"
154 * ```
155 *
156 * @see {@link changes} for reading the current value and subsequent updates
157 *
158 * @category getters
159 * @since 3.10.0
160 */
161export const get = <A>(self: TxSubscriptionRef<A>): Effect.Effect<A> => TxRef.get(self.ref)
162
163// =============================================================================
164// Mutations
165// =============================================================================
166
167/**
168 * Modifies the value of the TxSubscriptionRef using a function that returns both a
169 * result and the new value. The new value is published to all subscribers atomically.
170 *
171 * **When to use**
172 *
173 * Use to compute a separate return value and next `TxSubscriptionRef` state in
174 * one transactional update.
175 *
176 * **Example** (Modifying and returning a value)
177 *
178 * ```ts import.meta.vitest
179 * import { Effect, TxSubscriptionRef } from "effect"
180 *
181 * const program = Effect.gen(function*() {
182 * const ref = yield* TxSubscriptionRef.make(10)
183 * const result = yield* TxSubscriptionRef.modify(ref, (n) => [`was ${n}`, n + 1])
184 * return [result, yield* TxSubscriptionRef.get(ref)]
185 * })
186 *
187 * await Effect.runPromise(program) // => ["was 10", 11]
188 * ```
189 *
190 * @see {@link update} for deriving the next value without a separate return value
191 * @see {@link set} for replacing the value directly
192 *
193 * @category mutations
194 * @since 3.10.0
195 */
196export const modify: {
197 <A, B>(
198 f: (current: A) => [returnValue: B, newValue: A]
199 ): (self: TxSubscriptionRef<A>) => Effect.Effect<B>
200 <A, B>(
201 self: TxSubscriptionRef<A>,
202 f: (current: A) => [returnValue: B, newValue: A]
203 ): Effect.Effect<B>
204} = dual(
205 2,
206 <A, B>(
207 self: TxSubscriptionRef<A>,
208 f: (current: A) => [returnValue: B, newValue: A]
209 ): Effect.Effect<B> =>
210 Effect.gen(function*() {
211 const current = yield* TxRef.get(self.ref)
212 const [returnValue, newValue] = f(current)
213 yield* TxRef.set(self.ref, newValue)
214 yield* TxPubSub.publish(self.pubsub, newValue)
215 return returnValue
216 }).pipe(Effect.tx)
217)
218
219/**
220 * Sets the value of the TxSubscriptionRef and publishes the new value to all subscribers.
221 *
222 * **When to use**
223 *
224 * Use to replace the current `TxSubscriptionRef` value with a known value and
225 * publish it.
226 *
227 * **Example** (Setting a new value)
228 *
229 * ```ts import.meta.vitest
230 * import { Effect, TxSubscriptionRef } from "effect"
231 *
232 * const program = Effect.gen(function*() {
233 * const ref = yield* TxSubscriptionRef.make(0)
234 * yield* TxSubscriptionRef.set(ref, 42)
235 * return yield* TxSubscriptionRef.get(ref)
236 * })
237 *
238 * await Effect.runPromise(program) // => 42
239 * ```
240 *
241 * @see {@link update} for deriving the new value from the current value
242 * @see {@link getAndSet} for setting while returning the previous value
243 *
244 * @category mutations
245 * @since 3.10.0
246 */
247export const set: {
248 <A>(value: A): (self: TxSubscriptionRef<A>) => Effect.Effect<void>
249 <A>(self: TxSubscriptionRef<A>, value: A): Effect.Effect<void>
250} = dual(
251 2,
252 <A>(self: TxSubscriptionRef<A>, value: A): Effect.Effect<void> => modify(self, () => [void 0, value])
253)
254
255/**
256 * Updates the value of the TxSubscriptionRef using a function and publishes the new
257 * value to all subscribers.
258 *
259 * **When to use**
260 *
261 * Use to derive the next `TxSubscriptionRef` value from the current value and
262 * publish it.
263 *
264 * **Example** (Updating a value)
265 *
266 * ```ts import.meta.vitest
267 * import { Effect, TxSubscriptionRef } from "effect"
268 *
269 * const program = Effect.gen(function*() {
270 * const ref = yield* TxSubscriptionRef.make(5)
271 * yield* TxSubscriptionRef.update(ref, (n) => n * 2)
272 * return yield* TxSubscriptionRef.get(ref)
273 * })
274 *
275 * await Effect.runPromise(program) // => 10
276 * ```
277 *
278 * @see {@link set} for replacing the value directly
279 * @see {@link updateAndGet} for returning the new value after the update
280 *
281 * @category mutations
282 * @since 3.10.0
283 */
284export const update: {
285 <A>(f: (current: A) => A): (self: TxSubscriptionRef<A>) => Effect.Effect<void>
286 <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect.Effect<void>
287} = dual(
288 2,
289 <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect.Effect<void> =>
290 modify(self, (current) => [void 0, f(current)])
291)
292
293/**
294 * Gets the current value and sets a new value atomically. Publishes the new value
295 * to all subscribers.
296 *
297 * **When to use**
298 *
299 * Use to replace a `TxSubscriptionRef` value while returning the previous value
300 * and publishing the update to subscribers.
301 *
302 * **Example** (Getting and setting atomically)
303 *
304 * ```ts import.meta.vitest
305 * import { Effect, TxSubscriptionRef } from "effect"
306 *
307 * const program = Effect.gen(function*() {
308 * const ref = yield* TxSubscriptionRef.make("a")
309 * const old = yield* TxSubscriptionRef.getAndSet(ref, "b")
310 * return [old, yield* TxSubscriptionRef.get(ref)]
311 * })
312 *
313 * await Effect.runPromise(program) // => ["a", "b"]
314 * ```
315 *
316 * @see {@link set} for setting without returning the previous value
317 * @see {@link getAndUpdate} for deriving the new value from the previous value
318 *
319 * @category mutations
320 * @since 3.10.0
321 */
322export const getAndSet: {
323 <A>(value: A): (self: TxSubscriptionRef<A>) => Effect.Effect<A>
324 <A>(self: TxSubscriptionRef<A>, value: A): Effect.Effect<A>
325} = dual(
326 2,
327 <A>(self: TxSubscriptionRef<A>, value: A): Effect.Effect<A> => modify(self, (current) => [current, value])
328)
329
330/**
331 * Gets the current value and updates it using a function atomically. Publishes
332 * the new value to all subscribers.
333 *
334 * **When to use**
335 *
336 * Use to derive and publish a new `TxSubscriptionRef` value while returning the
337 * previous value.
338 *
339 * **Example** (Getting and updating atomically)
340 *
341 * ```ts import.meta.vitest
342 * import { Effect, TxSubscriptionRef } from "effect"
343 *
344 * const program = Effect.gen(function*() {
345 * const ref = yield* TxSubscriptionRef.make(1)
346 * const old = yield* TxSubscriptionRef.getAndUpdate(ref, (n) => n + 10)
347 * return [old, yield* TxSubscriptionRef.get(ref)]
348 * })
349 *
350 * await Effect.runPromise(program) // => [1, 11]
351 * ```
352 *
353 * @see {@link update} for updating without returning the previous value
354 * @see {@link updateAndGet} for returning the new value instead
355 *
356 * @category mutations
357 * @since 3.10.0
358 */
359export const getAndUpdate: {
360 <A>(f: (current: A) => A): (self: TxSubscriptionRef<A>) => Effect.Effect<A>
361 <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect.Effect<A>
362} = dual(
363 2,
364 <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect.Effect<A> =>
365 modify(self, (current) => [current, f(current)])
366)
367
368/**
369 * Updates the value using a function and returns the new value. Publishes the
370 * new value to all subscribers.
371 *
372 * **When to use**
373 *
374 * Use to derive and publish a new `TxSubscriptionRef` value while returning
375 * that new value.
376 *
377 * **Example** (Updating and reading atomically)
378 *
379 * ```ts import.meta.vitest
380 * import { Effect, TxSubscriptionRef } from "effect"
381 *
382 * const program = Effect.gen(function*() {
383 * const ref = yield* TxSubscriptionRef.make(3)
384 * return yield* TxSubscriptionRef.updateAndGet(ref, (n) => n * 3)
385 * })
386 *
387 * await Effect.runPromise(program) // => 9
388 * ```
389 *
390 * @see {@link update} for updating without returning the new value
391 * @see {@link getAndUpdate} for returning the previous value instead
392 *
393 * @category mutations
394 * @since 3.10.0
395 */
396export const updateAndGet: {
397 <A>(f: (current: A) => A): (self: TxSubscriptionRef<A>) => Effect.Effect<A>
398 <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect.Effect<A>
399} = dual(
400 2,
401 <A>(self: TxSubscriptionRef<A>, f: (current: A) => A): Effect.Effect<A> =>
402 modify(self, (current) => {
403 const newValue = f(current)
404 return [newValue, newValue]
405 })
406)
407
408// =============================================================================
409// Subscriptions
410// =============================================================================
411
412/**
413 * Subscribes to all changes of the TxSubscriptionRef. Returns a scoped TxDequeue
414 * that first yields the current value, then every subsequent update.
415 *
416 * **When to use**
417 *
418 * Use to subscribe to `TxSubscriptionRef` committed changes through a scoped
419 * transactional queue.
420 *
421 * **Example** (Subscribing to changes)
422 *
423 * ```ts import.meta.vitest
424 * import { Effect, TxQueue, TxSubscriptionRef } from "effect"
425 *
426 * const program = Effect.gen(function*() {
427 * const ref = yield* TxSubscriptionRef.make(0)
428 *
429 * return yield* Effect.scoped(
430 * Effect.gen(function*() {
431 * const sub = yield* TxSubscriptionRef.changes(ref)
432 * const initial = yield* TxQueue.take(sub)
433 *
434 * yield* TxSubscriptionRef.set(ref, 1)
435 * const next = yield* TxQueue.take(sub)
436 * return [initial, next]
437 * })
438 * )
439 * })
440 *
441 * await Effect.runPromise(program) // => [0, 1]
442 * ```
443 *
444 * @see {@link changesStream} for subscribing through a `Stream`
445 *
446 * @category subscriptions
447 * @since 3.10.0
448 */
449export const changes = <A>(
450 self: TxSubscriptionRef<A>
451): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope> =>
452 Effect.acquireRelease(
453 Effect.tx(
454 Effect.gen(function*() {
455 const sub = yield* TxPubSub.acquireSubscriber(self.pubsub)
456 const current = yield* TxRef.get(self.ref)
457 yield* TxQueue.offer(sub, current)
458 return sub
459 })
460 ),
461 (queue) => Effect.tx(TxPubSub.releaseSubscriber(self.pubsub, queue))
462 )
463
464/**
465 * Returns a Stream of all changes to the TxSubscriptionRef, starting with the
466 * current value followed by every subsequent update.
467 *
468 * **When to use**
469 *
470 * Use to consume `TxSubscriptionRef` committed changes as a `Stream`.
471 *
472 * **Example** (Streaming changes)
473 *
474 * ```ts import.meta.vitest
475 * import { Effect, Stream, TxSubscriptionRef } from "effect"
476 *
477 * const program = Effect.gen(function*() {
478 * const ref = yield* TxSubscriptionRef.make(0)
479 * yield* TxSubscriptionRef.set(ref, 1)
480 * yield* TxSubscriptionRef.set(ref, 2)
481 *
482 * const values = yield* Stream.runCollect(
483 * TxSubscriptionRef.changesStream(ref).pipe(Stream.take(1))
484 * )
485 * return Array.from(values)
486 * })
487 *
488 * await Effect.runPromise(program) // => [2]
489 * ```
490 *
491 * @see {@link changes} for subscribing through a transactional queue
492 *
493 * @category subscriptions
494 * @since 3.10.0
495 */
496export const changesStream = <A>(self: TxSubscriptionRef<A>): Stream.Stream<A, never, never> =>
497 Stream.unwrap(
498 Effect.map(
499 changes(self),
500 (sub) => Stream.fromEffectRepeat(Effect.tx(TxQueue.take(sub)))
501 )
502 )
503
504// =============================================================================
505// Guards
506// =============================================================================
507
508/**
509 * Checks whether the given value is a TxSubscriptionRef.
510 *
511 * **When to use**
512 *
513 * Use to narrow an unknown value before treating it as a `TxSubscriptionRef`.
514 *
515 * **Example** (Checking transactional subscription references)
516 *
517 * ```ts import.meta.vitest
518 * import { TxSubscriptionRef } from "effect"
519 *
520 * const someValue: unknown = {}
521 * TxSubscriptionRef.isTxSubscriptionRef(someValue) // => false
522 * ```
523 *
524 * @see {@link make} for creating a `TxSubscriptionRef`
525 *
526 * @category guards
527 * @since 4.0.0
528 */
529export const isTxSubscriptionRef = (u: unknown): u is TxSubscriptionRef<unknown> => hasProperty(u, TypeId)