Subscription<A>A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
Example (Taking messages from a subscription)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe within a scope for automatic cleanup
yield* Effect.scoped(Effect.gen(function*() {
const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(
pubsub
)
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Take individual messages
const message = yield* PubSub.take(subscription)
console.log(message) // "msg1"
// Take multiple messages
const messages = yield* PubSub.takeUpTo(subscription, 1)
console.log(messages) // ["msg2"]
const allMessages = yield* PubSub.takeAll(subscription)
console.log(allMessages) // ["msg3"]
}))
})export interface interface Subscription<out A>A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
Example (Taking messages from a subscription)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe within a scope for automatic cleanup
yield* Effect.scoped(Effect.gen(function*() {
const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(
pubsub
)
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Take individual messages
const message = yield* PubSub.take(subscription)
console.log(message) // "msg1"
// Take multiple messages
const messages = yield* PubSub.takeUpTo(subscription, 1)
console.log(messages) // ["msg2"]
const allMessages = yield* PubSub.takeAll(subscription)
console.log(allMessages) // ["msg3"]
}))
})
Subscription<out function (type parameter) A in Subscription<out A>A> extends Pipeable {
readonly [const SubscriptionTypeId: "~effect/PubSub/Subscription"SubscriptionTypeId]: {
readonly _A: Covariant<A>_A: type Covariant<A> = (_: never) => AFunction-type alias encoding covariant variance for a phantom type
parameter.
When to use
Use as a phantom field type to make a type parameter covariant in output
position.
Details
Covariant<A> is assignable to Covariant<B> when A extends B, following
the subtype direction.
Example (Defining a covariant phantom type)
import type { Types } from "effect"
interface Producer<T> {
readonly _phantom: Types.Covariant<T>
readonly get: () => T
}
Namespace for
Covariant
-related utilities.
When to use
Use when referring to type-level helpers nested under Covariant.
Covariant<function (type parameter) A in Subscription<out A>A>
}
readonly Subscription<out A>.pubsub: PubSub.Atomic<any>(property) Subscription<out A>.pubsub: {
capacity: number;
isEmpty: () => boolean;
isFull: () => boolean;
size: () => number;
publish: (value: any) => boolean;
publishAll: (elements: Iterable<any>) => Array<any>;
slide: () => void;
subscribe: () => PubSub.BackingSubscription<any>;
replayWindow: () => PubSub.ReplayWindow<any>;
}
interface PubSub<in out A>.Atomic<in out A>Low-level atomic PubSub interface that handles the core message storage and retrieval.
Atomic<any>
readonly Subscription<out A>.subscribers: PubSub.Subscribers<any>(property) Subscription<out A>.subscribers: {
clear: () => void;
delete: (key: PubSub.BackingSubscription<any>) => boolean;
forEach: (callbackfn: (value: Set<MutableList.MutableList<Deferred.Deferred<any, never>>>, key: PubSub.BackingSubscription<any>, map: Map<PubSub.BackingSubscription<any>, Set<MutableList.MutableList<Deferred.Deferred<any, never>>>>) => void, thisAr…;
get: (key: PubSub.BackingSubscription<any>) => Set<MutableList.MutableList<Deferred.Deferred<any, never>>> | undefined;
has: (key: PubSub.BackingSubscription<any>) => boolean;
set: (key: PubSub.BackingSubscription<any>, value: Set<MutableList.MutableList<Deferred.Deferred<any, never>>>) => PubSub.Subscribers<any>;
size: number;
entries: () => MapIterator<[PubSub.BackingSubscription<any>, Set<MutableList.MutableList<Deferred.Deferred<any, never>>>]>;
keys: () => MapIterator<PubSub.BackingSubscription<any>>;
values: () => MapIterator<Set<MutableList.MutableList<Deferred.Deferred<any, never>>>>;
}
type PubSub<in out A>.Subscribers<A> = Map<PubSub.BackingSubscription<A>, Set<MutableList.MutableList<Deferred.Deferred<A, never>>>>Tracks the pollers currently waiting on each backing subscription.
Details
This type is part of the low-level PubSub.Strategy contract. Most
application code should use subscribe, take, and the other PubSub
operations instead of manipulating subscriber maps directly.
Subscribers<any>
readonly Subscription<out A>.subscription: PubSub.BackingSubscription<A>(property) Subscription<out A>.subscription: {
isEmpty: () => boolean;
size: () => number;
poll: () => typeof MutableList.Empty | A;
pollUpTo: (n: number) => Array<A>;
unsubscribe: () => void;
}
interface PubSub<in out A>.BackingSubscription<out A>Low-level subscription interface that handles message polling for individual subscribers.
BackingSubscription<function (type parameter) A in Subscription<out A>A>
readonly Subscription<out A>.pollers: MutableList.MutableList<Deferred.Deferred<any>>(property) Subscription<out A>.pollers: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
import MutableListMutableList.interface MutableList<in out A>A mutable linked list data structure optimized for high-throughput operations.
MutableList provides efficient append/prepend operations and is ideal for
producer-consumer patterns, queues, and streaming scenarios.
Example (Creating and consuming a mutable list)
import { MutableList } from "effect"
// Create a mutable list
const list: MutableList.MutableList<number> = MutableList.make()
// Add elements
MutableList.append(list, 1)
MutableList.append(list, 2)
MutableList.prepend(list, 0)
// Access properties
console.log(list.length) // 3
console.log(list.head?.array) // Contains elements from head bucket
console.log(list.tail?.array) // Contains elements from tail bucket
// Take elements
console.log(MutableList.take(list)) // 0
console.log(MutableList.take(list)) // 1
console.log(MutableList.take(list)) // 2
The MutableList namespace contains type definitions and utilities for working
with mutable linked lists.
Example (Typing queue processors)
import { MutableList } from "effect"
// Type annotation using the namespace
const processQueue = (queue: MutableList.MutableList<string>) => {
while (queue.length > 0) {
const item = MutableList.take(queue)
if (item !== MutableList.Empty) {
console.log("Processing:", item)
}
}
}
// Using the namespace for type definitions
const createProcessor = <T>(): {
queue: MutableList.MutableList<T>
add: (item: T) => void
process: () => Array<T>
} => {
const queue = MutableList.make<T>()
return {
queue,
add: (item) => MutableList.append(queue, item),
process: () => MutableList.takeAll(queue)
}
}
MutableList<import DeferredDeferred.interface Deferred<in out A, in out E = never>A Deferred represents an asynchronous variable that can be set exactly
once, with the ability for an arbitrary number of fibers to suspend (by
calling Deferred.await) and automatically resume when the variable is set.
When to use
Use to coordinate multiple fibers around a value or failure that will be
supplied exactly once.
Example (Creating a Deferred for inter-fiber communication)
import { Deferred, Effect, Fiber } from "effect"
// Create and use a Deferred for inter-fiber communication
const program = Effect.gen(function*() {
// Create a Deferred that will hold a string value
const deferred: Deferred.Deferred<string> = yield* Deferred.make<string>()
// Fork a fiber that will set the deferred value
const producer = yield* Effect.forkChild(
Effect.gen(function*() {
yield* Effect.sleep("100 millis")
yield* Deferred.succeed(deferred, "Hello, World!")
})
)
// Fork a fiber that will await the deferred value
const consumer = yield* Effect.forkChild(
Effect.gen(function*() {
const value = yield* Deferred.await(deferred)
console.log("Received:", value)
return value
})
)
// Wait for both fibers to complete
yield* Fiber.join(producer)
const result = yield* Fiber.join(consumer)
return result
})
Companion namespace containing type-level metadata for Deferred.
When to use
Use to reference type-level metadata associated with Deferred.
Deferred<any>>
readonly Subscription<out A>.shutdownHook: Latch.Latch(property) Subscription<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 LatchLatch.Latch
readonly Subscription<out A>.shutdownFlag: MutableRef.MutableRef<boolean>(property) Subscription<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 MutableRefMutableRef.interface MutableRef<out T>A synchronous mutable reference that stores a current value.
When to use
Use to keep local mutable state in a stable, pipeable reference.
Details
Read or write the value directly through .current, or use the MutableRef
helpers for pipeable updates such as get, set, update, and
compareAndSet. All operations mutate the same reference in place.
Example (Creating and updating refs)
import { MutableRef } from "effect"
// Create a mutable reference
const ref: MutableRef.MutableRef<number> = MutableRef.make(42)
// Read the current value
console.log(ref.current) // 42
console.log(MutableRef.get(ref)) // 42
// Update the value
ref.current = 100
console.log(MutableRef.get(ref)) // 100
// Use with complex types
interface Config {
timeout: number
retries: number
}
const config: MutableRef.MutableRef<Config> = MutableRef.make({
timeout: 5000,
retries: 3
})
// Update through the interface
config.current = { timeout: 10000, retries: 5 }
console.log(config.current.timeout) // 10000
MutableRef<boolean>
readonly Subscription<out A>.strategy: PubSub.Strategy<any>(property) Subscription<out A>.strategy: {
shutdown: Effect.Effect<void>;
handleSurplus: (pubsub: PubSub.Atomic<any>, subscribers: PubSub.Subscribers<any>, elements: Iterable<any>, isShutdown: MutableRef.MutableRef<boolean>) => Effect.Effect<boolean>;
onPubSubEmptySpaceUnsafe: (pubsub: PubSub.Atomic<any>, subscribers: PubSub.Subscribers<any>) => void;
completePollersUnsafe: (pubsub: PubSub.Atomic<any>, subscribers: PubSub.Subscribers<any>, subscription: PubSub.BackingSubscription<any>, pollers: MutableList.MutableList<Deferred.Deferred<any, never>>) => void;
completeSubscribersUnsafe: (pubsub: PubSub.Atomic<any>, subscribers: PubSub.Subscribers<any>) => void;
}
interface PubSub<in out A>.Strategy<in out A>Strategy interface defining how PubSub handles backpressure and message distribution.
Strategy<any>
readonly Subscription<out A>.replayWindow: PubSub.ReplayWindow<A>(property) Subscription<out A>.replayWindow: {
take: () => A | undefined;
takeN: (n: number) => Array<A>;
takeAll: () => Array<A>;
remaining: number;
}
interface PubSub<in out A>.ReplayWindow<A>Interface for accessing replay buffer contents for late subscribers.
ReplayWindow<function (type parameter) A in Subscription<out A>A>
}