<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.
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.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<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.class Transactionclass Transaction {
key: Identifier;
Service: {
retry: boolean;
journal: Map<TxRef<any>, { readonly version: number; value: any }>;
};
}
Service that holds the current transaction state.
Details
It includes a journal that stores non-committed changes to TxRef values and
a retry flag that records whether the transaction should be retried.
Example (Building transactions)
import { Effect } from "effect"
// Transaction class for software transactional memory operations
const txEffect = Effect.gen(function*() {
const tx = yield* Effect.Transaction
// Use transaction for coordinated state changes
return "Transaction complete"
})
Transaction> =>
import EffectEffect.const gen: {
<Eff extends Effect<any, any, any>, AEff>(
f: () => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
<Self, Eff extends Effect<any, any, any>, AEff>(
options: { readonly self: Self },
f: (this: Self) => Generator<Eff, AEff, never>
): Effect<
AEff,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer E, infer _R>
]
? E
: never,
[Eff] extends [never]
? never
: [Eff] extends [
Effect<infer _A, infer _E, infer R>
]
? R
: never
>
}
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;
}