45 - PubSub: Fan-Out to Multiple Consumers

Previous · Next

A Queue is point-to-point: one producer, one consumer per message. A PubSub is broadcast: every active subscriber receives a copy of every published message.

Each subscriber gets its own independent buffer — no subscriber starves another, and a slow subscriber does not block the publisher (depending on the overflow strategy).

What you’ll work through

index.test.ts contains four beats: basic publish / subscribe, batching with publishAll, the Queue-vs-PubSub comparison that makes the choice explicit, and the subscriber lifecycle tied to a Scope so subscriptions clean up automatically.

By the end you should know when fan-out is needed (PubSub) vs when one producer feeds one consumer (Queue).

Mental Model

1
Queue (point-to-point)           PubSub (broadcast)
2
──────────────────────           ──────────────────
3
publisher ──► [A, B, C]          publisher ──► hub
4
                │                              │
5
         consumer-1 takes A              subscriber-1 receives [A, B, C]
6
         consumer-2 takes B              subscriber-2 receives [A, B, C]
7
         consumer-1 takes C              subscriber-3 receives [A, B, C]
8

9
Each message goes to ONE consumer.   Each message goes to ALL subscribers.

Use Queue when work should be distributed. Use PubSub when every consumer needs every event — cache invalidation, event sourcing, fan-out notifications.

1
import { Effect, PubSub, Queue } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const hub = yield* PubSub.bounded<string>(10)
5

6
  yield* Effect.scoped(
7
    Effect.gen(function* () {
8
      const sub1 = yield* PubSub.subscribe(hub)
9
      const sub2 = yield* PubSub.subscribe(hub)
10

11
      yield* PubSub.publish(hub, "hello")
12

13
      const a = yield* Queue.take(sub1) // "hello"
14
      const b = yield* Queue.take(sub2) // "hello"
15
      return [a, b]
16
    })
17
  )
18
})

1
//      ┌─── Effect<boolean, never, never>
2
//      ▼
3
yield* PubSub.publishAll(hub, [1, 2, 3])
4

5
//  Each subscriber sees the messages in their original order:
6
//  sub-A drains:  1, 2, 3
7
//  sub-B drains:  1, 2, 3   (independent copy)

1
// Queue — each take removes the item.
2
yield* Queue.offer(queue, "msg")
3
const a = yield* Queue.take(queue) // "msg" — gone
4
const b = yield* Queue.take(queue) // blocks — queue is empty
5

6
// PubSub — each subscriber has its own copy.
7
yield* PubSub.publish(hub, "msg")
8
const a = yield* Queue.take(sub1) // "msg"
9
const b = yield* Queue.take(sub2) // "msg"

1
yield* Effect.scoped(
2
  Effect.gen(function* () {
3
    const sub = yield* PubSub.subscribe(hub)
4
    yield* PubSub.publish(hub, "inside")
5
    yield* Queue.take(sub) // "inside"
6
  })
7
)
8
// sub is now unregistered
9
yield* PubSub.publish(hub, "outside") // not delivered to old sub

Overflow Strategies

PubSub offers the same capacity strategies as Queue:

Common Pitfalls

1, Subscribing after publishing. A subscriber only receives messages published after it subscribed. Subscribe first, then publish.

2, Forgetting Effect.scoped. PubSub.subscribe is a scoped resource. Without Effect.scoped, the subscription is never closed and the subscriber leaks.

3, Using PubSub where Queue fits. If you have one producer and one consumer and want load distribution, use Queue. PubSub clones every message for every subscriber, which has overhead.

See Also

• Previous, 38-semaphore: rate-limiting concurrent access to a shared resource.
• Previous, 42-queue: point-to-point producer/consumer buffering.
• Related, 36-fibers: fork and join for manual concurrency control.

Run the tests: pnpm test 45-pubsub

The full test suite

The index.test.ts for this lesson, verbatim from steps/45-pubsub/.

1
import { describe, expect, it } from "vite-plus/test"
2
import { Effect, PubSub } from "effect"
3

4
describe("PubSub", () => {
5
  it("broadcasts each message to every scoped subscriber", async () => {
6
    const program = Effect.scoped(Effect.gen(function*() {
7
      const bus = yield* PubSub.bounded<string>(4)
8
      const left = yield* PubSub.subscribe(bus)
9
      const right = yield* PubSub.subscribe(bus)
10
      yield* PubSub.publish(bus, "ready")
11
      return [yield* PubSub.take(left), yield* PubSub.take(right)]
12
    }))
13
    expect(await Effect.runPromise(program)).toEqual(["ready", "ready"])
14
  })
15
})