Stream 操作 · Effect 中文社区

探索 Stream 中用于操作与管理数据的常用操作，包括旁路、映射、过滤、合并等，帮助你高效地处理和转换流式数据。

在本指南中，我们将介绍一些可以在 Stream 上执行的基本操作。借助这些操作，你可以用多种方式操作和处理 Stream 的元素。

旁路（Tapping）

Stream.tap 操作允许你对 Stream 发出的每个元素运行一个 effect，从而观察或执行副作用，而不改变元素本身或返回类型。它适合用于记录日志、监控，或在每次发出元素时触发额外的动作。

示例（使用 Stream.tap 记录日志）

例如，可以用 Stream.tap 在映射操作的前后记录每个元素：

import { Stream, Console, Effect } from "effect"

const stream = Stream.make(1, 2, 3).pipe(
  Stream.tap((n) => Console.log(`before mapping: ${n}`)),
  Stream.map((n) => n * 2),
  Stream.tap((n) => Console.log(`after mapping: ${n}`)),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [2, 4, 6]

取出元素

Stream 中的「取出」操作让你按固定数量、条件或位置从 Stream 中提取特定的元素集合。下面介绍几种使用这些操作的方式：

API	说明
`take`	提取固定数量的元素。
`takeWhile`	在满足某个条件期间持续提取元素。
`takeUntil`	提取元素，直到满足某个条件为止。
`takeRight`	从末尾提取指定数量的元素。

示例（以不同方式提取元素）

import { Stream, Effect } from "effect"

const stream = Stream.iterate(0, (n) => n + 1)

// Using `take` to extract a fixed number of elements:
const s1 = Stream.take(stream, 5)
await Effect.runPromise(Stream.runCollect(s1)) // => [0, 1, 2, 3, 4]

// Using `takeWhile` to extract elements while a condition is met:
const s2 = Stream.takeWhile(stream, (n) => n < 5)
await Effect.runPromise(Stream.runCollect(s2)) // => [0, 1, 2, 3, 4]

// Using `takeUntil` to extract elements until a condition is met:
const s3 = Stream.takeUntil(stream, (n) => n === 5)
await Effect.runPromise(Stream.runCollect(s3)) // => [0, 1, 2, 3, 4, 5]

// Using `takeRight` to take elements from the end of the stream:
const s4 = Stream.takeRight(s3, 3)
await Effect.runPromise(Stream.runCollect(s4)) // => [3, 4, 5]

Stream 作为 Async Iterable 的替代方案

在处理异步数据源（例如 async iterable）时，你常常需要在循环中消费数据，直到满足某个条件为止。Stream 提供了类似的思路，并带来了额外的灵活性。

使用 async iterable 时，数据会在循环中处理，直到遇到 break 或 return 语句。要在 Stream 中复现这种行为，可以考虑以下选项：

API	说明
`takeUntil`	从 Stream 中取出元素，直到满足指定条件为止，类似于跳出循环。
`toPull`	返回一个 effect，它会持续从 Stream 中拉取由元素组成的数组。当 Stream 结束时，该 effect 会以 `Cause.Done` 完成信号失败；否则会以 Stream 自身的错误失败。

示例（使用 Stream.toPull）

import { Stream, Effect } from "effect"

// Simulate a chunked stream
const stream = Stream.fromIterable([1, 2, 3, 4, 5]).pipe(Stream.rechunk(2))

const program = Effect.gen(function* () {
  // Create an effect to get data chunks from the stream
  const getChunk = yield* Stream.toPull(stream)

  // Continuously fetch and process chunks
  while (true) {
    const chunk = yield* getChunk
    console.log(chunk)
  }
})

Effect.runPromise(Effect.scoped(program)).then(console.log, console.error)
/*
Output:
[ 1, 2 ]
[ 3, 4 ]
[ 5 ]
{
  '~effect/Cause/Done': '~effect/Cause/Done',
  _tag: 'Done',
  value: undefined
}
*/

映射

基本映射

Stream.map 操作会对 Stream 中的每个元素应用指定的函数，生成一个包含转换后值的新 Stream。

示例（把每个元素加 1）

import { Stream, Effect } from "effect"

const stream = Stream.make(1, 2, 3).pipe(
  Stream.map((n) => n + 1), // Increment each element by 1
)

await Effect.runPromise(Stream.runCollect(stream)) // => [2, 3, 4]

映射为常量值

Stream.map 方法允许你把 Stream 中的每个成功值替换为指定的常量值。当你希望 Stream 中的所有元素都发出统一的值、而不关心原始数据时，这会很有用。

示例（映射为 null）

import { Stream, Effect } from "effect"

const stream = Stream.range(1, 5).pipe(Stream.map(() => null))

await Effect.runPromise(Stream.runCollect(stream)) // => [null, null, null, null, null]

带 Effect 的映射

对于涉及 effect 的转换，请使用 Stream.mapEffect。该函数会对 Stream 中的每个元素应用一个带 effect 的操作，生成一个包含 effect 结果的新 Stream。

示例（生成随机数）

import { Stream, Random, Effect } from "effect"

const stream = Stream.make(10, 20, 30).pipe(
  // Generate a random number between 0 and each element
  Stream.mapEffect((n) => Random.nextIntBetween(0, n)),
)

const randomResults = await Effect.runPromise(Stream.runCollect(stream))
randomResults.length // => 3

要并发处理多个带 effect 的转换，可以使用 concurrency 选项。该选项允许指定数量的 effect 并发运行，结果会按原始顺序向下游发出。

示例（并发获取 URL）

import { Stream, Effect } from "effect"

const fetchUrl = (url: string) =>
  Effect.gen(function* () {
    console.log(`Fetching ${url}`)
    yield* Effect.sleep("100 millis")
    console.log(`Fetching ${url} done`)
    return [`Resource 0-${url}`, `Resource 1-${url}`, `Resource 2-${url}`]
  })

const stream = Stream.make("url1", "url2", "url3").pipe(
  // Fetch each URL concurrently with a limit of 2
  Stream.mapEffect(fetchUrl, { concurrency: 2 }),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [["Resource 0-url1", "Resource 1-url1", "Resource 2-url1"], ["Resource 0-url2", "Resource 1-url2", "Resource 2-url2"], ["Resource 0-url3", "Resource 1-url3", "Resource 2-url3"]]

有状态映射

Stream.mapAccum 与 Stream.map 类似，但它在应用转换时会跟踪状态，让你可以在一次操作中同时完成映射与累加。它适合用于计算 Stream 中的累计值这类任务。

示例（计算累计总和）

import { Stream, Effect } from "effect"

const stream = Stream.range(1, 5).pipe(
  //                                       ┌─── next state
  //                                       │          ┌─── emitted values
  //                                       ▼          ▼
  Stream.mapAccum(
    () => 0,
    (state, n) => [state + n, [state + n]],
  ),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 3, 6, 10, 15]

映射与扁平化

Stream.flattenIterable 操作与 Stream.map 类似，但它更进一步：先把每个元素映射为零个或多个元素（以 Iterable 形式），再把整个 Stream 扁平化。当需要把每个元素转换为多个值时，它尤其有用。

示例（拆分并扁平化 Stream）

import { Stream, Effect } from "effect"

const numbers = Stream.make("1-2-3", "4-5", "6").pipe(
  Stream.map((s) => s.split("-")),
  Stream.flattenIterable,
)

await Effect.runPromise(Stream.runCollect(numbers)) // => ["1", "2", "3", "4", "5", "6"]

过滤

Stream.filter 操作只放行满足特定条件的元素。它可以保留 Stream 中符合某项标准的元素，并丢弃其余元素。

示例（过滤偶数）

import { Stream, Effect } from "effect"

const stream = Stream.range(1, 11).pipe(Stream.filter((n) => n % 2 === 0))

await Effect.runPromise(Stream.runCollect(stream)) // => [2, 4, 6, 8, 10]

扫描

Stream 扫描让你可以累积地把一个函数应用到 Stream 的每个元素上，并发出每一个中间结果。与只给出最终结果的 reduce 不同，scan 提供了累积过程的逐步视图。

示例（累加求和）

import { Stream, Effect } from "effect"

const stream = Stream.range(1, 5).pipe(Stream.scan(0, (a, b) => a + b))

await Effect.runPromise(Stream.runCollect(stream)) // => [0, 1, 3, 6, 10, 15]

如果只需要最终的累积值，可以使用 Stream.runFold：

示例（最终的累积结果）

import { Stream, Effect } from "effect"

const fold = Stream.range(1, 5).pipe(
  Stream.runFold(
    () => 0,
    (a, b) => a + b,
  ),
)

await Effect.runPromise(fold) // => 15

排空

Stream 排空让你可以在 Stream 中执行带 effect 的操作，同时丢弃结果值。当你需要执行某些动作或副作用、但并不需要发出的值时，这会很有用。Stream.drain 函数通过忽略 Stream 中的所有元素并产出一个空的输出 Stream 来实现这一点。

示例（执行带 effect 的操作但不收集值）

import { Stream, Effect, Random } from "effect"

const stream = Stream.fromEffectRepeat(
  Effect.gen(function* () {
    const nextInt = yield* Random.nextInt
    const number = Math.abs(nextInt % 10)
    console.log(`random number: ${number}`)
    return number
  }),
).pipe(Stream.take(3))

const withValues = await Effect.runPromise(Stream.runCollect(stream))
withValues.length // => 3

const drained = Stream.drain(stream)

await Effect.runPromise(Stream.runCollect(drained)) // => []

检测 Stream 中的变化

Stream.changes 操作会检测并发出 Stream 中与其前一个元素不同的元素。它适合用于跟踪变化，或对连续重复的值去重。

示例（发出连续但不同的元素）

import { Stream, Effect } from "effect"

const stream = Stream.make(1, 1, 1, 2, 2, 3, 4).pipe(Stream.changes)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, 4]

组合（Zipping）

组合（Zipping）会把两个 Stream 的元素合并成一个新的 Stream，将来自每个输入 Stream 的元素配对。这可以通过 Stream.zip 或 Stream.zipWith 实现，后者允许自定义配对逻辑。

示例（基本的组合）

在这个示例中，两个 Stream 的元素会按顺序依次配对。当其中一个 Stream 耗尽时，结果 Stream 随之结束。

import { Stream, Effect } from "effect"

// Zip two streams together
const stream = Stream.zip(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c"),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [[1, "a"], [2, "b"], [3, "c"]]

示例（自定义组合逻辑）

这里，Stream.zipWith 会对每一对元素应用自定义逻辑，以用户定义的方式组合元素。

import { Stream, Effect } from "effect"

// Zip two streams with custom pairing logic
const stream = Stream.zipWith(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c"),
  (n, s) => [n + 10, s + "!"],
)

await Effect.runPromise(Stream.runCollect(stream)) // => [[11, "a!"], [12, "b!"], [13, "c!"]]

以不同速率组合 Stream

当组合的两个 Stream 以不同速度发出元素时，你可能不想等待较慢的那个 Stream 发出元素。使用 Stream.zipLatest 或 Stream.zipLatestWith，只要任一 Stream 产生新值，就可以立即进行配对。当较快的 Stream 有新值到达时，这些函数会使用较慢的那个 Stream 最近一次发出的元素。

示例（组合发出速率不同的 Stream）

import { Stream, Schedule, Effect } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.schedule(Schedule.spaced("1 second")),
)

const s2 = Stream.make("a", "b", "c", "d").pipe(
  Stream.schedule(Schedule.spaced("500 millis")),
)

const stream = Stream.zipLatest(s1, s2)

// The exact interleaving in the middle depends on real wall-clock timing,
// but `zipLatest` always waits for both sides to emit before starting (so
// the first pair is fixed) and both streams are exhausted together at the
// end (so the last pair is fixed too)
const zipLatestResults = await Effect.runPromise(Stream.runCollect(stream))
zipLatestResults[0] // => [1, "a"]
zipLatestResults.at(-1) // => [3, "d"]

与前一个和后一个元素配对

API	说明
`zipWithPrevious`	把 Stream 的每个元素与其前一个元素配对。
`zipWithNext`	把 Stream 的每个元素与其后一个元素配对。
`zipWithPreviousAndNext`	把每个元素同时与其前一个和后一个元素配对。

示例（把 Stream 元素与其后一个元素配对）

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

const stream = Stream.zipWithNext(Stream.make(1, 2, 3, 4))

await Effect.runPromise(Stream.runCollect(stream)) // => [[1, Option.some(2)], [2, Option.some(3)], [3, Option.some(4)], [4, Option.none()]]

为 Stream 元素建立索引

Stream.zipWithIndex 操作符是为 Stream 中每个元素建立索引的实用工具，它会把每个元素与其在序列中的位置配对。当你需要跟踪 Stream 中元素的顺序时，它尤其有用。

示例（为 Stream 中的每个元素建立索引）

import { Stream, Effect } from "effect"

const stream = Stream.zipWithIndex(
  Stream.make("Mary", "James", "Robert", "Patricia"),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [["Mary", 0], ["James", 1], ["Robert", 2], ["Patricia", 3]]

Stream 的笛卡尔积

Stream 模块包含计算两个 Stream 的_笛卡尔积_的功能，让你可以生成来自两个不同 Stream 的元素组合。当你需要把一组中的每个元素与另一组的所有元素配对时，这会很有用。

简单来说，假设你有两个集合，想从每个集合中各取一项来组成所有可能的配对，这个配对过程就是笛卡尔积。在 Stream 中，该操作会生成一个新的 Stream，其中包含两个输入 Stream 元素的所有可能配对。

要创建两个 Stream 的笛卡尔积，可以使用 Stream.cross 操作符及其类似变体。这些操作符会把两个 Stream 组合成一个包含所有可能元素组合的新 Stream。

示例（创建两个 Stream 的笛卡尔积）

import { Stream, Effect, Console } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(Stream.tap(Console.log))
const s2 = Stream.make("a", "b").pipe(Stream.tap(Console.log))

const cartesianProduct = Stream.cross(s1, s2)

await Effect.runPromise(Stream.runCollect(cartesianProduct)) // => [[1, "a"], [1, "b"], [2, "a"], [2, "b"], [3, "a"], [3, "b"]]

Multiple Iterations of Right Stream

注意，右侧的 Stream（本示例中的 s2）会被多次迭代，左侧 Stream（s1）中的每个元素都会触发一次。如果右侧的 Stream 涉及开销较大或会产生副作用的操作，这些操作将被重复执行。

分区

流的分区是指按照指定条件把一个流拆分为两个不同的流。Stream 模块为此提供了两个函数：Stream.partition 和 Stream.partitionEffect。下面来看看它们的工作方式，以及最适合使用它们的场景。

partition

Stream.partition 函数接收一个 Filter 作为输入，把原始流拆分为两个子流：一个子流包含满足条件的元素，另一个包含不满足条件的元素。得到的两个子流都被包装在 Scope 类型中。

示例（把流拆分为奇数和偶数）

import { Stream, Effect, Filter } from "effect"

//      ┌─── Effect<[Stream<number>, Stream<number>], never, Scope>
//      ▼
const program = Stream.range(1, 9).pipe(
  Stream.partition(
    Filter.fromPredicate((n) => n % 2 === 0),
    { bufferSize: 5 },
  ),
)

await Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* () {
      const [odds, evens] = yield* program
      return [yield* Stream.runCollect(odds), yield* Stream.runCollect(evens)]
    }),
  ),
) // => [[1, 3, 5, 7, 9], [2, 4, 6, 8]]

partitionEffect

有些情况下，你可能需要用涉及 effect 的条件来对流进行分区，这时 Stream.partitionEffect 函数正合适。它使用一个带 effect 的 Filter 把流拆分为两个子流：一个用于产生 Result.succeed 值的元素，另一个用于产生 Result.fail 值的元素。

示例（用带 effect 的谓词对流进行分区）

import { Stream, Effect, Filter, Result } from "effect"

//      ┌─── Effect<[Stream<number>, Stream<number>], never, Scope>
//      ▼
const program = Stream.range(1, 9).pipe(
  Stream.partitionEffect(
    // Simulate an effectful computation
    Filter.makeEffect((n: number) =>
      Effect.succeed(n % 2 === 0 ? Result.succeed(n) : Result.fail(n)),
    ),
    { capacity: 5 },
  ),
)

await Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* () {
      const [evens, odds] = yield* program
      return [yield* Stream.runCollect(odds), yield* Stream.runCollect(evens)]
    }),
  ),
) // => [[1, 3, 5, 7, 9], [2, 4, 6, 8]]

分组

处理数据流时，你可能需要按照特定条件对元素进行分组。Stream 模块为此提供了 groupByKey、groupBy、grouped 和 groupedWithin 四个函数。下面逐一看看它们的工作方式以及各自适用的场景。

groupByKey

Stream.groupByKey 函数根据一个类型为 (a: A) => K 的键函数对流进行分区，其中 A 是流中元素的类型，K 表示用于分组的键。该函数不涉及 effect，只是简单地应用所提供的键函数来对元素进行分组。

Stream.groupByKey 的结果是一个普通的 Stream，其元素为 readonly [K, Stream<V>] 对，表示分组后的流。要处理每个分组，可以配合 { concurrency: "unbounded" } 使用 Stream.flatMap，并传入一个类型为 ([key, stream]: [K, Stream<V, E>]) => Stream.Stream<...> 的函数。该函数会作用于所有分组，并以不确定的顺序把它们合并在一起。

示例（按考试成绩的十位数分组）

在下面的示例中，我们使用 Stream.groupByKey 按十位数对考试成绩进行分组，并统计每个分组中的成绩数量：

import { Stream, Effect } from "effect"

class Exam {
  constructor(
    readonly person: string,
    readonly score: number,
  ) {}
}

// Define a list of exam results
const examResults = [
  new Exam("Alex", 64),
  new Exam("Michael", 97),
  new Exam("Bill", 77),
  new Exam("John", 78),
  new Exam("Bobby", 71),
]

// Group exam results by the tens place in the score
const groupByKeyResult = Stream.fromIterable(examResults).pipe(
  Stream.groupByKey((exam) => Math.floor(exam.score / 10) * 10),
)

// Count the number of exam results in each group
const stream = groupByKeyResult.pipe(
  Stream.flatMap(
    ([key, stream]) =>
      Stream.fromEffect(
        Stream.runCollect(stream).pipe(
          Effect.map((values) => [key, values.length] as const),
        ),
      ),
    { concurrency: "unbounded" },
  ),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [[60, 1], [90, 1], [70, 3]]

groupBy

当分组需求更复杂、分区过程涉及 effect 时，可以使用 Stream.groupBy 函数。它接收一个带 effect 的分区函数，并返回一个普通的 Stream，其元素为 readonly [K, Stream<V>] 对，表示分组后的流。随后你可以像 Stream.groupByKey 那样，使用 Stream.flatMap 处理每个分组。

示例（按首字母对名字分组）

在下面的示例中，我们按名字的首字母进行分组，并统计每个分组中的名字数量。这里的分区操作是以带 effect 的方式设置的：

import { Stream, Effect } from "effect"

// Group names by their first letter
const groupByKeyResult = Stream.fromIterable([
  "Mary",
  "James",
  "Robert",
  "Patricia",
  "John",
  "Jennifer",
  "Rebecca",
  "Peter",
]).pipe(
  // Simulate an effectful groupBy operation
  Stream.groupBy((name) =>
    Effect.succeed([name.substring(0, 1), name] as const),
  ),
)

// Count the number of names in each group and display results
const stream = groupByKeyResult.pipe(
  Stream.flatMap(
    ([key, stream]) =>
      Stream.fromEffect(
        Stream.runCollect(stream).pipe(
          Effect.map((values) => [key, values.length] as const),
        ),
      ),
    { concurrency: "unbounded" },
  ),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [["M", 1], ["J", 3], ["R", 2], ["P", 2]]

grouped

Stream.grouped 函数适合把流划分为指定大小的块，从而更便于以更小、更规整的片段来处理数据。在批量处理或展示数据时，这尤其有用。

示例（把流划分为每 3 个元素的块）

import { Stream, Effect } from "effect"

// Create a stream of numbers and group them into chunks of 3
const stream = Stream.range(0, 8).pipe(Stream.grouped(3))

await Effect.runPromise(Stream.runCollect(stream)) // => [[0, 1, 2], [3, 4, 5], [6, 7, 8]]

groupedWithin

Stream.groupedWithin 函数提供了更灵活的分组方式：它根据指定的最大大小或时间间隔中先满足的那个条件来创建块。当处理的数据涉及时间约束时，这尤其有用。

示例（按大小或时间间隔分组）

在这个示例中，Stream.groupedWithin(18, "1.5 seconds") 会在累积满 18 个元素、或者距离上一块创建已过去 1.5 秒时，把流分成一块。

import { Stream, Schedule, Effect } from "effect"

// Create a stream that repeats every second and group by size or time
const stream = Stream.range(0, 9).pipe(
  Stream.repeat(Schedule.spaced("1 second")),
  Stream.groupedWithin(18, "1.5 seconds"),
  Stream.take(3),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7], [8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]]

拼接

在流处理中，你可能需要把多个流的内容组合起来。Stream 模块提供了若干操作符来实现这一点，包括 Stream.concat、Stream.flatten 和 Stream.flatMap。下面看看这些操作符各自的工作方式。

简单拼接

Stream.concat 操作符是连接两个流最直接的方式。它返回一个新的流，先发出第一个流（左侧）的元素，再发出第二个流（右侧）的元素。当你希望按特定顺序组合两个流时，这会很有用。

示例（按顺序拼接两个流）

import { Stream, Effect } from "effect"

const stream = Stream.concat(Stream.make(1, 2, 3), Stream.make("a", "b"))

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b"]

拼接多个流

如果要拼接多个流，Stream.flatten 提供了一种高效的方式，无需手动串联多个 Stream.concat 操作。该函数接收一个由流组成的流，并返回一个按顺序包含各个流中元素的单一流。

示例（拼接多个流）

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3)
const s2 = Stream.make("a", "b")
const s3 = Stream.make(true, false, false)

const stream = Stream.flatten(
  Stream.fromIterable<Stream.Stream<number | string | boolean>>([s1, s2, s3]),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b", true, false, false]

用 flatMap 进行高级拼接

Stream.flatMap 操作符支持更高级的拼接：它对源流的每个输出应用一个类型为 (a: A) => Stream<...> 的函数，从而生成一个新的流。随后该操作符会拼接所有得到的流，实际上把它们展平。

示例（用 Stream.flatMap 生成重复元素）

import { Stream, Effect } from "effect"

// Create a stream where each element is repeated 4 times
const stream = Stream.make(1, 2, 3).pipe(
  Stream.flatMap((a) => Stream.forever(Stream.succeed(a)).pipe(Stream.take(4))),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3]

如果需要并发执行 flatMap 操作，可以使用 concurrency 选项来控制同时运行多少个内部流。

此外，你还可以使用 switch 选项来实现“切换”行为：当源流有新的元素到达时，之前的流会被自动取消。当你只需要最新的结果，并希望通过取消过时的操作来节省资源时，这尤其有用。

示例（使用 switch 选项）

import { Stream, Effect, Console } from "effect"

// Helper function to create a stream with logging
const createStreamWithLogging = (n: number) =>
  Stream.fromEffect(
    Effect.gen(function* () {
      console.log(`Starting stream for value: ${n}`)
      const result = yield* Effect.delay(Effect.succeed(n), "500 millis")
      console.log(`Completed stream for value: ${result}`)
      return result
    }).pipe(
      Effect.onInterrupt(() =>
        Console.log(`Interrupted stream for value: ${n}`),
      ),
    ),
  )

// Without switch (default behavior):
// all streams run to completion
const stream1 = Stream.fromIterable([1, 2, 3]).pipe(
  Stream.flatMap(createStreamWithLogging),
)

// With switch behavior:
// only the last stream completes, previous streams
// are cancelled when new values arrive
const stream2 = Stream.fromIterable([1, 2, 3]).pipe(
  Stream.switchMap(createStreamWithLogging),
)

// Run examples sequentially to see the difference
await Effect.runPromise(
  Effect.gen(function* () {
    console.log("=== Without switch (all streams complete) ===")
    const result1 = yield* Stream.runCollect(stream1)
    console.log(result1)

    console.log("\n=== With switch (only last stream completes) ===")
    const result2 = yield* Stream.runCollect(stream2)
    console.log(result2)

    return [result1, result2]
  }),
) // => [[1, 2, 3], [3]]

switch 选项在搜索功能、实时数据处理等场景中尤其有价值：凡是希望在新输入到达时丢弃先前操作的情况，都很适用。

合并

有时你可能希望把两个流的元素交错在一起，生成一个单一的输出流。这时 Stream.concat 并不合适，因为它会等第一个流完成后才去消费第二个流。若要在元素可用时就交错它们，Stream.merge 及其变体正是为此设计的。

merge

Stream.merge 操作把两个源流的元素组合成一个流，并在元素产生时将它们交错输出。与 Stream.concat 不同，Stream.merge 不会等一个流结束后再开始另一个流。

示例（用 Stream.merge 交错两个流）

import { Schedule, Stream, Effect } from "effect"

// Create two streams with different emission intervals
const s1 = Stream.make(1, 2, 3).pipe(
  Stream.schedule(Schedule.spaced("100 millis")),
)
const s2 = Stream.make(4, 5, 6).pipe(
  Stream.schedule(Schedule.spaced("200 millis")),
)

// Merge s1 and s2 into a single stream that interleaves their values
const merged = Stream.merge(s1, s2)

// The relative order between the two streams can jitter under real
// scheduling, but each stream's own emission order is always preserved
const mergedValues = await Effect.runPromise(Stream.runCollect(merged))
mergedValues.filter((n) => n <= 3) // => [1, 2, 3]
mergedValues.filter((n) => n > 3) // => [4, 5, 6]

终止策略

合并两个流时，考虑终止策略很重要，尤其是在每个流生命周期不同的情况下。默认情况下，Stream.merge 会等待两个流都终止后才结束合并后的流。不过，你可以通过 haltStrategy 修改这一行为，它提供了四种终止策略：

终止策略	说明
`"left"`	当左侧的流终止时，合并后的流随之终止。
`"right"`	当右侧的流终止时，合并后的流随之终止。
`"both"`（默认）	只有当两个流都终止后，合并后的流才终止。
`"either"`	只要任意一个流终止，合并后的流就立即终止。

示例（用 haltStrategy: "left" 控制流的终止）

import { Stream, Schedule, Effect } from "effect"

const s1 = Stream.range(1, 5).pipe(
  Stream.schedule(Schedule.spaced("100 millis")),
)
const s2 = Stream.forever(Stream.succeed(0)).pipe(
  Stream.schedule(Schedule.spaced("200 millis")),
)

const merged = Stream.merge(s1, s2, { haltStrategy: "left" })

await Effect.runPromise(Stream.runCollect(merged)) // => [1, 0, 2, 3, 0, 4, 5]

mergeWith

有些情况下，你可能希望在合并两个流的同时把它们的元素转换为统一的类型。为此，可以把 Stream.merge 与作用于每个源流的 Stream.map 结合使用，从而为每个源流指定转换函数。

示例（合并并转换两个流）

import { Schedule, Stream, Effect } from "effect"

const s1 = Stream.make("1", "2", "3").pipe(
  Stream.schedule(Schedule.spaced("100 millis")),
)
const s2 = Stream.make(4.1, 5.3, 6.2).pipe(
  Stream.schedule(Schedule.spaced("200 millis")),
)

const merged = Stream.merge(
  // Convert string elements from `s1` to integers
  Stream.map(s1, (s) => parseInt(s)),
  // Round down decimal elements from `s2`
  Stream.map(s2, (n) => Math.floor(n)),
)

const mergedResults = await Effect.runPromise(Stream.runCollect(merged))
mergedResults.length // => 6

交替

interleave

Stream.interleave 操作符让你每次从两个流中各取出一个元素，从而生成一个新的交替流。如果其中一个流先结束，另一个流中剩余的元素会继续被取出，直到两个流都耗尽。

示例（两个流的基本交替）

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3)
const s2 = Stream.make(4, 5, 6)

const stream = Stream.interleave(s1, s2)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 4, 2, 5, 3, 6]

interleaveWith

对于更复杂的交替需求，Stream.interleaveWith 通过一个由 boolean 值组成的第三个流来指定交替模式，提供了额外的控制：当该流发出 true 时，从左侧的流取一个元素；否则从右侧的流取一个元素。

示例（用 Stream.interleaveWith 实现自定义交替逻辑）

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 3, 5, 7, 9)
const s2 = Stream.make(2, 4, 6, 8, 10)

// Define a boolean stream to control interleaving
const booleanStream = Stream.make(true, false, false).pipe(Stream.forever)

const stream = Stream.interleaveWith(s1, s2, booleanStream)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 4, 3, 6, 8, 5, 10, 7, 9]

穿插

穿插会在流中添加分隔元素或前后缀，这有助于对流中的数据进行格式化或结构化。

intersperse

Stream.intersperse 操作符会在流中每两个元素之间插入一个指定的分隔元素。这个分隔元素可以是任意选定的值，会被添加到每一对相邻元素之间。

示例（在流元素之间插入分隔元素）

import { Stream, Effect } from "effect"

// Create a stream of numbers and intersperse `0` between them
const stream = Stream.make(1, 2, 3, 4, 5).pipe(Stream.intersperse(0))

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 0, 2, 0, 3, 0, 4, 0, 5]

intersperseAffixes

对于更复杂的需求，Stream.intersperseAffixes 可以分别控制流开头、元素之间以及流末尾所添加的不同前后缀。

示例（为流添加前后缀）

import { Stream, Effect } from "effect"

// Create a stream and add affixes:
// - `[` at the start
// - `|` between elements
// - `]` at the end
const stream = Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.intersperseAffixes({
    start: "[",
    middle: "|",
    end: "]",
  }),
)

await Effect.runPromise(Stream.runCollect(stream)) // => ["[", 1, "|", 2, "|", 3, "|", 4, "|", 5, "]"]

广播

广播一个流会创建多个下游流，它们都会从源流接收到相同的元素。当你希望把每个元素同时发送给多个消费者时，这很有用。上游流有一个 capacity 参数，用于限制它在放慢速度以匹配最慢的下游流之前能领先多少。

示例（广播到多个下游流）

在下面的示例中，我们把一个数字流广播给两个下游消费者。第一个计算流中的最大值，第二个则带延迟地记录每个数字。上游流的速度会根据较慢的那个日志流进行调整：

import { Effect, Stream, Console, Schedule, Fiber } from "effect"

const numbers = Effect.scoped(
  Effect.gen(function* () {
    // Broadcast to 2 downstream consumers with a capacity of 5
    const [first, second] = yield* Stream.range(1, 20).pipe(
      Stream.tap((n) => Console.log(`Emit ${n} element before broadcasting`)),
      Stream.broadcastN({ n: 2, capacity: 5 }),
    )

    // First downstream stream: calculates maximum
    const fiber1 = yield* Stream.runFold(
      first,
      () => 0,
      (acc, e) => Math.max(acc, e),
    ).pipe(
      Effect.andThen((max) => Console.log(`Maximum: ${max}`)),
      Effect.forkChild,
    )

    // Second downstream stream: logs each element with a delay
    const fiber2 = yield* second.pipe(
      Stream.schedule(Schedule.spaced("1 second")),
      Stream.runForEach((n) => Console.log(`Logging to the Console: ${n}`)),
      Effect.forkChild,
    )

    // Wait for both fibers to complete
    yield* Fiber.join(fiber1).pipe(
      Effect.zip(Fiber.join(fiber2), { concurrent: true }),
    )
  }),
)

await Effect.runPromise(numbers) // => undefined

缓冲

Effect 的流采用拉取（pull-based）模型，下游消费者可以控制自己请求元素的速率。然而，当生产者与消费者的速度不匹配时，缓冲有助于平衡二者的交互。Stream.buffer 操作符正是为此设计的：即使消费者较慢，生产者也能继续工作。你可以通过 capacity 选项设置缓冲的最大容量。

buffer

Stream.buffer 操作符会把元素排入队列，让生产者能够在指定容量内独立于消费者工作。当较快的生产者与较慢的消费者需要顺畅运行、互不阻塞时，这很有帮助。

示例（用缓冲区应对速度不匹配）

import { Stream, Console, Schedule, Effect } from "effect"

const stream = Stream.range(1, 10).pipe(
  // Log each element before buffering
  Stream.tap((n) => Console.log(`before buffering: ${n}`)),
  // Buffer with a capacity of 4 elements
  Stream.buffer({ capacity: 4 }),
  // Log each element after buffering
  Stream.tap((n) => Console.log(`after buffering: ${n}`)),
  // Add a 5-second delay between each emission
  Stream.schedule(Schedule.spaced("5 seconds")),
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
before buffering: 1
before buffering: 2
before buffering: 3
before buffering: 4
before buffering: 5
before buffering: 6
after buffering: 1
after buffering: 2
before buffering: 7
after buffering: 3
before buffering: 8
after buffering: 4
before buffering: 9
after buffering: 5
before buffering: 10
...
*/

不同的缓冲选项让你可以根据使用场景定制缓冲策略：

缓冲类型	配置	说明
有界队列	`{ capacity: number }`	把队列限制为固定大小。
无界队列	`{ capacity: "unbounded" }`	允许缓冲的条目数量不受限制。
滑动队列	`{ capacity: number, strategy: "sliding" }`	保留最新的条目，队列满时丢弃较旧的条目。
丢弃队列	`{ capacity: number, strategy: "dropping" }`	保留最早的条目，队列满时丢弃新到的条目。

防抖

防抖是一种用来避免函数触发过于频繁的技术，当流快速发射值、而你只需要停顿之后的最后一个值时，它尤其有用。

Stream.debounce 函数通过推迟值的发射来实现这一点：只有在指定时间段内没有新值出现时，才会发射值。如果在等待期间有新值到达，计时器就会重置，最终只会在一次停顿之后发射最新的那个值。

示例（对快速发射值的流进行防抖）

import { Stream, Effect } from "effect"

// Helper function to log with elapsed time since the last log
let last = Date.now()
const log = (message: string) =>
  Effect.sync(() => {
    const end = Date.now()
    console.log(`${message} after ${end - last}ms`)
    last = end
  })

const stream = Stream.make(1, 2, 3).pipe(
  // Emit the value 4 after 200 ms
  Stream.concat(
    Stream.fromEffect(Effect.sleep("200 millis").pipe(Effect.as(4))),
  ),
  // Continue with more rapid values
  Stream.concat(Stream.make(5, 6)),
  // Emit 7 after 150 ms
  Stream.concat(
    Stream.fromEffect(Effect.sleep("150 millis").pipe(Effect.as(7))),
  ),
  Stream.concat(Stream.make(8)),
  Stream.tap((n) => log(`Received ${n}`)),
  // Only emit values after a pause of at least 100 milliseconds
  Stream.debounce("100 millis"),
  Stream.tap((n) => log(`> Emitted ${n}`)),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [3, 6, 8]

节流

节流是一种调节流元素发射速率的技术。它有助于保持稳定的数据输出节奏，在数据处理需要以恒定速率进行时很有价值。

Stream.throttle 函数使用令牌桶算法来控制流的发射速率。

示例（节流配置）

Stream.throttle({
  cost: () => 1,
  duration: "100 millis",
  units: 1,
})

在这个配置中：

每个被处理的 chunk 消耗一个令牌（cost = () => 1）。
令牌以每 100 毫秒（duration: "100 millis"）补充一个（units: 1）的速率重新填充。

Throttling Applies to Chunks, Not Elements

请注意，节流作用于 chunk 而不是单个元素。cost 函数会为每个 chunk 设置令牌开销。

shape 策略（默认）

shape 策略通过延迟 chunk 的发射来调节数据流，直到它们满足指定的带宽约束。该策略确保数据吞吐量不超过设定的上限，从而实现稳定可控的数据发射。

示例（使用 shape 策略进行节流）

import { Stream, Effect, Schedule } from "effect"

// Helper function to log with elapsed time since last log
let last = Date.now()
const log = (message: string) =>
  Effect.sync(() => {
    const end = Date.now()
    console.log(`${message} after ${end - last}ms`)
    last = end
  })

const stream = Stream.fromSchedule(Schedule.spaced("50 millis")).pipe(
  Stream.take(6),
  Stream.tap((n) => log(`Received ${n}`)),
  Stream.throttle({
    cost: (arr) => arr.length,
    duration: "100 millis",
    units: 1,
  }),
  Stream.tap((n) => log(`> Emitted ${n}`)),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [0, 1, 2, 3, 4, 5]

enforce 策略

enforce 策略通过丢弃超出带宽约束的 chunk 来严格调节数据流。

示例（使用 enforce 策略进行节流）

import { Stream, Effect, Schedule } from "effect"

// Helper function to log with elapsed time since last log
let last = Date.now()
const log = (message: string) =>
  Effect.sync(() => {
    const end = Date.now()
    console.log(`${message} after ${end - last}ms`)
    last = end
  })

const stream = Stream.make(1, 2, 3, 4, 5, 6).pipe(
  Stream.schedule(Schedule.exponential("100 millis")),
  Stream.tap((n) => log(`Received ${n}`)),
  Stream.throttle({
    cost: (arr) => arr.length,
    duration: "1 second",
    units: 1,
    strategy: "enforce",
  }),
  Stream.tap((n) => log(`> Emitted ${n}`)),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 4, 5, 6]

burst 选项

Stream.throttle 函数提供了 burst 选项，允许数据吞吐量暂时超过设定的速率上限。将该选项设为大于 0 即可启用 burst 能力（默认值为 0，表示不支持 burst）。 burst 容量会在令牌桶中提供额外的令牌，使流在出现数据突发时能够短暂超过其配置的速率。

示例（带 burst 容量的节流）

import { Effect, Schedule, Stream } from "effect"

// Helper function to log with elapsed time since last log
let last = Date.now()
const log = (message: string) =>
  Effect.sync(() => {
    const end = Date.now()
    console.log(`${message} after ${end - last}ms`)
    last = end
  })

const stream = Stream.fromSchedule(Schedule.spaced("10 millis")).pipe(
  Stream.take(20),
  Stream.tap((n) => log(`Received ${n}`)),
  Stream.throttle({
    cost: (arr) => arr.length,
    duration: "200 millis",
    units: 5,
    strategy: "enforce",
    burst: 2,
  }),
  Stream.tap((n) => log(`> Emitted ${n}`)),
)

// Exact emitted values depend on real wall-clock timing and vary between
// runs, but the "enforce" strategy only ever drops chunks. It never
// reorders or duplicates them, so the result is always a strictly
// increasing subsequence of 0..19 starting with 0
const burstResults = await Effect.runPromise(Stream.runCollect(stream))
burstResults[0] // => 0
burstResults.every((n, i) => i === 0 || n > burstResults[i - 1]) // => true

在这个配置中，流开始时令牌桶里有 5 个令牌，因此最初的五个 chunk 会被立即发射。额外的 burst 容量 2 可以暂时容纳更多发射，从而更灵活地处理后续数据。随着时间推移，令牌桶会按照节流配置重新填充，更多元素会被发射，这也展示了 burst 能力如何有效地应对不均匀的数据流。

调度

在处理流时，你可能需要在每个元素的发射之间引入特定的时间间隔。Stream.schedule 组合子允许你设置这些间隔。

示例（在流发射之间添加延迟）

import { Stream, Schedule, Console, Effect } from "effect"

// Create a stream that emits values with a 1-second delay between each
const stream = Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.schedule(Schedule.spaced("1 second")),
  Stream.tap(Console.log),
)

await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, 4, 5]

在这个示例中，我们使用 Schedule.spaced("1 second") 在流的每次发射之间引入了一秒的间隔。