<A>(
capacity:
| number
| { readonly capacity: number; readonly replay?: number | undefined }
): Effect.Effect<PubSub<A>>Creates a bounded PubSub with the dropping strategy. The PubSub will drop new
messages if the PubSub is at capacity.
Details
For best performance use capacities that are powers of two.
Example (Dropping messages when full)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create dropping PubSub that drops new messages when full
const pubsub = yield* PubSub.dropping<string>(3)
// With replay buffer for late subscribers
const pubsubWithReplay = yield* PubSub.dropping<string>({
capacity: 3,
replay: 5
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill the PubSub and see dropping behavior
yield* PubSub.publish(pubsub, "msg1") // succeeds
yield* PubSub.publish(pubsub, "msg2") // succeeds
yield* PubSub.publish(pubsub, "msg3") // succeeds
const dropped = yield* PubSub.publish(pubsub, "msg4") // returns false (dropped)
console.log("Message dropped:", !dropped) // true
const messages = yield* PubSub.takeAll(subscription)
console.log(messages) // ["msg1", "msg2", "msg3"]
}))
})export const const dropping: <A>(
capacity:
| number
| {
readonly capacity: number
readonly replay?: number | undefined
}
) => Effect.Effect<PubSub<A>>
Creates a bounded PubSub with the dropping strategy. The PubSub will drop new
messages if the PubSub is at capacity.
Details
For best performance use capacities that are powers of two.
Example (Dropping messages when full)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create dropping PubSub that drops new messages when full
const pubsub = yield* PubSub.dropping<string>(3)
// With replay buffer for late subscribers
const pubsubWithReplay = yield* PubSub.dropping<string>({
capacity: 3,
replay: 5
})
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Fill the PubSub and see dropping behavior
yield* PubSub.publish(pubsub, "msg1") // succeeds
yield* PubSub.publish(pubsub, "msg2") // succeeds
yield* PubSub.publish(pubsub, "msg3") // succeeds
const dropped = yield* PubSub.publish(pubsub, "msg4") // returns false (dropped)
console.log("Message dropped:", !dropped) // true
const messages = yield* PubSub.takeAll(subscription)
console.log(messages) // ["msg1", "msg2", "msg3"]
}))
})
dropping = <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 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()
})