TxPubSub
Broadcasts values to subscribers inside Effect transactions.
A TxPubSub<A> is a transactional publish/subscribe hub. Each subscriber owns a TxQueue, and each published value is offered to the subscriber queues that are registered at the time of publication. This module includes bounded, dropping, sliding, and unbounded hubs, publishing helpers, scoped subscriptions, shutdown operations, and a guard.
Constructors
Signature
declare function bounded<A = never>(capacity: number): Effect<TxPubSub<A>>;Creates a dropping TxPubSub with the specified capacity. When a subscriber's queue is full, the message is dropped for that subscriber.
Signature
declare function dropping<A = never>(capacity: number): Effect<TxPubSub<A>>;Creates a sliding TxPubSub with the specified capacity. When a subscriber's queue is full, the oldest message in that subscriber's queue is dropped.
Signature
declare function sliding<A = never>(capacity: number): Effect<TxPubSub<A>>;Creates an unbounded TxPubSub with unlimited capacity. Messages are always accepted.
Signature
declare function unbounded<A = never>(): Effect<TxPubSub<A>>;Getters
Returns the capacity of the TxPubSub.
Signature
declare function capacity<A>(self: TxPubSub<A>): number;Returns the current number of messages across all subscriber queues (the max).
Signature
declare function size<A>(self: TxPubSub<A>): Effect<number>;Guards
isTxPubSub
Checks whether the given value is a TxPubSub.
Signature
declare function isTxPubSub(u: unknown): u is TxPubSub<unknown>;Models
A TxPubSub represents a transactional publish/subscribe hub that broadcasts messages to all current subscribers using Software Transactional Memory (STM) semantics.
Signature
interface TxPubSub<in out A> extends Inspectable, Pipeable {
readonly "~effect/transactions/TxPubSub": "~effect/transactions/TxPubSub";
readonly capacity: number;
readonly strategy: "sliding" | "dropping" | "unbounded" | "bounded";
}Mutations
acquireSubscriber
Creates a subscriber queue and registers it with the pub/sub.
When to use
Use to create and register a subscriber queue inside a larger transaction when registration must be atomic with other Tx operations.
Details
This is the transactional acquire step of subscribe, exposed so that callers can compose it with other Tx operations in a single transaction, such as TxSubscriptionRef.changes.
See
subscribefor the scoped acquire and release wrapper when no custom transaction composition is neededreleaseSubscriberto remove and shut down a queue returned byacquireSubscriber
Signature
declare function acquireSubscriber<A>(
self: TxPubSub<A>,
): Effect<TxQueue<A, never>, never, Transaction>;awaitShutdown
Waits for the TxPubSub to be shut down.
Signature
declare function awaitShutdown<A>(self: TxPubSub<A>): Effect<void>;Publishes a message to all current subscribers.
Details
Returns true if the message was delivered to all subscribers, or false if the hub is shut down or the message was dropped for any subscriber. For the bounded strategy, the transaction retries if any subscriber queue is full. For the sliding strategy, full subscriber queues drop their oldest messages. For the dropping strategy, full subscriber queues drop the new message and the operation returns false.
Signature
declare const publish: {
<A>(value: A): (self: TxPubSub<A>) => Effect<boolean>;
<A>(self: TxPubSub<A>, value: A): Effect<boolean>;
};publishAll
Publishes all messages from an iterable to all current subscribers.
Details
Returns true if all messages were delivered to all subscribers.
Signature
declare const publishAll: {
<A>(values: Iterable<A>): (self: TxPubSub<A>) => Effect<boolean>;
<A>(self: TxPubSub<A>, values: Iterable<A>): Effect<boolean>;
};releaseSubscriber
Removes a subscriber queue from the pub/sub and shuts it down.
When to use
Use to release a manually acquired subscriber queue inside a larger transaction, removing it from the pub/sub and shutting it down together with related transactional cleanup.
Details
This is the transactional release step of subscribe, exposed so that callers can compose it with other Tx operations in a single transaction.
Gotchas
The supplied queue is shut down after being removed, so callers should pass a queue acquired for this pub/sub.
See
acquireSubscriberfor the matching transactional acquire stepsubscribefor the scoped acquire and release wrapper
Signature
declare const releaseSubscriber: {
<A>(queue: TxQueue<A>): (self: TxPubSub<A>) => Effect<void, never, Transaction>;
<A>(self: TxPubSub<A>, queue: TxQueue<A>): Effect<void, never, Transaction>;
};Shuts down the TxPubSub and all subscriber queues registered at the time of shutdown.
Details
After shutdown, publish and publishAll return false, and awaitShutdown completes. The operation is idempotent.
Gotchas
Subscribers acquired after shutdown are not automatically shut down by this call.
Signature
declare function shutdown<A>(self: TxPubSub<A>): Effect<void>;Subscribes to the TxPubSub, returning a scoped TxQueue for messages published after subscription.
Details
The returned queue uses the hub's capacity strategy: bounded subscriptions backpressure publishers when full, dropping subscriptions may miss new messages when full, and sliding subscriptions may evict older queued messages. The subscription is automatically removed when the scope is closed.
Signature
declare function subscribe<A>(self: TxPubSub<A>): Effect<TxQueue<A, never>, never, Scope>;Predicates
Checks whether the TxPubSub has no pending messages (all subscriber queues are empty).
Signature
declare function isEmpty<A>(self: TxPubSub<A>): Effect<boolean>;Checks whether any subscriber queue is at capacity.
Signature
declare function isFull<A>(self: TxPubSub<A>): Effect<boolean>;isShutdown
Checks whether the TxPubSub has been shut down.
Signature
declare function isShutdown<A>(self: TxPubSub<A>): Effect<boolean>;
Creates a bounded TxPubSub with the specified capacity. When a subscriber's queue is full, the publisher will retry the transaction until space is available.