# 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.

```ts
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 |

```ts
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.

```ts
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.

```ts
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:

```ts
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](https://effect.plants.sh/streaming/):
  [`Stream.fromQueue(queue)`](https://effect.plants.sh/streaming/creating-streams/) 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](https://effect.plants.sh/streaming/):

```ts
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:

```ts
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:

```ts
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](https://effect.plants.sh/cluster/) and the messaging built on top of these
primitives. For coordinating a single hand-off of one value between two fibers,
a [`Deferred`](https://effect.plants.sh/state-management/) is lighter than either.

---

## Queue reference

``. 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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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.

```ts
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`.

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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`.

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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).

```ts
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).

```ts
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`.

```ts
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

``. 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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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).

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`).

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`.

```ts
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).

```ts
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`.

```ts
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`.

```ts
import { PubSub } from "effect"

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

#### SlidingStrategy

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

```ts
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`.

```ts
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](https://effect.plants.sh/streaming/), see
  [`Stream.fromPubSub`, `Stream.fromSubscription`, and `Stream.fromQueue`](https://effect.plants.sh/streaming/creating-streams/).