<A>(self: TxPubSub<A>): Effect.Effect<
TxQueue.TxQueue<A>,
never,
Scope.Scope
>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.
Example (Subscribing multiple queues)
import { Effect, TxPubSub, TxQueue } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.unbounded<string>()
yield* Effect.scoped(
Effect.gen(function*() {
const sub1 = yield* TxPubSub.subscribe(hub)
const sub2 = yield* TxPubSub.subscribe(hub)
yield* TxPubSub.publish(hub, "broadcast")
const msg1 = yield* TxQueue.take(sub1)
const msg2 = yield* TxQueue.take(sub2)
console.log(msg1, msg2) // "broadcast" "broadcast"
})
)
})export const const subscribe: <A>(
self: TxPubSub<A>
) => Effect.Effect<
TxQueue.TxQueue<A>,
never,
Scope.Scope
>
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.
Example (Subscribing multiple queues)
import { Effect, TxPubSub, TxQueue } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.unbounded<string>()
yield* Effect.scoped(
Effect.gen(function*() {
const sub1 = yield* TxPubSub.subscribe(hub)
const sub2 = yield* TxPubSub.subscribe(hub)
yield* TxPubSub.publish(hub, "broadcast")
const msg1 = yield* TxQueue.take(sub1)
const msg2 = yield* TxQueue.take(sub2)
console.log(msg1, msg2) // "broadcast" "broadcast"
})
)
})
subscribe = <function (type parameter) A in <A>(self: TxPubSub<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope>A>(self: TxPubSub<A>(parameter) self: {
subscribersRef: TxRef.TxRef<Array<TxQueue.TxQueue<A>>>;
shutdownRef: TxRef.TxRef<boolean>;
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
toString: () => string;
toJSON: () => unknown;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
interface TxPubSub<in out A>A TxPubSub represents a transactional publish/subscribe hub that broadcasts messages
to all current subscribers using Software Transactional Memory (STM) semantics.
Example (Subscribing to a transactional pub/sub)
import { Effect, TxPubSub, TxQueue } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.unbounded<string>()
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxPubSub.subscribe(hub)
yield* TxPubSub.publish(hub, "hello")
const msg = yield* TxQueue.take(sub)
console.log(msg) // "hello"
})
)
})
TxPubSub<function (type parameter) A in <A>(self: TxPubSub<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope>A>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<import TxQueueTxQueue.interface TxQueue<in out A, in out E = never>Namespace containing type definitions for TxQueue variance annotations.
A TxQueue represents a transactional queue data structure that provides both
enqueue and dequeue operations with Software Transactional Memory (STM) semantics.
Example (Combining enqueue and dequeue operations)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
// Create a bounded transactional queue (E defaults to never)
const queue = yield* TxQueue.bounded<number>(10)
// Single operations - automatically transactional
const accepted = yield* TxQueue.offer(queue, 42)
const item = yield* TxQueue.take(queue) // Effect<number, never>
console.log(item) // 42
// Queue with error channel
const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
// Operations can handle queue-level failures
yield* TxQueue.fail(faultTolerantQueue, "queue failed")
const result = yield* Effect.flip(TxQueue.take(faultTolerantQueue))
console.log(result) // "queue failed"
})
TxQueue<function (type parameter) A in <A>(self: TxPubSub<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope>A>, never, import ScopeScope.type Scope.Scope = /*unresolved*/ anyScope> =>
import EffectEffect.acquireRelease(
import EffectEffect.tx(const acquireSubscriber: <A>(
self: TxPubSub<A>
) => Effect.Effect<
TxQueue.TxQueue<A>,
never,
Effect.Transaction
>
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.
acquireSubscriber(self: TxPubSub<A>(parameter) self: {
subscribersRef: TxRef.TxRef<Array<TxQueue.TxQueue<A>>>;
shutdownRef: TxRef.TxRef<boolean>;
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
toString: () => string;
toJSON: () => unknown;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
queue: TxQueue.TxQueue<A, never>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
import EffectEffect.tx(const releaseSubscriber: {
<A>(queue: TxQueue.TxQueue<A>): (
self: TxPubSub<A>
) => Effect.Effect<
void,
never,
Effect.Transaction
>
<A>(
self: TxPubSub<A>,
queue: TxQueue.TxQueue<A>
): Effect.Effect<
void,
never,
Effect.Transaction
>
}
self: TxPubSub<A>(parameter) self: {
subscribersRef: TxRef.TxRef<Array<TxQueue.TxQueue<A>>>;
shutdownRef: TxRef.TxRef<boolean>;
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
toString: () => string;
toJSON: () => unknown;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
queue: TxQueue.TxQueue<A, never>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}