Queue and PubSub

When fibers need to hand work to one another, two complementary primitives cover the field:

• A Queue is a buffer that distributes messages — each item is taken by exactly one consumer. Use it to spread a stream of jobs across a pool of workers.
• A PubSub broadcasts messages — each item is delivered to every subscriber. Use it to fan a domain event out to many independent listeners.

Both are asynchronous and fiber-safe: producers and consumers suspend rather than busy-wait, and a bounded variant applies backpressure so a fast producer can’t overwhelm slow consumers.

Queue — distributing work

Queue.offer adds a message; Queue.take removes and returns one, suspending if the queue is empty. Because each message is taken once, a queue is the natural backbone of a producer/consumer or worker-pool design.

import { Effect, Queue } from "effect"

const program = Effect.gen(function*() {
  // A bounded queue: producers suspend when it's full (backpressure).
  const queue = yield* Queue.bounded<number>(16)

  // A worker that pulls jobs forever, one at a time.
  const worker = (id: number) =>
    Effect.forever(
      Effect.gen(function*() {
        const job = yield* Queue.take(queue)
        yield* Effect.log(`worker ${id} handling job ${job}`)
        yield* Effect.sleep("100 millis")
      })
    )

  // Start three workers as supervised children.
  yield* Effect.forEach([1, 2, 3], (id) => Effect.forkChild(worker(id)))

  // Feed in ten jobs; the three workers share them out between themselves.
  yield* Effect.forEach([...Array(10).keys()], (n) => Queue.offer(queue, n))

  yield* Effect.sleep("1 second")
})

Effect.runFork(program)

Backpressure strategies

Queue.bounded(capacity) applies backpressure: once full, offer suspends the producer until a consumer makes room. That’s the safe default, but you can choose a different strategy when you’d rather drop messages than slow producers:

Constructor	When full, offer…
Queue.bounded(n)	suspends the producer (backpressure)
Queue.dropping(n)	discards the new message and returns false
Queue.sliding(n)	discards the oldest message to make room
Queue.unbounded()	never blocks; grows without limit

import { Effect, Queue } from "effect"

const program = Effect.gen(function*() {
  // A sliding queue of size 2 keeps only the most recent values.
  const queue = yield* Queue.sliding<string>(2)

  yield* Queue.offer(queue, "a")
  yield* Queue.offer(queue, "b")
  yield* Queue.offer(queue, "c") // evicts "a"

  const all = yield* Queue.takeAll(queue)
  yield* Effect.log(all) // ["b", "c"]
})

Effect.runFork(program)

The Enqueue / Dequeue split

A Queue<A, E> is both a write side and a read side. The module models these as two narrower interfaces, so you can hand a producer only the ability to offer and a consumer only the ability to take:

• Enqueue<A, E> — the write side (offer, offerAll, fail, end, shutdown, …). Contravariant in A.
• Dequeue<A, E> — the read side (take, takeAll, poll, peek, …). Covariant in A.
• Queue<A, E> — both at once.

Queue.asEnqueue / Queue.asDequeue narrow a full queue to one side (a type-level capability restriction — the same object is returned), and Queue.isQueue / Queue.isEnqueue / Queue.isDequeue are runtime guards.

import { Effect, Queue } from "effect"

// This function can only *write*; it has no way to take values.
const produce = (out: Queue.Enqueue<string>) =>
  Effect.gen(function*() {
    yield* Queue.offer(out, "hello")
    yield* Queue.offerAll(out, ["world", "!"])
  })

const program = Effect.gen(function*() {
  const queue = yield* Queue.bounded<string>(10)
  yield* produce(Queue.asEnqueue(queue))
})

Graceful end vs interrupt vs shutdown

The E type parameter is the queue’s error/termination channel. Producers can signal several different terminal states, and consumers observe them through the error channel of take:

• Queue.end — stop accepting offers but let consumers drain everything already buffered, then complete normally. Add Cause.Done to the queue’s E type to use it (take fails with Done once drained).
• Queue.fail / Queue.failCause — terminate with an error; buffered messages are still delivered first, then take fails with your error.
• Queue.interrupt — graceful interruption: stop offers, drain buffered messages, then complete with an interrupt cause.
• Queue.shutdown — immediate: discard buffered messages and resume all pending operations right away.

import { Cause, Effect, Queue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* Queue.make<number, Cause.Done>()

  yield* Queue.offerAll(queue, [1, 2, 3])

  // Stop accepting new work, but keep buffered items consumable.
  yield* Queue.end(queue)

  // collect drains every remaining item, then stops at Done.
  const items = yield* Queue.collect(queue)
  yield* Effect.log(items) // [1, 2, 3]
})

Effect.runFork(program)

Note

Queue.end requires Cause.Done in the error type because completion is delivered through the same channel as failures. Type the queue as Queue.Queue<A, Cause.Done> (or A, MyError | Cause.Done) when you intend to signal graceful completion.

A Queue inside a service

In real applications a queue usually lives behind a Context.Service so the rest of the program offers work without owning the queue’s lifecycle:

import { Context, Effect, Layer, Queue } from "effect"

export class JobQueue extends Context.Service<JobQueue, {
  readonly submit: (job: string) => Effect.Effect<void>
  readonly take: Effect.Effect<string>
}>()("app/JobQueue") {
  static readonly layer = Layer.effect(
    JobQueue,
    Effect.gen(function*() {
      const queue = yield* Queue.bounded<string>(128)
      yield* Effect.addFinalizer(() => Queue.shutdown(queue))

      const submit = Effect.fn("JobQueue.submit")(function*(job: string) {
        yield* Queue.offer(queue, job)
      })

      return JobQueue.of({ submit, take: Queue.take(queue) })
    })
  )
}

Tip

A queue pairs beautifully with Streaming: Stream.fromQueue(queue) turns a queue into a stream you can map, filter, and batch with the full streaming toolkit, while producers keep calling Queue.offer from elsewhere.

PubSub — broadcasting events

A PubSub is the broadcast counterpart of a queue. Publishing delivers the message to every current subscriber, so it’s the tool for an in-process event bus, live update notifications, or any one-to-many fan-out. Each subscriber gets its own buffered view via PubSub.subscribe, which is scoped — leaving the scope automatically unsubscribes and releases any messages retained for it.

The idiomatic shape is to wrap the PubSub in a service and expose a typed publish plus a subscribe stream:

import { Context, Effect, Layer, PubSub, Stream } from "effect"

export type OrderEvent =
  | { readonly _tag: "OrderPlaced"; readonly orderId: string }
  | { readonly _tag: "PaymentCaptured"; readonly orderId: string }
  | { readonly _tag: "OrderShipped"; readonly orderId: string }

export class OrderEvents extends Context.Service<OrderEvents, {
  publish(event: OrderEvent): Effect.Effect<void>
  publishAll(events: ReadonlyArray<OrderEvent>): Effect.Effect<void>
  readonly subscribe: Stream.Stream<OrderEvent>
}>()("acme/OrderEvents") {
  static readonly layer = Layer.effect(
    OrderEvents,
    Effect.gen(function*() {
      // Bounded for backpressure; `replay` lets late subscribers catch up on
      // the most recent events (handy after a reconnect).
      const pubsub = yield* PubSub.bounded<OrderEvent>({
        capacity: 256,
        replay: 50
      })

      // Shut the PubSub down when the service's scope closes.
      yield* Effect.addFinalizer(() => PubSub.shutdown(pubsub))

      const publish = Effect.fn("OrderEvents.publish")(
        function*(event: OrderEvent) {
          yield* PubSub.publish(pubsub, event)
        }
      )

      const publishAll = Effect.fn("OrderEvents.publishAll")(
        function*(events: ReadonlyArray<OrderEvent>) {
          yield* PubSub.publishAll(pubsub, events)
        }
      )

      // Every consumer of `subscribe` gets *its own* stream of all events
      // published after it subscribes (plus the replay buffer, if any).
      const subscribe = Stream.fromPubSub(pubsub)

      return OrderEvents.of({ publish, publishAll, subscribe })
    })
  )
}

Consumers each run their own stream and receive every event independently:

import { Effect, Stream } from "effect"

// One subscriber updates a read model; another sends notifications.
// Both see *all* events because PubSub broadcasts.
const projectionWorker = Effect.gen(function*() {
  const events = yield* OrderEvents
  yield* events.subscribe.pipe(
    Stream.tap((event) => Effect.log(`projection: ${event._tag}`)),
    Stream.runDrain
  )
})

const notificationWorker = Effect.gen(function*() {
  const events = yield* OrderEvents
  yield* events.subscribe.pipe(
    Stream.tap((event) => Effect.log(`notify: ${event._tag}`)),
    Stream.runDrain
  )
})

If you want to drive a subscription directly rather than as a stream, subscribe hands back a scoped Subscription you consume with PubSub.take and friends:

import { Effect, PubSub } from "effect"

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

    yield* PubSub.publish(pubsub, "ready")

    return yield* PubSub.take(subscription) // "ready"
  })
)

Note

Replay buffers ({ capacity, replay }) are a convenience for late subscribers to catch up on recent messages — they do not turn the PubSub into a durable event log. A PubSub only retains a message while at least one subscriber that was active for it has yet to take it.

Queue or PubSub?

The choice comes down to delivery:

	Queue	PubSub
Each message goes to	exactly one consumer	every subscriber
Typical use	work distribution, job pools	event bus, live fan-out
Adding a consumer	takes a share of the load	gets the full stream

If you need both — durable work distribution with delivery guarantees across processes — look at Cluster and the messaging built on top of these primitives. For coordinating a single hand-off of one value between two fibers, a Deferred is lighter than either.

---

Queue reference

import { Queue } from "effect". Most operations take the queue (or an Enqueue / Dequeue handle) as their first argument and return an Effect. The Unsafe variants are synchronous and bypass the Effect wrapper — prefer the effectful APIs in application code.

Constructors

make

Creates a queue with optional capacity and overflow strategy ("suspend", "dropping", or "sliding"). Defaults to unbounded with the "suspend" strategy.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.make<number>({ capacity: 8, strategy: "sliding" })
  yield* Queue.offer(queue, 1)
  // => true
})

bounded

Creates a bounded queue that applies backpressure: producers suspend when the queue is at capacity.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<string>(5)
  yield* Queue.offer(queue, "first")
  console.log(yield* Queue.size(queue)) // => 1
})

sliding

Creates a bounded queue that drops the oldest message when full so the newest is always retained.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.sliding<number>(3)
  yield* Queue.offerAll(queue, [1, 2, 3, 4])
  console.log(yield* Queue.takeAll(queue)) // => [2, 3, 4]
})

dropping

Creates a bounded queue that drops the new message when full; the offer returns false.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.dropping<number>(2)
  yield* Queue.offer(queue, 1) // => true
  yield* Queue.offer(queue, 2) // => true
  console.log(yield* Queue.offer(queue, 3)) // => false (dropped)
})

unbounded

Creates a queue with no capacity limit; producers never block.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.unbounded<string>()
  yield* Queue.offerAll(queue, ["a", "b", "c"])
  console.log(yield* Queue.size(queue)) // => 3
})

Producing

offer

Adds one message, returning false if the queue is already done. On a bounded "suspend" queue it suspends until there is room.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(3)
  console.log(yield* Queue.offer(queue, 1)) // => true
})

offerUnsafe

Synchronous, non-suspending version of offer. Returns false if the value can’t be admitted immediately.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(3)
  console.log(Queue.offerUnsafe(queue, 1)) // => true
})

offerAll

Adds many messages and returns the array of messages that could not be added (empty means all were accepted). On a "suspend" queue it waits for capacity.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.dropping<number>(3)
  console.log(yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])) // => [4, 5]
})

offerAllUnsafe

Synchronous version of offerAll; returns the remaining (unadmitted) messages without suspending.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(3)
  console.log(Queue.offerAllUnsafe(queue, [1, 2, 3, 4, 5])) // => [4, 5]
})

Signaling and termination

fail

Fails the queue with an error E; buffered messages are delivered first, then take fails. Returns false if already done.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, string>(10)
  console.log(yield* Queue.fail(queue, "boom")) // => true
  console.log(yield* Effect.flip(Queue.take(queue))) // => "boom"
})

failCause

Like fail but takes a full Cause<E>, so you can propagate defects or interrupts.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, string>(10)
  console.log(yield* Queue.failCause(queue, Cause.fail("nope"))) // => true
})

failCauseUnsafe

Synchronous version of failCause.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, string>(10)
  console.log(Queue.failCauseUnsafe(queue, Cause.fail("nope"))) // => true
})

end

Signals normal completion: no further offers are accepted but buffered messages remain consumable. Requires Cause.Done in the queue’s E type.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, Cause.Done>(10)
  yield* Queue.offer(queue, 1)
  console.log(yield* Queue.end(queue)) // => true
  console.log(yield* Queue.offer(queue, 2)) // => false (ended)
})

endUnsafe

Synchronous version of end.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, Cause.Done>(10)
  console.log(Queue.endUnsafe(queue)) // => true
})

interrupt

Gracefully interrupts: stop accepting offers, drain buffered messages, then complete with an interrupt cause.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  yield* Queue.offer(queue, 1)
  console.log(yield* Queue.interrupt(queue)) // => true
})

shutdown

Immediately discards buffered messages and resumes all pending operations. Idempotent; always returns true.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(2)
  yield* Queue.offerAll(queue, [1, 2])
  console.log(yield* Queue.shutdown(queue)) // => true
  console.log(yield* Queue.size(queue)) // => 0
})

clear

Removes and returns all currently buffered messages without waiting. Fails if the queue has failed.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  yield* Queue.offerAll(queue, [1, 2, 3])
  console.log(yield* Queue.clear(queue)) // => [1, 2, 3]
  console.log(yield* Queue.size(queue)) // => 0
})

Consuming

take

Takes one message, suspending until one is available. Fails with Done on a completed queue, or with the error on a failed one.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<string>(3)
  yield* Queue.offer(queue, "first")
  console.log(yield* Queue.take(queue)) // => "first"
})

takeAll

Takes all currently available messages, waiting until at least one is available. Returns a non-empty array.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(5)
  yield* Queue.offerAll(queue, [1, 2, 3])
  console.log(yield* Queue.takeAll(queue)) // => [1, 2, 3]
})

takeN

Takes exactly n messages, waiting until that many are available.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
  console.log(yield* Queue.takeN(queue, 3)) // => [1, 2, 3]
})

takeBetween

Takes between min and max messages, waiting until at least min are available and returning at most max.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  yield* Queue.offerAll(queue, [1, 2, 3, 4, 5, 6])
  console.log(yield* Queue.takeBetween(queue, 2, 4)) // => [1, 2, 3, 4]
})

collect

Takes every message until the queue is done or fails, then returns them. Requires Cause.Done in E.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, Cause.Done>(5)
  yield* Queue.offerAll(queue, [1, 2, 3])
  yield* Effect.forkChild(Queue.end(queue))
  console.log(yield* Queue.collect(queue)) // => [1, 2, 3]
})

poll

Tries to take one message without waiting, returning Option.some if available and Option.none otherwise.

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

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  console.log(Option.isNone(yield* Queue.poll(queue))) // => true
  yield* Queue.offer(queue, 42)
  console.log(Option.getOrNull(yield* Queue.poll(queue))) // => 42
})

peek

Returns the next message without removing it, suspending until one is available.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  yield* Queue.offer(queue, 42)
  console.log(yield* Queue.peek(queue)) // => 42
  console.log(yield* Queue.size(queue)) // => 1 (still buffered)
})

takeUnsafe

Synchronously takes one message, returning an Exit for an available message or terminal state, or undefined when nothing is immediately available.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  Queue.offerUnsafe(queue, 1)
  console.log(Queue.takeUnsafe(queue)) // => Exit.succeed(1)
  console.log(Queue.takeUnsafe(queue)) // => undefined
})

Introspection

size

Effectfully returns the number of buffered messages; completed queues report 0.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  yield* Queue.offerAll(queue, [1, 2, 3])
  console.log(yield* Queue.size(queue)) // => 3
})

sizeUnsafe

Synchronous version of size.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(10)
  Queue.offerUnsafe(queue, 1)
  console.log(Queue.sizeUnsafe(queue)) // => 1
})

isFull

Effectfully reports whether the queue is at capacity.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(3)
  yield* Queue.offerAll(queue, [1, 2, 3])
  console.log(yield* Queue.isFull(queue)) // => true
})

isFullUnsafe

Synchronous version of isFull.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(3)
  Queue.offerAllUnsafe(queue, [1, 2, 3])
  console.log(Queue.isFullUnsafe(queue)) // => true
})

await

Suspends until the queue reaches the Done state. Succeeds with void for normal completion; preserves other terminal causes.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, Cause.Done>(10)
  yield* Effect.forkChild(Queue.end(queue))
  yield* Queue.await(queue) // resumes once Done
})

Integration

into

Runs an Effect into a queue: success ends the queue, failure fails it. Handy for piping a producer effect’s outcome into the queue’s lifecycle.

import { Cause, Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, Cause.Done>(10)
  const work = Effect.as(Effect.sleep("10 millis"), "done")
  console.log(yield* Queue.into(work, queue)) // => true
  console.log(queue.state._tag) // => "Done"
})

Guards and conversions

isQueue

Type guard narrowing an unknown value to a full Queue.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(1)
  console.log(Queue.isQueue(queue)) // => true
  console.log(Queue.isQueue(42)) // => false
})

isEnqueue

Type guard for the write side. A full Queue also satisfies it.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(1)
  console.log(Queue.isEnqueue(queue)) // => true
})

isDequeue

Type guard for the read side.

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(1)
  console.log(Queue.isDequeue(queue)) // => true
})

asEnqueue

Narrows a Queue to its write-only Enqueue interface (type-level only).

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(1)
  const enqueue: Queue.Enqueue<number> = Queue.asEnqueue(queue)
  yield* Queue.offer(enqueue, 1) // => true
})

asDequeue

Narrows a Queue to its read-only Dequeue interface (type-level only).

import { Effect, Queue } from "effect"

Effect.gen(function*() {
  const queue = yield* Queue.bounded<number>(1)
  const dequeue: Queue.Dequeue<number> = Queue.asDequeue(queue)
  yield* Queue.offer(queue, 1)
  console.log(yield* Queue.take(dequeue)) // => 1
})

Interfaces

• Queue.Queue<A, E> — full read-write handle; invariant in A and E.
• Queue.Enqueue<A, E> — write side (offer, fail, end, …); contravariant in A.
• Queue.Dequeue<A, E> — read side (take, poll, peek, …); covariant in A. Used by Stream.fromQueue.

import { Queue } from "effect"

declare const queue: Queue.Queue<string, never>
declare const writeOnly: Queue.Enqueue<string>
declare const readOnly: Queue.Dequeue<string>

---

PubSub reference

import { PubSub } from "effect". Constructors return an Effect producing the hub; consume it through a scoped Subscription from subscribe.

Constructors

make

Builds a PubSub from a low-level atomic implementation and a delivery strategy. Use the higher-level constructors below unless you need a custom combination.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.make<string>({
    atomicPubSub: () => PubSub.makeAtomicBounded(100),
    strategy: () => new PubSub.BackPressureStrategy()
  })
  yield* PubSub.publish(pubsub, "Hello") // => true
})

bounded

Bounded hub with backpressure: publishers suspend when full. Pass a number, or { capacity, replay } to enable a replay buffer for late subscribers.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(100)
  const withReplay = yield* PubSub.bounded<string>({ capacity: 100, replay: 10 })
})

dropping

Bounded hub that drops new messages when full; publish returns false. Powers-of-two capacities perform best. Accepts an optional replay buffer.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.dropping<string>(3)
  yield* Effect.scoped(Effect.gen(function*() {
    yield* PubSub.subscribe(pubsub)
    yield* PubSub.publishAll(pubsub, ["a", "b", "c"])
    console.log(yield* PubSub.publish(pubsub, "d")) // => false (dropped)
  }))
})

sliding

Bounded hub that drops the oldest message when full. Accepts an optional replay buffer.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.sliding<string>(3)
  yield* Effect.scoped(Effect.gen(function*() {
    const sub = yield* PubSub.subscribe(pubsub)
    yield* PubSub.publishAll(pubsub, ["a", "b", "c", "d"]) // "a" evicted
    console.log(yield* PubSub.takeAll(sub)) // => ["b", "c", "d"]
  }))
})

unbounded

Hub with no capacity limit. Accepts an optional { replay } buffer.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<string>()
  const withReplay = yield* PubSub.unbounded<string>({ replay: 10 })
})

makeAtomicBounded

Low-level bounded storage layer (optionally with replay) for use with make and an explicit strategy. Capacity must be greater than zero.

import { PubSub } from "effect"

const atomic = PubSub.makeAtomicBounded<string>({ capacity: 64, replay: 8 })

makeAtomicUnbounded

Low-level unbounded storage layer (optionally with replay) for use with make.

import { PubSub } from "effect"

const atomic = PubSub.makeAtomicUnbounded<string>({ replay: 8 })

Publishing

publish

Publishes one message, returning whether it was accepted. Applies the configured strategy’s surplus handling (suspend / drop / slide). Returns false after shutdown.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  console.log(yield* PubSub.publish(pubsub, "Hello")) // => true
})

publishUnsafe

Synchronous, non-blocking publish; returns false if the message can’t be accepted immediately (full or shut down). Does not run effectful surplus handling.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  console.log(PubSub.publishUnsafe(pubsub, "Hello")) // => true
})

publishAll

Publishes many messages, returning whether all were accepted. May suspend on a bounded backpressure hub until there is room for all of them.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  console.log(yield* PubSub.publishAll(pubsub, ["a", "b", "c"])) // => true
})

Subscribing

subscribe

Returns a scoped Subscription. Leaving the scope unsubscribes automatically. Each subscription is an independent read side that receives every message published while it is active (plus any replay buffer).

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const sub = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publish(pubsub, "Hello")
  console.log(yield* PubSub.take(sub)) // => "Hello"
}))

take

Takes one message from a Subscription, suspending until one is available.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const sub = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publish(pubsub, "x")
  console.log(yield* PubSub.take(sub)) // => "x"
}))

takeAll

Takes all currently available messages from a Subscription, suspending if none are available. Returns a non-empty array.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const sub = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publishAll(pubsub, ["a", "b", "c"])
  console.log(yield* PubSub.takeAll(sub)) // => ["a", "b", "c"]
}))

takeUpTo

Takes up to max messages without suspending; returns [] if none are available.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const sub = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publishAll(pubsub, ["a", "b", "c"])
  console.log(yield* PubSub.takeUpTo(sub, 2)) // => ["a", "b"]
}))

takeBetween

Takes between min and max messages, suspending until at least min are available.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const sub = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publishAll(pubsub, ["a", "b", "c"])
  console.log(yield* PubSub.takeBetween(sub, 2, 5)) // => ["a", "b", "c"]
}))

Introspection

capacity

Returns the configured capacity (synchronous, not an Effect).

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(100)
  console.log(PubSub.capacity(pubsub)) // => 100
})

size

Effectfully returns the number of messages currently retained for active subscribers; 0 after shutdown.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  yield* PubSub.subscribe(pubsub)
  yield* PubSub.publishAll(pubsub, ["a", "b"])
  console.log(yield* PubSub.size(pubsub)) // => 2
}))

sizeUnsafe

Synchronous version of size.

import { PubSub } from "effect"

declare const pubsub: PubSub.PubSub<string>
console.log(PubSub.sizeUnsafe(pubsub)) // => current retained count

isFull

Effectfully reports whether the hub is at capacity.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(2)
  yield* PubSub.subscribe(pubsub)
  yield* PubSub.publishAll(pubsub, ["a", "b"])
  console.log(yield* PubSub.isFull(pubsub)) // => true
}))

isEmpty

Effectfully reports whether the hub retains zero messages.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  console.log(yield* PubSub.isEmpty(pubsub)) // => true
})

remaining

Effectfully returns the number of messages available to a Subscription (including its replay window); interrupts if the subscription is shut down.

import { Effect, PubSub } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const sub = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publishAll(pubsub, ["a", "b", "c"])
  console.log(yield* PubSub.remaining(sub)) // => 3
}))

remainingUnsafe

Synchronous version of remaining; returns Option.none() when the subscription is shut down.

import { Option, PubSub } from "effect"

declare const sub: PubSub.Subscription<string>
console.log(Option.getOrNull(PubSub.remainingUnsafe(sub))) // => number | null

Lifecycle

shutdown

Shuts the hub down, interrupting suspended publishers and subscribers and finalizing active subscriptions. Afterwards publish returns false and subscription takes interrupt.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(1)
  yield* PubSub.shutdown(pubsub)
  console.log(yield* PubSub.publish(pubsub, "x")) // => false
})

isShutdown

Effectfully reports whether shutdown has been called.

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  yield* PubSub.shutdown(pubsub)
  console.log(yield* PubSub.isShutdown(pubsub)) // => true
})

isShutdownUnsafe

Synchronous version of isShutdown.

import { PubSub } from "effect"

declare const pubsub: PubSub.PubSub<string>
console.log(PubSub.isShutdownUnsafe(pubsub)) // => boolean

awaitShutdown

Suspends until the hub is shut down (resolves immediately if it already is).

import { Effect, PubSub } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.bounded<string>(10)
  const waiter = yield* Effect.forkChild(PubSub.awaitShutdown(pubsub))
  yield* PubSub.shutdown(pubsub) // waiter now completes
})

Strategy classes

These implement PubSub.Strategy and are passed to make to control surplus handling. The higher-level constructors choose one for you (bounded → BackPressureStrategy, dropping / unbounded → DroppingStrategy, sliding → SlidingStrategy).

BackPressureStrategy

Suspends publishers until space is available — no messages are lost. Used by PubSub.bounded.

import { PubSub } from "effect"

const strategy = new PubSub.BackPressureStrategy<string>()

DroppingStrategy

Drops surplus messages so publishers never suspend. Used by PubSub.dropping and PubSub.unbounded.

import { PubSub } from "effect"

const strategy = new PubSub.DroppingStrategy<string>()

SlidingStrategy

Drops the oldest retained messages to admit new ones. Used by PubSub.sliding.

import { PubSub } from "effect"

const strategy = new PubSub.SlidingStrategy<string>()

Interfaces

• PubSub.PubSub<A> — the shared publish side; invariant in A.
• PubSub.Subscription<A> — an independent scoped read side returned by subscribe; covariant in A. Used by Stream.fromSubscription.

import { PubSub } from "effect"

declare const pubsub: PubSub.PubSub<string>
declare const subscription: PubSub.Subscription<string>

Tip

To turn a hub or subscription into a Stream, see Stream.fromPubSub, Stream.fromSubscription, and Stream.fromQueue.