<A>(options?: { readonly replay?: number | undefined }): Effect.Effect<
PubSub<A>
>Creates an unbounded PubSub.
Example (Creating an unbounded PubSub)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create unbounded PubSub
const pubsub = yield* PubSub.unbounded<string>()
// With replay buffer for late subscribers
const pubsubWithReplay = yield* PubSub.unbounded<string>({
replay: 10
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Can publish unlimited messages
for (let i = 0; i < 3; i++) {
yield* PubSub.publish(pubsub, `message-${i}`)
}
const message = yield* PubSub.take(subscription)
console.log("First message:", message) // "message-0"
}))
})export const const unbounded: <A>(options?: {
readonly replay?: number | undefined
}) => Effect.Effect<PubSub<A>>
Creates an unbounded PubSub.
Example (Creating an unbounded PubSub)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create unbounded PubSub
const pubsub = yield* PubSub.unbounded<string>()
// With replay buffer for late subscribers
const pubsubWithReplay = yield* PubSub.unbounded<string>({
replay: 10
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Can publish unlimited messages
for (let i = 0; i < 3; i++) {
yield* PubSub.publish(pubsub, `message-${i}`)
}
const message = yield* PubSub.take(subscription)
console.log("First message:", message) // "message-0"
}))
})
unbounded = <function (type parameter) A in <A>(options?: {
readonly replay?: number | undefined;
}): Effect.Effect<PubSub<A>>
options: | {
readonly replay?: number | undefined
}
| undefined
replay?: number | undefinedreplay?: number | undefined
}): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<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>(options?: {
readonly replay?: number | undefined;
}): Effect.Effect<PubSub<A>>
const make: <A>(options: {
readonly atomicPubSub: LazyArg<PubSub.Atomic<A>>
readonly strategy: LazyArg<PubSub.Strategy<A>>
}) => Effect.Effect<PubSub<A>>
Creates a PubSub with a custom atomic implementation and strategy.
Example (Creating a PubSub with a custom strategy)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create custom PubSub with specific atomic implementation and strategy
const pubsub = yield* PubSub.make<string>({
atomicPubSub: () => PubSub.makeAtomicBounded(100),
strategy: () => new PubSub.BackPressureStrategy()
})
// Use the created PubSub
yield* PubSub.publish(pubsub, "Hello")
})
make({
atomicPubSub: LazyArg<PubSub.Atomic<A>>atomicPubSub: () => const makeAtomicUnbounded: <unknown>(options?: {
readonly replay?: number | undefined;
}) => PubSub<in out A>.Atomic<unknown>
Creates an unbounded atomic PubSub implementation with optional replay buffer.
When to use
Use to create the low-level storage layer for a custom PubSub whose active
subscribers may retain an unbounded number of pending messages.
Gotchas
Messages published while subscribers are active can be retained without a
capacity limit until those subscribers take them or unsubscribe.
makeAtomicUnbounded(options: | {
readonly replay?: number | undefined
}
| undefined
strategy: LazyArg<PubSub.Strategy<A>>strategy: () => new constructor DroppingStrategy<unknown>(): DroppingStrategy<unknown>Represents the dropping strategy for bounded PubSub values.
When to use
Use to keep publishers fast by dropping new messages when the PubSub is at
capacity.
Details
A publish that arrives while the PubSub is full is dropped instead of
waiting for capacity.
Gotchas
Subscribers may miss messages published while they are subscribed.
Example (Applying a dropping strategy)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create PubSub with dropping strategy
const pubsub = yield* PubSub.dropping<string>(2)
// Or explicitly create with dropping strategy
const customPubsub = yield* PubSub.make<string>({
atomicPubSub: () => PubSub.makeAtomicBounded(2),
strategy: () => new PubSub.DroppingStrategy()
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill the PubSub
const pub1 = yield* PubSub.publish(pubsub, "msg1") // true
const pub2 = yield* PubSub.publish(pubsub, "msg2") // true
const pub3 = yield* PubSub.publish(pubsub, "msg3") // false (dropped)
console.log("Publication results:", [pub1, pub2, pub3]) // [true, true, false]
// Subscribers will only see the first two messages
const messages = yield* PubSub.takeAll(subscription)
console.log("Received messages:", messages) // ["msg1", "msg2"]
}))
})
DroppingStrategy()
})