操作

在本指南中，我们将探索您可以在流上执行的一些基本操作。这些操作允许您以各种方式操作和与流元素交互。

点击

Stream.tap 操作允许您对流发出的每个元素运行 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}`))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
before mapping: 1
after mapping: 2
before mapping: 2
after mapping: 4
before mapping: 3
after mapping: 6
{ _id: 'Chunk', values: [ 2, 4, 6 ] }
*/

提取元素

流中的”提取”操作让您可以提取特定的元素集合，可以按固定数量、条件或在流中的位置进行提取。以下是应用这些操作的几种方式：

API	描述
take	提取固定数量的元素。
takeWhile	在满足特定条件时提取元素。
takeUntil	提取元素直到满足特定条件。
takeRight	从末尾提取指定数量的元素。

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

import { Stream, Effect } from "effect"

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

// 使用 `take` 提取固定数量的元素：
const s1 = Stream.take(stream, 5)
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 0, 1, 2, 3, 4 ] }
*/

// 使用 `takeWhile` 在满足条件时提取元素：
const s2 = Stream.takeWhile(stream, (n) => n < 5)
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 0, 1, 2, 3, 4 ] }
*/

// 使用 `takeUntil` 提取元素直到满足条件：
const s3 = Stream.takeUntil(stream, (n) => n === 5)
Effect.runPromise(Stream.runCollect(s3)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 0, 1, 2, 3, 4, 5 ] }
*/

// 使用 `takeRight` 从流的末尾提取元素：
const s4 = Stream.takeRight(s3, 3)
Effect.runPromise(Stream.runCollect(s4)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 3, 4, 5 ] }
*/

流作为异步可迭代对象的替代方案

在处理异步数据源（如异步可迭代对象）时，您经常需要在循环中消费数据，直到满足某个条件。流提供了类似的方法并提供额外的灵活性。

对于异步可迭代对象，数据在循环中处理，直到遇到 break 或 return 语句。要使用流复制此行为，请考虑以下选项：

API	描述
takeUntil	从流中提取元素直到满足指定条件，类似于跳出循环。
toPull	返回一个持续从流中拉取数据块的 effect。当流结束时，此 effect 可能会失败并返回 None，或者在失败时返回 Some 错误。

示例（使用 Stream.toPull）

import { Stream, Effect } from "effect"

// 模拟一个分块流
const stream = Stream.fromIterable([1, 2, 3, 4, 5]).pipe(
  Stream.rechunk(2)
)

const program = Effect.gen(function* () {
  // 创建一个从流中获取数据块的 effect
  const getChunk = yield* Stream.toPull(stream)

  // 持续获取和处理块
  while (true) {
    const chunk = yield* getChunk
    console.log(chunk)
  }
})

Effect.runPromise(Effect.scoped(program)).then(console.log, console.error)
/*
输出：
{ _id: 'Chunk', values: [ 1, 2 ] }
{ _id: 'Chunk', values: [ 3, 4 ] }
{ _id: 'Chunk', values: [ 5 ] }
(FiberFailure) Error: {
  "_id": "Option",
  "_tag": "None"
}
*/

映射

基本映射

Stream.map 操作将指定的函数应用于流中的每个元素，创建一个包含转换值的新流。

示例（将每个元素递增 1）

import { Stream, Effect } from "effect"

const stream = Stream.make(1, 2, 3).pipe(
  Stream.map((n) => n + 1) // 将每个元素递增 1
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 2, 3, 4 ] }
*/

映射到常量值

Stream.as 方法允许您将流中的每个成功值替换为指定的常量值。当您希望流中的所有元素都发出统一的值时，这很有用，无论原始数据如何。

示例（映射到 null）

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ null, null, null, null, null ] }
*/

有效映射

对于涉及 effect 的转换，请使用 Stream.mapEffect。此函数将有效操作应用于流中的每个元素，产生一个包含有效结果的新流。

示例（随机数生成）

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

const stream = Stream.make(10, 20, 30).pipe(
  // 在 0 和每个元素之间生成一个随机数
  Stream.mapEffect((n) => Random.nextIntBetween(0, n))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
示例输出：
{ _id: 'Chunk', values: [ 5, 9, 22 ] }
*/

要并发处理多个有效转换，您可以使用并发选项。此选项允许指定数量的 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(
  // 并发获取每个 URL，限制为 2
  Stream.mapEffect(fetchUrl, { concurrency: 2 })
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
Fetching url1
Fetching url2
Fetching url1 done
Fetching url3
Fetching url2 done
Fetching url3 done
{
  _id: 'Chunk',
  values: [
    [ '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，但它应用带有状态跟踪的转换，允许您在单个操作中映射和累积值。这对于计算流中的运行总计等任务很有用。

示例（计算运行总计）

import { Stream, Effect } from "effect"

const stream = Stream.range(1, 5).pipe(
  //                                  ┌─── 下一个状态
  //                                  │          ┌─── 发出的值
  //                                  ▼          ▼
  Stream.mapAccum(0, (state, n) => [state + n, state + n])
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 1, 3, 6, 10, 15 ] }
*/

映射和扁平化

Stream.mapConcat 操作类似于 Stream.map，但它更进一步，将每个元素映射为零个或多个元素（作为 Iterable），然后扁平化整个流。这对于将每个元素转换为多个值特别有用。

示例（拆分和扁平化流）

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(numbers)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ '1', '2', '3', '4', '5', '6' ] }
*/

过滤

Stream.filter 操作允许您只通过满足特定条件的元素。这是一种在流中保留满足特定条件的元素同时丢弃其余元素的方法。

示例（过滤偶数）

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 2, 4, 6, 8, 10 ] }
*/

扫描

流扫描允许您将函数累积地应用于流中的每个元素，发出每个中间结果。与只提供最终结果的 reduce 不同，scan 提供累积过程的逐步视图。

示例（累积加法）

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 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))

Effect.runPromise(fold).then(console.log) // 输出：15

排空

流排空让您在流中执行有效操作，同时丢弃结果值。当您需要运行操作或执行副作用但不需要发出的值时，这很有用。Stream.drain 函数通过忽略流中的所有元素并产生空输出流来实现这一点。

示例（执行有效操作而不收集值）

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

const stream = Stream.repeatEffect(
  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))

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
示例输出：
random number: 7
random number: 5
random number: 0
{ _id: 'Chunk', values: [ 7, 5, 0 ] }
*/

const drained = Stream.drain(stream)

Effect.runPromise(Stream.runCollect(drained)).then(console.log)
/*
示例输出：
random number: 0
random number: 1
random number: 7
{ _id: 'Chunk', values: [] }
*/

检测流中的变化

Stream.changes 操作检测并发出与流中前面元素不同的元素。这对于跟踪变化或去重连续值很有用。

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

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 1, 2, 3, 4 ] }
*/

压缩

压缩将两个流中的元素组合成一个新流，配对每个输入流中的元素。这可以通过 Stream.zip 或 Stream.zipWith 实现，允许自定义配对逻辑。

示例（基本压缩）

在此示例中，两个流中的元素按顺序配对。当其中一个流耗尽时，结果流结束。

import { Stream, Effect } from "effect"

// 将两个流压缩在一起
const stream = Stream.zip(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c")
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ [ 1, 'a' ], [ 2, 'b' ], [ 3, 'c' ] ] }
*/

示例（自定义压缩逻辑）

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

import { Stream, Effect } from "effect"

// 使用自定义配对逻辑压缩两个流
const stream = Stream.zipWith(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c"),
  (n, s) => [n + 10, s + "!"]
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ [ 11, 'a!' ], [ 12, 'b!' ], [ 13, 'c!' ] ] }
*/

处理流结束

如果一个输入流在另一个之前结束，您可能希望使用默认值进行压缩以避免丢失配对。Stream.zipAll 和 Stream.zipAllWith 操作符提供此功能，允许您为任一流指定默认值。

示例（使用默认值压缩）

在此示例中，当第二个流完成时，第一个流继续使用 “x” 作为第二个流的默认值。

import { Stream, Effect } from "effect"

const stream = Stream.zipAll(Stream.make(1, 2, 3, 4, 5, 6), {
  other: Stream.make("a", "b", "c"),
  defaultSelf: -1,
  defaultOther: "x"
})

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    [ 1, 'a' ],
    [ 2, 'b' ],
    [ 3, 'c' ],
    [ 4, 'x' ],
    [ 5, 'x' ],
    [ 6, 'x' ]
  ]
}
*/

示例（使用 zipAllWith 的自定义逻辑）

使用 Stream.zipAllWith，自定义逻辑确定当任一流耗尽时如何组合元素，提供处理这些情况的灵活性。

import { Stream, Effect } from "effect"

const stream = Stream.zipAllWith(Stream.make(1, 2, 3, 4, 5, 6), {
  other: Stream.make("a", "b", "c"),
  onSelf: (n) => [n, "x"],
  onOther: (s) => [-1, s],
  onBoth: (n, s) => [n + 10, s + "!"]
})

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    [ 11, 'a!' ],
    [ 12, 'b!' ],
    [ 13, 'c!' ],
    [ 4, 'x' ],
    [ 5, 'x' ],
    [ 6, 'x' ]
  ]
}
*/

以不同速率压缩流

当组合以不同速度发出元素的流时，您可能不想等待较慢的流发出。使用 Stream.zipLatest 或 Stream.zipLatestWith，您可以在任一流产生新值时立即压缩元素。这些函数在较快流到达新值时使用较慢流的最新元素。

示例（组合具有不同发出速率的流）

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)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    [ 1, 'a' ],    // s1 发出 1 并与 s2 的最新值配对
    [ 1, 'b' ],    // s2 发出 'b'，与 s1 的最新值配对
    [ 2, 'b' ],    // s1 发出 2，与 s2 的最新值配对
    [ 2, 'c' ],    // s2 发出 'c'，与 s1 的最新值配对
    [ 2, 'd' ],    // s2 发出 'd'，与 s1 的最新值配对
    [ 3, 'd' ]     // s1 发出 3，与 s2 的最新值配对
  ]
}
*/

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

API	描述
zipWithPrevious	将流的每个元素与其前一个元素配对。
zipWithNext	将流的每个元素与其下一个元素配对。
zipWithPreviousAndNext	将每个元素与其前一个和下一个元素配对。

示例（将流元素与下一个元素配对）

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then((chunks) =>
  console.log("%o", chunks)
)
/*
输出：
{
  _id: 'Chunk',
  values: [
    [ 1, { _id: 'Option', _tag: 'Some', value: 2 }, [length]: 2 ],
    [ 2, { _id: 'Option', _tag: 'Some', value: 3 }, [length]: 2 ],
    [ 3, { _id: 'Option', _tag: 'Some', value: 4 }, [length]: 2 ],
    [ 4, { _id: 'Option', _tag: 'None' }, [length]: 2 ],
    [length]: 4
  ]
}
*/

索引流元素

Stream.zipWithIndex 操作符是为流中每个元素建立索引的有用工具，将每个项目与其在序列中的相应位置配对。当您想要跟踪流中元素的顺序时，这特别有用。

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

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    [ 'Mary', 0 ],
    [ 'James', 1 ],
    [ 'Robert', 2 ],
    [ 'Patricia', 3 ]
  ]
}
*/

流的笛卡尔积

Stream 模块包含计算两个流的_笛卡尔积_的功能，允许您创建来自两个不同流的元素组合。当您需要将一个集合中的每个元素与另一个集合中的每个元素配对时，这很有用。

简单来说，想象您有两个集合，并希望通过从每个集合中选择一个项目来形成所有可能的配对。这个配对过程就是笛卡尔积。在流中，此操作生成一个新流，其中包含来自两个输入流的元素的每种可能配对。

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

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

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)

Effect.runPromise(Stream.runCollect(cartesianProduct)).then(console.log)
/*
输出：
1
a
b
2
a
b
3
a
b
{
  _id: 'Chunk',
  values: [
    [ 1, 'a' ],
    [ 1, 'b' ],
    [ 2, 'a' ],
    [ 2, 'b' ],
    [ 3, 'a' ],
    [ 3, 'b' ]
  ]
}
*/

右侧流的多次迭代

请注意，右侧流（在此示例中为 s2）将被多次迭代， 左侧流（s1）中的每个元素都会迭代一次。如果右侧流涉及昂贵的或 产生副作用的操作，这些操作将被重复执行。

分区

分区流涉及根据指定条件将其分为两个不同的流。Stream 模块为此目的提供了两个函数：Stream.partition 和 Stream.partitionEither。让我们看看这些函数如何工作以及它们的最佳使用场景。

partition

Stream.partition 函数接受一个谓词（条件）作为输入，并将原始流分为两个子流。一个子流将包含满足条件的元素，而另一个包含不满足条件的元素。两个结果子流都包装在 Scope 类型中。

示例（将流分区为奇数和偶数）

import { Stream, Effect } from "effect"

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

Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* () {
      const [odds, evens] = yield* program
      console.log(yield* Stream.runCollect(odds))
      console.log(yield* Stream.runCollect(evens))
    })
  )
)
/*
输出：
{ _id: 'Chunk', values: [ 1, 3, 5, 7, 9 ] }
{ _id: 'Chunk', values: [ 2, 4, 6, 8 ] }
*/

partitionEither

在某些情况下，您可能需要使用涉及 effect 的条件来分区流。为此，Stream.partitionEither 函数是理想的选择。此函数使用有效谓词将流分为两个子流：一个用于产生 Either.left 值的元素，另一个用于产生 Either.right 值的元素。

示例（使用有效谓词分区流）

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

//      ┌─── Effect<[Stream<number>, Stream<number>], never, Scope>
//      ▼
const program = Stream.range(1, 9).pipe(
  Stream.partitionEither(
    // 模拟有效计算
    (n) => Effect.succeed(n % 2 === 0 ? Either.right(n) : Either.left(n)),
    { bufferSize: 5 }
  )
)

Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* () {
      const [odds, evens] = yield* program
      console.log(yield* Stream.runCollect(odds))
      console.log(yield* Stream.runCollect(evens))
    })
  )
)
/*
输出：
{ _id: 'Chunk', values: [ 1, 3, 5, 7, 9 ] }
{ _id: 'Chunk', values: [ 2, 4, 6, 8 ] }
*/

分组

在处理数据流时，您可能需要根据特定条件对元素进行分组。Stream 模块为此目的提供了几个函数：groupByKey、groupBy、grouped 和 groupedWithin。让我们回顾这些函数如何工作以及何时使用每个函数。

groupByKey

Stream.groupByKey 函数基于类型为 (a: A) => K 的键函数对流进行分区，其中 A 是流中元素的类型，K 表示分组的键。此函数是非有效的，通过简单地应用提供的键函数来对元素进行分组。

Stream.groupByKey 的结果是 GroupBy 数据类型，表示分组的流。要处理每个组，您可以使用 GroupBy.evaluate，它接受类型为 (key: K, stream: Stream<V, E>) => Stream.Stream<...> 的函数。此函数在所有组中操作并以非确定性顺序将它们合并在一起。

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

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

import { Stream, GroupBy, Effect, Chunk } from "effect"

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

// 定义考试结果列表
const examResults = [
  new Exam("Alex", 64),
  new Exam("Michael", 97),
  new Exam("Bill", 77),
  new Exam("John", 78),
  new Exam("Bobby", 71)
]

// 按分数的十位数对考试结果进行分组
const groupByKeyResult = Stream.fromIterable(examResults).pipe(
  Stream.groupByKey((exam) => Math.floor(exam.score / 10) * 10)
)

// 计算每组中考试结果的数量
const stream = GroupBy.evaluate(groupByKeyResult, (key, stream) =>
  Stream.fromEffect(
    Stream.runCollect(stream).pipe(
      Effect.andThen((chunk) => [key, Chunk.size(chunk)] as const)
    )
  )
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ [ 60, 1 ], [ 90, 1 ], [ 70, 3 ] ] }
*/

groupBy

对于分区涉及 effect 的更复杂分组需求，您可以使用 Stream.groupBy 函数。此函数接受有效分区函数并返回 GroupBy 数据类型，表示分组的流。然后，您可以使用 GroupBy.evaluate 处理每个组，类似于 Stream.groupByKey。

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

在以下示例中，我们按首字母对姓名进行分组，并计算每组中的姓名数量。在这里，分区操作设置为有效操作：

import { Stream, GroupBy, Effect, Chunk } from "effect"

// 按首字母对姓名进行分组
const groupByKeyResult = Stream.fromIterable([
  "Mary",
  "James",
  "Robert",
  "Patricia",
  "John",
  "Jennifer",
  "Rebecca",
  "Peter"
]).pipe(
  // 模拟有效的 groupBy 操作
  Stream.groupBy((name) => Effect.succeed([name.substring(0, 1), name]))
)

// 计算每组中的姓名数量并显示结果
const stream = GroupBy.evaluate(groupByKeyResult, (key, stream) =>
  Stream.fromEffect(
    Stream.runCollect(stream).pipe(
      Effect.andThen((chunk) => [key, Chunk.size(chunk)] as const)
    )
  )
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [ [ 'M', 1 ], [ 'J', 3 ], [ 'R', 2 ], [ 'P', 2 ] ]
}
*/

grouped

Stream.grouped 函数非常适合将流分为指定大小的块，使得在较小的有组织段中处理数据变得更容易。这在批量处理或显示数据时特别有用。

示例（将流分为 3 个元素的块）

import { Stream, Effect } from "effect"

// 创建数字流并将它们分组为 3 个元素的块
const stream = Stream.range(0, 8).pipe(Stream.grouped(3))

Effect.runPromise(Stream.runCollect(stream)).then((chunks) =>
  console.log("%o", chunks)
)
/*
输出：
{
  _id: 'Chunk',
  values: [
    { _id: 'Chunk', values: [ 0, 1, 2, [length]: 3 ] },
    { _id: 'Chunk', values: [ 3, 4, 5, [length]: 3 ] },
    { _id: 'Chunk', values: [ 6, 7, 8, [length]: 3 ] },
    [length]: 3
  ]
}
*/

groupedWithin

Stream.groupedWithin 函数允许灵活分组，通过基于指定的最大大小或时间间隔（以先满足的条件为准）创建块。这对于处理涉及时间约束的数据特别有用。

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

在此示例中，Stream.groupedWithin(18, "1.5 seconds") 在累积 18 个元素或自上次创建块以来经过 1.5 秒时将流分组为块。

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

// 创建每秒重复的流并按大小或时间分组
const stream = Stream.range(0, 9).pipe(
  Stream.repeat(Schedule.spaced("1 second")),
  Stream.groupedWithin(18, "1.5 seconds"),
  Stream.take(3)
)

Effect.runPromise(Stream.runCollect(stream)).then((chunks) =>
  console.log(Chunk.toArray(chunks))
)
/*
输出：
[
  {
    _id: 'Chunk',
    values: [
      0, 1, 2, 3, 4, 5, 6,
      7, 8, 9, 0, 1, 2, 3,
      4, 5, 6, 7
    ]
  },
  {
    _id: 'Chunk',
    values: [
      8, 9, 0, 1, 2,
      3, 4, 5, 6, 7,
      8, 9
    ]
  },
  {
    _id: 'Chunk',
    values: [
      0, 1, 2, 3, 4, 5, 6,
      7, 8, 9, 0, 1, 2, 3,
      4, 5, 6, 7
    ]
  }
]
*/

连接

在流处理中，您可能需要组合多个流的内容。Stream 模块提供了几个操作符来实现这一点，包括 Stream.concat、Stream.concatAll 和 Stream.flatMap。让我们看看这些操作符是如何工作的。

简单连接

Stream.concat 操作符是连接两个流的直接方法。它返回一个新流，该流发出第一个流（左侧）的元素，然后是第二个流（右侧）的元素。当您想要以特定顺序组合两个流时，这很有用。

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

import { Stream, Effect } from "effect"

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 1, 2, 3, 'a', 'b' ] }
*/

连接多个流

如果您有多个流要连接，Stream.concatAll 提供了一种高效的方法来组合它们，而无需手动链接多个 Stream.concat 操作。此函数接受一个流的 Chunk 并返回一个包含每个流元素的单个流。

示例（连接多个流）

import { Stream, Effect, Chunk } 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.concatAll<number | string | boolean, never, never>(
  Chunk.make(s1, s2, s3)
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    1,     2,     3,
    'a',   'b',   true,
    false, false
  ]
}
*/

使用 flatMap 的高级连接

Stream.flatMap 操作符允许高级连接，通过创建一个流，其中每个元素都是通过将类型为 (a: A) => Stream<...> 的函数应用于源流的每个输出而生成的。然后，此操作符连接所有结果流，有效地将它们扁平化。

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

import { Stream, Effect } from "effect"

// 创建一个流，其中每个元素重复 4 次
const stream = Stream.make(1, 2, 3).pipe(
  Stream.flatMap((a) => Stream.repeatValue(a).pipe(Stream.take(4)))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    1, 1, 1, 1, 2,
    2, 2, 2, 3, 3,
    3, 3
  ]
}
*/

如果您需要并发执行 flatMap 操作，可以使用并发选项来控制同时运行的内部流数量。

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

示例（使用 switch 选项）

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

// 创建带有日志记录的流的辅助函数
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}`)
      )
    )
  )

// 不使用 switch（默认行为）：
// 所有流都运行到完成
const stream1 = Stream.fromIterable([1, 2, 3]).pipe(
  Stream.flatMap(createStreamWithLogging)
)

// 使用 switch 行为：
// 只有最后一个流完成，当新值到达时
// 之前的流被取消
const stream2 = Stream.fromIterable([1, 2, 3]).pipe(
  Stream.flatMap(createStreamWithLogging, { switch: true })
)

// 按顺序运行示例以查看差异
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)
  })
)
/*
输出：
=== Without switch (all streams complete) ===
Starting stream for value: 1
Completed stream for value: 1
Starting stream for value: 2
Completed stream for value: 2
Starting stream for value: 3
Completed stream for value: 3
{ _id: 'Chunk', values: [ 1, 2, 3 ] }

=== With switch (only last stream completes) ===
Starting stream for value: 1
Interrupted stream for value: 1
Starting stream for value: 2
Interrupted stream for value: 2
Starting stream for value: 3
Completed stream for value: 3
{ _id: 'Chunk', values: [ 3 ] }
*/

switch 选项对于搜索功能、实时数据处理或任何希望在新输入到达时丢弃之前操作的情况特别有价值。

合并

有时，您可能希望交错来自两个流的元素并创建单个输出流。在这种情况下，Stream.concat 不合适，因为它等待第一个流完成后才消费第二个流。对于在元素可用时交错它们，Stream.merge 及其变体专为此目的而设计。

merge

Stream.merge 操作将来自两个源流的元素组合成单个流，在产生元素时交错它们。与 Stream.concat 不同，Stream.merge 不会等待一个流完成后才开始另一个流。

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

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, 5, 6).pipe(
  Stream.schedule(Schedule.spaced("200 millis"))
)

// 将 s1 和 s2 合并为交错其值的单个流
const merged = Stream.merge(s1, s2)

Effect.runPromise(Stream.runCollect(merged)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 1, 4, 2, 3, 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.repeatValue(0).pipe(
  Stream.schedule(Schedule.spaced("200 millis"))
)

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

Effect.runPromise(Stream.runCollect(merged)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    1, 0, 2, 3,
    0, 4, 5
  ]
}
*/

mergeWith

在某些情况下，您可能希望在合并两个流的同时将它们的元素转换为统一类型。Stream.mergeWith 专为此目的而设计，允许您为每个源流指定转换函数。

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

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.mergeWith(s1, s2, {
  // 将 `s1` 中的字符串元素转换为整数
  onSelf: (s) => parseInt(s),
  // 将 `s2` 中的小数元素向下舍入
  onOther: (n) => Math.floor(n)
})

Effect.runPromise(Stream.runCollect(merged)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 1, 4, 2, 3, 5, 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)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{ _id: 'Chunk', values: [ 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)

// 定义一个布尔流来控制交错
const booleanStream = Stream.make(true, false, false).pipe(Stream.forever)

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    1, 2,  4, 3, 6,
    8, 5, 10, 7, 9
  ]
}
*/

穿插

穿插在流中添加分隔符或词缀，对于格式化或构造流中的数据很有用。

intersperse

Stream.intersperse 操作符在流中每对元素之间插入指定的分隔符元素。此分隔符可以是任何选择的值，并在每个连续对之间添加。

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

import { Stream, Effect } from "effect"

// 创建数字流并在它们之间穿插 `0`
const stream = Stream.make(1, 2, 3, 4, 5).pipe(Stream.intersperse(0))

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    1, 0, 2, 0, 3,
    0, 4, 0, 5
  ]
}
*/

intersperseAffixes

对于更复杂的需求，Stream.intersperseAffixes 提供对流开始、元素之间和流结束处不同词缀的控制。

示例（向流添加词缀）

import { Stream, Effect } from "effect"

// 创建流并添加词缀：
// - 开始处为 `[`
// - 元素之间为 `|`
// - 结束处为 `]`
const stream = Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.intersperseAffixes({
    start: "[",
    middle: "|",
    end: "]"
  })
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Chunk',
  values: [
    '[', 1,   '|', 2,   '|',
    3,   '|', 4,   '|', 5,
    ']'
  ]
}
*/

广播

广播流创建多个下游流，每个流都从源流接收相同的元素。当您想要同时将每个元素发送给多个消费者时，这很有用。上游流有一个 maximumLag 参数，设置它在放慢速度以匹配最慢下游流的速度之前可以领先多少的限制。

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

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

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

const numbers = Effect.scoped(
  Stream.range(1, 20).pipe(
    Stream.tap((n) =>
      Console.log(`Emit ${n} element before broadcasting`)
    ),
    // 向 2 个下游消费者广播，最大延迟为 5
    Stream.broadcast(2, 5),
    Stream.flatMap(([first, second]) =>
      Effect.gen(function* () {
        // 第一个下游流：计算最大值
        const fiber1 = yield* Stream.runFold(first, 0, (acc, e) =>
          Math.max(acc, e)
        ).pipe(
          Effect.andThen((max) => Console.log(`Maximum: ${max}`)),
          Effect.fork
        )

        // 第二个下游流：延迟记录每个元素
        const fiber2 = yield* second.pipe(
          Stream.schedule(Schedule.spaced("1 second")),
          Stream.runForEach((n) =>
            Console.log(`Logging to the Console: ${n}`)
          ),
          Effect.fork
        )

        // 等待两个纤程完成
        yield* Fiber.join(fiber1).pipe(
          Effect.zip(Fiber.join(fiber2), { concurrent: true })
        )
      })
    ),
    Stream.runCollect
  )
)

Effect.runPromise(numbers).then(console.log)
/*
输出：
Emit 1 element before broadcasting
Emit 2 element before broadcasting
Emit 3 element before broadcasting
Emit 4 element before broadcasting
Emit 5 element before broadcasting
Emit 6 element before broadcasting
Emit 7 element before broadcasting
Emit 8 element before broadcasting
Emit 9 element before broadcasting
Emit 10 element before broadcasting
Emit 11 element before broadcasting
Logging to the Console: 1
Logging to the Console: 2
Logging to the Console: 3
Logging to the Console: 4
Logging to the Console: 5
Emit 12 element before broadcasting
Emit 13 element before broadcasting
Emit 14 element before broadcasting
Emit 15 element before broadcasting
Emit 16 element before broadcasting
Logging to the Console: 6
Logging to the Console: 7
Logging to the Console: 8
Logging to the Console: 9
Logging to the Console: 10
Emit 17 element before broadcasting
Emit 18 element before broadcasting
Emit 19 element before broadcasting
Emit 20 element before broadcasting
Logging to the Console: 11
Logging to the Console: 12
Logging to the Console: 13
Logging to the Console: 14
Logging to the Console: 15
Maximum: 20
Logging to the Console: 16
Logging to the Console: 17
Logging to the Console: 18
Logging to the Console: 19
Logging to the Console: 20
{ _id: 'Chunk', values: [ undefined ] }
*/

缓冲

Effect 流使用基于拉取的模型，允许下游消费者控制请求元素的速率。但是，当生产者和消费者之间的速度不匹配时，缓冲可以帮助平衡它们的交互。Stream.buffer 操作符专为管理这种情况而设计，允许生产者即使在消费者较慢时也能继续工作。您可以使用 capacity 选项设置最大缓冲容量。

buffer

Stream.buffer 操作符将元素排队，允许生产者独立于消费者工作，直到达到指定容量。当较快的生产者和较慢的消费者需要平稳运行而不相互阻塞时，这很有帮助。

示例（使用缓冲处理速度不匹配）

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

const stream = Stream.range(1, 10).pipe(
  // 在缓冲前记录每个元素
  Stream.tap((n) => Console.log(`before buffering: ${n}`)),
  // 缓冲容量为 4 个元素
  Stream.buffer({ capacity: 4 }),
  // 在缓冲后记录每个元素
  Stream.tap((n) => Console.log(`after buffering: ${n}`)),
  // 在每次发射之间添加 5 秒延迟
  Stream.schedule(Schedule.spaced("5 seconds"))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
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"

// 辅助函数，记录自上次记录以来的经过时间
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(
  // 在 200 毫秒后发出值 4
  Stream.concat(
    Stream.fromEffect(Effect.sleep("200 millis").pipe(Effect.as(4)))
  ),
  // 继续更多快速值
  Stream.concat(Stream.make(5, 6)),
  // 在 150 毫秒后发出 7
  Stream.concat(
    Stream.fromEffect(Effect.sleep("150 millis").pipe(Effect.as(7)))
  ),
  Stream.concat(Stream.make(8)),
  Stream.tap((n) => log(`Received ${n}`)),
  // 只在至少 100 毫秒的暂停后发出值
  Stream.debounce("100 millis"),
  Stream.tap((n) => log(`> Emitted ${n}`))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
示例输出：
Received 1 after 5ms
Received 2 after 2ms
Received 3 after 0ms
> Emitted 3 after 104ms
Received 4 after 99ms
Received 5 after 1ms
Received 6 after 0ms
> Emitted 6 after 101ms
Received 7 after 50ms
Received 8 after 1ms
> Emitted 8 after 101ms
{ _id: 'Chunk', values: [ 3, 6, 8 ] }
*/

节流

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

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

示例（节流配置）

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

在此配置中：

• 每个处理的块使用一个令牌（cost = () => 1）。
• 令牌以每 100 毫秒（duration: "100 millis"）一个令牌（units: 1）的速率补充。

节流适用于块，而非元素

请注意，节流操作的是块而不是单个元素。 cost 函数设置每个块的令牌成本。

塑形策略（默认）

“塑形”策略通过延迟块发射直到它们符合指定的带宽约束来调节数据流。 此策略确保数据吞吐量不超过定义的限制，允许稳定和受控的数据发射。

示例（使用塑形策略应用节流）

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

// 辅助函数，记录自上次记录以来的经过时间
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: Chunk.size,
    duration: "100 millis",
    units: 1
  }),
  Stream.tap((n) => log(`> Emitted ${n}`))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
示例输出：
Received 0 after 56ms
> Emitted 0 after 0ms
Received 1 after 52ms
> Emitted 1 after 48ms
Received 2 after 52ms
> Emitted 2 after 49ms
Received 3 after 52ms
> Emitted 3 after 48ms
Received 4 after 52ms
> Emitted 4 after 47ms
Received 5 after 52ms
> Emitted 5 after 49ms
{ _id: 'Chunk', values: [ 0, 1, 2, 3, 4, 5 ] }
*/

强制策略

“强制”策略通过丢弃超出带宽约束的块来严格调节数据流。

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

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

// 辅助函数，记录自上次记录以来的经过时间
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: Chunk.size,
    duration: "1 second",
    units: 1,
    strategy: "enforce"
  }),
  Stream.tap((n) => log(`> Emitted ${n}`))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
示例输出：
Received 1 after 106ms
> Emitted 1 after 1ms
Received 2 after 200ms
Received 3 after 402ms
Received 4 after 801ms
> Emitted 4 after 1ms
Received 5 after 1601ms
> Emitted 5 after 1ms
Received 6 after 3201ms
> Emitted 6 after 0ms
{ _id: 'Chunk', values: [ 1, 4, 5, 6 ] }
*/

burst 选项

Stream.throttle 函数提供了一个 burst 选项，允许临时增加数据吞吐量超出设定的速率限制。 此选项设置为大于 0 以激活突发能力（默认为 0，表示不支持突发）。 突发容量在令牌桶中提供额外的令牌，使流能够在数据突发时暂时超过其配置的速率。

示例（使用突发容量进行节流）

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

// 辅助函数，记录自上次记录以来的经过时间
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: Chunk.size,
    duration: "200 millis",
    units: 5,
    strategy: "enforce",
    burst: 2
  }),
  Stream.tap((n) => log(`> Emitted ${n}`))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
示例输出：
Received 0 after 16ms
> Emitted 0 after 0ms
Received 1 after 12ms
> Emitted 1 after 0ms
Received 2 after 11ms
> Emitted 2 after 0ms
Received 3 after 11ms
> Emitted 3 after 0ms
Received 4 after 11ms
> Emitted 4 after 1ms
Received 5 after 11ms
> Emitted 5 after 0ms
Received 6 after 12ms
> Emitted 6 after 0ms
Received 7 after 11ms
Received 8 after 12ms
Received 9 after 11ms
Received 10 after 11ms
> Emitted 10 after 0ms
Received 11 after 11ms
Received 12 after 11ms
Received 13 after 12ms
> Emitted 13 after 0ms
Received 14 after 11ms
Received 15 after 12ms
Received 16 after 11ms
Received 17 after 11ms
> Emitted 17 after 0ms
Received 18 after 12ms
Received 19 after 10ms
{
  _id: 'Chunk',
  values: [
    0, 1,  2,  3,  4,
    5, 6, 10, 13, 17
  ]
}
*/

在此设置中，流开始时桶中包含 5 个令牌，允许前五个块立即发出。 额外的 2 个突发容量暂时容纳进一步的发射，允许更灵活地处理后续数据。 随着时间的推移，当桶根据节流配置重新填充时，会发出额外的元素，展示了突发能力如何有效管理不均匀的数据流。

调度

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

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

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

// 创建一个在每个值之间有 1 秒延迟的流
const stream = Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.schedule(Schedule.spaced("1 second")),
  Stream.tap(Console.log)
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
1
2
3
4
5
{
  _id: "Chunk",
  values: [ 1, 2, 3, 4, 5 ]
}
*/

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