<A>(self: TxSubscriptionRef<A>): Effect.Effect<
TxQueue.TxQueue<A>,
never,
Scope.Scope
>Subscribes to all changes of the TxSubscriptionRef. Returns a scoped TxDequeue that first yields the current value, then every subsequent update.
When to use
Use to subscribe to TxSubscriptionRef committed changes through a scoped
transactional queue.
Example (Subscribing to changes)
import { Effect, TxQueue, TxSubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* TxSubscriptionRef.make(0)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxSubscriptionRef.changes(ref)
const initial = yield* TxQueue.take(sub)
console.log(initial) // 0
yield* TxSubscriptionRef.set(ref, 1)
const next = yield* TxQueue.take(sub)
console.log(next) // 1
})
)
})export const const changes: <A>(
self: TxSubscriptionRef<A>
) => Effect.Effect<
TxQueue.TxQueue<A>,
never,
Scope.Scope
>
Subscribes to all changes of the TxSubscriptionRef. Returns a scoped TxDequeue
that first yields the current value, then every subsequent update.
When to use
Use to subscribe to TxSubscriptionRef committed changes through a scoped
transactional queue.
Example (Subscribing to changes)
import { Effect, TxQueue, TxSubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* TxSubscriptionRef.make(0)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxSubscriptionRef.changes(ref)
const initial = yield* TxQueue.take(sub)
console.log(initial) // 0
yield* TxSubscriptionRef.set(ref, 1)
const next = yield* TxQueue.take(sub)
console.log(next) // 1
})
)
})
changes = <function (type parameter) A in <A>(self: TxSubscriptionRef<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope>A>(
self: TxSubscriptionRef<A>(parameter) self: {
ref: TxRef.TxRef<A>;
pubsub: TxPubSub.TxPubSub<A>;
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 TxSubscriptionRef<in out A>A TxSubscriptionRef is a transactional reference that allows subscribing to all
committed changes. Subscribers receive the current value followed by every subsequent
update via a transactional dequeue.
When to use
Use to store transactional state whose committed changes must be observable by
subscribers.
Example (Subscribing to transactional changes)
import { Effect, TxQueue, TxSubscriptionRef } from "effect"
const program = Effect.gen(function*() {
const ref = yield* TxSubscriptionRef.make(0)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxSubscriptionRef.changes(ref)
const initial = yield* TxQueue.take(sub)
console.log(initial) // 0
yield* TxSubscriptionRef.set(ref, 1)
const next = yield* TxQueue.take(sub)
console.log(next) // 1
})
)
})
TxSubscriptionRef<function (type parameter) A in <A>(self: TxSubscriptionRef<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope>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: TxSubscriptionRef<A>): Effect.Effect<TxQueue.TxQueue<A>, never, Scope.Scope>A>, never, import ScopeScope.Scope> =>
import EffectEffect.const acquireRelease: <A, E, R, R2>(
acquire: Effect<A, E, R>,
release: (
a: A,
exit: Exit.Exit<unknown, unknown>
) => Effect<unknown, never, R2>,
options?: { readonly interruptible?: boolean }
) => Effect<A, E, R | R2 | Scope>
Constructs a scoped resource from an acquisition effect and a release
finalizer.
When to use
Use to acquire a scoped resource with an explicit release finalizer.
Details
If acquisition succeeds, the release finalizer is added to the current scope
and is guaranteed to run when that scope closes. The finalizer receives the
Exit value used to close the scope.
By default, acquisition is protected by an uninterruptible region. Pass
{ interruptible: true } to allow the acquisition effect to be interrupted.
Example (Acquiring and releasing a resource)
import { Console, Effect, Exit } from "effect"
// Simulate a resource that needs cleanup
interface FileHandle {
readonly path: string
readonly content: string
}
// Acquire a file handle
const acquire = Effect.gen(function*() {
yield* Console.log("Opening file")
return { path: "/tmp/file.txt", content: "file content" }
})
// Release the file handle
const release = (handle: FileHandle, exit: Exit.Exit<unknown, unknown>) =>
Console.log(
`Closing file ${handle.path} with exit: ${
Exit.isSuccess(exit) ? "success" : "failure"
}`
)
// Create a scoped resource
const resource = Effect.acquireRelease(acquire, release)
// Use the resource within a scope
const program = Effect.scoped(
Effect.gen(function*() {
const handle = yield* resource
yield* Console.log(`Using file: ${handle.path}`)
return handle.content
})
)
acquireRelease(
import EffectEffect.const tx: <A, E, R>(
effect: Effect<A, E, R>
) => Effect<A, E, Exclude<R, Transaction>>
Defines a transaction boundary. Transactions are "all or nothing" with respect to changes
made to transactional values (i.e. TxRef) that occur within the transaction body.
Details
If called inside an active transaction, tx composes with the current transaction and reuses
its journal and retry state instead of creating a nested boundary.
Effect transactions are optimistic with retry. A transaction is retried when
its body explicitly calls Effect.txRetry and any accessed transactional
value changes, or when any accessed transactional value changes because a
different transaction commits before the current one.
The outermost tx call creates the transaction boundary and commits or rolls back the full
composed transaction.
Example (Running a transaction)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
const ref1 = yield* TxRef.make(0)
const ref2 = yield* TxRef.make(0)
// Nested tx calls compose into the same transaction
yield* Effect.tx(Effect.gen(function*() {
yield* TxRef.set(ref1, 10)
yield* Effect.tx(TxRef.set(ref2, 20))
const sum = (yield* TxRef.get(ref1)) + (yield* TxRef.get(ref2))
console.log(`Transaction sum: ${sum}`)
}))
console.log(`Final ref1: ${yield* TxRef.get(ref1)}`) // 10
console.log(`Final ref2: ${yield* TxRef.get(ref2)}`) // 20
})
tx(
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 sub: TxQueue.TxQueue<A, never>const sub: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
import TxPubSubTxPubSub.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: TxSubscriptionRef<A>(parameter) self: {
ref: TxRef.TxRef<A>;
pubsub: TxPubSub.TxPubSub<A>;
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; <…;
}
TxSubscriptionRef<A>.pubsub: TxPubSub.TxPubSub<A>(property) TxSubscriptionRef<A>.pubsub: {
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; <…;
}
const current: Acurrent = yield* import TxRefTxRef.const get: <A>(
self: TxRef<A>
) => Effect.Effect<A>
Reads the current value of the TxRef.
When to use
Use to read the current value of a TxRef.
Example (Reading transactional references)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
const counter = yield* TxRef.make(42)
// Read the value within a transaction
const value = yield* Effect.tx(
TxRef.get(counter)
)
console.log(value) // 42
})
get(self: TxSubscriptionRef<A>(parameter) self: {
ref: TxRef.TxRef<A>;
pubsub: TxPubSub.TxPubSub<A>;
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; <…;
}
TxSubscriptionRef<A>.ref: TxRef.TxRef<A>(property) TxSubscriptionRef<A>.ref: {
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; <…;
}
import TxQueueTxQueue.const offer: {
<A, E>(value: A): (
self: TxEnqueue<A, E>
) => Effect.Effect<boolean>
<A, E>(
self: TxEnqueue<A, E>,
value: A
): Effect.Effect<boolean>
}
const sub: TxQueue.TxQueue<A, never>const sub: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
const current: Acurrent)
return const sub: TxQueue.TxQueue<A, never>const sub: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
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.const tx: <A, E, R>(
effect: Effect<A, E, R>
) => Effect<A, E, Exclude<R, Transaction>>
Defines a transaction boundary. Transactions are "all or nothing" with respect to changes
made to transactional values (i.e. TxRef) that occur within the transaction body.
Details
If called inside an active transaction, tx composes with the current transaction and reuses
its journal and retry state instead of creating a nested boundary.
Effect transactions are optimistic with retry. A transaction is retried when
its body explicitly calls Effect.txRetry and any accessed transactional
value changes, or when any accessed transactional value changes because a
different transaction commits before the current one.
The outermost tx call creates the transaction boundary and commits or rolls back the full
composed transaction.
Example (Running a transaction)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
const ref1 = yield* TxRef.make(0)
const ref2 = yield* TxRef.make(0)
// Nested tx calls compose into the same transaction
yield* Effect.tx(Effect.gen(function*() {
yield* TxRef.set(ref1, 10)
yield* Effect.tx(TxRef.set(ref2, 20))
const sum = (yield* TxRef.get(ref1)) + (yield* TxRef.get(ref2))
console.log(`Transaction sum: ${sum}`)
}))
console.log(`Final ref1: ${yield* TxRef.get(ref1)}`) // 10
console.log(`Final ref2: ${yield* TxRef.get(ref2)}`) // 20
})
tx(import TxPubSubTxPubSub.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: TxSubscriptionRef<A>(parameter) self: {
ref: TxRef.TxRef<A>;
pubsub: TxPubSub.TxPubSub<A>;
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; <…;
}
TxSubscriptionRef<A>.pubsub: TxPubSub.TxPubSub<A>(property) TxSubscriptionRef<A>.pubsub: {
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;
}