流中的错误处理

从失败中恢复

在处理可能遇到错误的流时，了解如何优雅地处理这些错误至关重要。Stream.orElse 函数是从失败中恢复并在出现错误时切换到替代流的强大工具。

示例

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("Oh! Error!")),
  Stream.concat(Stream.make(4, 5))
)

const s2 = Stream.make("a", "b", "c")

const stream = Stream.orElse(s1, () => s2)

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

在此示例中，s1 遇到错误，但我们不是终止流，而是使用 Stream.orElse 优雅地切换到 s2。这确保即使一个流失败，我们也可以继续处理数据。

还有一个名为 Stream.orElseEither 的变体，它使用 Either 数据类型根据成功或失败来区分来自两个流的元素：

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("Oh! Error!")),
  Stream.concat(Stream.make(4, 5))
)

const s2 = Stream.make("a", "b", "c")

const stream = Stream.orElseEither(s1, () => s2)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: "Chunk",
  values: [
    {
      _id: "Either",
      _tag: "Left",
      left: 1
    }, {
      _id: "Either",
      _tag: "Left",
      left: 2
    }, {
      _id: "Either",
      _tag: "Left",
      left: 3
    }, {
      _id: "Either",
      _tag: "Right",
      right: "a"
    }, {
      _id: "Either",
      _tag: "Right",
      right: "b"
    }, {
      _id: "Either",
      _tag: "Right",
      right: "c"
    }
  ]
}
*/

与 Stream.orElse 相比，Stream.catchAll 函数提供了高级错误处理功能。使用 Stream.catchAll，您可以根据遇到的失败的类型和值做出决策。

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("Uh Oh!" as const)),
  Stream.concat(Stream.make(4, 5)),
  Stream.concat(Stream.fail("Ouch" as const))
)

const s2 = Stream.make("a", "b", "c")

const s3 = Stream.make(true, false, false)

const stream = Stream.catchAll(
  s1,
  (error): Stream.Stream<string | boolean> => {
    switch (error) {
      case "Uh Oh!":
        return s2
      case "Ouch":
        return s3
    }
  }
)

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

在此示例中，我们有一个流 s1，它可能遇到两种不同类型的错误。我们不是像使用 Stream.orElse 那样直接切换到替代流，而是使用 Stream.catchAll 来精确确定如何处理每种类型的错误。这种对错误恢复的控制级别使您能够根据特定的错误条件选择不同的流或操作。

从缺陷中恢复

在处理流时，必须为各种失败场景做好准备，包括在流处理过程中可能发生的缺陷。为了解决这个问题，Stream.catchAllCause 函数提供了一个强健的解决方案。它使您能够优雅地处理和恢复可能出现的任何类型的失败。

示例

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.dieMessage("Boom!")),
  Stream.concat(Stream.make(4, 5))
)

const s2 = Stream.make("a", "b", "c")

const stream = Stream.catchAllCause(s1, () => s2)

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

在此示例中，s1 可能遇到缺陷，但我们不是让应用程序崩溃，而是使用 Stream.catchAllCause 优雅地切换到替代流 s2。这确保您的应用程序保持强健，即使面对意外问题也能继续处理数据。

从某些错误中恢复

在流处理中，可能存在需要从特定类型的失败中恢复的情况。Stream.catchSome 和 Stream.catchSomeCause 函数来拯救，允许您有选择地处理和缓解错误。

如果您想从特定错误中恢复，可以使用 Stream.catchSome：

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

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("Oh! Error!")),
  Stream.concat(Stream.make(4, 5))
)

const s2 = Stream.make("a", "b", "c")

const stream = Stream.catchSome(s1, (error) => {
  if (error === "Oh! Error!") {
    return Option.some(s2)
  }
  return Option.none()
})

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

要从特定原因中恢复，您可以使用 Stream.catchSomeCause 函数：

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

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.dieMessage("Oh! Error!")),
  Stream.concat(Stream.make(4, 5))
)

const s2 = Stream.make("a", "b", "c")

const stream = Stream.catchSomeCause(s1, (cause) => {
  if (Cause.isDie(cause)) {
    return Option.some(s2)
  }
  return Option.none()
})

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

恢复到 Effect

在流处理中，优雅地处理错误并在需要时执行清理任务至关重要。Stream.onError 函数允许我们做到这一点。如果我们的流遇到错误，我们可以指定要执行的清理任务。

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

const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.dieMessage("Oh! Boom!")),
  Stream.concat(Stream.make(4, 5)),
  Stream.onError(() =>
    Console.log(
      "Stream application closed! We are doing some cleanup jobs."
    ).pipe(Effect.orDie)
  )
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
输出：
Stream application closed! We are doing some cleanup jobs.
error: RuntimeException: Oh! Boom!
*/

重试失败的流

有时，流可能遇到临时或可恢复的失败。在这种情况下，Stream.retry 操作符很有用。它允许您指定重试调度，流将根据该调度进行重试。

示例

import { Stream, Effect, Schedule } from "effect"
import * as NodeReadLine from "node:readline"

const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(
    Stream.fromEffect(
      Effect.gen(function* () {
        const s = yield* readLine("Enter a number: ")
        const n = parseInt(s)
        if (Number.isNaN(n)) {
          return yield* Effect.fail("NaN")
        }
        return n
      })
    ).pipe(Stream.retry(Schedule.exponential("1 second")))
  )
)

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

const readLine = (message: string): Effect.Effect<string> =>
  Effect.promise(
    () =>
      new Promise((resolve) => {
        const rl = NodeReadLine.createInterface({
          input: process.stdin,
          output: process.stdout
        })
        rl.question(message, (answer) => {
          rl.close()
          resolve(answer)
        })
      })
  )

在此示例中，流要求用户输入一个数字，但如果输入了无效值（例如 “a”、“b”、“c”），它会因 “NaN” 而失败。但是，我们使用带有指数退避调度的 Stream.retry，这意味着它将在延迟递增的持续时间后重试。这允许我们处理临时错误并最终收集有效输入。

细化错误

在处理流时，可能存在您想要有选择地保留某些错误并用剩余错误终止流的情况。您可以使用 Stream.refineOrDie 函数来实现这一点。

示例

import { Stream, Option } from "effect"

const stream = Stream.fail(new Error())

const res = Stream.refineOrDie(stream, (error) => {
  if (error instanceof SyntaxError) {
    return Option.some(error)
  }
  return Option.none()
})

在此示例中，stream 最初因通用 Error 而失败。但是，我们使用 Stream.refineOrDie 来过滤并仅保留 SyntaxError 类型的错误。任何其他错误都将被终止，而 SyntaxErrors 将保留在 refinedStream 中。

超时

在处理流时，存在您可能想要处理超时的场景，例如如果流在特定持续时间内不产生值则终止流。在本节中，我们将探索如何使用各种操作符管理超时。

timeout

Stream.timeout 操作符允许您在流上设置超时。如果流在指定持续时间内不产生值，它将终止。

import { Stream, Effect } from "effect"

const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeout("2 seconds")
)

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

timeoutFail

Stream.timeoutFail 操作符将超时与自定义失败消息结合起来。如果流超时，它会因指定的错误消息而失败。

import { Stream, Effect } from "effect"

const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutFail(() => "timeout", "2 seconds")
)

Effect.runPromiseExit(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Exit',
  _tag: 'Failure',
  cause: { _id: 'Cause', _tag: 'Fail', failure: 'timeout' }
}
*/

timeoutFailCause

与 Stream.timeoutFail 类似，Stream.timeoutFailCause 将超时与自定义失败原因结合起来。如果流超时，它会因指定的原因而失败。

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

const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutFailCause(() => Cause.die("timeout"), "2 seconds")
)

Effect.runPromiseExit(Stream.runCollect(stream)).then(console.log)
/*
输出：
{
  _id: 'Exit',
  _tag: 'Failure',
  cause: { _id: 'Cause', _tag: 'Die', defect: 'timeout' }
}
*/

timeoutTo

The Stream.timeoutTo operator allows you to switch to another stream if the first stream does not produce a value within the specified duration.

import { Stream, Effect } from "effect"

const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutTo("2 seconds", Stream.make(1, 2, 3))
)

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