<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.interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<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: <A>(capacity: number | {
readonly capacity: number;
readonly replay?: number | undefined;
}) => PubSub<in out A>.Atomic<A>
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<A>(): BackPressureStrategy<A>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()
})