1/**
2 * Broadcasts values to subscribers inside Effect transactions.
3 *
4 * A `TxPubSub<A>` is a transactional publish/subscribe hub. Each subscriber
5 * owns a `TxQueue`, and each published value is offered to the subscriber
6 * queues that are registered at the time of publication. This module includes
7 * bounded, dropping, sliding, and unbounded hubs, publishing helpers, scoped
8 * subscriptions, shutdown operations, and a guard.
9 *
10 * @since 4.0.0
11 */
12import * as Arr from "./Array.ts"
13import * as Effect from "./Effect.ts"
14import { dual } from "./Function.ts"
15import type { Inspectable } from "./Inspectable.ts"
16import { NodeInspectSymbol, toJson } from "./Inspectable.ts"
17import type { Pipeable } from "./Pipeable.ts"
18import { pipeArguments } from "./Pipeable.ts"
19import { hasProperty } from "./Predicate.ts"
20import type * as Scope from "./Scope.ts"
21import * as TxQueue from "./TxQueue.ts"
22import * as TxRef from "./TxRef.ts"
23
24const TypeId = "~effect/TxPubSub"
25
26/**
27 * A TxPubSub represents a transactional publish/subscribe hub that broadcasts messages
28 * to all current subscribers using Software Transactional Memory (STM) semantics.
29 *
30 * **Example** (Subscribing to a transactional pub/sub)
31 *
32 * ```ts import.meta.vitest
33 * import { Effect, TxPubSub, TxQueue } from "effect"
34 *
35 * const program = Effect.gen(function*() {
36 * const hub = yield* TxPubSub.unbounded<string>()
37 *
38 * return yield* Effect.scoped(
39 * Effect.gen(function*() {
40 * const sub = yield* TxPubSub.subscribe(hub)
41 * yield* TxPubSub.publish(hub, "hello")
42 * return yield* TxQueue.take(sub)
43 * })
44 * )
45 * })
46 *
47 * await Effect.runPromise(program) // => "hello"
48 * ```
49 *
50 * @category models
51 * @since 4.0.0
52 */
53export interface TxPubSub<in out A> extends Inspectable, Pipeable {
54 readonly [TypeId]: typeof TypeId
55 /** @internal */
56 readonly subscribersRef: TxRef.TxRef<Array<TxQueue.TxQueue<A>>>
57 /** @internal */
58 readonly shutdownRef: TxRef.TxRef<boolean>
59 readonly strategy: "bounded" | "unbounded" | "dropping" | "sliding"
60 readonly capacity: number
61}
62
63const TxPubSubProto: Omit<TxPubSub<any>, typeof TypeId | "subscribersRef" | "shutdownRef" | "strategy" | "capacity"> = {
64 [NodeInspectSymbol](this: TxPubSub<unknown>) {
65 return toJson(this)
66 },
67 toJSON(this: TxPubSub<unknown>) {
68 return {
69 _id: "TxPubSub",
70 strategy: this.strategy,
71 capacity: this.capacity
72 }
73 },
74 toString(this: TxPubSub<unknown>) {
75 return `TxPubSub(${this.strategy}, ${this.capacity})`
76 },
77 pipe() {
78 return pipeArguments(this, arguments)
79 }
80}
81
82const makeTxPubSub = <A>(
83 subscribersRef: TxRef.TxRef<Array<TxQueue.TxQueue<A>>>,
84 shutdownRef: TxRef.TxRef<boolean>,
85 strategy: "bounded" | "unbounded" | "dropping" | "sliding",
86 cap: number
87): TxPubSub<A> => {
88 const self = Object.create(TxPubSubProto)
89 self[TypeId] = TypeId
90 self.subscribersRef = subscribersRef
91 self.shutdownRef = shutdownRef
92 self.strategy = strategy
93 self.capacity = cap
94 return self
95}
96
97// =============================================================================
98// Constructors
99// =============================================================================
100
101/**
102 * Creates a bounded TxPubSub with the specified capacity. When a subscriber's
103 * queue is full, the publisher will retry the transaction until space is available.
104 *
105 * **Example** (Creating a bounded pub/sub)
106 *
107 * ```ts import.meta.vitest
108 * import { Effect, TxPubSub, TxQueue } from "effect"
109 *
110 * const program = Effect.gen(function*() {
111 * const hub = yield* TxPubSub.bounded<number>(16)
112 *
113 * return yield* Effect.scoped(
114 * Effect.gen(function*() {
115 * const sub = yield* TxPubSub.subscribe(hub)
116 * yield* TxPubSub.publish(hub, 42)
117 * return yield* TxQueue.take(sub)
118 * })
119 * )
120 * })
121 *
122 * await Effect.runPromise(program) // => 42
123 * ```
124 *
125 * @category constructors
126 * @since 2.0.0
127 */
128export const bounded = <A = never>(capacity: number): Effect.Effect<TxPubSub<A>> =>
129 Effect.gen(function*() {
130 const subscribersRef = yield* TxRef.make<Array<TxQueue.TxQueue<A>>>([])
131 const shutdownRef = yield* TxRef.make(false)
132 return makeTxPubSub(subscribersRef, shutdownRef, "bounded", capacity)
133 }).pipe(Effect.tx)
134
135/**
136 * Creates a dropping TxPubSub with the specified capacity. When a subscriber's
137 * queue is full, the message is dropped for that subscriber.
138 *
139 * **Example** (Creating a dropping pub/sub)
140 *
141 * ```ts import.meta.vitest
142 * import { Effect, TxPubSub, TxQueue } from "effect"
143 *
144 * const program = Effect.gen(function*() {
145 * const hub = yield* TxPubSub.dropping<number>(2)
146 *
147 * return yield* Effect.scoped(
148 * Effect.gen(function*() {
149 * const sub = yield* TxPubSub.subscribe(hub)
150 * yield* TxPubSub.publish(hub, 1)
151 * yield* TxPubSub.publish(hub, 2)
152 * yield* TxPubSub.publish(hub, 3) // dropped
153 * const v1 = yield* TxQueue.take(sub)
154 * const v2 = yield* TxQueue.take(sub)
155 * return [v1, v2]
156 * })
157 * )
158 * })
159 *
160 * await Effect.runPromise(program) // => [1, 2]
161 * ```
162 *
163 * @category constructors
164 * @since 2.0.0
165 */
166export const dropping = <A = never>(capacity: number): Effect.Effect<TxPubSub<A>> =>
167 Effect.gen(function*() {
168 const subscribersRef = yield* TxRef.make<Array<TxQueue.TxQueue<A>>>([])
169 const shutdownRef = yield* TxRef.make(false)
170 return makeTxPubSub(subscribersRef, shutdownRef, "dropping", capacity)
171 }).pipe(Effect.tx)
172
173/**
174 * Creates a sliding TxPubSub with the specified capacity. When a subscriber's
175 * queue is full, the oldest message in that subscriber's queue is dropped.
176 *
177 * **Example** (Creating a sliding pub/sub)
178 *
179 * ```ts import.meta.vitest
180 * import { Effect, TxPubSub, TxQueue } from "effect"
181 *
182 * const program = Effect.gen(function*() {
183 * const hub = yield* TxPubSub.sliding<number>(2)
184 *
185 * return yield* Effect.scoped(
186 * Effect.gen(function*() {
187 * const sub = yield* TxPubSub.subscribe(hub)
188 * yield* TxPubSub.publish(hub, 1)
189 * yield* TxPubSub.publish(hub, 2)
190 * yield* TxPubSub.publish(hub, 3) // evicts 1
191 * return yield* TxQueue.take(sub)
192 * })
193 * )
194 * })
195 *
196 * await Effect.runPromise(program) // => 2
197 * ```
198 *
199 * @category constructors
200 * @since 2.0.0
201 */
202export const sliding = <A = never>(capacity: number): Effect.Effect<TxPubSub<A>> =>
203 Effect.gen(function*() {
204 const subscribersRef = yield* TxRef.make<Array<TxQueue.TxQueue<A>>>([])
205 const shutdownRef = yield* TxRef.make(false)
206 return makeTxPubSub(subscribersRef, shutdownRef, "sliding", capacity)
207 }).pipe(Effect.tx)
208
209/**
210 * Creates an unbounded TxPubSub with unlimited capacity. Messages are always accepted.
211 *
212 * **Example** (Creating an unbounded pub/sub)
213 *
214 * ```ts import.meta.vitest
215 * import { Effect, TxPubSub, TxQueue } from "effect"
216 *
217 * const program = Effect.gen(function*() {
218 * const hub = yield* TxPubSub.unbounded<string>()
219 *
220 * return yield* Effect.scoped(
221 * Effect.gen(function*() {
222 * const sub = yield* TxPubSub.subscribe(hub)
223 * yield* TxPubSub.publish(hub, "msg")
224 * return yield* TxQueue.take(sub)
225 * })
226 * )
227 * })
228 *
229 * await Effect.runPromise(program) // => "msg"
230 * ```
231 *
232 * @category constructors
233 * @since 2.0.0
234 */
235export const unbounded = <A = never>(): Effect.Effect<TxPubSub<A>> =>
236 Effect.gen(function*() {
237 const subscribersRef = yield* TxRef.make<Array<TxQueue.TxQueue<A>>>([])
238 const shutdownRef = yield* TxRef.make(false)
239 return makeTxPubSub(subscribersRef, shutdownRef, "unbounded", Number.POSITIVE_INFINITY)
240 }).pipe(Effect.tx)
241
242// =============================================================================
243// Getters
244// =============================================================================
245
246/**
247 * Returns the capacity of the TxPubSub.
248 *
249 * **Example** (Reading pub/sub capacity)
250 *
251 * ```ts import.meta.vitest
252 * import { Effect, TxPubSub } from "effect"
253 *
254 * const program = Effect.gen(function*() {
255 * const hub = yield* TxPubSub.bounded<number>(16)
256 * return TxPubSub.capacity(hub)
257 * })
258 *
259 * await Effect.runPromise(program) // => 16
260 * ```
261 *
262 * @category getters
263 * @since 2.0.0
264 */
265export const capacity = <A>(self: TxPubSub<A>): number => self.capacity
266
267/**
268 * Returns the current number of messages across all subscriber queues (the max).
269 *
270 * **Example** (Reading subscriber queue size)
271 *
272 * ```ts import.meta.vitest
273 * import { Effect, TxPubSub, TxQueue } from "effect"
274 *
275 * const program = Effect.gen(function*() {
276 * const hub = yield* TxPubSub.unbounded<number>()
277 *
278 * return yield* Effect.scoped(
279 * Effect.gen(function*() {
280 * const sub = yield* TxPubSub.subscribe(hub)
281 * yield* TxPubSub.publish(hub, 1)
282 * yield* TxPubSub.publish(hub, 2)
283 * return yield* TxPubSub.size(hub)
284 * })
285 * )
286 * })
287 *
288 * await Effect.runPromise(program) // => 2
289 * ```
290 *
291 * @category getters
292 * @since 2.0.0
293 */
294export const size = <A>(self: TxPubSub<A>): Effect.Effect<number> =>
295 Effect.gen(function*() {
296 const subscribers = yield* TxRef.get(self.subscribersRef)
297 let maxSize = 0
298 for (const queue of subscribers) {
299 const s = yield* TxQueue.size(queue)
300 if (s > maxSize) maxSize = s
301 }
302 return maxSize
303 }).pipe(Effect.tx)
304
305/**
306 * Checks whether the TxPubSub has no pending messages (all subscriber queues are empty).
307 *
308 * **Example** (Checking whether a pub/sub is empty)
309 *
310 * ```ts import.meta.vitest
311 * import { Effect, TxPubSub } from "effect"
312 *
313 * const program = Effect.gen(function*() {
314 * const hub = yield* TxPubSub.unbounded<number>()
315 * return yield* TxPubSub.isEmpty(hub)
316 * })
317 *
318 * await Effect.runPromise(program) // => true
319 * ```
320 *
321 * @category predicates
322 * @since 2.0.0
323 */
324export const isEmpty = <A>(self: TxPubSub<A>): Effect.Effect<boolean> => Effect.map(size(self), (s) => s === 0)
325
326/**
327 * Checks whether any subscriber queue has pending messages.
328 * A pub/sub with no subscribers is empty, even if values have been published.
329 *
330 * **Example** (Checking whether a pub/sub is non-empty)
331 *
332 * ```ts import.meta.vitest
333 * import { Effect, TxPubSub } from "effect"
334 *
335 * const program = Effect.gen(function*() {
336 * const hub = yield* TxPubSub.unbounded<number>()
337 * const empty = yield* TxPubSub.isNonEmpty(hub)
338 *
339 * const nonEmpty = yield* Effect.scoped(Effect.gen(function*() {
340 * yield* TxPubSub.subscribe(hub)
341 * yield* TxPubSub.publish(hub, 1)
342 * return yield* TxPubSub.isNonEmpty(hub)
343 * }))
344 * return [empty, nonEmpty] as const
345 * })
346 *
347 * await Effect.runPromise(program) // => [false, true]
348 * ```
349 *
350 * @category predicates
351 * @since 4.0.0
352 */
353export const isNonEmpty = <A>(self: TxPubSub<A>): Effect.Effect<boolean> => Effect.map(size(self), (s) => s > 0)
354
355/**
356 * Checks whether any subscriber queue is at capacity.
357 *
358 * **Example** (Checking whether a pub/sub is full)
359 *
360 * ```ts import.meta.vitest
361 * import { Effect, TxPubSub } from "effect"
362 *
363 * const program = Effect.gen(function*() {
364 * const hub = yield* TxPubSub.bounded<number>(2)
365 * return yield* TxPubSub.isFull(hub)
366 * })
367 *
368 * await Effect.runPromise(program) // => false
369 * ```
370 *
371 * @category predicates
372 * @since 2.0.0
373 */
374export const isFull = <A>(self: TxPubSub<A>): Effect.Effect<boolean> =>
375 Effect.gen(function*() {
376 if (self.capacity === Number.POSITIVE_INFINITY) return false
377 const subscribers = yield* TxRef.get(self.subscribersRef)
378 for (const queue of subscribers) {
379 if (yield* TxQueue.isFull(queue)) return true
380 }
381 return false
382 }).pipe(Effect.tx)
383
384/**
385 * Checks whether the TxPubSub has been shut down.
386 *
387 * **Example** (Checking whether a pub/sub is shut down)
388 *
389 * ```ts import.meta.vitest
390 * import { Effect, TxPubSub } from "effect"
391 *
392 * const program = Effect.gen(function*() {
393 * const hub = yield* TxPubSub.unbounded<number>()
394 * const before = yield* TxPubSub.isShutdown(hub)
395 * yield* TxPubSub.shutdown(hub)
396 * return [before, yield* TxPubSub.isShutdown(hub)]
397 * })
398 *
399 * await Effect.runPromise(program) // => [false, true]
400 * ```
401 *
402 * @category predicates
403 * @since 2.0.0
404 */
405export const isShutdown = <A>(self: TxPubSub<A>): Effect.Effect<boolean> => TxRef.get(self.shutdownRef)
406
407// =============================================================================
408// Mutations
409// =============================================================================
410
411/**
412 * Publishes a message to all current subscribers.
413 *
414 * **Details**
415 *
416 * Returns `true` if the message was delivered to all subscribers, or `false` if the hub is shut down or the message was dropped for any subscriber. For the bounded strategy, the transaction retries if any subscriber queue is full. For the sliding strategy, full subscriber queues drop their oldest messages. For the dropping strategy, full subscriber queues drop the new message and the operation returns `false`.
417 *
418 * **Example** (Publishing a message to subscribers)
419 *
420 * ```ts import.meta.vitest
421 * import { Effect, TxPubSub, TxQueue } from "effect"
422 *
423 * const program = Effect.gen(function*() {
424 * const hub = yield* TxPubSub.unbounded<string>()
425 *
426 * // No subscribers - publish is a no-op
427 * const r1 = yield* TxPubSub.publish(hub, "no one listening")
428 *
429 * const msg = yield* Effect.scoped(
430 * Effect.gen(function*() {
431 * const sub = yield* TxPubSub.subscribe(hub)
432 * yield* TxPubSub.publish(hub, "hello")
433 * return yield* TxQueue.take(sub)
434 * })
435 * )
436 * return [r1, msg]
437 * })
438 *
439 * await Effect.runPromise(program) // => [true, "hello"]
440 * ```
441 *
442 * @category mutations
443 * @since 2.0.0
444 */
445export const publish: {
446 <A>(value: A): (self: TxPubSub<A>) => Effect.Effect<boolean>
447 <A>(self: TxPubSub<A>, value: A): Effect.Effect<boolean>
448} = dual(
449 2,
450 <A>(self: TxPubSub<A>, value: A): Effect.Effect<boolean> =>
451 Effect.gen(function*() {
452 if (yield* TxRef.get(self.shutdownRef)) return false
453
454 const subscribers = yield* TxRef.get(self.subscribersRef)
455 let allAccepted = true
456
457 for (const queue of subscribers) {
458 const accepted = yield* TxQueue.offer(queue, value)
459 if (!accepted) allAccepted = false
460 }
461
462 return allAccepted
463 }).pipe(Effect.tx)
464)
465
466/**
467 * Publishes all messages from an iterable to all current subscribers.
468 *
469 * **Details**
470 *
471 * Returns `true` if all messages were delivered to all subscribers.
472 *
473 * **Example** (Publishing multiple messages to subscribers)
474 *
475 * ```ts import.meta.vitest
476 * import { Effect, TxPubSub, TxQueue } from "effect"
477 *
478 * const program = Effect.gen(function*() {
479 * const hub = yield* TxPubSub.unbounded<number>()
480 *
481 * return yield* Effect.scoped(
482 * Effect.gen(function*() {
483 * const sub = yield* TxPubSub.subscribe(hub)
484 * yield* TxPubSub.publishAll(hub, [1, 2, 3])
485 * const v1 = yield* TxQueue.take(sub)
486 * const v2 = yield* TxQueue.take(sub)
487 * const v3 = yield* TxQueue.take(sub)
488 * return [v1, v2, v3]
489 * })
490 * )
491 * })
492 *
493 * await Effect.runPromise(program) // => [1, 2, 3]
494 * ```
495 *
496 * @category mutations
497 * @since 2.0.0
498 */
499export const publishAll: {
500 <A>(values: Iterable<A>): (self: TxPubSub<A>) => Effect.Effect<boolean>
501 <A>(self: TxPubSub<A>, values: Iterable<A>): Effect.Effect<boolean>
502} = dual(
503 2,
504 <A>(self: TxPubSub<A>, values: Iterable<A>): Effect.Effect<boolean> => {
505 const valuesArray = Arr.fromIterable(values)
506 return Effect.gen(function*() {
507 if (yield* TxRef.get(self.shutdownRef)) return false
508
509 let allAccepted = true
510 for (const value of valuesArray) {
511 const accepted = yield* publish(self, value)
512 if (!accepted) allAccepted = false
513 }
514 return allAccepted
515 }).pipe(Effect.tx)
516 }
517)
518
519/**
520 * Subscribes to the TxPubSub, returning a scoped `TxQueue` for messages published after subscription.
521 *
522 * **Details**
523 *
524 * The returned queue uses the hub's capacity strategy: bounded subscriptions backpressure publishers when full, dropping subscriptions may miss new messages when full, and sliding subscriptions may evict older queued messages. The subscription is automatically removed when the scope is closed.
525 *
526 * **Example** (Subscribing multiple queues)
527 *
528 * ```ts import.meta.vitest
529 * import { Effect, TxPubSub, TxQueue } from "effect"
530 *
531 * const program = Effect.gen(function*() {
532 * const hub = yield* TxPubSub.unbounded<string>()
533 *
534 * return yield* Effect.scoped(
535 * Effect.gen(function*() {
536 * const sub1 = yield* TxPubSub.subscribe(hub)
537 * const sub2 = yield* TxPubSub.subscribe(hub)
538 *
539 * yield* TxPubSub.publish(hub, "broadcast")
540 *
541 * const msg1 = yield* TxQueue.take(sub1)
542 * const msg2 = yield* TxQueue.take(sub2)
543 * return [msg1, msg2]
544 * })
545 * )
546 * })
547 *
548 * await Effect.runPromise(program) // => ["broadcast", "broadcast"]
549 * ```
550 *
551 * @category mutations
552 * @since 2.0.0
553 */
554export const subscribe = <A>(self: TxPubSub<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope> =>
555 Effect.acquireRelease(
556 Effect.tx(acquireSubscriber(self)),
557 (queue) => Effect.tx(releaseSubscriber(self, queue))
558 )
559
560/**
561 * Creates a subscriber queue and registers it with the pub/sub.
562 *
563 * **When to use**
564 *
565 * Use to create and register a subscriber queue inside a larger transaction
566 * when registration must be atomic with other Tx operations.
567 *
568 * **Details**
569 *
570 * This is the transactional acquire step of `subscribe`, exposed so that callers can compose it with other Tx operations in a single transaction, such as `TxSubscriptionRef.changes`.
571 *
572 * @see {@link subscribe} for the scoped acquire and release wrapper when no custom transaction composition is needed
573 * @see {@link releaseSubscriber} to remove and shut down a queue returned by `acquireSubscriber`
574 *
575 * @category mutations
576 * @since 4.0.0
577 */
578export const acquireSubscriber = <A>(
579 self: TxPubSub<A>
580): Effect.Effect<TxQueue.TxQueue<A>, never, Effect.Transaction> =>
581 Effect.gen(function*() {
582 const queue = yield* makeSubscriberQueue<A>(self.strategy, self.capacity)
583 yield* TxRef.update(self.subscribersRef, (subs) => [...subs, queue])
584 return queue
585 })
586
587/**
588 * Removes a subscriber queue from the pub/sub and shuts it down.
589 *
590 * **When to use**
591 *
592 * Use to release a manually acquired subscriber queue inside a larger
593 * transaction, removing it from the pub/sub and shutting it down together with
594 * related transactional cleanup.
595 *
596 * **Details**
597 *
598 * This is the transactional release step of `subscribe`, exposed so that callers can compose it with other Tx operations in a single transaction.
599 *
600 * **Gotchas**
601 *
602 * The supplied queue is shut down after being removed, so callers should pass a
603 * queue acquired for this pub/sub.
604 *
605 * @see {@link acquireSubscriber} for the matching transactional acquire step
606 * @see {@link subscribe} for the scoped acquire and release wrapper
607 *
608 * @category mutations
609 * @since 4.0.0
610 */
611export const releaseSubscriber: {
612 <A>(queue: TxQueue.TxQueue<A>): (self: TxPubSub<A>) => Effect.Effect<void, never, Effect.Transaction>
613 <A>(self: TxPubSub<A>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, never, Effect.Transaction>
614} = dual(
615 2,
616 <A>(
617 self: TxPubSub<A>,
618 queue: TxQueue.TxQueue<A>
619 ): Effect.Effect<void, never, Effect.Transaction> =>
620 Effect.gen(function*() {
621 yield* TxRef.update(self.subscribersRef, (subs) => subs.filter((q) => q !== queue))
622 yield* TxQueue.shutdown(queue)
623 })
624)
625
626const makeSubscriberQueue = <A>(
627 strategy: "bounded" | "unbounded" | "dropping" | "sliding",
628 cap: number
629): Effect.Effect<TxQueue.TxQueue<A>> => {
630 switch (strategy) {
631 case "bounded":
632 return TxQueue.bounded<A>(cap)
633 case "dropping":
634 return TxQueue.dropping<A>(cap)
635 case "sliding":
636 return TxQueue.sliding<A>(cap)
637 case "unbounded":
638 return TxQueue.unbounded<A>()
639 }
640}
641
642/**
643 * Shuts down the TxPubSub and all subscriber queues registered at the time of shutdown.
644 *
645 * **Details**
646 *
647 * After shutdown, `publish` and `publishAll` return `false`, and `awaitShutdown` completes. The operation is idempotent.
648 *
649 * **Gotchas**
650 *
651 * Subscribers acquired after shutdown are not automatically shut down by this call.
652 *
653 * **Example** (Shutting down a pub/sub)
654 *
655 * ```ts import.meta.vitest
656 * import { Effect, TxPubSub } from "effect"
657 *
658 * const program = Effect.gen(function*() {
659 * const hub = yield* TxPubSub.unbounded<number>()
660 * yield* TxPubSub.shutdown(hub)
661 *
662 * const shut = yield* TxPubSub.isShutdown(hub)
663 * const accepted = yield* TxPubSub.publish(hub, 1)
664 * return [shut, accepted]
665 * })
666 *
667 * await Effect.runPromise(program) // => [true, false]
668 * ```
669 *
670 * @category mutations
671 * @since 2.0.0
672 */
673export const shutdown = <A>(self: TxPubSub<A>): Effect.Effect<void> =>
674 Effect.gen(function*() {
675 const alreadyShutdown = yield* TxRef.get(self.shutdownRef)
676 if (alreadyShutdown) return
677
678 yield* TxRef.set(self.shutdownRef, true)
679 const subscribers = yield* TxRef.get(self.subscribersRef)
680 for (const queue of subscribers) {
681 yield* TxQueue.shutdown(queue)
682 }
683 }).pipe(Effect.tx)
684
685/**
686 * Waits for the TxPubSub to be shut down.
687 *
688 * **Example** (Waiting for shutdown)
689 *
690 * ```ts import.meta.vitest
691 * import { Effect, Fiber, TxPubSub } from "effect"
692 *
693 * const program = Effect.gen(function*() {
694 * const hub = yield* TxPubSub.unbounded<number>()
695 *
696 * const fiber = yield* Effect.forkChild(TxPubSub.awaitShutdown(hub))
697 * yield* TxPubSub.shutdown(hub)
698 * yield* Fiber.await(fiber)
699 * return yield* TxPubSub.isShutdown(hub)
700 * })
701 *
702 * await Effect.runPromise(program) // => true
703 * ```
704 *
705 * @category mutations
706 * @since 2.0.0
707 */
708export const awaitShutdown = <A>(self: TxPubSub<A>): Effect.Effect<void> =>
709 Effect.gen(function*() {
710 const shut = yield* TxRef.get(self.shutdownRef)
711 if (shut) return
712 return yield* Effect.txRetry
713 }).pipe(Effect.tx)
714
715// =============================================================================
716// Guards
717// =============================================================================
718
719/**
720 * Checks whether the given value is a TxPubSub.
721 *
722 * **Example** (Checking for a TxPubSub)
723 *
724 * ```ts import.meta.vitest
725 * import { TxPubSub } from "effect"
726 *
727 * const someValue: unknown = {}
728 * TxPubSub.isTxPubSub(someValue) // => false
729 * ```
730 *
731 * @category guards
732 * @since 4.0.0
733 */
734export const isTxPubSub = (u: unknown): u is TxPubSub<unknown> => hasProperty(u, TypeId)