<A>(
capacity:
| number
| { readonly capacity: number; readonly replay?: number | undefined }
): Effect.Effect<PubSub<A>>Creates a bounded PubSub that applies backpressure when it reaches
capacity.
Details
Published messages are retained until all current subscribers have taken
them. When the capacity is full, publishers suspend until space is available.
Pass an options object to configure both capacity and an optional replay
buffer for late subscribers.
Example (Creating a bounded PubSub)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create bounded PubSub with capacity 100
const pubsub = yield* PubSub.bounded<string>(100)
// Create with replay buffer for late subscribers
const pubsubWithReplay = yield* PubSub.bounded<string>({
capacity: 100,
replay: 10 // Last 10 messages replayed to new subscribers
})
})export const const bounded: <A>(
capacity:
| number
| {
readonly capacity: number
readonly replay?: number | undefined
}
) => Effect.Effect<PubSub<A>>
Creates a bounded PubSub that applies backpressure when it reaches
capacity.
Details
Published messages are retained until all current subscribers have taken
them. When the capacity is full, publishers suspend until space is available.
Pass an options object to configure both capacity and an optional replay
buffer for late subscribers.
Example (Creating a bounded PubSub)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create bounded PubSub with capacity 100
const pubsub = yield* PubSub.bounded<string>(100)
// Create with replay buffer for late subscribers
const pubsubWithReplay = yield* PubSub.bounded<string>({
capacity: 100,
replay: 10 // Last 10 messages replayed to new subscribers
})
})
bounded = <function (type parameter) A in <A>(capacity: number | {
readonly capacity: number;
readonly replay?: number | undefined;
}): Effect.Effect<PubSub<A>>
capacity: | number
| {
readonly capacity: number
readonly replay?: number | undefined
}
capacity: numbercapacity: number
readonly 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>(capacity: number | {
readonly capacity: number;
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 makeAtomicBounded: <unknown>(capacity: number | {
readonly capacity: number;
readonly replay?: number | undefined;
}) => PubSub<in out A>.Atomic<unknown>
Creates a bounded atomic PubSub implementation with optional replay buffer.
When to use
Use to provide bounded message storage when building a custom PubSub with
make and an explicit delivery strategy.
Details
Pass either a capacity number or an options object with capacity and
optional replay. A positive replay value enables a replay buffer for late
subscribers, and fractional replay sizes are rounded up.
Gotchas
The capacity must be greater than zero; invalid capacities throw
synchronously before an atomic implementation is created.
makeAtomicBounded(capacity: | number
| {
readonly capacity: number
readonly replay?: number | undefined
}
strategy: LazyArg<PubSub.Strategy<A>>strategy: () => new constructor BackPressureStrategy<unknown>(): BackPressureStrategy<unknown>Represents the back-pressure strategy for bounded PubSub values.
When to use
Use to preserve every message for current subscribers when a bounded custom
PubSub should make publishers wait for capacity instead of dropping or
evicting messages.
Details
Publishers wait when the PubSub is at capacity, so all current subscribers
can receive every published message.
Gotchas
A slow subscriber can slow down publishers and other subscribers.
BackPressureStrategy()
})