资源流

在 Stream 模块中，您会发现大多数构造函数都提供了一个特殊变体，专门用于将作用域资源提升到 Stream 中。当您使用这些特定构造函数时，您实际上是在创建在资源管理方面本质上安全的流。这些构造函数在创建流之前处理资源获取，在流使用后确保其正确关闭。

Stream 还为我们提供了 Stream.acquireRelease 和 Stream.finalizer 构造函数，它们与 Effect.acquireRelease 和 Effect.addFinalizer 有相似之处。这些工具使我们能够在流结束其操作之前执行清理或终结化任务。

获取释放

在本节中，我们将探索一个示例，演示在处理文件操作时使用 Stream.acquireRelease。

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

// 模拟文件操作
const open = (filename: string) =>
  Effect.gen(function* () {
    yield* Console.log(`Opening ${filename}`)
    return {
      getLines: Effect.succeed(["Line 1", "Line 2", "Line 3"]),
      close: Console.log(`Closing ${filename}`)
    }
  })

const stream = Stream.acquireRelease(
  open("file.txt"),
  (file) => file.close
).pipe(Stream.flatMap((file) => file.getLines))

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

在此代码片段中，我们使用 open 函数模拟文件操作。使用 Stream.acquireRelease 函数来确保文件正确打开和关闭，然后我们使用获取的资源处理文件的行。

终结化

在本节中，我们将探索流中终结化的概念。终结化允许我们在流结束之前执行特定操作。当我们想要执行清理任务或为流添加最后的修饰时，这特别有用。

想象一个场景，我们的流应用程序在完成执行时需要清理临时目录。我们可以使用 Stream.finalizer 函数来实现这一点：

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

const application = Stream.fromEffect(Console.log("Application Logic."))

const deleteDir = (dir: string) => Console.log(`Deleting dir: ${dir}`)

const program = application.pipe(
  Stream.concat(
    Stream.finalizer(
      deleteDir("tmp").pipe(
        Effect.andThen(Console.log("Temporary directory was deleted."))
      )
    )
  )
)

Effect.runPromise(Stream.runCollect(program)).then(console.log)
/*
输出：
Application Logic.
Deleting dir: tmp
Temporary directory was deleted.
{
  _id: "Chunk",
  values: [ undefined, undefined ]
}
*/

在此代码示例中，我们从由 application 流表示的应用程序逻辑开始。然后我们使用 Stream.finalizer 定义终结化步骤，该步骤删除临时目录并记录消息。这确保当应用程序完成执行时临时目录得到正确清理。

确保

在本节中，我们将探索一个需要在流终结化后执行操作的场景。为了实现这一点，我们可以使用 Stream.ensuring 操作符。

考虑这样一种情况：我们的应用程序已经完成了其主要逻辑并终结化了一些资源，但我们还需要在之后执行其他操作。我们可以为此目的使用 Stream.ensuring：

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

const program = Stream.fromEffect(Console.log("Application Logic.")).pipe(
  Stream.concat(Stream.finalizer(Console.log("Finalizing the stream"))),
  Stream.ensuring(
    Console.log("Doing some other works after stream's finalization")
  )
)

Effect.runPromise(Stream.runCollect(program)).then(console.log)
/*
输出：
Application Logic.
Finalizing the stream
Doing some other works after stream's finalization
{
  _id: "Chunk",
  values: [ undefined, undefined ]
}
*/

在此代码示例中，我们从由 Application Logic. 消息表示的应用程序逻辑开始。然后我们使用 Stream.finalizer 指定终结化步骤，该步骤记录 Finalizing the stream。之后，我们使用 Stream.ensuring 表示我们想要在流终结化后执行其他任务，产生消息 Performing additional tasks after stream's finalization。这确保我们的终结化后操作按预期执行。