<A>(
capacity:
| number
| { readonly capacity: number; readonly replay?: number | undefined }
): Effect.Effect<PubSub<A>>Creates a bounded PubSub with the sliding strategy. The PubSub will add new
messages and drop old messages if the PubSub is at capacity.
Details
For best performance use capacities that are powers of two.
Example (Sliding old messages when full)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create sliding PubSub that evicts old messages when full
const pubsub = yield* PubSub.sliding<string>(3)
// With replay buffer
const pubsubWithReplay = yield* PubSub.sliding<string>({
capacity: 3,
replay: 2
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill and overflow the PubSub
yield* PubSub.publish(pubsub, "msg1")
yield* PubSub.publish(pubsub, "msg2")
yield* PubSub.publish(pubsub, "msg3")
yield* PubSub.publish(pubsub, "msg4") // "msg1" is evicted
const messages = yield* PubSub.takeAll(subscription)
console.log(messages) // ["msg2", "msg3", "msg4"]
}))
})export const const sliding: <A>(
capacity:
| number
| {
readonly capacity: number
readonly replay?: number | undefined
}
) => Effect.Effect<PubSub<A>>
Creates a bounded PubSub with the sliding strategy. The PubSub will add new
messages and drop old messages if the PubSub is at capacity.
Details
For best performance use capacities that are powers of two.
Example (Sliding old messages when full)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create sliding PubSub that evicts old messages when full
const pubsub = yield* PubSub.sliding<string>(3)
// With replay buffer
const pubsubWithReplay = yield* PubSub.sliding<string>({
capacity: 3,
replay: 2
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill and overflow the PubSub
yield* PubSub.publish(pubsub, "msg1")
yield* PubSub.publish(pubsub, "msg2")
yield* PubSub.publish(pubsub, "msg3")
yield* PubSub.publish(pubsub, "msg4") // "msg1" is evicted
const messages = yield* PubSub.takeAll(subscription)
console.log(messages) // ["msg2", "msg3", "msg4"]
}))
})
sliding = <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 SlidingStrategy<unknown>(): SlidingStrategy<unknown>Represents the sliding strategy for bounded PubSub values.
When to use
Use to keep the most recent messages when the PubSub is at capacity.
Details
New messages are accepted by evicting older messages from the bounded
PubSub.
Gotchas
Slow subscribers may miss older messages that are evicted before they are
consumed.
Example (Applying a sliding strategy)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create PubSub with sliding strategy
const pubsub = yield* PubSub.sliding<string>(2)
// Or explicitly create with sliding strategy
const customPubsub = yield* PubSub.make<string>({
atomicPubSub: () => PubSub.makeAtomicBounded(2),
strategy: () => new PubSub.SlidingStrategy()
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Publish messages that exceed capacity
yield* PubSub.publish(pubsub, "msg1") // stored
yield* PubSub.publish(pubsub, "msg2") // stored
yield* PubSub.publish(pubsub, "msg3") // "msg1" evicted, "msg3" stored
yield* PubSub.publish(pubsub, "msg4") // "msg2" evicted, "msg4" stored
// Subscribers will see the most recent messages
const messages = yield* PubSub.takeAll(subscription)
console.log("Recent messages:", messages) // ["msg3", "msg4"]
}))
})
SlidingStrategy()
})