创建流

在本节中，我们将探索创建 Effect Stream 的各种方法。这些方法将帮助您生成满足需求的流。

常见构造函数

make

您可以使用 Stream.make 构造函数创建一个纯流。此构造函数接受可变数量的值作为其参数。

import { Stream, Effect } from "effect"

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

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

empty

有时，您可能需要一个不产生任何值的流。在这种情况下，您可以使用 Stream.empty。此构造函数创建一个保持为空的流。

import { Stream, Effect } from "effect"

const stream = Stream.empty

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

void

如果您需要一个包含单个 void 值的流，可以使用 Stream.void。当您想要表示具有单个事件或信号的流时，此构造函数很方便。

import { Stream, Effect } from "effect"

const stream = Stream.void

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// { _id: 'Chunk', values: [ undefined ] }

range

要创建指定范围 [min, max]（包括两个端点 min 和 max）内的整数流，可以使用 Stream.range。这对于生成连续数字流特别有用。

import { Stream, Effect } from "effect"

// 创建从 1 到 5 的数字流
const stream = Stream.range(1, 5)

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

iterate

使用 Stream.iterate，您可以通过对初始值迭代应用函数来生成流。初始值成为流产生的第一个元素，随后是由 f(init)、f(f(init)) 等产生的后续值。

import { Stream, Effect } from "effect"

// 创建递增数字流
const stream = Stream.iterate(1, (n) => n + 1) // 产生 1, 2, 3, ...

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// { _id: 'Chunk', values: [ 1, 2, 3, 4, 5 ] }

scoped

Stream.scoped 用于从作用域资源创建单值流。在处理需要显式获取、使用和释放的资源时，这很方便。

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

// 从作用域资源创建单值流
const stream = Stream.scoped(
  Effect.acquireUseRelease(
    Console.log("acquire"),
    () => Console.log("use"),
    () => Console.log("release")
  )
)

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

从成功和失败创建

与 Effect 数据类型非常相似，您可以使用 fail 和 succeed 函数生成 Stream：

import { Stream, Effect } from "effect"

// 创建可以发出错误的流
const streamWithError: Stream.Stream<never, string> =
  Stream.fail("Uh oh!")

Effect.runPromise(Stream.runCollect(streamWithError))
// throws Error: Uh oh!

// 创建发出数值的流
const streamWithNumber: Stream.Stream<number> = Stream.succeed(5)

Effect.runPromise(Stream.runCollect(streamWithNumber)).then(console.log)
// { _id: 'Chunk', values: [ 5 ] }

从块创建

您可以像这样从 Chunk 构造流：

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

// 从单个块创建包含值的流
const stream = Stream.fromChunk(Chunk.make(1, 2, 3))

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

此外，您也可以从多个 Chunk 创建流：

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

// 从多个块创建包含值的流
const stream = Stream.fromChunks(Chunk.make(1, 2, 3), Chunk.make(4, 5, 6))

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

从 Effect 创建

您可以通过使用 Stream.fromEffect 构造函数从 Effect 工作流生成流。例如，考虑以下生成单个随机数的流：

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

const stream = Stream.fromEffect(Random.nextInt)

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

此方法允许您无缝地将 Effect 的输出转换为流，为在流中处理异步操作提供了一种直接的方式。

从异步回调创建

想象您有一个依赖回调的异步函数。如果您想将这些回调发出的结果捕获为流，可以使用 Stream.async 函数。此函数旨在适配多次调用其回调并将结果作为流发出的函数。

让我们在以下示例中分解如何使用它：

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

const events = [1, 2, 3, 4]

const stream = Stream.async(
  (emit: StreamEmit.Emit<never, never, number, void>) => {
    events.forEach((n) => {
      setTimeout(() => {
        if (n === 3) {
          // 终止流
          emit(Effect.fail(Option.none()))
        } else {
          // 将当前项添加到流中
          emit(Effect.succeed(Chunk.of(n)))
        }
      }, 100 * n)
    })
  }
)

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

StreamEmit.Emit<R, E, A, void> 类型表示可以多次调用的异步回调。此回调接受类型为 Effect<Chunk<A>, Option<E>, R> 的值。以下是每种可能结果的含义：

• 当提供给回调的值在成功时产生 Chunk<A> 时，它表示指定的元素应作为流的一部分发出。

• 如果传递给回调的值导致 Some<E> 失败，它表示流以指定错误终止。

• 当传递给回调的值导致 None 失败时，它作为流结束的信号，本质上终止流。

简单来说，此类型允许您指定异步回调如何与流交互，确定何时发出元素、何时以错误终止或何时发出流结束信号。

从可迭代对象创建

fromIterable

您可以使用 Stream.fromIterable 构造函数从值的 Iterable 创建纯流。这是将值集合转换为流的直接方式。

import { Stream, Effect } from "effect"

const numbers = [1, 2, 3]

const stream = Stream.fromIterable(numbers)

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

fromIterableEffect

当您有一个产生 Iterable 类型值的 effect 时，可以使用 Stream.fromIterableEffect 构造函数从该 effect 生成流。

例如，假设您有一个检索用户列表的数据库操作。由于此操作涉及 effects，您可以利用 Stream.fromIterableEffect 将结果转换为 Stream：

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

class Database extends Context.Tag("Database")<
  Database,
  { readonly getUsers: Effect.Effect<Array<string>> }
>() {}

const getUsers = Database.pipe(Effect.andThen((_) => _.getUsers))

const stream = Stream.fromIterableEffect(getUsers)

Effect.runPromise(
  Stream.runCollect(
    stream.pipe(
      Stream.provideService(Database, {
        getUsers: Effect.succeed(["user1", "user2"])
      })
    )
  )
).then(console.log)
// { _id: 'Chunk', values: [ 'user1', 'user2' ] }

这使您能够无缝地处理 effects 并将其结果转换为流以进行进一步处理。

fromAsyncIterable

异步可迭代对象是另一种可以转换为流的数据源类型。使用 Stream.fromAsyncIterable 构造函数，您可以处理异步数据源并优雅地处理潜在错误。

import { Stream, Effect } from "effect"

const myAsyncIterable = async function* () {
  yield 1
  yield 2
}

const stream = Stream.fromAsyncIterable(
  myAsyncIterable(),
  (e) => new Error(String(e)) // 错误处理
)

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

在此代码中，我们定义了一个异步可迭代对象，然后从中创建了一个名为 stream 的流。此外，我们提供了一个错误处理函数来管理转换过程中可能发生的任何潜在错误。

从重复创建

重复单个值

您可以使用 Stream.repeatValue 构造函数创建无限重复特定值的流：

import { Stream, Effect } from "effect"

const stream = Stream.repeatValue(0)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// { _id: 'Chunk', values: [ 0, 0, 0, 0, 0 ] }

重复流的内容

Stream.repeat 允许您创建根据调度重复指定流内容的流。这对于生成重复事件或值很有用。

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

// 创建无限重复值的流
const stream = Stream.repeat(Stream.succeed(1), Schedule.forever)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// { _id: 'Chunk', values: [ 1, 1, 1, 1, 1 ] }

重复 Effect 的结果

想象您有一个有副作用的 API 调用，并且您想使用该调用的结果来创建流。您可以通过从 effect 创建流并无限重复它来实现这一点。

以下是生成随机数流的示例：

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

const stream = Stream.repeatEffect(Random.nextInt)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
/*
示例输出：
{
  _id: 'Chunk',
  values: [ 1666935266, 604851965, 2194299958, 3393707011, 4090317618 ]
}
*/

带终止条件的重复 Effect

您可以重复评估给定的 effect 并根据特定条件终止流。

在此示例中，我们正在耗尽 Iterator 以从中创建流：

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

const drainIterator = <A>(it: Iterator<A>): Stream.Stream<A> =>
  Stream.repeatEffectOption(
    Effect.sync(() => it.next()).pipe(
      Effect.andThen((res) => {
        if (res.done) {
          return Effect.fail(Option.none())
        }
        return Effect.succeed(res.value)
      })
    )
  )

生成时钟信号

您可以使用 Stream.tick 构造函数创建在指定间隔发出 void 值的流。这对于创建周期性事件很有用。

import { Stream, Effect } from "effect"

const stream = Stream.tick("100 millis")

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

从展开/分页创建

在函数式编程中，unfold 的概念可以被认为是 fold 的对应物。

使用 fold，我们处理数据结构并产生返回值。例如，我们可以取一个 Array<number> 并计算其元素的总和。

另一方面，unfold 表示一个操作，我们从初始值开始，使用指定的状态函数一次添加一个元素来生成递归数据结构。例如，我们可以创建从 1 开始的自然数序列，并使用 increment 函数作为状态函数。

展开

unfold

Stream 模块包含一个定义如下的 unfold 函数：

declare const unfold: <S, A>(
  initialState: S,
  step: (s: S) => Option.Option<readonly [A, S]>
) => Stream<A>

它的工作原理如下：

• initialState：这是初始状态值。
• step：状态函数 step 将当前状态 s 作为输入。如果此函数的结果是 None，流结束。如果是 Some<[A, S]>，流中的下一个元素是 A，状态 S 为下一步处理更新。

例如，让我们使用 Stream.unfold 创建自然数流：

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

const stream = Stream.unfold(1, (n) => Option.some([n, n + 1]))

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// { _id: 'Chunk', values: [ 1, 2, 3, 4, 5 ] }

unfoldEffect

有时，我们可能需要在展开操作期间执行有副作用的状态转换。这就是 Stream.unfoldEffect 派上用场的地方。它允许我们在生成流时处理 effects。

以下是使用 Stream.unfoldEffect 创建随机 1 和 -1 值的无限流的示例：

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

const stream = Stream.unfoldEffect(1, (n) =>
  Random.nextBoolean.pipe(
    Effect.map((b) => (b ? Option.some([n, -n]) : Option.some([n, n])))
  )
)

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

其他变体

还有类似的操作，如 Stream.unfoldChunk 和 Stream.unfoldChunkEffect，专为处理 Chunk 数据类型而定制。

分页

paginate

Stream.paginate 类似于 Stream.unfold，但允许进一步发出值。

例如，以下流发出 0, 1, 2, 3 元素：

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

const stream = Stream.paginate(0, (n) => [
  n,
  n < 3 ? Option.some(n + 1) : Option.none()
])

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

它的工作原理如下：

• 我们从初始值 0 开始。
• 提供的函数接受当前值 n 并返回一个元组。元组的第一个元素是要发出的值（n），第二个元素确定是否继续（Option.some(n + 1)）或停止（Option.none()）。

其他变体

还有类似的操作，如 Stream.paginateChunk 和 Stream.paginateChunkEffect，专为处理 Chunk 数据类型而定制。

展开与分页的对比

您可能想知道 unfold 和 paginate 组合器之间的区别以及何时使用其中一个。让我们通过深入一个示例来探索这一点。

想象我们有一个分页 API，以分页方式提供大量数据。当我们向此 API 发出请求时，它返回一个 ResultPage 对象，其中包含当前页面的结果和一个标志，指示这是否是最后一页或下一页是否有更多数据要检索。以下是我们 API 的简化表示：

import { Chunk, Effect } from "effect"

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

我们的目标是将此分页 API 转换为 RowData 事件流。对于我们的初次尝试，我们可能认为使用 Stream.unfold 操作是正确的方法：

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

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // 返回 3 页
    )
  )
}

const firstAttempt = Stream.unfoldChunkEffect(0, (pageNumber) =>
  listPaginated(pageNumber).pipe(
    Effect.map((page) => {
      if (page.isLast) {
        return Option.none()
      }
      return Option.some([page.results, pageNumber + 1] as const)
    })
  )
)

Effect.runPromise(Stream.runCollect(firstAttempt)).then(console.log)
/*
输出：
{
  _id: "Chunk",
  values: [ "Result 0-1", "Result 0-2", "Result 1-1", "Result 1-2" ]
}
*/

然而，这种方法有一个缺点，它不包括最后一页的结果。为了解决这个问题，我们执行额外的 API 调用来包含那些缺失的结果：

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

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

const secondAttempt = Stream.unfoldChunkEffect(
  Option.some(0),
  (pageNumber) =>
    Option.match(pageNumber, {
      // We already hit the last page
      onNone: () => Effect.succeed(Option.none()),
      // We did not hit the last page yet
      onSome: (pageNumber) =>
        listPaginated(pageNumber).pipe(
          Effect.map((page) =>
            Option.some([
              page.results,
              page.isLast ? Option.none() : Option.some(pageNumber + 1)
            ])
          )
        )
    })
)

Effect.runPromise(Stream.runCollect(secondAttempt)).then(console.log)
/*
Output:
{
  _id: 'Chunk',
  values: [
    'Result 0-1',
    'Result 0-2',
    'Result 1-1',
    'Result 1-2',
    'Result 2-1',
    'Result 2-2'
  ]
}
*/

虽然这种方法有效，但很明显 Stream.unfold 不是从分页 API 检索数据的最友好选项。它需要额外的解决方法来包含最后一页的结果。

这就是 Stream.paginate 发挥作用的地方。它提供了一种更符合人体工程学的方式将分页 API 转换为 Effect 流。让我们使用 Stream.paginate 重写我们的解决方案：

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

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

const finalAttempt = Stream.paginateChunkEffect(0, (pageNumber) =>
  listPaginated(pageNumber).pipe(
    Effect.andThen((page) => {
      return [
        page.results,
        page.isLast ? Option.none<number>() : Option.some(pageNumber + 1)
      ]
    })
  )
)

Effect.runPromise(Stream.runCollect(finalAttempt)).then(console.log)
/*
Output:
{
  _id: 'Chunk',
  values: [
    'Result 0-1',
    'Result 0-2',
    'Result 1-1',
    'Result 1-2',
    'Result 2-1',
    'Result 2-2'
  ]
}
*/

从队列和发布订阅创建

在 Effect 中，有两种基本的异步消息数据类型：Queue 和 PubSub。您可以分别利用 Stream.fromQueue 和 Stream.fromPubSub 轻松地将这些数据类型转换为 Stream。

从调度创建

我们可以从不需要任何进一步输入的 Schedule 创建流。流将为调度输出的每个值发出一个元素，只要调度继续就会继续：

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

// 每 1 秒发出值，总共发出 10 次
const schedule = Schedule.spaced("1 second").pipe(
  Schedule.compose(Schedule.recurs(10))
)

const stream = Stream.fromSchedule(schedule)

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