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, a type guard, and low-level types for custom hub strategies.

35 exports Since v2.0.0 Source

Constructors

bounded

Since v2.0.0

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. The capacity must be a positive integer or Infinity for unbounded storage. Invalid capacities cause the effect to die.

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

Since v2.0.0

Creates a bounded PubSub with the dropping strategy. The PubSub will drop new messages if the PubSub is at capacity.

Details

The capacity must be a positive integer or Infinity for unbounded storage. Invalid capacities cause the effect to die. 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

Since v4.0.0

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

Since v4.0.0

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 a positive integer or Infinity for unbounded storage; 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

Since v4.0.0

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

Since v2.0.0

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

The capacity must be a positive integer or Infinity for unbounded storage. Invalid capacities cause the effect to die. 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

Since v2.0.0

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

Since v2.0.0

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

Since v4.0.0

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

Since v4.0.0

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

Since v2.0.0

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

Since v4.0.0

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

Guards

isPubSub

Since v4.0.0

Returns true if a value is a PubSub.

Details

This is a type guard that checks for the PubSub runtime marker.

Signature

declare function isPubSub<A = unknown>(u: unknown): u is PubSub<A>

Example Checking if a value is a PubSub

import { Effect, PubSub } from "effect"

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  return [PubSub.isPubSub(pubsub), PubSub.isPubSub({}), PubSub.isPubSub(null)]
})

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

Lifecycle

awaitShutdown

Since v2.0.0

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!"

end

Since v4.0.0

Ends the PubSub with a final message.

When to use

Use to tell every subscriber that no more messages will follow, without losing the messages they have not consumed yet.

Details

Later publishes return false, as do publishers waiting for capacity when end is called. Each subscriber receives the messages already buffered for it, then the final message. Subscribers that arrive after the end receive the replayed messages, if any, and then the final message. The final message never occupies capacity, so a bounded PubSub cannot drop it. Returns false if the PubSub was already ended or shut down.

take, takeAll, and takeBetween deliver the final message; non-suspending takeUpTo does not.

Gotchas

The final message is sticky: subsequent takes return it again. Consumers must treat it as terminal rather than continuing to take messages.

See

shutdown for interrupting subscribers instead of delivering a final message

Signature

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

Example Ending a PubSub

import { Effect, PubSub } from "effect"

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

  yield* PubSub.publish(pubsub, "Hello")
  const ended = yield* PubSub.end(pubsub, "Bye")

  // Later publishes are rejected
  const published = yield* PubSub.publish(pubsub, "World")

  // Buffered messages are delivered before the final message
  const first = yield* PubSub.take(subscription)
  const last = yield* PubSub.take(subscription)

  // Late subscribers receive the final message too
  const late = yield* PubSub.subscribe(pubsub)
  const lateMessage = yield* PubSub.take(late)
  return [ended, published, first, last, lateMessage]
}))

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

endUnsafe

Since v4.0.0

Ends the PubSub with a final message synchronously.

Details

See end for the semantics.

Signature

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

shutdown

Since v2.0.0

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

Since v4.0.0

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>, ended: MutableRef<Option<A>>): Effect<boolean>;
  onPubSubEmptySpaceUnsafe(pubsub: Atomic<A>, subscribers: Subscribers<A>): void;
  removeUnsafe(deferred: Deferred<boolean>): void;
}

DroppingStrategy

Since v4.0.0

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

Since v2.0.0

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 ended: MutableRef<Option<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

Since v4.0.0

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>, ended: MutableRef<Option<A>>): 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

Since v4.0.0

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 ended: MutableRef<Option<any>>;
  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

Since v2.0.0

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

Predicates

isEmpty

Since v2.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v4.0.0

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

Since v2.0.0

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

Since v2.0.0

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

Since v4.0.0

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

Since v2.0.0

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

Since v4.0.0

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

Since v4.0.0

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

Since v4.0.0

Takes between the specified minimum and maximum number of messages from the subscription. Finite fractional bounds are rounded down, while NaN and non-positive bounds are treated as 0. Will suspend if the normalized 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

Since v4.0.0

Takes up to the specified number of messages from the subscription without suspending. Finite fractional values are rounded down, while NaN and non-positive values are treated as 0.

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"], []]