<A = never>(capacity: number): Effect.Effect<TxPubSub<A>>Creates a sliding TxPubSub with the specified capacity. When a subscriber's queue is full, the oldest message in that subscriber's queue is dropped.
Example (Creating a sliding pub/sub)
import { Effect, TxPubSub, TxQueue } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.sliding<number>(2)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxPubSub.subscribe(hub)
yield* TxPubSub.publish(hub, 1)
yield* TxPubSub.publish(hub, 2)
yield* TxPubSub.publish(hub, 3) // evicts 1
const v1 = yield* TxQueue.take(sub)
console.log(v1) // 2
})
)
})constructors
Source effect/TxPubSub.ts:1966 lines
export const const sliding: <A = never>(
capacity: number
) => Effect.Effect<TxPubSub<A>>
Creates a sliding TxPubSub with the specified capacity. When a subscriber's
queue is full, the oldest message in that subscriber's queue is dropped.
Example (Creating a sliding pub/sub)
import { Effect, TxPubSub, TxQueue } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.sliding<number>(2)
yield* Effect.scoped(
Effect.gen(function*() {
const sub = yield* TxPubSub.subscribe(hub)
yield* TxPubSub.publish(hub, 1)
yield* TxPubSub.publish(hub, 2)
yield* TxPubSub.publish(hub, 3) // evicts 1
const v1 = yield* TxQueue.take(sub)
console.log(v1) // 2
})
)
})
sliding = <function (type parameter) A in <A = never>(capacity: number): Effect.Effect<TxPubSub<A>>A = never>(capacity: numbercapacity: number): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<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 = never>(capacity: number): Effect.Effect<TxPubSub<A>>A>> =>
import EffectEffect.gen(function*() {
const const subscribersRef: TxRef.TxRef<
Array<TxQueue.TxQueue<A, never>>
>
const 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; <…;
}
import TxRefTxRef.const make: <TxQueue.TxQueue<A, never>[]>(initial: TxQueue.TxQueue<A, never>[]) => anyCreates a new TxRef with the specified initial value.
When to use
Use to create a TxRef inside an Effect workflow.
Example (Creating transactional references)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
// Create a transactional reference with initial value
const counter = yield* TxRef.make(0)
const name = yield* TxRef.make("Alice")
// Use in transactions
yield* Effect.tx(Effect.gen(function*() {
yield* TxRef.set(counter, 42)
yield* TxRef.set(name, "Bob")
}))
console.log(yield* TxRef.get(counter)) // 42
console.log(yield* TxRef.get(name)) // "Bob"
})
make<interface Array<T>Array<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 = never>(capacity: number): Effect.Effect<TxPubSub<A>>A>>>([])
const const shutdownRef: TxRef.TxRef<boolean>const shutdownRef: {
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 TxRefTxRef.const make: <boolean>(
initial: boolean
) => any
Creates a new TxRef with the specified initial value.
When to use
Use to create a TxRef inside an Effect workflow.
Example (Creating transactional references)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
// Create a transactional reference with initial value
const counter = yield* TxRef.make(0)
const name = yield* TxRef.make("Alice")
// Use in transactions
yield* Effect.tx(Effect.gen(function*() {
yield* TxRef.set(counter, 42)
yield* TxRef.set(name, "Bob")
}))
console.log(yield* TxRef.get(counter)) // 42
console.log(yield* TxRef.get(name)) // "Bob"
})
make(false)
return const makeTxPubSub: <A>(
subscribersRef: TxRef.TxRef<
Array<TxQueue.TxQueue<A>>
>,
shutdownRef: TxRef.TxRef<boolean>,
strategy:
| "bounded"
| "unbounded"
| "dropping"
| "sliding",
cap: number
) => TxPubSub<A>
const subscribersRef: TxRef.TxRef<
Array<TxQueue.TxQueue<A, never>>
>
const 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; <…;
}
const shutdownRef: TxRef.TxRef<boolean>const shutdownRef: {
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; <…;
}
capacity: numbercapacity)
}).pipe(import EffectEffect.tx)