<A>(self: TxPubSub<A>): Effect.Effect<void>Shuts down the TxPubSub and all subscriber queues registered at the time of shutdown.
Details
After shutdown, publish and publishAll return false, and awaitShutdown completes. The operation is idempotent.
Gotchas
Subscribers acquired after shutdown are not automatically shut down by this call.
Example (Shutting down a pub/sub)
import { Effect, TxPubSub } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.unbounded<number>()
yield* TxPubSub.shutdown(hub)
const shut = yield* TxPubSub.isShutdown(hub)
console.log(shut) // true
const accepted = yield* TxPubSub.publish(hub, 1)
console.log(accepted) // false
})mutations
Source effect/TxPubSub.ts:62311 lines
export const const shutdown: <A>(
self: TxPubSub<A>
) => Effect.Effect<void>
Shuts down the TxPubSub and all subscriber queues registered at the time of shutdown.
Details
After shutdown, publish and publishAll return false, and awaitShutdown completes. The operation is idempotent.
Gotchas
Subscribers acquired after shutdown are not automatically shut down by this call.
Example (Shutting down a pub/sub)
import { Effect, TxPubSub } from "effect"
const program = Effect.gen(function*() {
const hub = yield* TxPubSub.unbounded<number>()
yield* TxPubSub.shutdown(hub)
const shut = yield* TxPubSub.isShutdown(hub)
console.log(shut) // true
const accepted = yield* TxPubSub.publish(hub, 1)
console.log(accepted) // false
})
shutdown = <function (type parameter) A in <A>(self: TxPubSub<A>): Effect.Effect<void>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<void>A>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<void> =>
import EffectEffect.gen(function*() {
const const alreadyShutdown: anyalreadyShutdown = 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: 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>.shutdownRef: TxRef.TxRef<boolean>(property) TxPubSub<in out A>.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; <…;
}
const alreadyShutdown: anyalreadyShutdown) return
yield* import TxRefTxRef.const set: {
<A>(value: A): (
self: TxRef<A>
) => Effect.Effect<void>
<A>(
self: TxRef<A>,
value: 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<in out A>.shutdownRef: TxRef.TxRef<boolean>(property) TxPubSub<in out A>.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; <…;
}
const subscribers: Array<
TxQueue.TxQueue<A, never>
>
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: 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; <…;
}
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 subscribers: Array<
TxQueue.TxQueue<A, never>
>
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(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;
}
import EffectEffect.tx)