PubSub

Broadcasts values from publishers to many subscribers.

Publishers add messages with publish or publishAll, and each active Subscription receives its own copy of every accepted message. Unlike a queue, subscribers do not compete for messages. This module includes bounded, dropping, sliding, and unbounded hubs, optional replay buffers for late subscribers, message-taking helpers, capacity and shutdown operations, and low-level types for custom hub strategies.

32 exports Added in v2.0.0 Source

Constructors

bounded

Added in v2.0.0 Source

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.

Signature

declare function bounded<A>(capacity: number | {
  readonly capacity: number;
  readonly replay?: number;
}): Effect<PubSub<A>>

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
  })

  const capacities = [PubSub.capacity(pubsub), PubSub.capacity(pubsubWithReplay)]
  yield* PubSub.shutdown(pubsub)
  yield* PubSub.shutdown(pubsubWithReplay)
  return capacities
})

const actual = await Effect.runPromise(program)
actual // => [100, 100]

dropping

Added in v2.0.0 Source

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.

Signature

declare function dropping<A>(capacity: number | {
  readonly capacity: number;
  readonly replay?: number;
}): Effect<PubSub<A>>

Example

(Dropping messages when full)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Create dropping PubSub that drops new messages when full
  const pubsub = yield* PubSub.dropping<string>(3)

  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)

  const messages = yield* PubSub.takeAll(subscription)
  return { dropped: !dropped, messages }
}))

const actual = await Effect.runPromise(program)
actual // => { dropped: true, messages: ["msg1", "msg2", "msg3"] }

make

Added in v4.0.0 Source

Creates a PubSub with a custom atomic implementation and strategy.

Signature

declare function make<A>(options: {
  readonly atomicPubSub: LazyArg<Atomic<A>>;
  readonly strategy: LazyArg<Strategy<A>>;
}): Effect<PubSub<A>>

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
  const published = yield* PubSub.publish(pubsub, "Hello")
  yield* PubSub.shutdown(pubsub)
  return published
})

const actual = await Effect.runPromise(program)
actual // => true

makeAtomicBounded

Added in v4.0.0 Source

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.

See

make for constructing a PubSub from an atomic implementation and delivery strategy
makeAtomicUnbounded for an atomic implementation without a bounded capacity
bounded for the higher-level backpressure constructor
dropping for the higher-level dropping constructor
sliding for the higher-level sliding constructor

Signature

declare function makeAtomicBounded<A>(capacity: number | {
  readonly capacity: number;
  readonly replay?: number;
}): Atomic<A>

makeAtomicUnbounded

Added in v4.0.0 Source

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.

See

makeAtomicBounded for a bounded atomic implementation that enforces capacity
make for wrapping an atomic implementation with a delivery strategy
unbounded for the high-level effectful constructor for unbounded PubSub values

Signature

declare function makeAtomicUnbounded<A>(options?: {
  readonly replay?: number;
}): Atomic<A>

sliding

Added in v2.0.0 Source

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.

Signature

declare function sliding<A>(capacity: number | {
  readonly capacity: number;
  readonly replay?: number;
}): Effect<PubSub<A>>

Example

(Sliding old messages when full)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Create sliding PubSub that evicts old messages when full
  const pubsub = yield* PubSub.sliding<string>(3)

  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

  return yield* PubSub.takeAll(subscription)
}))

const actual = await Effect.runPromise(program)
actual // => ["msg2", "msg3", "msg4"]

unbounded

Added in v2.0.0 Source

Creates an unbounded PubSub.

Signature

declare function unbounded<A>(options?: {
  readonly replay?: number;
}): Effect<PubSub<A>>

Example

(Creating an unbounded PubSub)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Create unbounded PubSub
  const pubsub = yield* PubSub.unbounded<string>()

  const subscription = yield* PubSub.subscribe(pubsub)

  // Can publish unlimited messages
  for (let i = 0; i < 3; i++) {
    yield* PubSub.publish(pubsub, `message-${i}`)
  }

  return yield* PubSub.takeAll(subscription)
}))

const actual = await Effect.runPromise(program)
actual // => ["message-0", "message-1", "message-2"]

Getters

capacity

Added in v2.0.0 Source

Returns the number of elements the queue can hold.

Signature

declare function capacity<A>(self: PubSub<A>): number

Example

(Getting PubSub capacity)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(100)
  const unboundedPubsub = yield* PubSub.unbounded<string>()
  return [PubSub.capacity(pubsub), PubSub.capacity(unboundedPubsub)]
})

const actual = await Effect.runPromise(program)
actual // => [100, Number.MAX_SAFE_INTEGER]

remaining

Added in v4.0.0 Source

Returns the number of messages currently available in the subscription as an Effect.

When to use

Use when checking a subscription from effectful code and shutdown should interrupt the effect.

Details

The count includes replay-buffered messages. If the subscription has been shut down, the effect interrupts.

See

remainingUnsafe for a synchronous check that reports shutdown as Option.none()

Signature

declare function remaining<A>(self: Subscription<A>): Effect<number>

Example

(Checking remaining messages)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Publish some messages
  yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])

  // Check how many messages are available
  const count = yield* PubSub.remaining(subscription)

  // Take one message
  const message = yield* PubSub.take(subscription)

  const remaining = yield* PubSub.remaining(subscription)
  return { count, message, remaining }
}))

const actual = await Effect.runPromise(program)
actual // => { count: 3, message: "msg1", remaining: 2 }

remainingUnsafe

Added in v4.0.0 Source

Synchronously returns the number of messages currently available in the subscription, or Option.none() when it is shut down.

When to use

Use when you need synchronous polling outside a managed workflow and want shutdown observed as data instead of interruption.

See

remaining for the effectful variant that interrupts on shutdown

Signature

declare function remainingUnsafe<A>(self: Subscription<A>): Option<number>

Example

(Checking remaining messages synchronously)

import { Effect, Option, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(2)
  const subscription = yield* PubSub.subscribe(pubsub)
  return PubSub.remainingUnsafe(subscription)
}))

const actual = await Effect.runPromise(program)
actual // => Option.some(0)

size

Added in v2.0.0 Source

Returns the current number of messages retained by the PubSub for active subscribers.

Details

If the PubSub has been shut down, the returned effect succeeds with 0. The size is not a count of waiting subscribers or suspended publishers.

Signature

declare function size<A>(self: PubSub<A>): Effect<number>

Example

(Getting PubSub size)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Initially empty
  const initialSize = yield* PubSub.size(pubsub)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Publish some messages for the active subscription
  yield* PubSub.publish(pubsub, "msg1")
  yield* PubSub.publish(pubsub, "msg2")

  const afterPublish = yield* PubSub.size(pubsub)
  const messages = yield* PubSub.takeAll(subscription)
  return { initialSize, afterPublish, messages }
}))

const actual = await Effect.runPromise(program)
actual // => { initialSize: 0, afterPublish: 2, messages: ["msg1", "msg2"] }

sizeUnsafe

Added in v4.0.0 Source

Returns the current number of messages retained by the PubSub for active subscribers synchronously.

When to use

Use when an immediate PubSub size snapshot is needed outside effectful code and concurrent changes between the check and later use are acceptable.

Details

Returns 0 after shutdown. Because this is an unsafe synchronous snapshot, prefer size in effectful code.

Signature

declare function sizeUnsafe<A>(self: PubSub<A>): number

Example

(Reading size synchronously)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(2)
  return PubSub.sizeUnsafe(pubsub)
})

const actual = await Effect.runPromise(program)
actual // => 0

Lifecycle

awaitShutdown

Added in v2.0.0 Source

Waits until the queue is shutdown. The Effect returned by this method will not resume until the queue has been shutdown. If the queue is already shutdown, the Effect will resume right away.

Signature

declare function awaitShutdown<A>(self: PubSub<A>): Effect<void>

Example

(Waiting for shutdown)

import { Effect, Fiber, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Start a fiber that waits for shutdown
  const waiterFiber = yield* Effect.forkChild(
    Effect.gen(function*() {
      yield* PubSub.awaitShutdown(pubsub)
      return "PubSub has been shutdown!"
    })
  )

  // Shutdown the PubSub
  yield* PubSub.shutdown(pubsub)

  // The waiter will now complete
  return yield* Fiber.join(waiterFiber)
})

const actual = await Effect.runPromise(program)
actual // => "PubSub has been shutdown!"

shutdown

Added in v2.0.0 Source

Shuts down the PubSub, interrupting suspended publishers and subscribers and finalizing active subscriptions.

Details

After shutdown, publish and publishAll succeed with false, publishUnsafe returns false, and subscription operations such as take interrupt.

Signature

declare function shutdown<A>(self: PubSub<A>): Effect<void>

Example

(Shutting down a PubSub)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(1)

  // Shutdown the PubSub
  yield* PubSub.shutdown(pubsub)

  const isShutdown = yield* PubSub.isShutdown(pubsub)

  // Publishing after shutdown returns false
  const published = yield* PubSub.publish(pubsub, "msg1")
  return { isShutdown, published }
})

const actual = await Effect.runPromise(program)
actual // => { isShutdown: true, published: false }

Models

BackPressureStrategy

Added in v4.0.0 Source

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.

See

bounded for creating bounded PubSubs with back pressure by default
DroppingStrategy for dropping new messages when capacity is full
SlidingStrategy for evicting old messages when capacity is full

Signature

declare class BackPressureStrategy<in out A> implements Strategy<A> {
  constructor<in out A>();
  publishers: MutableList<readonly [A, Deferred<boolean, never>, boolean]>;
  shutdown: Effect<void>;
  completePollersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>, subscription: BackingSubscription<A>, pollers: MutableList<Deferred<A, never>>): void;
  completeSubscribersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
  handleSurplus(pubsub: Atomic<A>, subscribers: Subscribers<A>, elements: Iterable<A>, isShutdown: MutableRef<boolean>): Effect<boolean>;
  onPubSubEmptySpaceUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
  removeUnsafe(deferred: Deferred<boolean>): void;
}

DroppingStrategy

Added in v4.0.0 Source

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.

Signature

declare class DroppingStrategy<in out A> implements Strategy<A> {
  constructor<in out A>();
  shutdown: Effect<void>;
  completePollersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>, subscription: BackingSubscription<A>, pollers: MutableList<Deferred<A, never>>): void;
  completeSubscribersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
  handleSurplus(_pubsub: Atomic<A>, _subscribers: Subscribers<A>, _elements: Iterable<A>, _isShutdown: MutableRef<boolean>): Effect<boolean>;
  onPubSubEmptySpaceUnsafe(_pubsub: Atomic<A>, _subscribers: Subscribers<A>): void;
}

Example

(Applying a dropping strategy)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Explicitly create a PubSub with a dropping strategy
  const pubsub = yield* PubSub.make<string>({
    atomicPubSub: () => PubSub.makeAtomicBounded(2),
    strategy: () => new PubSub.DroppingStrategy()
  })

  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)

  // Subscribers will only see the first two messages
  const messages = yield* PubSub.takeAll(subscription)
  return { published: [pub1, pub2, pub3], messages }
}))

const actual = await Effect.runPromise(program)
actual // => { published: [true, true, false], messages: ["msg1", "msg2"] }

PubSub interface

Added in v2.0.0 Source

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.

Signature

interface PubSub<in out A> extends Pipeable {
  readonly "~effect/PubSub": {
    readonly _A: Invariant<A>;
  };
  readonly pubsub: Atomic<A>;
  readonly scope: Closeable;
  readonly shutdownFlag: MutableRef<boolean>;
  readonly shutdownHook: Latch;
  readonly strategy: Strategy<A>;
  readonly subscribers: Subscribers<A>;
}

Example

(Publishing and subscribing to messages)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Create a bounded PubSub with capacity 10
  const pubsub = yield* PubSub.bounded<string>(10)

  // Subscribe and consume messages
  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)
  return [message1, message2]
}))

const actual = await Effect.runPromise(program)
actual // => ["Hello", "World"]

SlidingStrategy

Added in v4.0.0 Source

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.

Signature

declare class SlidingStrategy<in out A> implements Strategy<A> {
  constructor<in out A>();
  shutdown: Effect<void>;
  completePollersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>, subscription: BackingSubscription<A>, pollers: MutableList<Deferred<A, never>>): void;
  completeSubscribersUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
  handleSurplus(pubsub: Atomic<A>, subscribers: Subscribers<A>, elements: Iterable<A>, _isShutdown: MutableRef<boolean>): Effect<boolean>;
  onPubSubEmptySpaceUnsafe(_pubsub: Atomic<A>, _subscribers: Subscribers<A>): void;
  slidingPublishUnsafe(pubsub: Atomic<A>, elements: Iterable<A>): void;
}

Example

(Applying a sliding strategy)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  // Explicitly create a PubSub with a sliding strategy
  const pubsub = yield* PubSub.make<string>({
    atomicPubSub: () => PubSub.makeAtomicBounded(2),
    strategy: () => new PubSub.SlidingStrategy()
  })

  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
  return yield* PubSub.takeAll(subscription)
}))

const actual = await Effect.runPromise(program)
actual // => ["msg3", "msg4"]

Subscription interface

Added in v4.0.0 Source

A subscription represents a consumer's connection to a PubSub, allowing them to take messages.

Signature

interface Subscription<out A> extends Pipeable {
  readonly "~effect/PubSub/Subscription": {
    readonly _A: Covariant<A>;
  };
  readonly pollers: MutableList<Deferred<any, never>>;
  readonly pubsub: Atomic<any>;
  readonly replayWindow: ReplayWindow<A>;
  readonly shutdownFlag: MutableRef<boolean>;
  readonly shutdownHook: Latch;
  readonly strategy: Strategy<any>;
  readonly subscribers: Subscribers<any>;
  readonly subscription: BackingSubscription<A>;
}

Example

(Taking messages from a subscription)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Subscribe within a scope for automatic cleanup
  const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(pubsub)

  yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])

  // Take individual messages
  const message = yield* PubSub.take(subscription)

  // Take multiple messages
  const messages = yield* PubSub.takeUpTo(subscription, 1)
  const allMessages = yield* PubSub.takeAll(subscription)
  return { message, messages, allMessages }
}))

const actual = await Effect.runPromise(program)
actual // => { message: "msg1", messages: ["msg2"], allMessages: ["msg3"] }

Other

PubSub

Added in v2.0.0 Source

Companion namespace containing the low-level building blocks used by PubSub, including atomic implementations, backing subscriptions, replay windows, and delivery strategies.

Predicates

isEmpty

Added in v2.0.0 Source

Returns true if the Pubsub contains zero elements, false otherwise.

Signature

declare function isEmpty<A>(self: PubSub<A>): Effect<boolean>

Example

(Checking whether a PubSub is empty)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Initially empty
  const initiallyEmpty = yield* PubSub.isEmpty(pubsub)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Publish a message for the active subscription
  yield* PubSub.publish(pubsub, "Hello")

  const nowEmpty = yield* PubSub.isEmpty(pubsub)
  const message = yield* PubSub.take(subscription)
  return { initiallyEmpty, nowEmpty, message }
}))

const actual = await Effect.runPromise(program)
actual // => { initiallyEmpty: true, nowEmpty: false, message: "Hello" }

isFull

Added in v2.0.0 Source

Returns true when the PubSub has reached its configured capacity.

Details

For unbounded PubSubs this is normally false.

Signature

declare function isFull<A>(self: PubSub<A>): Effect<boolean>

Example

(Checking whether a PubSub is full)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(2)

  // Initially not full
  const initiallyFull = yield* PubSub.isFull(pubsub)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Fill the PubSub for the active subscription
  yield* PubSub.publish(pubsub, "msg1")
  yield* PubSub.publish(pubsub, "msg2")

  const nowFull = yield* PubSub.isFull(pubsub)
  const messages = yield* PubSub.takeAll(subscription)
  return { initiallyFull, nowFull, messages }
}))

const actual = await Effect.runPromise(program)
actual // => { initiallyFull: false, nowFull: true, messages: ["msg1", "msg2"] }

isShutdown

Added in v2.0.0 Source

Checks effectfully whether shutdown has been called, returning true after shutdown and false otherwise.

Signature

declare function isShutdown<A>(self: PubSub<A>): Effect<boolean>

Example

(Checking whether a PubSub is shut down)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Initially not shutdown
  const initiallyShutdown = yield* PubSub.isShutdown(pubsub)

  // Shutdown the PubSub
  yield* PubSub.shutdown(pubsub)

  const nowShutdown = yield* PubSub.isShutdown(pubsub)
  return [initiallyShutdown, nowShutdown]
})

const actual = await Effect.runPromise(program)
actual // => [false, true]

isShutdownUnsafe

Added in v4.0.0 Source

Checks synchronously whether shutdown has been called, returning true after shutdown and false otherwise.

When to use

Use when an immediate PubSub shutdown-state snapshot is needed outside effectful code and racing shutdown changes are acceptable.

Signature

declare function isShutdownUnsafe<A>(self: PubSub<A>): boolean

Example

(Checking shutdown synchronously)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(2)
  const initiallyShutdown = PubSub.isShutdownUnsafe(pubsub)
  yield* PubSub.shutdown(pubsub)
  return [initiallyShutdown, PubSub.isShutdownUnsafe(pubsub)]
})

const actual = await Effect.runPromise(program)
actual // => [false, true]

Publishing

publish

Added in v2.0.0 Source

Publishes a message to the PubSub as an Effect, returning whether the message was accepted.

When to use

Use when you need to publish from effectful code and let the configured PubSub strategy handle surplus messages.

Details

The effect succeeds with false if the PubSub is shut down. If the message cannot be accepted immediately, the configured strategy decides how surplus messages are handled.

See

publishUnsafe for a synchronous non-blocking attempt that does not run effectful surplus handling

Signature

declare const publish: {
  <A>(value: A): (self: PubSub<A>) => Effect<boolean>;
  <A>(self: PubSub<A>, value: A): Effect<boolean>;
}

Example

(Publishing a message)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Publish a message
  const published = yield* PubSub.publish(pubsub, "Hello World")

  const subscription = yield* PubSub.subscribe(pubsub)

  yield* PubSub.publish(pubsub, "Hello")
  const message = yield* PubSub.take(subscription)
  return { published, message }
}))

const actual = await Effect.runPromise(program)
actual // => { published: true, message: "Hello" }

publishAll

Added in v2.0.0 Source

Publishes all of the specified messages to the PubSub, returning whether they were published to the PubSub.

Signature

declare const publishAll: {
  <A>(elements: Iterable<A>): (self: PubSub<A>) => Effect<boolean>;
  <A>(self: PubSub<A>, elements: Iterable<A>): Effect<boolean>;
}

Example

(Publishing multiple messages)

import { Effect, Fiber, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Publish multiple messages at once
  const allPublished = yield* PubSub.publishAll(pubsub, ["Hello", "World", "from", "Effect"])

  // With a smaller capacity and an active subscription
  const smallPubsub = yield* PubSub.bounded<string>(2)
  const subscription = yield* PubSub.subscribe(smallPubsub)

  // Will suspend until space becomes available for all messages
  const fiber = yield* Effect.forkChild(PubSub.publishAll(smallPubsub, ["msg1", "msg2", "msg3", "msg4"]))

  const firstBatch = yield* PubSub.takeBetween(subscription, 2, 2)
  const result = yield* Fiber.join(fiber)
  const secondBatch = yield* PubSub.takeAll(subscription)
  return { allPublished, firstBatch, result, secondBatch }
}))

const actual = await Effect.runPromise(program)
actual // => { allPublished: true, firstBatch: ["msg1", "msg2"], result: true, secondBatch: ["msg3", "msg4"] }

publishUnsafe

Added in v4.0.0 Source

Attempts to publish a message synchronously without applying the PubSub strategy's effectful surplus handling.

When to use

Use when you need a non-blocking synchronous publish attempt where false is an acceptable result when the message cannot be accepted immediately.

Details

Returns false if the PubSub is shut down or the message cannot be accepted immediately, for example when a bounded PubSub is full. Prefer publish when backpressure or sliding behavior should be honored.

See

publish for effectful publishing that honors the configured surplus strategy

Signature

declare const publishUnsafe: {
  <A>(value: A): (self: PubSub<A>) => boolean;
  <A>(self: PubSub<A>, value: A): boolean;
}

Example

(Publishing without suspending)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(2)
  return PubSub.publishUnsafe(pubsub, "Hello")
})

const actual = await Effect.runPromise(program)
actual // => true

Subscriptions

Added in v2.0.0 Source

Subscribes to receive messages from the PubSub. The resulting subscription can be evaluated multiple times within the scope to take a message from the PubSub each time.

Signature

declare function subscribe<A>(self: PubSub<A>): Effect<Subscription<A>, never, Scope>

Example

(Subscribing to messages)

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  // Subscribe within a scope for automatic cleanup
  const first = yield* Effect.scoped(Effect.gen(function*() {
    const subscription = yield* PubSub.subscribe(pubsub)

    // Publish some messages
    yield* PubSub.publish(pubsub, "Hello")
    yield* PubSub.publish(pubsub, "World")

    // Take messages one by one
    const msg1 = yield* PubSub.take(subscription)
    const msg2 = yield* PubSub.take(subscription)

    // Subscription is automatically cleaned up when scope exits
    return [msg1, msg2]
  }))

  const second = yield* Effect.scoped(Effect.gen(function*() {
    const sub1 = yield* PubSub.subscribe(pubsub)
    const sub2 = yield* PubSub.subscribe(pubsub)

    // Multiple subscribers can receive the same messages
    yield* PubSub.publish(pubsub, "Broadcast")

    return yield* Effect.all([
      PubSub.take(sub1),
      PubSub.take(sub2)
    ])
  }))
  return [first, second]
})

const actual = await Effect.runPromise(program)
actual // => [["Hello", "World"], ["Broadcast", "Broadcast"]]

take

Added in v4.0.0 Source

Takes a single message from the subscription. If no messages are available, this will suspend until a message becomes available.

Signature

declare function take<A>(self: Subscription<A>): Effect<A>

Example

(Taking a message)

import { Effect, Fiber, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Start a fiber to take a message (will suspend)
  const takeFiber = yield* Effect.forkChild(PubSub.take(subscription))

  // Publish a message
  yield* PubSub.publish(pubsub, "Hello")

  // The take will now complete
  return yield* Fiber.join(takeFiber)
}))

const actual = await Effect.runPromise(program)
actual // => "Hello"

takeAll

Added in v4.0.0 Source

Takes all available messages from the subscription, suspending if no items are available.

Signature

declare function takeAll<A>(self: Subscription<A>): Effect<[A, ...Array<A>]>

Example

(Taking all available messages)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Publish multiple messages
  yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])

  // Take all available messages at once
  return yield* PubSub.takeAll(subscription)
}))

const actual = await Effect.runPromise(program)
actual // => ["msg1", "msg2", "msg3"]

takeBetween

Added in v4.0.0 Source

Takes between the specified minimum and maximum number of messages from the subscription. Will suspend if the minimum number is not immediately available.

Signature

declare const takeBetween: {
  (min: number, max: number): <A>(self: Subscription<A>) => Effect<Array<A>>;
  <A>(self: Subscription<A>, min: number, max: number): Effect<Array<A>>;
}

Example

(Taking between a minimum and maximum)

import { Effect, Fiber, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Start taking between 2 and 5 messages (will suspend)
  const takeFiber = yield* Effect.forkChild(PubSub.takeBetween(subscription, 2, 5))

  // Publish 3 messages
  yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])

  // Now the take will complete with 3 messages
  return yield* Fiber.join(takeFiber)
}))

const actual = await Effect.runPromise(program)
actual // => ["msg1", "msg2", "msg3"]

takeUpTo

Added in v4.0.0 Source

Takes up to the specified number of messages from the subscription without suspending.

Signature

declare const takeUpTo: {
  (max: number): <A>(self: Subscription<A>) => Effect<Array<A>>;
  <A>(self: Subscription<A>, max: number): Effect<Array<A>>;
}

Example

(Taking up to a maximum number of messages)

import { Effect, PubSub } from "effect"

const program = Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)

  const subscription = yield* PubSub.subscribe(pubsub)

  // Publish multiple messages
  yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3", "msg4", "msg5"])

  // Take up to 3 messages
  const upTo3 = yield* PubSub.takeUpTo(subscription, 3)

  // Take up to 5 more (only 2 remaining)
  const upTo5 = yield* PubSub.takeUpTo(subscription, 5)

  // No more messages available
  const noMore = yield* PubSub.takeUpTo(subscription, 10)
  return [upTo3, upTo5, noMore]
}))

const actual = await Effect.runPromise(program)
actual // => [["msg1", "msg2", "msg3"], ["msg4", "msg5"], []]