发布订阅

一个 PubSub 充当异步消息中心，允许发布者发送可被所有当前订阅者接收的消息。

与队列不同，队列中每个值只传递给一个消费者，而 PubSub 将每个发布的消息广播给所有订阅者。这使得 PubSub 非常适合需要消息广播而不是负载分发的场景。

基本操作

一个 PubSub<A> 存储类型为 A 的消息，并提供两个基本操作：

API	描述
PubSub.publish	向 PubSub 发送类型为 A 的消息，返回一个指示消息是否成功发布的 effect。
PubSub.subscribe	创建一个允许订阅 PubSub 的作用域 effect，当作用域结束时自动取消订阅。订阅者通过一个保存已发布消息的 Dequeue 接收消息。

示例（向多个订阅者发布消息）

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

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

    // 两个订阅者
    const dequeue1 = yield* PubSub.subscribe(pubsub)
    const dequeue2 = yield* PubSub.subscribe(pubsub)

    // 向 pubsub 发布消息
    yield* PubSub.publish(pubsub, "Hello from a PubSub!")

    // 每个订阅者都接收到消息
    console.log("订阅者 1: " + (yield* Queue.take(dequeue1)))
    console.log("订阅者 2: " + (yield* Queue.take(dequeue2)))
  })
)

Effect.runFork(program)
/*
输出:
订阅者 1: Hello from a PubSub!
订阅者 2: Hello from a PubSub!
*/

在发布前订阅

订阅者只接收在其活跃订阅期间发布的消息。 为确保订阅者接收特定消息， 请在发布消息前建立订阅。

创建 PubSub

有界 PubSub

有界 PubSub 在达到容量时对发布者应用背压，暂停额外的发布直到有空间可用。

背压确保所有订阅者在订阅期间接收所有消息。但是，如果订阅者较慢，可能导致消息传递变慢。

示例（有界 PubSub 创建）

import { PubSub } from "effect"

// 创建一个容量为 2 的有界 PubSub
const boundedPubSub = PubSub.bounded<string>(2)

丢弃 PubSub

丢弃 PubSub 在满时丢弃新值。如果消息被丢弃，PubSub.publish 操作返回 false。

在丢弃 pubsub 中，发布者可以继续发布新值，但不保证订阅者接收所有消息。

示例（丢弃 PubSub 创建）

import { PubSub } from "effect"

// 创建一个容量为 2 的丢弃 PubSub
const droppingPubSub = PubSub.dropping<string>(2)

滑动 PubSub

滑动 PubSub 移除最旧的消息以为新消息腾出空间，确保发布永不阻塞。

滑动 pubsub 防止慢订阅者影响消息传递速率。但是，慢订阅者仍有可能错过一些消息。

示例（滑动 PubSub 创建）

import { PubSub } from "effect"

// 创建一个容量为 2 的滑动 PubSub
const slidingPubSub = PubSub.sliding<string>(2)

无界 PubSub

无界 PubSub 没有容量限制，因此发布总是立即成功。

无界 pubsub 保证所有订阅者接收所有消息而不减慢消息传递。但是，如果消息发布速度快于消费速度，它们可能无限增长。

通常，建议使用有界、丢弃或滑动 pubsub，除非您有无界 pubsub 的特定用例。

示例

import { PubSub } from "effect"

// 创建一个具有无限容量的无界 PubSub
const unboundedPubSub = PubSub.unbounded<string>()

PubSub 操作符

publishAll

PubSub.publishAll 函数让您一次向 pubsub 发布多个值。

示例（发布多个消息）

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

const program = Effect.scoped(
  Effect.gen(function* () {
    const pubsub = yield* PubSub.bounded<string>(2)
    const dequeue = yield* PubSub.subscribe(pubsub)
    yield* PubSub.publishAll(pubsub, ["Message 1", "Message 2"])
    console.log(yield* Queue.takeAll(dequeue))
  })
)

Effect.runFork(program)
/*
输出:
{ _id: 'Chunk', values: [ 'Message 1', 'Message 2' ] }
*/

capacity / size

您可以分别使用 PubSub.capacity 和 PubSub.size 检查 pubsub 的容量和当前大小。

注意 PubSub.capacity 返回一个 number，因为容量在 pubsub 创建时设置且永不改变。 相比之下，PubSub.size 返回一个确定 pubsub 当前大小的 effect，因为 pubsub 中的消息数量可能随时间变化。

示例（检索 PubSub 容量和大小）

import { Effect, PubSub } from "effect"

const program = Effect.gen(function* () {
  const pubsub = yield* PubSub.bounded<number>(2)
  console.log(`capacity: ${PubSub.capacity(pubsub)}`)
  console.log(`size: ${yield* PubSub.size(pubsub)}`)
})

Effect.runFork(program)
/*
输出:
capacity: 2
size: 0
*/

关闭 PubSub

要关闭 pubsub，请使用 PubSub.shutdown。您还可以使用 PubSub.isShutdown 验证它是否已关闭，或使用 PubSub.awaitShutdown 等待关闭完成。关闭 pubsub 还会终止所有关联的队列，确保关闭信号得到有效传达。

PubSub 作为 Enqueue

PubSub 操作符与队列的操作符相似，主要区别是使用 PubSub.publish 和 PubSub.subscribe 代替 Queue.offer 和 Queue.take。如果您已经熟悉使用 Queue，您会发现 PubSub 很简单。

本质上，PubSub 可以看作是只允许写入的 Enqueue：

import type { Queue } from "effect"

interface PubSub<A> extends Queue.Enqueue<A> {}

这里，Enqueue 类型指的是只接受入队（或写入）的队列。在这里入队的任何值都会发布到 pubsub，关闭等操作也会影响 pubsub。

这种设计使 PubSub 高度灵活，让您可以在任何需要只接受发布值的 Enqueue 的地方使用它。