1/**
2 * Broadcasts values from publishers to many subscribers.
3 *
4 * Publishers add messages with `publish` or `publishAll`, and each active
5 * `Subscription` receives its own copy of every accepted message. Unlike a
6 * queue, subscribers do not compete for messages. This module includes bounded,
7 * dropping, sliding, and unbounded hubs, optional replay buffers for late
8 * subscribers, message-taking helpers, capacity and shutdown operations, a
9 * type guard, and low-level types for custom hub strategies.
10 *
11 * @since 2.0.0
12 */
13import * as Arr from "./Array.ts"
14import * as Context from "./Context.ts"
15import * as Deferred from "./Deferred.ts"
16import * as Effect from "./Effect.ts"
17import * as Exit from "./Exit.ts"
18import type { LazyArg } from "./Function.ts"
19import { dual, identity } from "./Function.ts"
20import * as Count from "./internal/count.ts"
21import * as Latch from "./Latch.ts"
22import * as MutableList from "./MutableList.ts"
23import * as MutableRef from "./MutableRef.ts"
24import { nextPow2 } from "./Number.ts"
25import * as Option from "./Option.ts"
26import { type Pipeable, pipeArguments } from "./Pipeable.ts"
27import { hasProperty } from "./Predicate.ts"
28import * as Scope from "./Scope.ts"
29import type { Covariant, Invariant } from "./Types.ts"
30
31const TypeId = "~effect/PubSub"
32
33/**
34 * A `PubSub<A>` is an asynchronous message hub into which publishers can publish
35 * messages of type `A` and subscribers can subscribe to take messages of type
36 * `A`.
37 *
38 * **Example** (Publishing and subscribing to messages)
39 *
40 * ```ts import.meta.vitest
41 * import { Effect, PubSub } from "effect"
42 *
43 * const program = Effect.scoped(Effect.gen(function*() {
44 * // Create a bounded PubSub with capacity 10
45 * const pubsub = yield* PubSub.bounded<string>(10)
46 *
47 * // Subscribe and consume messages
48 * const subscription = yield* PubSub.subscribe(pubsub)
49 *
50 * // Publish messages
51 * yield* PubSub.publish(pubsub, "Hello")
52 * yield* PubSub.publish(pubsub, "World")
53 *
54 * const message1 = yield* PubSub.take(subscription)
55 * const message2 = yield* PubSub.take(subscription)
56 * return [message1, message2]
57 * }))
58 *
59 * const actual = await Effect.runPromise(program)
60 * actual // => ["Hello", "World"]
61 * ```
62 *
63 * @category models
64 * @since 2.0.0
65 */
66export interface PubSub<in out A> extends Pipeable {
67 readonly [TypeId]: {
68 readonly _A: Invariant<A>
69 }
70 readonly pubsub: PubSub.Atomic<A>
71 readonly subscribers: PubSub.Subscribers<A>
72 readonly scope: Scope.Closeable
73 readonly shutdownHook: Latch.Latch
74 readonly shutdownFlag: MutableRef.MutableRef<boolean>
75 readonly strategy: PubSub.Strategy<A>
76 readonly ended: MutableRef.MutableRef<Option.Option<A>>
77}
78
79/**
80 * Companion namespace containing the low-level building blocks used by
81 * `PubSub`, including atomic implementations, backing subscriptions, replay
82 * windows, and delivery strategies.
83 *
84 * @since 2.0.0
85 */
86export declare namespace PubSub {
87 /**
88 * Low-level atomic PubSub interface that handles the core message storage and retrieval.
89 *
90 * @category models
91 * @since 4.0.0
92 */
93 export interface Atomic<in out A> {
94 readonly capacity: number
95 isEmpty(): boolean
96 isFull(): boolean
97 size(): number
98 publish(value: A): boolean
99 publishAll(elements: Iterable<A>): Array<A>
100 slide(): void
101 subscribe(): BackingSubscription<A>
102 replayWindow(): ReplayWindow<A>
103 }
104
105 /**
106 * Low-level subscription interface that handles message polling for individual subscribers.
107 *
108 * @category models
109 * @since 4.0.0
110 */
111 export interface BackingSubscription<out A> {
112 isEmpty(): boolean
113 size(): number
114 poll(): A | MutableList.Empty
115 pollUpTo(n: number): Array<A>
116 unsubscribe(): void
117 }
118
119 /**
120 * Tracks the pollers currently waiting on each backing subscription.
121 *
122 * **Details**
123 *
124 * This type is part of the low-level `PubSub.Strategy` contract. Most
125 * application code should use `subscribe`, `take`, and the other `PubSub`
126 * operations instead of manipulating subscriber maps directly.
127 *
128 * @category models
129 * @since 4.0.0
130 */
131 export type Subscribers<A> = Map<
132 BackingSubscription<A>,
133 Set<MutableList.MutableList<Deferred.Deferred<A>>>
134 >
135
136 /**
137 * Interface for accessing replay buffer contents for late subscribers.
138 *
139 * @category models
140 * @since 4.0.0
141 */
142 export interface ReplayWindow<A> {
143 take(): A | undefined
144 takeN(n: number): Array<A>
145 takeAll(): Array<A>
146 close(): void
147 readonly remaining: number
148 }
149
150 /**
151 * Strategy interface defining how PubSub handles backpressure and message distribution.
152 *
153 * @category models
154 * @since 4.0.0
155 */
156 export interface Strategy<in out A> {
157 /**
158 * Describes any finalization logic associated with this strategy.
159 */
160 readonly shutdown: Effect.Effect<void>
161
162 /**
163 * Describes how publishers should signal to subscribers that they are
164 * waiting for space to become available in the `PubSub`.
165 */
166 handleSurplus(
167 pubsub: Atomic<A>,
168 subscribers: Subscribers<A>,
169 elements: Iterable<A>,
170 isShutdown: MutableRef.MutableRef<boolean>
171 ): Effect.Effect<boolean>
172
173 /**
174 * Describes how subscribers should signal to publishers waiting for space
175 * to become available in the `PubSub` that space may be available.
176 */
177 onPubSubEmptySpaceUnsafe(
178 pubsub: Atomic<A>,
179 subscribers: Subscribers<A>
180 ): void
181
182 /**
183 * Describes how subscribers waiting for additional values from the `PubSub`
184 * should take those values and signal to publishers that they are no
185 * longer waiting for additional values.
186 */
187 completePollersUnsafe(
188 pubsub: Atomic<A>,
189 subscribers: Subscribers<A>,
190 subscription: BackingSubscription<A>,
191 pollers: MutableList.MutableList<Deferred.Deferred<A>>
192 ): void
193
194 /**
195 * Describes how publishers should signal to subscribers waiting for
196 * additional values from the `PubSub` that new values are available.
197 */
198 completeSubscribersUnsafe(
199 pubsub: Atomic<A>,
200 subscribers: Subscribers<A>
201 ): void
202 }
203}
204
205/**
206 * Returns `true` if a value is a `PubSub`.
207 *
208 * **Details**
209 *
210 * This is a type guard that checks for the `PubSub` runtime marker.
211 *
212 * **Example** (Checking if a value is a PubSub)
213 *
214 * ```ts import.meta.vitest
215 * import { Effect, PubSub } from "effect"
216 *
217 * const program = Effect.gen(function*() {
218 * const pubsub = yield* PubSub.bounded<string>(10)
219 * return [PubSub.isPubSub(pubsub), PubSub.isPubSub({}), PubSub.isPubSub(null)]
220 * })
221 *
222 * const actual = await Effect.runPromise(program)
223 * actual // => [true, false, false]
224 * ```
225 *
226 * @category guards
227 * @since 4.0.0
228 */
229export const isPubSub = <A = unknown>(u: unknown): u is PubSub<A> => hasProperty(u, TypeId)
230
231const SubscriptionTypeId = "~effect/PubSub/Subscription"
232
233/**
234 * A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
235 *
236 * **Example** (Taking messages from a subscription)
237 *
238 * ```ts import.meta.vitest
239 * import { Effect, PubSub } from "effect"
240 *
241 * const program = Effect.scoped(Effect.gen(function*() {
242 * const pubsub = yield* PubSub.bounded<string>(10)
243 *
244 * // Subscribe within a scope for automatic cleanup
245 * const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(pubsub)
246 *
247 * yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
248 *
249 * // Take individual messages
250 * const message = yield* PubSub.take(subscription)
251 *
252 * // Take multiple messages
253 * const messages = yield* PubSub.takeUpTo(subscription, 1)
254 * const allMessages = yield* PubSub.takeAll(subscription)
255 * return { message, messages, allMessages }
256 * }))
257 *
258 * const actual = await Effect.runPromise(program)
259 * actual // => { message: "msg1", messages: ["msg2"], allMessages: ["msg3"] }
260 * ```
261 *
262 * @category models
263 * @since 4.0.0
264 */
265export interface Subscription<out A> extends Pipeable {
266 readonly [SubscriptionTypeId]: {
267 readonly _A: Covariant<A>
268 }
269 readonly pubsub: PubSub.Atomic<any>
270 readonly subscribers: PubSub.Subscribers<any>
271 readonly subscription: PubSub.BackingSubscription<A>
272 readonly pollers: MutableList.MutableList<Deferred.Deferred<any>>
273 readonly shutdownHook: Latch.Latch
274 readonly shutdownFlag: MutableRef.MutableRef<boolean>
275 readonly strategy: PubSub.Strategy<any>
276 readonly replayWindow: PubSub.ReplayWindow<A>
277 readonly ended: MutableRef.MutableRef<Option.Option<any>>
278}
279
280/**
281 * Creates a PubSub with a custom atomic implementation and strategy.
282 *
283 * **Example** (Creating a PubSub with a custom strategy)
284 *
285 * ```ts import.meta.vitest
286 * import { Effect, PubSub } from "effect"
287 *
288 * const program = Effect.gen(function*() {
289 * // Create custom PubSub with specific atomic implementation and strategy
290 * const pubsub = yield* PubSub.make<string>({
291 * atomicPubSub: () => PubSub.makeAtomicBounded(100),
292 * strategy: () => new PubSub.BackPressureStrategy()
293 * })
294 *
295 * // Use the created PubSub
296 * const published = yield* PubSub.publish(pubsub, "Hello")
297 * yield* PubSub.shutdown(pubsub)
298 * return published
299 * })
300 *
301 * const actual = await Effect.runPromise(program)
302 * actual // => true
303 * ```
304 *
305 * @category constructors
306 * @since 4.0.0
307 */
308export const make = <A>(
309 options: {
310 readonly atomicPubSub: LazyArg<PubSub.Atomic<A>>
311 readonly strategy: LazyArg<PubSub.Strategy<A>>
312 }
313): Effect.Effect<PubSub<A>> =>
314 Effect.sync(() =>
315 makePubSubUnsafe(
316 options.atomicPubSub(),
317 new Map(),
318 Scope.makeUnsafe(),
319 Latch.makeUnsafe(false),
320 MutableRef.make(false),
321 options.strategy(),
322 MutableRef.make(Option.none())
323 )
324 )
325
326/**
327 * Creates a bounded `PubSub` that applies backpressure when it reaches
328 * capacity.
329 *
330 * **Details**
331 *
332 * Published messages are retained until all current subscribers have taken
333 * them. When the capacity is full, publishers suspend until space is available.
334 * Pass an options object to configure both `capacity` and an optional replay
335 * buffer for late subscribers.
336 *
337 * **Example** (Creating a bounded PubSub)
338 *
339 * ```ts import.meta.vitest
340 * import { Effect, PubSub } from "effect"
341 *
342 * const program = Effect.gen(function*() {
343 * // Create bounded PubSub with capacity 100
344 * const pubsub = yield* PubSub.bounded<string>(100)
345 *
346 * // Create with replay buffer for late subscribers
347 * const pubsubWithReplay = yield* PubSub.bounded<string>({
348 * capacity: 100,
349 * replay: 10 // Last 10 messages replayed to new subscribers
350 * })
351 *
352 * const capacities = [PubSub.capacity(pubsub), PubSub.capacity(pubsubWithReplay)]
353 * yield* PubSub.shutdown(pubsub)
354 * yield* PubSub.shutdown(pubsubWithReplay)
355 * return capacities
356 * })
357 *
358 * const actual = await Effect.runPromise(program)
359 * actual // => [100, 100]
360 * ```
361 *
362 * @category constructors
363 * @since 2.0.0
364 */
365export const bounded = <A>(
366 capacity: number | {
367 readonly capacity: number
368 readonly replay?: number | undefined
369 }
370): Effect.Effect<PubSub<A>> =>
371 make({
372 atomicPubSub: () => makeAtomicBounded(capacity),
373 strategy: () => new BackPressureStrategy()
374 })
375
376/**
377 * Creates a bounded `PubSub` with the dropping strategy. The `PubSub` will drop new
378 * messages if the `PubSub` is at capacity.
379 *
380 * **Details**
381 *
382 * For best performance use capacities that are powers of two.
383 *
384 * **Example** (Dropping messages when full)
385 *
386 * ```ts import.meta.vitest
387 * import { Effect, PubSub } from "effect"
388 *
389 * const program = Effect.scoped(Effect.gen(function*() {
390 * // Create dropping PubSub that drops new messages when full
391 * const pubsub = yield* PubSub.dropping<string>(3)
392 *
393 * const subscription = yield* PubSub.subscribe(pubsub)
394 *
395 * // Fill the PubSub and see dropping behavior
396 * yield* PubSub.publish(pubsub, "msg1") // succeeds
397 * yield* PubSub.publish(pubsub, "msg2") // succeeds
398 * yield* PubSub.publish(pubsub, "msg3") // succeeds
399 * const dropped = yield* PubSub.publish(pubsub, "msg4") // returns false (dropped)
400 *
401 * const messages = yield* PubSub.takeAll(subscription)
402 * return { dropped: !dropped, messages }
403 * }))
404 *
405 * const actual = await Effect.runPromise(program)
406 * actual // => { dropped: true, messages: ["msg1", "msg2", "msg3"] }
407 * ```
408 *
409 * @category constructors
410 * @since 2.0.0
411 */
412export const dropping = <A>(
413 capacity: number | {
414 readonly capacity: number
415 readonly replay?: number | undefined
416 }
417): Effect.Effect<PubSub<A>> =>
418 make({
419 atomicPubSub: () => makeAtomicBounded(capacity),
420 strategy: () => new DroppingStrategy()
421 })
422
423/**
424 * Creates a bounded `PubSub` with the sliding strategy. The `PubSub` will add new
425 * messages and drop old messages if the `PubSub` is at capacity.
426 *
427 * **Details**
428 *
429 * For best performance use capacities that are powers of two.
430 *
431 * **Example** (Sliding old messages when full)
432 *
433 * ```ts import.meta.vitest
434 * import { Effect, PubSub } from "effect"
435 *
436 * const program = Effect.scoped(Effect.gen(function*() {
437 * // Create sliding PubSub that evicts old messages when full
438 * const pubsub = yield* PubSub.sliding<string>(3)
439 *
440 * const subscription = yield* PubSub.subscribe(pubsub)
441 *
442 * // Fill and overflow the PubSub
443 * yield* PubSub.publish(pubsub, "msg1")
444 * yield* PubSub.publish(pubsub, "msg2")
445 * yield* PubSub.publish(pubsub, "msg3")
446 * yield* PubSub.publish(pubsub, "msg4") // "msg1" is evicted
447 *
448 * return yield* PubSub.takeAll(subscription)
449 * }))
450 *
451 * const actual = await Effect.runPromise(program)
452 * actual // => ["msg2", "msg3", "msg4"]
453 * ```
454 *
455 * @category constructors
456 * @since 2.0.0
457 */
458export const sliding = <A>(
459 capacity: number | {
460 readonly capacity: number
461 readonly replay?: number | undefined
462 }
463): Effect.Effect<PubSub<A>> =>
464 make({
465 atomicPubSub: () => makeAtomicBounded(capacity),
466 strategy: () => new SlidingStrategy()
467 })
468
469/**
470 * Creates an unbounded `PubSub`.
471 *
472 * **Example** (Creating an unbounded PubSub)
473 *
474 * ```ts import.meta.vitest
475 * import { Effect, PubSub } from "effect"
476 *
477 * const program = Effect.scoped(Effect.gen(function*() {
478 * // Create unbounded PubSub
479 * const pubsub = yield* PubSub.unbounded<string>()
480 *
481 * const subscription = yield* PubSub.subscribe(pubsub)
482 *
483 * // Can publish unlimited messages
484 * for (let i = 0; i < 3; i++) {
485 * yield* PubSub.publish(pubsub, `message-${i}`)
486 * }
487 *
488 * return yield* PubSub.takeAll(subscription)
489 * }))
490 *
491 * const actual = await Effect.runPromise(program)
492 * actual // => ["message-0", "message-1", "message-2"]
493 * ```
494 *
495 * @category constructors
496 * @since 2.0.0
497 */
498export const unbounded = <A>(options?: {
499 readonly replay?: number | undefined
500}): Effect.Effect<PubSub<A>> =>
501 make({
502 atomicPubSub: () => makeAtomicUnbounded(options),
503 strategy: () => new DroppingStrategy()
504 })
505
506/**
507 * Creates a bounded atomic PubSub implementation with optional replay buffer.
508 *
509 * **When to use**
510 *
511 * Use to provide bounded message storage when building a custom `PubSub` with
512 * `make` and an explicit delivery strategy.
513 *
514 * **Details**
515 *
516 * Pass either a capacity number or an options object with `capacity` and
517 * optional `replay`. A positive `replay` value enables a replay buffer for late
518 * subscribers, and fractional replay sizes are rounded up.
519 *
520 * **Gotchas**
521 *
522 * The capacity must be greater than zero; invalid capacities throw
523 * synchronously before an atomic implementation is created.
524 *
525 * @see {@link make} for constructing a `PubSub` from an atomic implementation and delivery strategy
526 * @see {@link makeAtomicUnbounded} for an atomic implementation without a bounded capacity
527 * @see {@link bounded} for the higher-level backpressure constructor
528 * @see {@link dropping} for the higher-level dropping constructor
529 * @see {@link sliding} for the higher-level sliding constructor
530 *
531 * @category constructors
532 * @since 4.0.0
533 */
534export const makeAtomicBounded = <A>(
535 capacity: number | {
536 readonly capacity: number
537 readonly replay?: number | undefined
538 }
539): PubSub.Atomic<A> => {
540 const options = typeof capacity === "number" ? { capacity } : capacity
541 ensureCapacity(options.capacity)
542 if (options.capacity === Infinity) {
543 return makeAtomicUnbounded(options)
544 }
545 const replayBuffer = options.replay && options.replay > 0 ? new ReplayBuffer<A>(Math.ceil(options.replay)) : undefined
546 if (options.capacity === 1) {
547 return new BoundedPubSubSingle(replayBuffer)
548 } else if (nextPow2(options.capacity) === options.capacity) {
549 return new BoundedPubSubPow2(options.capacity, replayBuffer)
550 } else {
551 return new BoundedPubSubArb(options.capacity, replayBuffer)
552 }
553}
554
555/**
556 * Creates an unbounded atomic PubSub implementation with optional replay buffer.
557 *
558 * **When to use**
559 *
560 * Use to create the low-level storage layer for a custom `PubSub` whose active
561 * subscribers may retain an unbounded number of pending messages.
562 *
563 * **Gotchas**
564 *
565 * Messages published while subscribers are active can be retained without a
566 * capacity limit until those subscribers take them or unsubscribe.
567 *
568 * @see {@link makeAtomicBounded} for a bounded atomic implementation that enforces capacity
569 * @see {@link make} for wrapping an atomic implementation with a delivery strategy
570 * @see {@link unbounded} for the high-level effectful constructor for unbounded `PubSub` values
571 *
572 * @category constructors
573 * @since 4.0.0
574 */
575export const makeAtomicUnbounded = <A>(options?: {
576 readonly replay?: number | undefined
577}): PubSub.Atomic<A> => {
578 const replay = options?.replay
579 return new UnboundedPubSub(
580 replay && replay > 0 ? new ReplayBuffer<A>(Math.ceil(replay)) : undefined
581 )
582}
583
584/**
585 * Returns the number of elements the queue can hold.
586 *
587 * **Example** (Getting PubSub capacity)
588 *
589 * ```ts import.meta.vitest
590 * import { Effect, PubSub } from "effect"
591 *
592 * const program = Effect.gen(function*() {
593 * const pubsub = yield* PubSub.bounded<string>(100)
594 * const unboundedPubsub = yield* PubSub.unbounded<string>()
595 * return [PubSub.capacity(pubsub), PubSub.capacity(unboundedPubsub)]
596 * })
597 *
598 * const actual = await Effect.runPromise(program)
599 * actual // => [100, Number.MAX_SAFE_INTEGER]
600 * ```
601 *
602 * @category getters
603 * @since 2.0.0
604 */
605export const capacity = <A>(self: PubSub<A>): number => self.pubsub.capacity
606
607/**
608 * Returns the current number of messages retained by the `PubSub` for active
609 * subscribers.
610 *
611 * **Details**
612 *
613 * If the `PubSub` has been shut down, the returned effect succeeds with `0`.
614 * The size is not a count of waiting subscribers or suspended publishers.
615 *
616 * **Example** (Getting PubSub size)
617 *
618 * ```ts import.meta.vitest
619 * import { Effect, PubSub } from "effect"
620 *
621 * const program = Effect.scoped(Effect.gen(function*() {
622 * const pubsub = yield* PubSub.bounded<string>(10)
623 *
624 * // Initially empty
625 * const initialSize = yield* PubSub.size(pubsub)
626 *
627 * const subscription = yield* PubSub.subscribe(pubsub)
628 *
629 * // Publish some messages for the active subscription
630 * yield* PubSub.publish(pubsub, "msg1")
631 * yield* PubSub.publish(pubsub, "msg2")
632 *
633 * const afterPublish = yield* PubSub.size(pubsub)
634 * const messages = yield* PubSub.takeAll(subscription)
635 * return { initialSize, afterPublish, messages }
636 * }))
637 *
638 * const actual = await Effect.runPromise(program)
639 * actual // => { initialSize: 0, afterPublish: 2, messages: ["msg1", "msg2"] }
640 * ```
641 *
642 * @category getters
643 * @since 2.0.0
644 */
645export const size = <A>(self: PubSub<A>): Effect.Effect<number> => Effect.sync(() => sizeUnsafe(self))
646/**
647 * Returns the current number of messages retained by the `PubSub` for active
648 * subscribers synchronously.
649 *
650 * **When to use**
651 *
652 * Use when an immediate `PubSub` size snapshot is needed outside effectful code
653 * and concurrent changes between the check and later use are acceptable.
654 *
655 * **Details**
656 *
657 * Returns `0` after shutdown. Because this is an unsafe synchronous snapshot,
658 * prefer `size` in effectful code.
659 *
660 * **Example** (Reading size synchronously)
661 *
662 * ```ts import.meta.vitest
663 * import { Effect, PubSub } from "effect"
664 *
665 * const program = Effect.gen(function*() {
666 * const pubsub = yield* PubSub.bounded<string>(2)
667 * return PubSub.sizeUnsafe(pubsub)
668 * })
669 *
670 * const actual = await Effect.runPromise(program)
671 * actual // => 0
672 * ```
673 *
674 * @category getters
675 * @since 4.0.0
676 */
677export const sizeUnsafe = <A>(self: PubSub<A>): number => {
678 if (MutableRef.get(self.shutdownFlag)) {
679 return 0
680 }
681 return self.pubsub.size()
682}
683
684/**
685 * Returns `true` when the `PubSub` has reached its configured capacity.
686 *
687 * **Details**
688 *
689 * For unbounded PubSubs this is normally `false`.
690 *
691 * **Example** (Checking whether a PubSub is full)
692 *
693 * ```ts import.meta.vitest
694 * import { Effect, PubSub } from "effect"
695 *
696 * const program = Effect.scoped(Effect.gen(function*() {
697 * const pubsub = yield* PubSub.bounded<string>(2)
698 *
699 * // Initially not full
700 * const initiallyFull = yield* PubSub.isFull(pubsub)
701 *
702 * const subscription = yield* PubSub.subscribe(pubsub)
703 *
704 * // Fill the PubSub for the active subscription
705 * yield* PubSub.publish(pubsub, "msg1")
706 * yield* PubSub.publish(pubsub, "msg2")
707 *
708 * const nowFull = yield* PubSub.isFull(pubsub)
709 * const messages = yield* PubSub.takeAll(subscription)
710 * return { initiallyFull, nowFull, messages }
711 * }))
712 *
713 * const actual = await Effect.runPromise(program)
714 * actual // => { initiallyFull: false, nowFull: true, messages: ["msg1", "msg2"] }
715 * ```
716 *
717 * @category predicates
718 * @since 2.0.0
719 */
720export const isFull = <A>(self: PubSub<A>): Effect.Effect<boolean> =>
721 Effect.map(size(self), (size) => size === self.pubsub.capacity)
722
723/**
724 * Returns `true` if the `Pubsub` contains zero elements, `false` otherwise.
725 *
726 * **Example** (Checking whether a PubSub is empty)
727 *
728 * ```ts import.meta.vitest
729 * import { Effect, PubSub } from "effect"
730 *
731 * const program = Effect.scoped(Effect.gen(function*() {
732 * const pubsub = yield* PubSub.bounded<string>(10)
733 *
734 * // Initially empty
735 * const initiallyEmpty = yield* PubSub.isEmpty(pubsub)
736 *
737 * const subscription = yield* PubSub.subscribe(pubsub)
738 *
739 * // Publish a message for the active subscription
740 * yield* PubSub.publish(pubsub, "Hello")
741 *
742 * const nowEmpty = yield* PubSub.isEmpty(pubsub)
743 * const message = yield* PubSub.take(subscription)
744 * return { initiallyEmpty, nowEmpty, message }
745 * }))
746 *
747 * const actual = await Effect.runPromise(program)
748 * actual // => { initiallyEmpty: true, nowEmpty: false, message: "Hello" }
749 * ```
750 *
751 * @category predicates
752 * @since 2.0.0
753 */
754export const isEmpty = <A>(self: PubSub<A>): Effect.Effect<boolean> => Effect.map(size(self), (size) => size === 0)
755
756/**
757 * Shuts down the `PubSub`, interrupting suspended publishers and subscribers
758 * and finalizing active subscriptions.
759 *
760 * **Details**
761 *
762 * After shutdown, `publish` and `publishAll` succeed with `false`,
763 * `publishUnsafe` returns `false`, and subscription operations such as `take`
764 * interrupt.
765 *
766 * **Example** (Shutting down a PubSub)
767 *
768 * ```ts import.meta.vitest
769 * import { Effect, PubSub } from "effect"
770 *
771 * const program = Effect.gen(function*() {
772 * const pubsub = yield* PubSub.bounded<string>(1)
773 *
774 * // Shutdown the PubSub
775 * yield* PubSub.shutdown(pubsub)
776 *
777 * const isShutdown = yield* PubSub.isShutdown(pubsub)
778 *
779 * // Publishing after shutdown returns false
780 * const published = yield* PubSub.publish(pubsub, "msg1")
781 * return { isShutdown, published }
782 * })
783 *
784 * const actual = await Effect.runPromise(program)
785 * actual // => { isShutdown: true, published: false }
786 * ```
787 *
788 * @category lifecycle
789 * @since 2.0.0
790 */
791export const shutdown = <A>(self: PubSub<A>): Effect.Effect<void> =>
792 Effect.uninterruptible(Effect.withFiber((fiber) => {
793 MutableRef.set(self.shutdownFlag, true)
794 return Scope.close(self.scope, Exit.interrupt(fiber.id)).pipe(
795 Effect.andThen(self.strategy.shutdown),
796 Effect.when(self.shutdownHook.open),
797 Effect.asVoid
798 )
799 }))
800
801/**
802 * Ends the `PubSub` with a final message.
803 *
804 * **When to use**
805 *
806 * Use to tell every subscriber that no more messages will follow, without
807 * losing the messages they have not consumed yet.
808 *
809 * **Details**
810 *
811 * Later publishes return `false`, as do publishers waiting for capacity when
812 * `end` is called. Each subscriber receives the messages already buffered
813 * for it, then the final message. Subscribers that arrive
814 * after the end receive the replayed messages, if any, and then the final
815 * message. The final message never occupies capacity, so a bounded `PubSub`
816 * cannot drop it. Returns `false` if the `PubSub` was already ended or shut
817 * down.
818 *
819 * `take`, `takeAll`, and `takeBetween` deliver the final message;
820 * non-suspending `takeUpTo` does not.
821 *
822 * **Gotchas**
823 *
824 * The final message is sticky: subsequent takes return it again. Consumers
825 * must treat it as terminal rather than continuing to take messages.
826 *
827 * **Example** (Ending a PubSub)
828 *
829 * ```ts import.meta.vitest
830 * import { Effect, PubSub } from "effect"
831 *
832 * const program = Effect.scoped(Effect.gen(function*() {
833 * const pubsub = yield* PubSub.bounded<string>(2)
834 * const subscription = yield* PubSub.subscribe(pubsub)
835 *
836 * yield* PubSub.publish(pubsub, "Hello")
837 * const ended = yield* PubSub.end(pubsub, "Bye")
838 *
839 * // Later publishes are rejected
840 * const published = yield* PubSub.publish(pubsub, "World")
841 *
842 * // Buffered messages are delivered before the final message
843 * const first = yield* PubSub.take(subscription)
844 * const last = yield* PubSub.take(subscription)
845 *
846 * // Late subscribers receive the final message too
847 * const late = yield* PubSub.subscribe(pubsub)
848 * const lateMessage = yield* PubSub.take(late)
849 * return [ended, published, first, last, lateMessage]
850 * }))
851 *
852 * const actual = await Effect.runPromise(program)
853 * actual // => [true, false, "Hello", "Bye", "Bye"]
854 * ```
855 *
856 * @see {@link shutdown} for interrupting subscribers instead of delivering a final message
857 *
858 * @category lifecycle
859 * @since 4.0.0
860 */
861export const end: {
862 <A>(value: A): (self: PubSub<A>) => Effect.Effect<boolean>
863 <A>(self: PubSub<A>, value: A): Effect.Effect<boolean>
864} = dual(2, <A>(self: PubSub<A>, value: A): Effect.Effect<boolean> => Effect.sync(() => endUnsafe(self, value)))
865
866/**
867 * Ends the `PubSub` with a final message synchronously.
868 *
869 * **Details**
870 *
871 * See {@link end} for the semantics.
872 *
873 * @category lifecycle
874 * @since 4.0.0
875 */
876export const endUnsafe: {
877 <A>(value: A): (self: PubSub<A>) => boolean
878 <A>(self: PubSub<A>, value: A): boolean
879} = dual(2, <A>(self: PubSub<A>, value: A): boolean => {
880 if (self.shutdownFlag.current || Option.isSome(self.ended.current)) return false
881 MutableRef.set(self.ended, Option.some(value))
882 if (self.strategy instanceof BackPressureStrategy) {
883 for (const [_, deferred, last] of MutableList.takeAll(self.strategy.publishers)) {
884 if (last) Deferred.doneUnsafe(deferred, Exit.succeed(false))
885 }
886 }
887 // A waiting subscriber has nothing buffered, so it receives the final
888 // message right away.
889 const exit = Exit.succeed(value)
890 for (const pollersSet of self.subscribers.values()) {
891 for (const pollers of pollersSet) {
892 let poller: Deferred.Deferred<A> | MutableList.Empty
893 while ((poller = MutableList.take(pollers)) !== MutableList.Empty) {
894 Deferred.doneUnsafe(poller, exit)
895 }
896 }
897 }
898 self.subscribers.clear()
899 return true
900})
901
902/**
903 * Checks effectfully whether `shutdown` has been called, returning `true`
904 * after shutdown and `false` otherwise.
905 *
906 * **Example** (Checking whether a PubSub is shut down)
907 *
908 * ```ts import.meta.vitest
909 * import { Effect, PubSub } from "effect"
910 *
911 * const program = Effect.gen(function*() {
912 * const pubsub = yield* PubSub.bounded<string>(10)
913 *
914 * // Initially not shutdown
915 * const initiallyShutdown = yield* PubSub.isShutdown(pubsub)
916 *
917 * // Shutdown the PubSub
918 * yield* PubSub.shutdown(pubsub)
919 *
920 * const nowShutdown = yield* PubSub.isShutdown(pubsub)
921 * return [initiallyShutdown, nowShutdown]
922 * })
923 *
924 * const actual = await Effect.runPromise(program)
925 * actual // => [false, true]
926 * ```
927 *
928 * @category predicates
929 * @since 2.0.0
930 */
931export const isShutdown = <A>(self: PubSub<A>): Effect.Effect<boolean> => Effect.sync(() => isShutdownUnsafe(self))
932
933/**
934 * Checks synchronously whether `shutdown` has been called, returning `true`
935 * after shutdown and `false` otherwise.
936 *
937 * **When to use**
938 *
939 * Use when an immediate `PubSub` shutdown-state snapshot is needed outside
940 * effectful code and racing shutdown changes are acceptable.
941 *
942 * **Example** (Checking shutdown synchronously)
943 *
944 * ```ts import.meta.vitest
945 * import { Effect, PubSub } from "effect"
946 *
947 * const program = Effect.gen(function*() {
948 * const pubsub = yield* PubSub.bounded<string>(2)
949 * const initiallyShutdown = PubSub.isShutdownUnsafe(pubsub)
950 * yield* PubSub.shutdown(pubsub)
951 * return [initiallyShutdown, PubSub.isShutdownUnsafe(pubsub)]
952 * })
953 *
954 * const actual = await Effect.runPromise(program)
955 * actual // => [false, true]
956 * ```
957 *
958 * @category predicates
959 * @since 4.0.0
960 */
961export const isShutdownUnsafe = <A>(self: PubSub<A>): boolean => self.shutdownFlag.current
962
963/**
964 * Waits until the queue is shutdown. The `Effect` returned by this method will
965 * not resume until the queue has been shutdown. If the queue is already
966 * shutdown, the `Effect` will resume right away.
967 *
968 * **Example** (Waiting for shutdown)
969 *
970 * ```ts import.meta.vitest
971 * import { Effect, Fiber, PubSub } from "effect"
972 *
973 * const program = Effect.gen(function*() {
974 * const pubsub = yield* PubSub.bounded<string>(10)
975 *
976 * // Start a fiber that waits for shutdown
977 * const waiterFiber = yield* Effect.forkChild(
978 * Effect.gen(function*() {
979 * yield* PubSub.awaitShutdown(pubsub)
980 * return "PubSub has been shutdown!"
981 * })
982 * )
983 *
984 * // Shutdown the PubSub
985 * yield* PubSub.shutdown(pubsub)
986 *
987 * // The waiter will now complete
988 * return yield* Fiber.join(waiterFiber)
989 * })
990 *
991 * const actual = await Effect.runPromise(program)
992 * actual // => "PubSub has been shutdown!"
993 * ```
994 *
995 * @category lifecycle
996 * @since 2.0.0
997 */
998export const awaitShutdown = <A>(self: PubSub<A>): Effect.Effect<void> => self.shutdownHook.await
999
1000/**
1001 * Publishes a message to the `PubSub` as an `Effect`, returning whether the
1002 * message was accepted.
1003 *
1004 * **When to use**
1005 *
1006 * Use when you need to publish from effectful code and let the configured
1007 * PubSub strategy handle surplus messages.
1008 *
1009 * **Details**
1010 *
1011 * The effect succeeds with `false` if the `PubSub` is shut down. If the message
1012 * cannot be accepted immediately, the configured strategy decides how surplus
1013 * messages are handled.
1014 *
1015 * **Example** (Publishing a message)
1016 *
1017 * ```ts import.meta.vitest
1018 * import { Effect, PubSub } from "effect"
1019 *
1020 * const program = Effect.scoped(Effect.gen(function*() {
1021 * const pubsub = yield* PubSub.bounded<string>(10)
1022 *
1023 * // Publish a message
1024 * const published = yield* PubSub.publish(pubsub, "Hello World")
1025 *
1026 * const subscription = yield* PubSub.subscribe(pubsub)
1027 *
1028 * yield* PubSub.publish(pubsub, "Hello")
1029 * const message = yield* PubSub.take(subscription)
1030 * return { published, message }
1031 * }))
1032 *
1033 * const actual = await Effect.runPromise(program)
1034 * actual // => { published: true, message: "Hello" }
1035 * ```
1036 *
1037 * @see {@link publishUnsafe} for a synchronous non-blocking attempt that does not run effectful surplus handling
1038 *
1039 * @category publishing
1040 * @since 2.0.0
1041 */
1042export const publish: {
1043 <A>(value: A): (self: PubSub<A>) => Effect.Effect<boolean>
1044 <A>(self: PubSub<A>, value: A): Effect.Effect<boolean>
1045} = dual(2, <A>(self: PubSub<A>, value: A): Effect.Effect<boolean> =>
1046 Effect.suspend(() => {
1047 if (self.shutdownFlag.current || Option.isSome(self.ended.current)) {
1048 return Effect.succeed(false)
1049 }
1050
1051 if (self.pubsub.publish(value)) {
1052 self.strategy.completeSubscribersUnsafe(self.pubsub, self.subscribers)
1053 return Effect.succeed(true)
1054 }
1055
1056 return self.strategy.handleSurplus(
1057 self.pubsub,
1058 self.subscribers,
1059 [value],
1060 self.shutdownFlag
1061 )
1062 }))
1063
1064/**
1065 * Attempts to publish a message synchronously without applying the PubSub
1066 * strategy's effectful surplus handling.
1067 *
1068 * **When to use**
1069 *
1070 * Use when you need a non-blocking synchronous publish attempt where `false`
1071 * is an acceptable result when the message cannot be accepted immediately.
1072 *
1073 * **Details**
1074 *
1075 * Returns `false` if the `PubSub` is shut down or the message cannot be
1076 * accepted immediately, for example when a bounded PubSub is full. Prefer
1077 * `publish` when backpressure or sliding behavior should be honored.
1078 *
1079 * **Example** (Publishing without suspending)
1080 *
1081 * ```ts import.meta.vitest
1082 * import { Effect, PubSub } from "effect"
1083 *
1084 * const program = Effect.gen(function*() {
1085 * const pubsub = yield* PubSub.bounded<string>(2)
1086 * return PubSub.publishUnsafe(pubsub, "Hello")
1087 * })
1088 *
1089 * const actual = await Effect.runPromise(program)
1090 * actual // => true
1091 * ```
1092 *
1093 * @see {@link publish} for effectful publishing that honors the configured surplus strategy
1094 *
1095 * @category publishing
1096 * @since 4.0.0
1097 */
1098export const publishUnsafe: {
1099 <A>(value: A): (self: PubSub<A>) => boolean
1100 <A>(self: PubSub<A>, value: A): boolean
1101} = dual(2, <A>(self: PubSub<A>, value: A): boolean => {
1102 if (self.shutdownFlag.current || Option.isSome(self.ended.current)) return false
1103 if (self.pubsub.publish(value)) {
1104 self.strategy.completeSubscribersUnsafe(self.pubsub, self.subscribers)
1105 return true
1106 }
1107 return false
1108})
1109
1110/**
1111 * Publishes all of the specified messages to the `PubSub`, returning whether they
1112 * were published to the `PubSub`.
1113 *
1114 * **Example** (Publishing multiple messages)
1115 *
1116 * ```ts import.meta.vitest
1117 * import { Effect, Fiber, PubSub } from "effect"
1118 *
1119 * const program = Effect.scoped(Effect.gen(function*() {
1120 * const pubsub = yield* PubSub.bounded<string>(10)
1121 *
1122 * // Publish multiple messages at once
1123 * const allPublished = yield* PubSub.publishAll(pubsub, ["Hello", "World", "from", "Effect"])
1124 *
1125 * // With a smaller capacity and an active subscription
1126 * const smallPubsub = yield* PubSub.bounded<string>(2)
1127 * const subscription = yield* PubSub.subscribe(smallPubsub)
1128 *
1129 * // Will suspend until space becomes available for all messages
1130 * const fiber = yield* Effect.forkChild(PubSub.publishAll(smallPubsub, ["msg1", "msg2", "msg3", "msg4"]))
1131 *
1132 * const firstBatch = yield* PubSub.takeBetween(subscription, 2, 2)
1133 * const result = yield* Fiber.join(fiber)
1134 * const secondBatch = yield* PubSub.takeAll(subscription)
1135 * return { allPublished, firstBatch, result, secondBatch }
1136 * }))
1137 *
1138 * const actual = await Effect.runPromise(program)
1139 * actual // => { allPublished: true, firstBatch: ["msg1", "msg2"], result: true, secondBatch: ["msg3", "msg4"] }
1140 * ```
1141 *
1142 * @category publishing
1143 * @since 2.0.0
1144 */
1145export const publishAll: {
1146 <A>(elements: Iterable<A>): (self: PubSub<A>) => Effect.Effect<boolean>
1147 <A>(self: PubSub<A>, elements: Iterable<A>): Effect.Effect<boolean>
1148} = dual(2, <A>(self: PubSub<A>, elements: Iterable<A>): Effect.Effect<boolean> =>
1149 Effect.suspend(() => {
1150 if (self.shutdownFlag.current || Option.isSome(self.ended.current)) {
1151 return Effect.succeed(false)
1152 }
1153 const surplus = self.pubsub.publishAll(elements)
1154 self.strategy.completeSubscribersUnsafe(self.pubsub, self.subscribers)
1155 if (surplus.length === 0) {
1156 return Effect.succeed(true)
1157 }
1158 return self.strategy.handleSurplus(
1159 self.pubsub,
1160 self.subscribers,
1161 surplus,
1162 self.shutdownFlag
1163 )
1164 }))
1165
1166/**
1167 * Subscribes to receive messages from the `PubSub`. The resulting subscription can
1168 * be evaluated multiple times within the scope to take a message from the `PubSub`
1169 * each time.
1170 *
1171 * **Example** (Subscribing to messages)
1172 *
1173 * ```ts import.meta.vitest
1174 * import { Effect, PubSub } from "effect"
1175 *
1176 * const program = Effect.gen(function*() {
1177 * const pubsub = yield* PubSub.bounded<string>(10)
1178 *
1179 * // Subscribe within a scope for automatic cleanup
1180 * const first = yield* Effect.scoped(Effect.gen(function*() {
1181 * const subscription = yield* PubSub.subscribe(pubsub)
1182 *
1183 * // Publish some messages
1184 * yield* PubSub.publish(pubsub, "Hello")
1185 * yield* PubSub.publish(pubsub, "World")
1186 *
1187 * // Take messages one by one
1188 * const msg1 = yield* PubSub.take(subscription)
1189 * const msg2 = yield* PubSub.take(subscription)
1190 *
1191 * // Subscription is automatically cleaned up when scope exits
1192 * return [msg1, msg2]
1193 * }))
1194 *
1195 * const second = yield* Effect.scoped(Effect.gen(function*() {
1196 * const sub1 = yield* PubSub.subscribe(pubsub)
1197 * const sub2 = yield* PubSub.subscribe(pubsub)
1198 *
1199 * // Multiple subscribers can receive the same messages
1200 * yield* PubSub.publish(pubsub, "Broadcast")
1201 *
1202 * return yield* Effect.all([
1203 * PubSub.take(sub1),
1204 * PubSub.take(sub2)
1205 * ])
1206 * }))
1207 * return [first, second]
1208 * })
1209 *
1210 * const actual = await Effect.runPromise(program)
1211 * actual // => [["Hello", "World"], ["Broadcast", "Broadcast"]]
1212 * ```
1213 *
1214 * @category subscriptions
1215 * @since 2.0.0
1216 */
1217export const subscribe = <A>(self: PubSub<A>): Effect.Effect<Subscription<A>, never, Scope.Scope> =>
1218 Effect.uninterruptible(
1219 Effect.contextWith((services) => {
1220 const localScope = Context.get(services, Scope.Scope)
1221 const scope = Scope.forkUnsafe(self.scope)
1222 const subscription = makeSubscriptionUnsafe(self.pubsub, self.subscribers, self.strategy, self.ended)
1223 return Scope.addFinalizer(scope, unsubscribe(subscription)).pipe(
1224 Effect.andThen(Scope.addFinalizerExit(localScope, (exit) => Scope.close(scope, exit))),
1225 Effect.as(subscription)
1226 )
1227 })
1228 )
1229
1230const unsubscribe = <A>(self: Subscription<A>): Effect.Effect<void> =>
1231 Effect.uninterruptible(
1232 Effect.withFiber<void>((state) => {
1233 MutableRef.set(self.shutdownFlag, true)
1234 return Effect.forEach(
1235 MutableList.takeAll(self.pollers),
1236 (d) => Deferred.interruptWith(d, state.id),
1237 { discard: true, concurrency: "unbounded" }
1238 ).pipe(
1239 Effect.tap(() =>
1240 Effect.sync(() => {
1241 self.subscribers.delete(self.subscription)
1242 self.subscription.unsubscribe()
1243 self.replayWindow.close()
1244 self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers)
1245 })
1246 ),
1247 Effect.when(self.shutdownHook.open),
1248 Effect.asVoid
1249 )
1250 })
1251 )
1252
1253/**
1254 * Takes a single message from the subscription. If no messages are available,
1255 * this will suspend until a message becomes available.
1256 *
1257 * **Example** (Taking a message)
1258 *
1259 * ```ts import.meta.vitest
1260 * import { Effect, Fiber, PubSub } from "effect"
1261 *
1262 * const program = Effect.scoped(Effect.gen(function*() {
1263 * const pubsub = yield* PubSub.bounded<string>(10)
1264 *
1265 * const subscription = yield* PubSub.subscribe(pubsub)
1266 *
1267 * // Start a fiber to take a message (will suspend)
1268 * const takeFiber = yield* Effect.forkChild(PubSub.take(subscription))
1269 *
1270 * // Publish a message
1271 * yield* PubSub.publish(pubsub, "Hello")
1272 *
1273 * // The take will now complete
1274 * return yield* Fiber.join(takeFiber)
1275 * }))
1276 *
1277 * const actual = await Effect.runPromise(program)
1278 * actual // => "Hello"
1279 * ```
1280 *
1281 * @category subscriptions
1282 * @since 4.0.0
1283 */
1284export const take = <A>(self: Subscription<A>): Effect.Effect<A> =>
1285 Effect.suspend(() => {
1286 if (self.shutdownFlag.current) {
1287 return Effect.interrupt
1288 }
1289 if (self.replayWindow.remaining > 0) {
1290 const message = self.replayWindow.take()!
1291 return Effect.succeed(message)
1292 }
1293 const message = self.pollers.length === 0
1294 ? self.subscription.poll()
1295 : MutableList.Empty
1296 if (message === MutableList.Empty) {
1297 return pollForItem(self)
1298 } else {
1299 self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers)
1300 return Effect.succeed(message)
1301 }
1302 })
1303
1304/**
1305 * Takes all available messages from the subscription, suspending if no items
1306 * are available.
1307 *
1308 * **Example** (Taking all available messages)
1309 *
1310 * ```ts import.meta.vitest
1311 * import { Effect, PubSub } from "effect"
1312 *
1313 * const program = Effect.scoped(Effect.gen(function*() {
1314 * const pubsub = yield* PubSub.bounded<string>(10)
1315 *
1316 * const subscription = yield* PubSub.subscribe(pubsub)
1317 *
1318 * // Publish multiple messages
1319 * yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
1320 *
1321 * // Take all available messages at once
1322 * return yield* PubSub.takeAll(subscription)
1323 * }))
1324 *
1325 * const actual = await Effect.runPromise(program)
1326 * actual // => ["msg1", "msg2", "msg3"]
1327 * ```
1328 *
1329 * @category subscriptions
1330 * @since 4.0.0
1331 */
1332export const takeAll = <A>(self: Subscription<A>): Effect.Effect<Arr.NonEmptyArray<A>> =>
1333 Effect.suspend(function loop(value?: [A]): Effect.Effect<Arr.NonEmptyArray<A>> {
1334 if (self.shutdownFlag.current) {
1335 return Effect.interrupt
1336 }
1337 let as = self.pollers.length === 0
1338 ? self.subscription.pollUpTo(Number.POSITIVE_INFINITY)
1339 : []
1340 if (value) {
1341 as = value.concat(as)
1342 }
1343 self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers)
1344 if (self.replayWindow.remaining > 0) {
1345 return Effect.succeed(self.replayWindow.takeAll().concat(as) as Arr.NonEmptyArray<A>)
1346 } else if (!Arr.isArrayNonEmpty(as)) {
1347 return Effect.flatMap(pollForItem(self), (item) => loop([item]))
1348 }
1349 return Effect.succeed(as)
1350 })
1351
1352const pollForItem = <A>(self: Subscription<A>) =>
1353 Effect.callback<A>((resume) => {
1354 if (self.shutdownFlag.current) return resume(Effect.interrupt)
1355 if (Option.isSome(self.ended.current)) return resume(Effect.succeed(self.ended.current.value))
1356 const deferred = Deferred.makeUnsafe<A>()
1357 let set = self.subscribers.get(self.subscription)
1358 if (!set) {
1359 set = new Set()
1360 self.subscribers.set(self.subscription, set)
1361 }
1362 set.add(self.pollers)
1363 MutableList.append(self.pollers, deferred)
1364 self.strategy.completePollersUnsafe(
1365 self.pubsub,
1366 self.subscribers,
1367 self.subscription,
1368 self.pollers
1369 )
1370 if (deferred.effect) return resume(deferred.effect)
1371 deferred.resumes = [resume]
1372 return Effect.sync(() => MutableList.remove(self.pollers, deferred))
1373 })
1374
1375/**
1376 * Takes up to the specified number of messages from the subscription without
1377 * suspending. Finite fractional values are rounded down, while `NaN` and
1378 * non-positive values are treated as `0`.
1379 *
1380 * **Example** (Taking up to a maximum number of messages)
1381 *
1382 * ```ts import.meta.vitest
1383 * import { Effect, PubSub } from "effect"
1384 *
1385 * const program = Effect.scoped(Effect.gen(function*() {
1386 * const pubsub = yield* PubSub.bounded<string>(10)
1387 *
1388 * const subscription = yield* PubSub.subscribe(pubsub)
1389 *
1390 * // Publish multiple messages
1391 * yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3", "msg4", "msg5"])
1392 *
1393 * // Take up to 3 messages
1394 * const upTo3 = yield* PubSub.takeUpTo(subscription, 3)
1395 *
1396 * // Take up to 5 more (only 2 remaining)
1397 * const upTo5 = yield* PubSub.takeUpTo(subscription, 5)
1398 *
1399 * // No more messages available
1400 * const noMore = yield* PubSub.takeUpTo(subscription, 10)
1401 * return [upTo3, upTo5, noMore]
1402 * }))
1403 *
1404 * const actual = await Effect.runPromise(program)
1405 * actual // => [["msg1", "msg2", "msg3"], ["msg4", "msg5"], []]
1406 * ```
1407 *
1408 * @category subscriptions
1409 * @since 4.0.0
1410 */
1411export const takeUpTo: {
1412 (max: number): <A>(self: Subscription<A>) => Effect.Effect<Array<A>>
1413 <A>(self: Subscription<A>, max: number): Effect.Effect<Array<A>>
1414} = dual(2, <A>(self: Subscription<A>, max: number): Effect.Effect<Array<A>> =>
1415 Effect.suspend(() => {
1416 if (self.shutdownFlag.current) return Effect.interrupt
1417 max = Count.normalize(max)
1418 let replay: Array<A> | undefined = undefined
1419 if (self.replayWindow.remaining >= max) {
1420 return Effect.succeed(self.replayWindow.takeN(max))
1421 } else if (self.replayWindow.remaining > 0) {
1422 replay = self.replayWindow.takeAll()
1423 max = max - replay.length
1424 }
1425 const as = self.pollers.length === 0
1426 ? self.subscription.pollUpTo(max)
1427 : []
1428 self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers)
1429 return replay ? Effect.succeed(replay.concat(as)) : Effect.succeed(as)
1430 }))
1431
1432/**
1433 * Takes between the specified minimum and maximum number of messages from the
1434 * subscription. Finite fractional bounds are rounded down, while `NaN` and
1435 * non-positive bounds are treated as `0`. Will suspend if the normalized
1436 * minimum number is not immediately available.
1437 *
1438 * **Example** (Taking between a minimum and maximum)
1439 *
1440 * ```ts import.meta.vitest
1441 * import { Effect, Fiber, PubSub } from "effect"
1442 *
1443 * const program = Effect.scoped(Effect.gen(function*() {
1444 * const pubsub = yield* PubSub.bounded<string>(10)
1445 *
1446 * const subscription = yield* PubSub.subscribe(pubsub)
1447 *
1448 * // Start taking between 2 and 5 messages (will suspend)
1449 * const takeFiber = yield* Effect.forkChild(PubSub.takeBetween(subscription, 2, 5))
1450 *
1451 * // Publish 3 messages
1452 * yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
1453 *
1454 * // Now the take will complete with 3 messages
1455 * return yield* Fiber.join(takeFiber)
1456 * }))
1457 *
1458 * const actual = await Effect.runPromise(program)
1459 * actual // => ["msg1", "msg2", "msg3"]
1460 * ```
1461 *
1462 * @category subscriptions
1463 * @since 4.0.0
1464 */
1465export const takeBetween: {
1466 (min: number, max: number): <A>(self: Subscription<A>) => Effect.Effect<Array<A>>
1467 <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>
1468} = dual(
1469 3,
1470 <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>> =>
1471 Effect.suspend(() => takeRemainderLoop(self, Count.normalize(min), Count.normalize(max), []))
1472)
1473
1474const takeRemainderLoop = <A>(
1475 self: Subscription<A>,
1476 min: number,
1477 max: number,
1478 acc: Array<A>
1479): Effect.Effect<Array<A>> => {
1480 if (max < min) {
1481 return Effect.succeed(acc)
1482 }
1483 return Effect.flatMap(takeUpTo(self, max), (bs) => {
1484 acc.push(...bs)
1485 const remaining = min - bs.length
1486 if (remaining === 1) {
1487 return Effect.map(take(self), (b) => {
1488 acc.push(b)
1489 return acc
1490 })
1491 }
1492 if (remaining > 1) {
1493 return Effect.flatMap(take(self), (b) => {
1494 acc.push(b)
1495 return takeRemainderLoop(
1496 self,
1497 remaining - 1,
1498 max - bs.length - 1,
1499 acc
1500 )
1501 })
1502 }
1503 return Effect.succeed(acc)
1504 })
1505}
1506
1507/**
1508 * Returns the number of messages currently available in the subscription as an
1509 * `Effect`.
1510 *
1511 * **When to use**
1512 *
1513 * Use when checking a subscription from effectful code and shutdown should
1514 * interrupt the effect.
1515 *
1516 * **Details**
1517 *
1518 * The count includes replay-buffered messages. If the subscription has been
1519 * shut down, the effect interrupts.
1520 *
1521 * **Example** (Checking remaining messages)
1522 *
1523 * ```ts import.meta.vitest
1524 * import { Effect, PubSub } from "effect"
1525 *
1526 * const program = Effect.scoped(Effect.gen(function*() {
1527 * const pubsub = yield* PubSub.bounded<string>(10)
1528 *
1529 * const subscription = yield* PubSub.subscribe(pubsub)
1530 *
1531 * // Publish some messages
1532 * yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
1533 *
1534 * // Check how many messages are available
1535 * const count = yield* PubSub.remaining(subscription)
1536 *
1537 * // Take one message
1538 * const message = yield* PubSub.take(subscription)
1539 *
1540 * const remaining = yield* PubSub.remaining(subscription)
1541 * return { count, message, remaining }
1542 * }))
1543 *
1544 * const actual = await Effect.runPromise(program)
1545 * actual // => { count: 3, message: "msg1", remaining: 2 }
1546 * ```
1547 *
1548 * @see {@link remainingUnsafe} for a synchronous check that reports shutdown as `Option.none()`
1549 *
1550 * @category getters
1551 * @since 4.0.0
1552 */
1553export const remaining = <A>(self: Subscription<A>): Effect.Effect<number> =>
1554 Effect.suspend(() =>
1555 self.shutdownFlag.current
1556 ? Effect.interrupt
1557 : Effect.succeed(self.subscription.size() + self.replayWindow.remaining)
1558 )
1559
1560/**
1561 * Synchronously returns the number of messages currently available in the
1562 * subscription, or `Option.none()` when it is shut down.
1563 *
1564 * **When to use**
1565 *
1566 * Use when you need synchronous polling outside a managed workflow and want
1567 * shutdown observed as data instead of interruption.
1568 *
1569 * **Example** (Checking remaining messages synchronously)
1570 *
1571 * ```ts import.meta.vitest
1572 * import { Effect, Option, PubSub } from "effect"
1573 *
1574 * const program = Effect.scoped(Effect.gen(function*() {
1575 * const pubsub = yield* PubSub.bounded<string>(2)
1576 * const subscription = yield* PubSub.subscribe(pubsub)
1577 * return PubSub.remainingUnsafe(subscription)
1578 * }))
1579 *
1580 * const actual = await Effect.runPromise(program)
1581 * actual // => Option.some(0)
1582 * ```
1583 *
1584 * @see {@link remaining} for the effectful variant that interrupts on shutdown
1585 *
1586 * @category getters
1587 * @since 4.0.0
1588 */
1589export const remainingUnsafe = <A>(self: Subscription<A>): Option.Option<number> => {
1590 if (self.shutdownFlag.current) {
1591 return Option.none()
1592 }
1593 return Option.some(self.subscription.size() + self.replayWindow.remaining)
1594}
1595
1596// -----------------------------------------------------------------------------
1597// internal
1598// -----------------------------------------------------------------------------
1599
1600const AbsentValue = Symbol.for("effect/PubSub/AbsentValue")
1601type AbsentValue = typeof AbsentValue
1602
1603const addSubscribers = <A>(
1604 subscribers: PubSub.Subscribers<A>,
1605 subscription: PubSub.BackingSubscription<A>,
1606 pollers: MutableList.MutableList<Deferred.Deferred<A>>
1607) => {
1608 if (!subscribers.has(subscription)) {
1609 subscribers.set(subscription, new Set())
1610 }
1611 const set = subscribers.get(subscription)!
1612 set.add(pollers)
1613}
1614
1615const removeSubscribers = <A>(
1616 subscribers: PubSub.Subscribers<A>,
1617 subscription: PubSub.BackingSubscription<A>,
1618 pollers: MutableList.MutableList<Deferred.Deferred<A>>
1619) => {
1620 if (!subscribers.has(subscription)) {
1621 return
1622 }
1623 const set = subscribers.get(subscription)!
1624 set.delete(pollers)
1625 if (set.size === 0) {
1626 subscribers.delete(subscription)
1627 }
1628}
1629
1630const makeSubscriptionUnsafe = <A>(
1631 pubsub: PubSub.Atomic<A>,
1632 subscribers: PubSub.Subscribers<A>,
1633 strategy: PubSub.Strategy<A>,
1634 ended: MutableRef.MutableRef<Option.Option<A>>
1635): Subscription<A> =>
1636 new SubscriptionImpl(
1637 pubsub,
1638 subscribers,
1639 pubsub.subscribe(),
1640 MutableList.make<Deferred.Deferred<A>>(),
1641 Latch.makeUnsafe(false),
1642 MutableRef.make(false),
1643 strategy,
1644 pubsub.replayWindow(),
1645 ended
1646 )
1647
1648class BoundedPubSubArb<in out A> implements PubSub.Atomic<A> {
1649 array: Array<A>
1650 replayIndices: Array<number>
1651 publisherIndex = 0
1652 subscribers: Array<number>
1653 subscriberCount = 0
1654 subscribersIndex = 0
1655
1656 readonly capacity: number
1657 readonly replayBuffer: ReplayBuffer<A> | undefined
1658
1659 constructor(capacity: number, replayBuffer: ReplayBuffer<A> | undefined) {
1660 this.capacity = capacity
1661 this.replayBuffer = replayBuffer
1662 this.array = Array.from({ length: capacity })
1663 this.replayIndices = replayBuffer ? Array.from({ length: capacity }) : []
1664 this.subscribers = Array.from({ length: capacity })
1665 }
1666
1667 replayWindow(): PubSub.ReplayWindow<A> {
1668 return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
1669 }
1670
1671 isEmpty(): boolean {
1672 return this.publisherIndex === this.subscribersIndex
1673 }
1674
1675 isFull(): boolean {
1676 return this.publisherIndex === this.subscribersIndex + this.capacity
1677 }
1678
1679 size(): number {
1680 return this.publisherIndex - this.subscribersIndex
1681 }
1682
1683 publish(value: A): boolean {
1684 if (this.isFull()) {
1685 return false
1686 }
1687 const replayIndex = this.replayBuffer?.offer(value)
1688 if (this.subscriberCount !== 0) {
1689 const index = this.publisherIndex % this.capacity
1690 this.array[index] = value
1691 if (replayIndex !== undefined) {
1692 this.replayIndices[index] = replayIndex
1693 }
1694 this.subscribers[index] = this.subscriberCount
1695 this.publisherIndex += 1
1696 }
1697 return true
1698 }
1699
1700 publishAll(elements: Iterable<A>): Array<A> {
1701 if (this.subscriberCount === 0) {
1702 if (this.replayBuffer) {
1703 this.replayBuffer.offerAll(elements)
1704 }
1705 return []
1706 }
1707 const chunk = Arr.fromIterable(elements)
1708 const n = chunk.length
1709 const size = this.publisherIndex - this.subscribersIndex
1710 const available = this.capacity - size
1711 const forPubSub = Math.min(n, available)
1712 if (forPubSub === 0) {
1713 return chunk
1714 }
1715 let iteratorIndex = 0
1716 const publishAllIndex = this.publisherIndex + forPubSub
1717 while (this.publisherIndex !== publishAllIndex) {
1718 const a = chunk[iteratorIndex++]
1719 const index = this.publisherIndex % this.capacity
1720 this.array[index] = a
1721 const replayIndex = this.replayBuffer?.offer(a)
1722 if (replayIndex !== undefined) {
1723 this.replayIndices[index] = replayIndex
1724 }
1725 this.subscribers[index] = this.subscriberCount
1726 this.publisherIndex += 1
1727 }
1728 return chunk.slice(iteratorIndex)
1729 }
1730
1731 slide(): void {
1732 if (this.subscribersIndex !== this.publisherIndex) {
1733 const index = this.subscribersIndex % this.capacity
1734 const value = this.array[index]
1735 this.array[index] = AbsentValue as unknown as A
1736 this.subscribers[index] = 0
1737 this.subscribersIndex += 1
1738 this.replayBuffer?.slide(value, this.replayIndices[index])
1739 }
1740 }
1741
1742 subscribe(): PubSub.BackingSubscription<A> {
1743 this.subscriberCount += 1
1744 return new BoundedPubSubArbSubscription(this, this.publisherIndex, false)
1745 }
1746}
1747
1748class BoundedPubSubArbSubscription<in out A> implements PubSub.BackingSubscription<A> {
1749 private self: BoundedPubSubArb<A>
1750 private subscriberIndex: number
1751 private unsubscribed: boolean
1752
1753 constructor(
1754 self: BoundedPubSubArb<A>,
1755 subscriberIndex: number,
1756 unsubscribed: boolean
1757 ) {
1758 this.self = self
1759 this.subscriberIndex = subscriberIndex
1760 this.unsubscribed = unsubscribed
1761 }
1762
1763 isEmpty(): boolean {
1764 return (
1765 this.unsubscribed ||
1766 this.self.publisherIndex === this.subscriberIndex ||
1767 this.self.publisherIndex === this.self.subscribersIndex
1768 )
1769 }
1770
1771 size() {
1772 if (this.unsubscribed) {
1773 return 0
1774 }
1775 return this.self.publisherIndex - Math.max(this.subscriberIndex, this.self.subscribersIndex)
1776 }
1777
1778 poll(): A | MutableList.Empty {
1779 if (this.unsubscribed) {
1780 return MutableList.Empty
1781 }
1782 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex)
1783 if (this.subscriberIndex !== this.self.publisherIndex) {
1784 const index = this.subscriberIndex % this.self.capacity
1785 const elem = this.self.array[index]!
1786 this.self.subscribers[index] -= 1
1787 if (this.self.subscribers[index] === 0) {
1788 this.self.array[index] = AbsentValue as unknown as A
1789 this.self.subscribersIndex += 1
1790 }
1791 this.subscriberIndex += 1
1792 return elem
1793 }
1794 return MutableList.Empty
1795 }
1796
1797 pollUpTo(n: number): Array<A> {
1798 n = Count.normalize(n)
1799 if (this.unsubscribed) {
1800 return []
1801 }
1802 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex)
1803 const size = this.self.publisherIndex - this.subscriberIndex
1804 const toPoll = Math.min(n, size)
1805 if (toPoll <= 0) {
1806 return []
1807 }
1808 const builder: Array<A> = []
1809 const pollUpToIndex = this.subscriberIndex + toPoll
1810 while (this.subscriberIndex !== pollUpToIndex) {
1811 const index = this.subscriberIndex % this.self.capacity
1812 const a = this.self.array[index] as A
1813 this.self.subscribers[index] -= 1
1814 if (this.self.subscribers[index] === 0) {
1815 this.self.array[index] = AbsentValue as unknown as A
1816 this.self.subscribersIndex += 1
1817 }
1818 builder.push(a)
1819 this.subscriberIndex += 1
1820 }
1821
1822 return builder
1823 }
1824
1825 unsubscribe(): void {
1826 if (!this.unsubscribed) {
1827 this.unsubscribed = true
1828 this.self.subscriberCount -= 1
1829 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex)
1830 while (this.subscriberIndex !== this.self.publisherIndex) {
1831 const index = this.subscriberIndex % this.self.capacity
1832 this.self.subscribers[index] -= 1
1833 if (this.self.subscribers[index] === 0) {
1834 this.self.array[index] = AbsentValue as unknown as A
1835 this.self.subscribersIndex += 1
1836 }
1837 this.subscriberIndex += 1
1838 }
1839 }
1840 }
1841}
1842
1843class BoundedPubSubPow2<in out A> implements PubSub.Atomic<A> {
1844 array: Array<A>
1845 replayIndices: Array<number>
1846 mask: number
1847 publisherIndex = 0
1848 subscribers: Array<number>
1849 subscriberCount = 0
1850 subscribersIndex = 0
1851
1852 readonly capacity: number
1853 readonly replayBuffer: ReplayBuffer<A> | undefined
1854
1855 constructor(capacity: number, replayBuffer: ReplayBuffer<A> | undefined) {
1856 this.capacity = capacity
1857 this.replayBuffer = replayBuffer
1858 this.array = Array.from({ length: capacity })
1859 this.replayIndices = replayBuffer ? Array.from({ length: capacity }) : []
1860 this.mask = capacity - 1
1861 this.subscribers = Array.from({ length: capacity })
1862 }
1863
1864 replayWindow(): PubSub.ReplayWindow<A> {
1865 return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
1866 }
1867
1868 isEmpty(): boolean {
1869 return this.publisherIndex === this.subscribersIndex
1870 }
1871
1872 isFull(): boolean {
1873 return this.publisherIndex === this.subscribersIndex + this.capacity
1874 }
1875
1876 size(): number {
1877 return this.publisherIndex - this.subscribersIndex
1878 }
1879
1880 publish(value: A): boolean {
1881 if (this.isFull()) {
1882 return false
1883 }
1884 const replayIndex = this.replayBuffer?.offer(value)
1885 if (this.subscriberCount !== 0) {
1886 const index = this.publisherIndex & this.mask
1887 this.array[index] = value
1888 if (replayIndex !== undefined) {
1889 this.replayIndices[index] = replayIndex
1890 }
1891 this.subscribers[index] = this.subscriberCount
1892 this.publisherIndex += 1
1893 }
1894 return true
1895 }
1896
1897 publishAll(elements: Iterable<A>): Array<A> {
1898 if (this.subscriberCount === 0) {
1899 if (this.replayBuffer) {
1900 this.replayBuffer.offerAll(elements)
1901 }
1902 return []
1903 }
1904 const chunk = Arr.fromIterable(elements)
1905 const n = chunk.length
1906 const size = this.publisherIndex - this.subscribersIndex
1907 const available = this.capacity - size
1908 const forPubSub = Math.min(n, available)
1909 if (forPubSub === 0) {
1910 return chunk
1911 }
1912 let iteratorIndex = 0
1913 const publishAllIndex = this.publisherIndex + forPubSub
1914 while (this.publisherIndex !== publishAllIndex) {
1915 const elem = chunk[iteratorIndex++]
1916 const index = this.publisherIndex & this.mask
1917 this.array[index] = elem
1918 const replayIndex = this.replayBuffer?.offer(elem)
1919 if (replayIndex !== undefined) {
1920 this.replayIndices[index] = replayIndex
1921 }
1922 this.subscribers[index] = this.subscriberCount
1923 this.publisherIndex += 1
1924 }
1925 return chunk.slice(iteratorIndex)
1926 }
1927
1928 slide(): void {
1929 if (this.subscribersIndex !== this.publisherIndex) {
1930 const index = this.subscribersIndex & this.mask
1931 const value = this.array[index]
1932 this.array[index] = AbsentValue as unknown as A
1933 this.subscribers[index] = 0
1934 this.subscribersIndex += 1
1935 this.replayBuffer?.slide(value, this.replayIndices[index])
1936 }
1937 }
1938
1939 subscribe(): PubSub.BackingSubscription<A> {
1940 this.subscriberCount += 1
1941 return new BoundedPubSubPow2Subscription(this, this.publisherIndex, false)
1942 }
1943}
1944
1945class BoundedPubSubPow2Subscription<in out A> implements PubSub.BackingSubscription<A> {
1946 private self: BoundedPubSubPow2<A>
1947 private subscriberIndex: number
1948 private unsubscribed: boolean
1949
1950 constructor(
1951 self: BoundedPubSubPow2<A>,
1952 subscriberIndex: number,
1953 unsubscribed: boolean
1954 ) {
1955 this.self = self
1956 this.subscriberIndex = subscriberIndex
1957 this.unsubscribed = unsubscribed
1958 }
1959
1960 isEmpty(): boolean {
1961 return (
1962 this.unsubscribed ||
1963 this.self.publisherIndex === this.subscriberIndex ||
1964 this.self.publisherIndex === this.self.subscribersIndex
1965 )
1966 }
1967
1968 size() {
1969 if (this.unsubscribed) {
1970 return 0
1971 }
1972 return this.self.publisherIndex - Math.max(this.subscriberIndex, this.self.subscribersIndex)
1973 }
1974
1975 poll(): A | MutableList.Empty {
1976 if (this.unsubscribed) {
1977 return MutableList.Empty
1978 }
1979 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex)
1980 if (this.subscriberIndex !== this.self.publisherIndex) {
1981 const index = this.subscriberIndex & this.self.mask
1982 const elem = this.self.array[index]!
1983 this.self.subscribers[index] -= 1
1984 if (this.self.subscribers[index] === 0) {
1985 this.self.array[index] = AbsentValue as unknown as A
1986 this.self.subscribersIndex += 1
1987 }
1988 this.subscriberIndex += 1
1989 return elem
1990 }
1991 return MutableList.Empty
1992 }
1993
1994 pollUpTo(n: number): Array<A> {
1995 n = Count.normalize(n)
1996 if (this.unsubscribed) {
1997 return []
1998 }
1999 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex)
2000 const size = this.self.publisherIndex - this.subscriberIndex
2001 const toPoll = Math.min(n, size)
2002 if (toPoll <= 0) {
2003 return []
2004 }
2005 const builder: Array<A> = []
2006 const pollUpToIndex = this.subscriberIndex + toPoll
2007 while (this.subscriberIndex !== pollUpToIndex) {
2008 const index = this.subscriberIndex & this.self.mask
2009 const elem = this.self.array[index] as A
2010 this.self.subscribers[index] -= 1
2011 if (this.self.subscribers[index] === 0) {
2012 this.self.array[index] = AbsentValue as unknown as A
2013 this.self.subscribersIndex += 1
2014 }
2015 builder.push(elem)
2016 this.subscriberIndex += 1
2017 }
2018 return builder
2019 }
2020
2021 unsubscribe(): void {
2022 if (!this.unsubscribed) {
2023 this.unsubscribed = true
2024 this.self.subscriberCount -= 1
2025 this.subscriberIndex = Math.max(this.subscriberIndex, this.self.subscribersIndex)
2026 while (this.subscriberIndex !== this.self.publisherIndex) {
2027 const index = this.subscriberIndex & this.self.mask
2028 this.self.subscribers[index] -= 1
2029 if (this.self.subscribers[index] === 0) {
2030 this.self.array[index] = AbsentValue as unknown as A
2031 this.self.subscribersIndex += 1
2032 }
2033 this.subscriberIndex += 1
2034 }
2035 }
2036 }
2037}
2038
2039class BoundedPubSubSingle<in out A> implements PubSub.Atomic<A> {
2040 publisherIndex = 0
2041 subscriberCount = 0
2042 subscribers = 0
2043 value: A = AbsentValue as unknown as A
2044 replayIndex = 0
2045
2046 readonly capacity = 1
2047 readonly replayBuffer: ReplayBuffer<A> | undefined
2048
2049 constructor(replayBuffer: ReplayBuffer<A> | undefined) {
2050 this.replayBuffer = replayBuffer
2051 }
2052
2053 replayWindow(): PubSub.ReplayWindow<A> {
2054 return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
2055 }
2056
2057 pipe() {
2058 return pipeArguments(this, arguments)
2059 }
2060
2061 isEmpty(): boolean {
2062 return this.subscribers === 0
2063 }
2064
2065 isFull(): boolean {
2066 return !this.isEmpty()
2067 }
2068
2069 size(): number {
2070 return this.isEmpty() ? 0 : 1
2071 }
2072
2073 publish(value: A): boolean {
2074 if (this.isFull()) {
2075 return false
2076 }
2077 const replayIndex = this.replayBuffer?.offer(value)
2078 if (this.subscriberCount !== 0) {
2079 this.value = value
2080 if (replayIndex !== undefined) {
2081 this.replayIndex = replayIndex
2082 }
2083 this.subscribers = this.subscriberCount
2084 this.publisherIndex += 1
2085 }
2086 return true
2087 }
2088
2089 publishAll(elements: Iterable<A>): Array<A> {
2090 if (this.subscriberCount === 0) {
2091 if (this.replayBuffer) {
2092 this.replayBuffer.offerAll(elements)
2093 }
2094 return []
2095 }
2096 const chunk = Arr.fromIterable(elements)
2097 if (chunk.length === 0) {
2098 return chunk
2099 }
2100 if (this.publish(chunk[0])) {
2101 return chunk.slice(1)
2102 } else {
2103 return chunk
2104 }
2105 }
2106
2107 slide(): void {
2108 if (this.isFull()) {
2109 const value = this.value
2110 this.subscribers = 0
2111 this.value = AbsentValue as unknown as A
2112 this.replayBuffer?.slide(value, this.replayIndex)
2113 }
2114 }
2115
2116 subscribe(): PubSub.BackingSubscription<A> {
2117 this.subscriberCount += 1
2118 return new BoundedPubSubSingleSubscription(this, this.publisherIndex, false)
2119 }
2120}
2121
2122class BoundedPubSubSingleSubscription<in out A> implements PubSub.BackingSubscription<A> {
2123 private self: BoundedPubSubSingle<A>
2124 private subscriberIndex: number
2125 private unsubscribed: boolean
2126
2127 constructor(
2128 self: BoundedPubSubSingle<A>,
2129 subscriberIndex: number,
2130 unsubscribed: boolean
2131 ) {
2132 this.self = self
2133 this.subscriberIndex = subscriberIndex
2134 this.unsubscribed = unsubscribed
2135 }
2136
2137 isEmpty(): boolean {
2138 return (
2139 this.unsubscribed ||
2140 this.self.subscribers === 0 ||
2141 this.subscriberIndex === this.self.publisherIndex
2142 )
2143 }
2144
2145 size() {
2146 return this.isEmpty() ? 0 : 1
2147 }
2148
2149 poll(): A | MutableList.Empty {
2150 if (this.isEmpty()) {
2151 return MutableList.Empty
2152 }
2153 const elem = this.self.value
2154 this.self.subscribers -= 1
2155 if (this.self.subscribers === 0) {
2156 this.self.value = AbsentValue as unknown as A
2157 }
2158 this.subscriberIndex = this.self.publisherIndex
2159 return elem
2160 }
2161
2162 pollUpTo(n: number): Array<A> {
2163 if (Count.normalize(n) < 1 || this.isEmpty()) {
2164 return []
2165 }
2166 return [this.poll() as A]
2167 }
2168
2169 unsubscribe(): void {
2170 if (!this.unsubscribed) {
2171 this.unsubscribed = true
2172 this.self.subscriberCount -= 1
2173 if (this.self.subscribers !== 0 && this.subscriberIndex !== this.self.publisherIndex) {
2174 this.self.subscribers -= 1
2175 if (this.self.subscribers === 0) {
2176 this.self.value = AbsentValue as unknown as A
2177 }
2178 }
2179 }
2180 }
2181}
2182
2183interface Node<out A> {
2184 value: A | AbsentValue
2185 replayIndex: number | undefined
2186 subscribers: number
2187 next: Node<A> | null
2188}
2189
2190class UnboundedPubSub<in out A> implements PubSub.Atomic<A> {
2191 publisherHead: Node<A> = {
2192 value: AbsentValue,
2193 replayIndex: undefined,
2194 subscribers: 0,
2195 next: null
2196 }
2197 publisherTail = this.publisherHead
2198 publisherIndex = 0
2199 subscribersIndex = 0
2200
2201 readonly capacity = Number.MAX_SAFE_INTEGER
2202 readonly replayBuffer: ReplayBuffer<A> | undefined
2203
2204 constructor(replayBuffer: ReplayBuffer<A> | undefined) {
2205 this.replayBuffer = replayBuffer
2206 }
2207
2208 replayWindow(): PubSub.ReplayWindow<A> {
2209 return this.replayBuffer ? new ReplayWindowImpl(this.replayBuffer) : emptyReplayWindow
2210 }
2211
2212 isEmpty(): boolean {
2213 return this.publisherHead === this.publisherTail
2214 }
2215
2216 isFull(): boolean {
2217 return false
2218 }
2219
2220 size(): number {
2221 return this.publisherIndex - this.subscribersIndex
2222 }
2223
2224 publish(value: A): boolean {
2225 const replayIndex = this.replayBuffer?.offer(value)
2226 const subscribers = this.publisherTail.subscribers
2227 if (subscribers !== 0) {
2228 const node: Node<A> = {
2229 value,
2230 replayIndex,
2231 subscribers,
2232 next: null
2233 }
2234 this.publisherTail.next = node
2235 this.publisherTail = this.publisherTail.next
2236 this.publisherIndex += 1
2237 }
2238 return true
2239 }
2240
2241 publishAll(elements: Iterable<A>): Array<A> {
2242 if (this.publisherTail.subscribers !== 0) {
2243 for (const a of elements) {
2244 this.publish(a)
2245 }
2246 } else if (this.replayBuffer) {
2247 this.replayBuffer.offerAll(elements)
2248 }
2249 return []
2250 }
2251
2252 slide(): void {
2253 if (this.publisherHead !== this.publisherTail) {
2254 const node = this.publisherHead.next!
2255 const value = node.value as A
2256 this.publisherHead = this.publisherHead.next!
2257 this.publisherHead.value = AbsentValue
2258 this.subscribersIndex += 1
2259 this.replayBuffer?.slide(value, node.replayIndex!)
2260 }
2261 }
2262
2263 subscribe(): PubSub.BackingSubscription<A> {
2264 this.publisherTail.subscribers += 1
2265 return new UnboundedPubSubSubscription(
2266 this,
2267 this.publisherTail,
2268 this.publisherIndex,
2269 false
2270 )
2271 }
2272}
2273
2274class UnboundedPubSubSubscription<in out A> implements PubSub.BackingSubscription<A> {
2275 private self: UnboundedPubSub<A>
2276 private subscriberHead: Node<A>
2277 private subscriberIndex: number
2278 private unsubscribed: boolean
2279
2280 constructor(
2281 self: UnboundedPubSub<A>,
2282 subscriberHead: Node<A>,
2283 subscriberIndex: number,
2284 unsubscribed: boolean
2285 ) {
2286 this.self = self
2287 this.subscriberHead = subscriberHead
2288 this.subscriberIndex = subscriberIndex
2289 this.unsubscribed = unsubscribed
2290 }
2291
2292 isEmpty(): boolean {
2293 if (this.unsubscribed) {
2294 return true
2295 }
2296 let empty = true
2297 let loop = true
2298 while (loop) {
2299 if (this.subscriberHead === this.self.publisherTail) {
2300 loop = false
2301 } else {
2302 if (this.subscriberHead.next!.value !== AbsentValue) {
2303 empty = false
2304 loop = false
2305 } else {
2306 this.subscriberHead = this.subscriberHead.next!
2307 this.subscriberIndex += 1
2308 }
2309 }
2310 }
2311 return empty
2312 }
2313
2314 size() {
2315 if (this.unsubscribed) {
2316 return 0
2317 }
2318 return this.self.publisherIndex - Math.max(this.subscriberIndex, this.self.subscribersIndex)
2319 }
2320
2321 poll(): A | MutableList.Empty {
2322 if (this.unsubscribed) {
2323 return MutableList.Empty
2324 }
2325 let loop = true
2326 let polled: A | MutableList.Empty = MutableList.Empty
2327 while (loop) {
2328 if (this.subscriberHead === this.self.publisherTail) {
2329 loop = false
2330 } else {
2331 const elem = this.subscriberHead.next!.value
2332 if (elem !== AbsentValue) {
2333 polled = elem
2334 this.subscriberHead.subscribers -= 1
2335 if (this.subscriberHead.subscribers === 0) {
2336 this.self.publisherHead = this.self.publisherHead.next!
2337 this.self.publisherHead.value = AbsentValue
2338 this.self.subscribersIndex += 1
2339 }
2340 loop = false
2341 }
2342 this.subscriberHead = this.subscriberHead.next!
2343 this.subscriberIndex += 1
2344 }
2345 }
2346 return polled
2347 }
2348
2349 pollUpTo(n: number): Array<A> {
2350 n = Count.normalize(n)
2351 const builder: Array<A> = []
2352 let i = 0
2353 while (i !== n) {
2354 const a = this.poll()
2355 if (a === MutableList.Empty) {
2356 i = n
2357 } else {
2358 builder.push(a)
2359 i += 1
2360 }
2361 }
2362 return builder
2363 }
2364
2365 unsubscribe(): void {
2366 if (!this.unsubscribed) {
2367 this.unsubscribed = true
2368 this.self.publisherTail.subscribers -= 1
2369 while (this.subscriberHead !== this.self.publisherTail) {
2370 if (this.subscriberHead.next!.value !== AbsentValue) {
2371 this.subscriberHead.subscribers -= 1
2372 if (this.subscriberHead.subscribers === 0) {
2373 this.self.publisherHead = this.self.publisherHead.next!
2374 this.self.publisherHead.value = AbsentValue
2375 this.self.subscribersIndex += 1
2376 }
2377 }
2378 this.subscriberHead = this.subscriberHead.next!
2379 }
2380 }
2381 }
2382}
2383
2384class SubscriptionImpl<in out A> implements Subscription<A> {
2385 readonly [SubscriptionTypeId] = {
2386 _A: identity
2387 }
2388
2389 readonly pubsub: PubSub.Atomic<A>
2390 readonly subscribers: PubSub.Subscribers<A>
2391 readonly subscription: PubSub.BackingSubscription<A>
2392 readonly pollers: MutableList.MutableList<Deferred.Deferred<A>>
2393 readonly shutdownHook: Latch.Latch
2394 readonly shutdownFlag: MutableRef.MutableRef<boolean>
2395 readonly strategy: PubSub.Strategy<A>
2396 readonly replayWindow: PubSub.ReplayWindow<A>
2397 readonly ended: MutableRef.MutableRef<Option.Option<A>>
2398
2399 constructor(
2400 pubsub: PubSub.Atomic<A>,
2401 subscribers: PubSub.Subscribers<A>,
2402 subscription: PubSub.BackingSubscription<A>,
2403 pollers: MutableList.MutableList<Deferred.Deferred<A>>,
2404 shutdownHook: Latch.Latch,
2405 shutdownFlag: MutableRef.MutableRef<boolean>,
2406 strategy: PubSub.Strategy<A>,
2407 replayWindow: PubSub.ReplayWindow<A>,
2408 ended: MutableRef.MutableRef<Option.Option<A>>
2409 ) {
2410 this.pubsub = pubsub
2411 this.subscribers = subscribers
2412 this.subscription = subscription
2413 this.pollers = pollers
2414 this.shutdownHook = shutdownHook
2415 this.shutdownFlag = shutdownFlag
2416 this.strategy = strategy
2417 this.replayWindow = replayWindow
2418 this.ended = ended
2419 }
2420
2421 pipe() {
2422 return pipeArguments(this, arguments)
2423 }
2424}
2425
2426class PubSubImpl<in out A> implements PubSub<A> {
2427 readonly [TypeId] = {
2428 _A: identity
2429 }
2430
2431 readonly pubsub: PubSub.Atomic<A>
2432 readonly subscribers: PubSub.Subscribers<A>
2433 readonly scope: Scope.Closeable
2434 readonly shutdownHook: Latch.Latch
2435 readonly shutdownFlag: MutableRef.MutableRef<boolean>
2436 readonly strategy: PubSub.Strategy<A>
2437 readonly ended: MutableRef.MutableRef<Option.Option<A>>
2438
2439 constructor(
2440 pubsub: PubSub.Atomic<A>,
2441 subscribers: PubSub.Subscribers<A>,
2442 scope: Scope.Closeable,
2443 shutdownHook: Latch.Latch,
2444 shutdownFlag: MutableRef.MutableRef<boolean>,
2445 strategy: PubSub.Strategy<A>,
2446 ended: MutableRef.MutableRef<Option.Option<A>>
2447 ) {
2448 this.pubsub = pubsub
2449 this.subscribers = subscribers
2450 this.scope = scope
2451 this.shutdownHook = shutdownHook
2452 this.shutdownFlag = shutdownFlag
2453 this.strategy = strategy
2454 this.ended = ended
2455 }
2456
2457 pipe() {
2458 return pipeArguments(this, arguments)
2459 }
2460}
2461
2462const makePubSubUnsafe = <A>(
2463 pubsub: PubSub.Atomic<A>,
2464 subscribers: PubSub.Subscribers<A>,
2465 scope: Scope.Closeable,
2466 shutdownHook: Latch.Latch,
2467 shutdownFlag: MutableRef.MutableRef<boolean>,
2468 strategy: PubSub.Strategy<A>,
2469 ended: MutableRef.MutableRef<Option.Option<A>>
2470): PubSub<A> => new PubSubImpl(pubsub, subscribers, scope, shutdownHook, shutdownFlag, strategy, ended)
2471
2472const ensureCapacity = (capacity: number): void => {
2473 if (capacity <= 0) {
2474 throw new Error(`Cannot construct PubSub with capacity of ${capacity}`)
2475 }
2476}
2477
2478// -----------------------------------------------------------------------------
2479// PubSub.Strategy
2480// -----------------------------------------------------------------------------
2481
2482/**
2483 * Represents the back-pressure strategy for bounded `PubSub` values.
2484 *
2485 * **When to use**
2486 *
2487 * Use to preserve every message for current subscribers when a bounded custom
2488 * `PubSub` should make publishers wait for capacity instead of dropping or
2489 * evicting messages.
2490 *
2491 * **Details**
2492 *
2493 * Publishers wait when the `PubSub` is at capacity, so all current subscribers
2494 * can receive every published message.
2495 *
2496 * **Gotchas**
2497 *
2498 * A slow subscriber can slow down publishers and other subscribers.
2499 *
2500 * @see {@link bounded} for creating bounded PubSubs with back pressure by default
2501 * @see {@link DroppingStrategy} for dropping new messages when capacity is full
2502 * @see {@link SlidingStrategy} for evicting old messages when capacity is full
2503 *
2504 * @category models
2505 * @since 4.0.0
2506 */
2507export class BackPressureStrategy<in out A> implements PubSub.Strategy<A> {
2508 publishers: MutableList.MutableList<
2509 readonly [A, Deferred.Deferred<boolean>, boolean]
2510 > = MutableList.make()
2511
2512 get shutdown(): Effect.Effect<void> {
2513 return Effect.withFiber((fiber) =>
2514 Effect.forEach(
2515 MutableList.takeAll(this.publishers),
2516 ([_, deferred, last]) => last ? Deferred.interruptWith(deferred, fiber.id) : Effect.void,
2517 { concurrency: "unbounded", discard: true }
2518 )
2519 )
2520 }
2521
2522 handleSurplus(
2523 pubsub: PubSub.Atomic<A>,
2524 subscribers: PubSub.Subscribers<A>,
2525 elements: Iterable<A>,
2526 isShutdown: MutableRef.MutableRef<boolean>
2527 ): Effect.Effect<boolean> {
2528 return Effect.callback<boolean>((resume) => {
2529 const deferred = Deferred.makeUnsafe<boolean>()
2530 this.offerUnsafe(elements, deferred)
2531 this.onPubSubEmptySpaceUnsafe(pubsub, subscribers)
2532 this.completeSubscribersUnsafe(pubsub, subscribers)
2533 if (MutableRef.get(isShutdown)) {
2534 this.removeUnsafe(deferred)
2535 return resume(Effect.interrupt)
2536 }
2537 if (deferred.effect) return resume(deferred.effect)
2538 deferred.resumes = [resume]
2539 return Effect.sync(() => this.removeUnsafe(deferred))
2540 })
2541 }
2542
2543 onPubSubEmptySpaceUnsafe(
2544 pubsub: PubSub.Atomic<A>,
2545 subscribers: PubSub.Subscribers<A>
2546 ): void {
2547 let keepPolling = true
2548 while (keepPolling && !pubsub.isFull()) {
2549 const publisher = MutableList.take(this.publishers)
2550 if (publisher === MutableList.Empty) {
2551 keepPolling = false
2552 } else {
2553 const [value, deferred] = publisher
2554 const published = pubsub.publish(value)
2555 if (published && publisher[2]) {
2556 Deferred.doneUnsafe(deferred, Exit.succeed(true))
2557 } else if (!published) {
2558 MutableList.prepend(this.publishers, publisher)
2559 }
2560 this.completeSubscribersUnsafe(pubsub, subscribers)
2561 }
2562 }
2563 }
2564
2565 completePollersUnsafe(
2566 pubsub: PubSub.Atomic<A>,
2567 subscribers: PubSub.Subscribers<A>,
2568 subscription: PubSub.BackingSubscription<A>,
2569 pollers: MutableList.MutableList<Deferred.Deferred<A>>
2570 ): void {
2571 return strategyCompletePollersUnsafe(this, pubsub, subscribers, subscription, pollers)
2572 }
2573
2574 completeSubscribersUnsafe(pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>): void {
2575 return strategyCompleteSubscribersUnsafe(this, pubsub, subscribers)
2576 }
2577
2578 private offerUnsafe(elements: Iterable<A>, deferred: Deferred.Deferred<boolean>): void {
2579 const iterator = elements[Symbol.iterator]()
2580 let next: IteratorResult<A> = iterator.next()
2581 if (!next.done) {
2582 // oxlint-disable-next-line no-constant-condition
2583 while (1) {
2584 const value = next.value
2585 next = iterator.next()
2586 if (next.done) {
2587 MutableList.append(this.publishers, [value, deferred, true])
2588 break
2589 }
2590 MutableList.append(this.publishers, [value, deferred, false])
2591 }
2592 }
2593 }
2594
2595 removeUnsafe(deferred: Deferred.Deferred<boolean>): void {
2596 MutableList.filter(this.publishers, ([_, d]) => d !== deferred)
2597 }
2598}
2599
2600/**
2601 * Represents the dropping strategy for bounded `PubSub` values.
2602 *
2603 * **When to use**
2604 *
2605 * Use to keep publishers fast by dropping new messages when the `PubSub` is at
2606 * capacity.
2607 *
2608 * **Details**
2609 *
2610 * A publish that arrives while the `PubSub` is full is dropped instead of
2611 * waiting for capacity.
2612 *
2613 * **Gotchas**
2614 *
2615 * Subscribers may miss messages published while they are subscribed.
2616 *
2617 * **Example** (Applying a dropping strategy)
2618 *
2619 * ```ts import.meta.vitest
2620 * import { Effect, PubSub } from "effect"
2621 *
2622 * const program = Effect.scoped(Effect.gen(function*() {
2623 * // Explicitly create a PubSub with a dropping strategy
2624 * const pubsub = yield* PubSub.make<string>({
2625 * atomicPubSub: () => PubSub.makeAtomicBounded(2),
2626 * strategy: () => new PubSub.DroppingStrategy()
2627 * })
2628 *
2629 * const subscription = yield* PubSub.subscribe(pubsub)
2630 *
2631 * // Fill the PubSub
2632 * const pub1 = yield* PubSub.publish(pubsub, "msg1") // true
2633 * const pub2 = yield* PubSub.publish(pubsub, "msg2") // true
2634 * const pub3 = yield* PubSub.publish(pubsub, "msg3") // false (dropped)
2635 *
2636 * // Subscribers will only see the first two messages
2637 * const messages = yield* PubSub.takeAll(subscription)
2638 * return { published: [pub1, pub2, pub3], messages }
2639 * }))
2640 *
2641 * const actual = await Effect.runPromise(program)
2642 * actual // => { published: [true, true, false], messages: ["msg1", "msg2"] }
2643 * ```
2644 *
2645 * @category models
2646 * @since 4.0.0
2647 */
2648export class DroppingStrategy<in out A> implements PubSub.Strategy<A> {
2649 get shutdown(): Effect.Effect<void> {
2650 return Effect.void
2651 }
2652
2653 handleSurplus(
2654 _pubsub: PubSub.Atomic<A>,
2655 _subscribers: PubSub.Subscribers<A>,
2656 _elements: Iterable<A>,
2657 _isShutdown: MutableRef.MutableRef<boolean>
2658 ): Effect.Effect<boolean> {
2659 return Effect.succeed(false)
2660 }
2661
2662 onPubSubEmptySpaceUnsafe(
2663 _pubsub: PubSub.Atomic<A>,
2664 _subscribers: PubSub.Subscribers<A>
2665 ): void {
2666 //
2667 }
2668
2669 completePollersUnsafe(
2670 pubsub: PubSub.Atomic<A>,
2671 subscribers: PubSub.Subscribers<A>,
2672 subscription: PubSub.BackingSubscription<A>,
2673 pollers: MutableList.MutableList<Deferred.Deferred<A>>
2674 ): void {
2675 return strategyCompletePollersUnsafe(this, pubsub, subscribers, subscription, pollers)
2676 }
2677
2678 completeSubscribersUnsafe(pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>): void {
2679 return strategyCompleteSubscribersUnsafe(this, pubsub, subscribers)
2680 }
2681}
2682
2683/**
2684 * Represents the sliding strategy for bounded `PubSub` values.
2685 *
2686 * **When to use**
2687 *
2688 * Use to keep the most recent messages when the `PubSub` is at capacity.
2689 *
2690 * **Details**
2691 *
2692 * New messages are accepted by evicting older messages from the bounded
2693 * `PubSub`.
2694 *
2695 * **Gotchas**
2696 *
2697 * Slow subscribers may miss older messages that are evicted before they are
2698 * consumed.
2699 *
2700 * **Example** (Applying a sliding strategy)
2701 *
2702 * ```ts import.meta.vitest
2703 * import { Effect, PubSub } from "effect"
2704 *
2705 * const program = Effect.scoped(Effect.gen(function*() {
2706 * // Explicitly create a PubSub with a sliding strategy
2707 * const pubsub = yield* PubSub.make<string>({
2708 * atomicPubSub: () => PubSub.makeAtomicBounded(2),
2709 * strategy: () => new PubSub.SlidingStrategy()
2710 * })
2711 *
2712 * const subscription = yield* PubSub.subscribe(pubsub)
2713 *
2714 * // Publish messages that exceed capacity
2715 * yield* PubSub.publish(pubsub, "msg1") // stored
2716 * yield* PubSub.publish(pubsub, "msg2") // stored
2717 * yield* PubSub.publish(pubsub, "msg3") // "msg1" evicted, "msg3" stored
2718 * yield* PubSub.publish(pubsub, "msg4") // "msg2" evicted, "msg4" stored
2719 *
2720 * // Subscribers will see the most recent messages
2721 * return yield* PubSub.takeAll(subscription)
2722 * }))
2723 *
2724 * const actual = await Effect.runPromise(program)
2725 * actual // => ["msg3", "msg4"]
2726 * ```
2727 *
2728 * @category models
2729 * @since 4.0.0
2730 */
2731export class SlidingStrategy<in out A> implements PubSub.Strategy<A> {
2732 get shutdown(): Effect.Effect<void> {
2733 return Effect.void
2734 }
2735
2736 handleSurplus(
2737 pubsub: PubSub.Atomic<A>,
2738 subscribers: PubSub.Subscribers<A>,
2739 elements: Iterable<A>,
2740 _isShutdown: MutableRef.MutableRef<boolean>
2741 ): Effect.Effect<boolean> {
2742 return Effect.sync(() => {
2743 this.slidingPublishUnsafe(pubsub, elements)
2744 this.completeSubscribersUnsafe(pubsub, subscribers)
2745 return true
2746 })
2747 }
2748
2749 onPubSubEmptySpaceUnsafe(
2750 _pubsub: PubSub.Atomic<A>,
2751 _subscribers: PubSub.Subscribers<A>
2752 ): void {
2753 //
2754 }
2755
2756 completePollersUnsafe(
2757 pubsub: PubSub.Atomic<A>,
2758 subscribers: PubSub.Subscribers<A>,
2759 subscription: PubSub.BackingSubscription<A>,
2760 pollers: MutableList.MutableList<Deferred.Deferred<A>>
2761 ): void {
2762 return strategyCompletePollersUnsafe(this, pubsub, subscribers, subscription, pollers)
2763 }
2764
2765 completeSubscribersUnsafe(pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>): void {
2766 return strategyCompleteSubscribersUnsafe(this, pubsub, subscribers)
2767 }
2768
2769 slidingPublishUnsafe(pubsub: PubSub.Atomic<A>, elements: Iterable<A>): void {
2770 const it = elements[Symbol.iterator]()
2771 let next = it.next()
2772 if (!next.done && pubsub.capacity > 0) {
2773 let a = next.value
2774 let loop = true
2775 while (loop) {
2776 pubsub.slide()
2777 const pub = pubsub.publish(a)
2778 if (pub && (next = it.next()) && !next.done) {
2779 a = next.value
2780 } else if (pub) {
2781 loop = false
2782 }
2783 }
2784 }
2785 }
2786}
2787
2788const strategyCompletePollersUnsafe = <A>(
2789 strategy: PubSub.Strategy<A>,
2790 pubsub: PubSub.Atomic<A>,
2791 subscribers: PubSub.Subscribers<A>,
2792 subscription: PubSub.BackingSubscription<A>,
2793 pollers: MutableList.MutableList<Deferred.Deferred<A>>
2794): void => {
2795 let keepPolling = true
2796 while (keepPolling && !subscription.isEmpty()) {
2797 const poller = MutableList.take(pollers)
2798 if (poller === MutableList.Empty) {
2799 removeSubscribers(subscribers, subscription, pollers)
2800 if (pollers.length === 0) {
2801 keepPolling = false
2802 } else {
2803 addSubscribers(subscribers, subscription, pollers)
2804 }
2805 } else {
2806 const pollResult = subscription.poll()
2807 if (pollResult === MutableList.Empty) {
2808 MutableList.prepend(pollers, poller)
2809 } else {
2810 Deferred.doneUnsafe(poller, Exit.succeed(pollResult))
2811 strategy.onPubSubEmptySpaceUnsafe(pubsub, subscribers)
2812 }
2813 }
2814 }
2815}
2816
2817const strategyCompleteSubscribersUnsafe = <A>(
2818 strategy: PubSub.Strategy<A>,
2819 pubsub: PubSub.Atomic<A>,
2820 subscribers: PubSub.Subscribers<A>
2821): void => {
2822 for (
2823 const [subscription, pollersSet] of subscribers
2824 ) {
2825 for (const pollers of pollersSet) {
2826 strategy.completePollersUnsafe(pubsub, subscribers, subscription, pollers)
2827 }
2828 }
2829}
2830
2831interface ReplayNode<A> {
2832 value: A | AbsentValue
2833 index: number
2834 next: ReplayNode<A> | null
2835}
2836
2837class ReplayBuffer<A> {
2838 readonly capacity: number
2839 head: ReplayNode<A> = { value: AbsentValue, index: 0, next: null }
2840 tail: ReplayNode<A> = this.head
2841 readonly slideValues: Array<{
2842 readonly value: A
2843 readonly index: number
2844 }> = []
2845 size = 0
2846 index = 0
2847 publisherIndex = 0
2848
2849 constructor(capacity: number) {
2850 this.capacity = capacity
2851 }
2852
2853 slide(value: A, publisherIndex: number): void {
2854 this.slideValues[this.index % this.capacity] = {
2855 value,
2856 index: publisherIndex
2857 }
2858 this.index++
2859 }
2860 offer(a: A): number {
2861 const index = this.publisherIndex++
2862 this.tail.value = a
2863 this.tail.index = index
2864 this.tail.next = {
2865 value: AbsentValue,
2866 index: 0,
2867 next: null
2868 }
2869 this.tail = this.tail.next
2870 if (this.size === this.capacity) {
2871 this.head = this.head.next!
2872 } else {
2873 this.size += 1
2874 }
2875 return index
2876 }
2877 offerAll(as: Iterable<A>): void {
2878 for (const a of as) {
2879 this.offer(a)
2880 }
2881 }
2882}
2883
2884class ReplayWindowImpl<A> implements PubSub.ReplayWindow<A> {
2885 readonly buffer: ReplayBuffer<A>
2886 readonly values: Array<A>
2887 index = 0
2888 remaining: number
2889 slideIndex: number
2890 newestIndex = -1
2891
2892 constructor(buffer: ReplayBuffer<A>) {
2893 this.buffer = buffer
2894 this.remaining = buffer.size
2895 this.slideIndex = buffer.index
2896 this.values = new Array(this.remaining)
2897 let node = buffer.head
2898 for (let i = 0; i < this.remaining; i++) {
2899 this.values[i] = node.value as A
2900 this.newestIndex = node.index
2901 node = node.next!
2902 }
2903 }
2904 close(): void {
2905 this.values.length = 0
2906 this.remaining = 0
2907 }
2908 sync(): void {
2909 const slides = this.buffer.index - this.slideIndex
2910 if (slides === 0 || this.remaining === 0) {
2911 return
2912 }
2913 const count = Math.min(slides, this.buffer.capacity)
2914 const start = this.buffer.index - count
2915 for (let i = 0; i < count; i++) {
2916 const entry = this.buffer.slideValues[(start + i) % this.buffer.capacity]
2917 if (entry.index > this.newestIndex) {
2918 this.index = (this.index + 1) % this.values.length
2919 this.values[(this.index + this.remaining - 1) % this.values.length] = entry.value
2920 this.newestIndex = entry.index
2921 }
2922 }
2923 this.slideIndex = this.buffer.index
2924 }
2925 take(): A | undefined {
2926 if (this.remaining === 0) {
2927 return undefined
2928 }
2929 this.sync()
2930 const value = this.values[this.index]
2931 this.values[this.index] = AbsentValue as unknown as A
2932 this.index = (this.index + 1) % this.values.length
2933 this.remaining--
2934 if (this.remaining === 0) {
2935 this.close()
2936 }
2937 return value as A
2938 }
2939 takeN(n: number): Array<A> {
2940 n = Count.normalize(n)
2941 const len = Math.min(n, this.remaining)
2942 const items = new Array(len)
2943 for (let i = 0; i < len; i++) {
2944 items[i] = this.take()!
2945 }
2946 return items
2947 }
2948 takeAll(): Array<A> {
2949 return this.takeN(this.remaining)
2950 }
2951}
2952
2953const emptyReplayWindow: PubSub.ReplayWindow<never> = {
2954 remaining: 0,
2955 take: () => undefined,
2956 takeN: () => [],
2957 takeAll: () => [],
2958 close: () => void 0
2959}