<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.
Source effect/TxPubSub.ts:5288 lines
export const 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 = <function (type parameter) A in <A>(self: TxPubSub<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Effect.Transaction>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, Effect.Transaction>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, Effect.Transaction>A>, never, import EffectEffect.type Effect.Transaction = /*unresolved*/ anyTransaction> =>
import EffectEffect.gen(function*() {
const const queue: TxQueue.TxQueue<A, never>const queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
const makeSubscriberQueue: <A>(
strategy:
| "bounded"
| "unbounded"
| "dropping"
| "sliding",
cap: number
) => Effect.Effect<TxQueue.TxQueue<A>>
function (type parameter) A in <A>(self: TxPubSub<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Effect.Transaction>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; <…;
}
TxPubSub<in out A>.strategy: "bounded" | "unbounded" | "dropping" | "sliding"strategy, 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; <…;
}
TxPubSub<in out A>.capacity: numbercapacity)
yield* import TxRefTxRef.const update: {
<A>(f: (current: NoInfer<A>) => A): (
self: TxRef<A>
) => Effect.Effect<void>
<A>(
self: TxRef<A>,
f: (current: A) => A
): Effect.Effect<void>
}
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; <…;
}
TxPubSub<A>.subscribersRef: TxRef.TxRef<Array<TxQueue.TxQueue<A>>>(property) TxPubSub<A>.subscribersRef: {
version: number;
pending: Map<unknown, () => void>;
value: A;
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; <…;
}
subs: Array<TxQueue.TxQueue<A, never>>subs) => [...subs: Array<TxQueue.TxQueue<A, never>>subs, const queue: TxQueue.TxQueue<A, never>const queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
const queue: TxQueue.TxQueue<A, never>const queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
Referenced by 2 symbols