<A>(self: PubSub<A>): Effect.Effect<void>Shuts down the PubSub, interrupting suspended publishers and subscribers
and finalizing active subscriptions.
Details
After shutdown, publish and publishAll succeed with false,
publishUnsafe returns false, and subscription operations such as take
interrupt.
Example (Shutting down a PubSub)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(1)
// Shutdown the PubSub
yield* PubSub.shutdown(pubsub)
const isShutdown = yield* PubSub.isShutdown(pubsub)
console.log("Is shutdown:", isShutdown) // true
// Publishing after shutdown returns false
const published = yield* PubSub.publish(pubsub, "msg1")
console.log("Published after shutdown:", published) // false
})lifecycle
Source effect/PubSub.ts:7579 lines
export const const shutdown: <A>(
self: PubSub<A>
) => Effect.Effect<void>
Shuts down the PubSub, interrupting suspended publishers and subscribers
and finalizing active subscriptions.
Details
After shutdown, publish and publishAll succeed with false,
publishUnsafe returns false, and subscription operations such as take
interrupt.
Example (Shutting down a PubSub)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(1)
// Shutdown the PubSub
yield* PubSub.shutdown(pubsub)
const isShutdown = yield* PubSub.isShutdown(pubsub)
console.log("Is shutdown:", isShutdown) // true
// Publishing after shutdown returns false
const published = yield* PubSub.publish(pubsub, "msg1")
console.log("Published after shutdown:", published) // false
})
shutdown = <function (type parameter) A in <A>(self: PubSub<A>): Effect.Effect<void>A>(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<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; <…;
}
interface PubSub<in out A>A PubSub<A> is an asynchronous message hub into which publishers can publish
messages of type A and subscribers can subscribe to take messages of type
A.
Example (Publishing and subscribing to messages)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create a bounded PubSub with capacity 10
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe and consume messages
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Publish messages
yield* PubSub.publish(pubsub, "Hello")
yield* PubSub.publish(pubsub, "World")
const message1 = yield* PubSub.take(subscription)
const message2 = yield* PubSub.take(subscription)
console.log(message1, message2) // "Hello", "World"
}))
})
Companion namespace containing the low-level building blocks used by
PubSub, including atomic implementations, backing subscriptions, replay
windows, and delivery strategies.
PubSub<function (type parameter) A in <A>(self: PubSub<A>): Effect.Effect<void>A>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<void> =>
import EffectEffect.uninterruptible(import EffectEffect.withFiber((fiber: Fiber<unknown, unknown>(parameter) fiber: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (context: Context.Context<never>) => void;
currentScheduler: Scheduler;
currentDispatcher: SchedulerDispatcher;
currentSpan: AnySpan | undefined;
currentLogLevel: LogLevel;
minimumLogLevel: LogLevel;
currentStackFrame: StackFrame | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
addObserver: (cb: (exit: Exit<A, E>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<A, E> | undefined;
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 MutableRefMutableRef.set(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<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; <…;
}
PubSub<in out A>.shutdownFlag: MutableRef.MutableRef<boolean>(property) PubSub<in out A>.shutdownFlag: {
current: T;
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; <…;
toString: () => string;
toJSON: () => unknown;
}
import ScopeScope.const close: <A, E>(
self: Scope,
exit: Exit<A, E>
) => Effect<void>
Closes a scope and runs its registered finalizers.
When to use
Use to close a scope manually with a specific exit value.
Details
Finalizers run in the scope's configured order and receive the supplied
Exit.
Example (Running scope finalizers)
import { Console, Effect, Exit, Scope } from "effect"
const resourceManagement = Effect.gen(function*() {
const scope = yield* Scope.make("sequential")
// Add multiple finalizers
yield* Scope.addFinalizer(scope, Console.log("Close database connection"))
yield* Scope.addFinalizer(scope, Console.log("Close file handle"))
yield* Scope.addFinalizer(scope, Console.log("Release memory"))
// Do some work...
yield* Console.log("Performing operations...")
// Close scope - finalizers run in reverse order of registration
yield* Scope.close(scope, Exit.succeed("Success!"))
// Output: "Release memory", "Close file handle", "Close database connection"
})
close(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<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; <…;
}
PubSub<in out A>.scope: Scope.Closeable(property) PubSub<in out A>.scope: {
strategy: "sequential" | "parallel";
state: State.Open | State.Closed | State.Empty;
}
import ExitExit.interrupt(fiber: Fiber<unknown, unknown>(parameter) fiber: {
id: number;
currentOpCount: number;
getRef: <A>(ref: Context.Reference<A>) => A;
context: Context.Context<never>;
setContext: (context: Context.Context<never>) => void;
currentScheduler: Scheduler;
currentDispatcher: SchedulerDispatcher;
currentSpan: AnySpan | undefined;
currentLogLevel: LogLevel;
minimumLogLevel: LogLevel;
currentStackFrame: StackFrame | undefined;
maxOpsBeforeYield: number;
currentPreventYield: boolean;
addObserver: (cb: (exit: Exit<A, E>) => void) => () => void;
interruptUnsafe: (fiberId?: number | undefined, annotations?: Context.Context<never> | undefined) => void;
pollUnsafe: () => Exit<A, E> | undefined;
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 EffectEffect.andThen(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<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; <…;
}
PubSub<A>.strategy: PubSub.Strategy<A>(property) PubSub<A>.strategy: {
shutdown: Effect.Effect<void>;
handleSurplus: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>, elements: Iterable<A>, isShutdown: MutableRef.MutableRef<boolean>) => Effect.Effect<boolean>;
onPubSubEmptySpaceUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>) => void;
completePollersUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>, subscription: PubSub.BackingSubscription<A>, pollers: MutableList.MutableList<Deferred.Deferred<A, never>>) => void;
completeSubscribersUnsafe: (pubsub: PubSub.Atomic<A>, subscribers: PubSub.Subscribers<A>) => void;
}
PubSub<in out A>.Strategy<in out A>.shutdown: Effect.Effect<void>(property) PubSub<in out A>.Strategy<in out A>.shutdown: {
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; <…;
toString: () => string;
toJSON: () => unknown;
}
Describes any finalization logic associated with this strategy.
shutdown),
import EffectEffect.when(self: PubSub<A>(parameter) self: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<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; <…;
}
PubSub<in out A>.shutdownHook: Latch.Latch(property) PubSub<in out A>.shutdownHook: {
open: Effect.Effect<boolean>;
openUnsafe: (this: Latch) => boolean;
release: Effect.Effect<boolean>;
await: Effect.Effect<void>;
close: Effect.Effect<boolean>;
closeUnsafe: (this: Latch) => boolean;
whenOpen: <A, E, R>(self: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
isOpen: (this: Latch) => boolean;
}
import EffectEffect.asVoid
)
}))Referenced by 2 symbols