PubSub
Broadcasts values from publishers to many subscribers.
Publishers add messages with publish or publishAll, and each active Subscription receives its own copy of every accepted message. Unlike a queue, subscribers do not compete for messages. This module includes bounded, dropping, sliding, and unbounded hubs, optional replay buffers for late subscribers, message-taking helpers, capacity and shutdown operations, and low-level types for custom hub strategies.
Constructors
Signature
declare function bounded<A>(
capacity:
| number
| {
readonly capacity: number;
readonly replay?: number;
},
): Effect<PubSub<A>>;Creates a bounded PubSub with the dropping strategy. The PubSub will drop new messages if the PubSub is at capacity.
Details
For best performance use capacities that are powers of two.
Signature
declare function dropping<A>(
capacity:
| number
| {
readonly capacity: number;
readonly replay?: number;
},
): Effect<PubSub<A>>;Creates a PubSub with a custom atomic implementation and strategy.
Signature
declare function make<A>(options: {
readonly atomicPubSub: LazyArg<Atomic<A>>;
readonly strategy: LazyArg<Strategy<A>>;
}): Effect<PubSub<A>>;makeAtomicBounded
Creates a bounded atomic PubSub implementation with optional replay buffer.
When to use
Use to provide bounded message storage when building a custom PubSub with make and an explicit delivery strategy.
Details
Pass either a capacity number or an options object with capacity and optional replay. A positive replay value enables a replay buffer for late subscribers, and fractional replay sizes are rounded up.
Gotchas
The capacity must be greater than zero; invalid capacities throw synchronously before an atomic implementation is created.
See
makefor constructing aPubSubfrom an atomic implementation and delivery strategymakeAtomicUnboundedfor an atomic implementation without a bounded capacityboundedfor the higher-level backpressure constructordroppingfor the higher-level dropping constructorslidingfor the higher-level sliding constructor
Signature
declare function makeAtomicBounded<A>(
capacity:
| number
| {
readonly capacity: number;
readonly replay?: number;
},
): Atomic<A>;makeAtomicUnbounded
Creates an unbounded atomic PubSub implementation with optional replay buffer.
When to use
Use to create the low-level storage layer for a custom PubSub whose active subscribers may retain an unbounded number of pending messages.
Gotchas
Messages published while subscribers are active can be retained without a capacity limit until those subscribers take them or unsubscribe.
See
makeAtomicBoundedfor a bounded atomic implementation that enforces capacitymakefor wrapping an atomic implementation with a delivery strategyunboundedfor the high-level effectful constructor for unboundedPubSubvalues
Signature
declare function makeAtomicUnbounded<A>(options?: { readonly replay?: number }): Atomic<A>;Creates a bounded PubSub with the sliding strategy. The PubSub will add new messages and drop old messages if the PubSub is at capacity.
Details
For best performance use capacities that are powers of two.
Signature
declare function sliding<A>(
capacity:
| number
| {
readonly capacity: number;
readonly replay?: number;
},
): Effect<PubSub<A>>;Creates an unbounded PubSub.
Signature
declare function unbounded<A>(options?: { readonly replay?: number }): Effect<PubSub<A>>;Getters
Returns the number of elements the queue can hold.
Signature
declare function capacity<A>(self: PubSub<A>): number;Returns the number of messages currently available in the subscription as an Effect.
When to use
Use when checking a subscription from effectful code and shutdown should interrupt the effect.
Details
The count includes replay-buffered messages. If the subscription has been shut down, the effect interrupts.
See
remainingUnsafefor a synchronous check that reports shutdown asOption.none()
Signature
declare function remaining<A>(self: Subscription<A>): Effect<number>;remainingUnsafe
Synchronously returns the number of messages currently available in the subscription, or Option.none() when it is shut down.
When to use
Use when you need synchronous polling outside a managed workflow and want shutdown observed as data instead of interruption.
See
remainingfor the effectful variant that interrupts on shutdown
Signature
declare function remainingUnsafe<A>(self: Subscription<A>): Option<number>;Returns the current number of messages retained by the PubSub for active subscribers.
Details
If the PubSub has been shut down, the returned effect succeeds with 0. The size is not a count of waiting subscribers or suspended publishers.
Signature
declare function size<A>(self: PubSub<A>): Effect<number>;sizeUnsafe
Returns the current number of messages retained by the PubSub for active subscribers synchronously.
When to use
Use when an immediate PubSub size snapshot is needed outside effectful code and concurrent changes between the check and later use are acceptable.
Details
Returns 0 after shutdown. Because this is an unsafe synchronous snapshot, prefer size in effectful code.
Signature
declare function sizeUnsafe<A>(self: PubSub<A>): number;Lifecycle
awaitShutdown
Waits until the queue is shutdown. The Effect returned by this method will not resume until the queue has been shutdown. If the queue is already shutdown, the Effect will resume right away.
Signature
declare function awaitShutdown<A>(self: PubSub<A>): Effect<void>;Shuts down the PubSub, interrupting suspended publishers and subscribers and finalizing active subscriptions.
Details
After shutdown, publish and publishAll succeed with false, publishUnsafe returns false, and subscription operations such as take interrupt.
Signature
declare function shutdown<A>(self: PubSub<A>): Effect<void>;Models
BackPressureStrategy
Represents the back-pressure strategy for bounded PubSub values.
When to use
Use to preserve every message for current subscribers when a bounded custom PubSub should make publishers wait for capacity instead of dropping or evicting messages.
Details
Publishers wait when the PubSub is at capacity, so all current subscribers can receive every published message.
Gotchas
A slow subscriber can slow down publishers and other subscribers.
See
boundedfor creating bounded PubSubs with back pressure by defaultDroppingStrategyfor dropping new messages when capacity is fullSlidingStrategyfor evicting old messages when capacity is full
Signature
declare class BackPressureStrategy<in out A> implements Strategy<A> {
constructor<in out A>();
publishers: MutableList<readonly [A, Deferred<boolean, never>, boolean]>;
shutdown: Effect<void>;
completePollersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>, subscription: BackingSubscription<A>, pollers: MutableList<Deferred<A, never>>): void;
completeSubscribersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
handleSurplus(pubsub: Atomic<A>, subscribers: Subscribers<A>, elements: Iterable<A>, isShutdown: MutableRef<boolean>): Effect<boolean>;
onPubSubEmptySpaceUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
removeUnsafe(deferred: Deferred<boolean>): void;
}DroppingStrategy
Represents the dropping strategy for bounded PubSub values.
When to use
Use to keep publishers fast by dropping new messages when the PubSub is at capacity.
Details
A publish that arrives while the PubSub is full is dropped instead of waiting for capacity.
Gotchas
Subscribers may miss messages published while they are subscribed.
Signature
declare class DroppingStrategy<in out A> implements Strategy<A> {
constructor<in out A>();
shutdown: Effect<void>;
completePollersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>, subscription: BackingSubscription<A>, pollers: MutableList<Deferred<A, never>>): void;
completeSubscribersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
handleSurplus(_pubsub: Atomic<A>, _subscribers: Subscribers<A>, _elements: Iterable<A>, _isShutdown: MutableRef<boolean>): Effect<boolean>;
onPubSubEmptySpaceUnsafe(_pubsub: Atomic<A>, _subscribers: Subscribers<A>): void;
}A PubSub<A> is an asynchronous message hub into which publishers can publish messages of type A and subscribers can subscribe to take messages of type A.
Signature
interface PubSub<in out A> extends Pipeable {
readonly "~effect/PubSub": {
readonly _A: Invariant<A>;
};
readonly pubsub: Atomic<A>;
readonly scope: Closeable;
readonly shutdownFlag: MutableRef<boolean>;
readonly shutdownHook: Latch;
readonly strategy: Strategy<A>;
readonly subscribers: Subscribers<A>;
}SlidingStrategy
Represents the sliding strategy for bounded PubSub values.
When to use
Use to keep the most recent messages when the PubSub is at capacity.
Details
New messages are accepted by evicting older messages from the bounded PubSub.
Gotchas
Slow subscribers may miss older messages that are evicted before they are consumed.
Signature
declare class SlidingStrategy<in out A> implements Strategy<A> {
constructor<in out A>();
shutdown: Effect<void>;
completePollersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>, subscription: BackingSubscription<A>, pollers: MutableList<Deferred<A, never>>): void;
completeSubscribersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
handleSurplus(pubsub: Atomic<A>, subscribers: Subscribers<A>, elements: Iterable<A>, _isShutdown: MutableRef<boolean>): Effect<boolean>;
onPubSubEmptySpaceUnsafe(_pubsub: Atomic<A>, _subscribers: Subscribers<A>): void;
slidingPublishUnsafe(pubsub: Atomic<A>, elements: Iterable<A>): void;
}Subscription interface
A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
Signature
interface Subscription<out A> extends Pipeable {
readonly "~effect/PubSub/Subscription": {
readonly _A: Covariant<A>;
};
readonly pollers: MutableList<Deferred<any, never>>;
readonly pubsub: Atomic<any>;
readonly replayWindow: ReplayWindow<A>;
readonly shutdownFlag: MutableRef<boolean>;
readonly shutdownHook: Latch;
readonly strategy: Strategy<any>;
readonly subscribers: Subscribers<any>;
readonly subscription: BackingSubscription<A>;
}Other
Predicates
Returns true if the Pubsub contains zero elements, false otherwise.
Signature
declare function isEmpty<A>(self: PubSub<A>): Effect<boolean>;Returns true when the PubSub has reached its configured capacity.
Details
For unbounded PubSubs this is normally false.
Signature
declare function isFull<A>(self: PubSub<A>): Effect<boolean>;isShutdown
Checks effectfully whether shutdown has been called, returning true after shutdown and false otherwise.
Signature
declare function isShutdown<A>(self: PubSub<A>): Effect<boolean>;isShutdownUnsafe
Checks synchronously whether shutdown has been called, returning true after shutdown and false otherwise.
When to use
Use when an immediate PubSub shutdown-state snapshot is needed outside effectful code and racing shutdown changes are acceptable.
Signature
declare function isShutdownUnsafe<A>(self: PubSub<A>): boolean;Publishing
Publishes a message to the PubSub as an Effect, returning whether the message was accepted.
When to use
Use when you need to publish from effectful code and let the configured PubSub strategy handle surplus messages.
Details
The effect succeeds with false if the PubSub is shut down. If the message cannot be accepted immediately, the configured strategy decides how surplus messages are handled.
See
publishUnsafefor a synchronous non-blocking attempt that does not run effectful surplus handling
Signature
declare const publish: {
<A>(value: A): (self: PubSub<A>) => Effect<boolean>;
<A>(self: PubSub<A>, value: A): Effect<boolean>;
};publishAll
Publishes all of the specified messages to the PubSub, returning whether they were published to the PubSub.
Signature
declare const publishAll: {
<A>(elements: Iterable<A>): (self: PubSub<A>) => Effect<boolean>;
<A>(self: PubSub<A>, elements: Iterable<A>): Effect<boolean>;
};publishUnsafe
Attempts to publish a message synchronously without applying the PubSub strategy's effectful surplus handling.
When to use
Use when you need a non-blocking synchronous publish attempt where false is an acceptable result when the message cannot be accepted immediately.
Details
Returns false if the PubSub is shut down or the message cannot be accepted immediately, for example when a bounded PubSub is full. Prefer publish when backpressure or sliding behavior should be honored.
See
publishfor effectful publishing that honors the configured surplus strategy
Signature
declare const publishUnsafe: {
<A>(value: A): (self: PubSub<A>) => boolean;
<A>(self: PubSub<A>, value: A): boolean;
};Subscriptions
Subscribes to receive messages from the PubSub. The resulting subscription can be evaluated multiple times within the scope to take a message from the PubSub each time.
Signature
declare function subscribe<A>(self: PubSub<A>): Effect<Subscription<A>, never, Scope>;Takes a single message from the subscription. If no messages are available, this will suspend until a message becomes available.
Signature
declare function take<A>(self: Subscription<A>): Effect<A>;Takes all available messages from the subscription, suspending if no items are available.
Signature
declare function takeAll<A>(self: Subscription<A>): Effect<[A, ...Array<A>]>;takeBetween
Takes between the specified minimum and maximum number of messages from the subscription. Will suspend if the minimum number is not immediately available.
Signature
declare const takeBetween: {
(min: number, max: number): <A>(self: Subscription<A>) => Effect<Array<A>>;
<A>(self: Subscription<A>, min: number, max: number): Effect<Array<A>>;
};Takes up to the specified number of messages from the subscription without suspending.
Signature
declare const takeUpTo: {
(max: number): <A>(self: Subscription<A>) => Effect<Array<A>>;
<A>(self: Subscription<A>, max: number): Effect<Array<A>>;
};
Creates a bounded
PubSubthat applies backpressure when it reaches capacity.Details
Published messages are retained until all current subscribers have taken them. When the capacity is full, publishers suspend until space is available. Pass an options object to configure both
capacityand an optional replay buffer for late subscribers.