<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
>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.
export const 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
>
}
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.
releaseSubscriber: {
<function (type parameter) A in <A>(queue: TxQueue.TxQueue<A>): (self: TxPubSub<A>) => Effect.Effect<void, never, Effect.Transaction>A>(queue: TxQueue.TxQueue<A>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
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>(queue: TxQueue.TxQueue<A>): (self: TxPubSub<A>) => Effect.Effect<void, 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>(queue: TxQueue.TxQueue<A>): (self: TxPubSub<A>) => Effect.Effect<void, 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<void, 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>
<function (type parameter) A in <A>(self: TxPubSub<A>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, 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>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, never, Effect.Transaction>A>, queue: TxQueue.TxQueue<A>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
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>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, 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<void, 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>
} = dual<(...args: Array<any>) => any, <A>(self: TxPubSub<A>, queue: TxQueue.TxQueue<A>) => Effect.Effect<void, never, Effect.Transaction>>(arity: 2, body: <A>(self: TxPubSub<A>, queue: TxQueue.TxQueue<A>) => Effect.Effect<void, never, Effect.Transaction>): ((...args: Array<any>) => any) & (<A>(self: TxPubSub<A>, queue: TxQueue.TxQueue<A>) => Effect.Effect<void, never, Effect.Transaction>) (+1 overload)Creates a function that can be called in data-first style or data-last
(pipe-friendly) style.
When to use
Use to expose one implementation through both direct and pipe-friendly
call styles.
Details
Pass either the arity of the uncurried function or a predicate that decides
whether the current call is data-first. Arity is the common case. Use a
predicate when optional arguments make arity ambiguous.
Example (Selecting data-first or data-last style by arity)
import { Function, pipe } from "effect"
const sum = Function.dual<
(that: number) => (self: number) => number,
(self: number, that: number) => number
>(2, (self, that) => self + that)
console.log(sum(2, 3)) // 5
console.log(pipe(2, sum(3))) // 5
Example (Defining overloads with call signatures)
import { Function, pipe } from "effect"
const sum: {
(that: number): (self: number) => number
(self: number, that: number): number
} = Function.dual(2, (self: number, that: number): number => self + that)
console.log(sum(2, 3)) // 5
console.log(pipe(2, sum(3))) // 5
Example (Selecting data-first or data-last style with a predicate)
import { Function, pipe } from "effect"
const sum = Function.dual<
(that: number) => (self: number) => number,
(self: number, that: number) => number
>(
(args) => args.length === 2,
(self, that) => self + that
)
console.log(sum(2, 3)) // 5
console.log(pipe(2, sum(3))) // 5
dual(
2,
<function (type parameter) A in <A>(self: TxPubSub<A>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, 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>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, never, Effect.Transaction>A>,
queue: TxQueue.TxQueue<A>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
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>, queue: TxQueue.TxQueue<A>): Effect.Effect<void, 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<void, 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
>
}
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.Array<TxQueue<A, never>>.filter(predicate: (value: TxQueue.TxQueue<A, never>, index: number, array: TxQueue.TxQueue<A, never>[]) => unknown, thisArg?: any): TxQueue.TxQueue<A, never>[] (+1 overload)Returns the elements of an array that meet the condition specified in a callback function.
filter((q: TxQueue.TxQueue<A, never>(parameter) q: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
q: TxQueue.TxQueue<A, never>(parameter) q: {
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>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
import TxQueueTxQueue.const shutdown: <A, E>(
self: TxEnqueue<A, E>
) => Effect.Effect<boolean>
Shuts down the queue immediately by clearing all items and interrupting it (legacy compatibility).
Details
This operation clears all items from the queue using clear, then interrupts the queue using interrupt. This function mutates the original TxQueue by clearing its contents and marking it as shutdown. It does not return a new TxQueue reference.
Example (Shutting down queues)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number>(10)
yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
const sizeBefore = yield* TxQueue.size(queue)
console.log(sizeBefore) // 5
yield* TxQueue.shutdown(queue)
const sizeAfter = yield* TxQueue.size(queue)
console.log(sizeAfter) // 0 (cleared)
const isShutdown = yield* TxQueue.isShutdown(queue)
console.log(isShutdown) // true (interrupted)
})
shutdown(queue: TxQueue.TxQueue<A>(parameter) queue: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}