Resourceful Streams

Resources acquired by a stream must remain open for the whole period in which the stream is consumed. Compose Effect.acquireRelease, Stream.fromEffect, and Stream.scoped to tie the resource lifetime to the stream. Use Stream.ensuring when a stream only needs a finalizer.

Acquire Release

The following example acquires a file, emits its lines, and closes it when stream consumption ends.

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

3
// Simulating File operations
4
const open = (filename: string) =>
5
  Effect.gen(function* () {
6
    yield* Console.log(`Opening ${filename}`)
7
    return {
8
      getLines: Effect.succeed(["Line 1", "Line 2", "Line 3"]),
9
      close: Console.log(`Closing ${filename}`),
10
    }
11
  })
12

13
const stream = Stream.scoped(
14
  Stream.fromEffect(
15
    Effect.acquireRelease(open("file.txt"), (file) => file.close),
16
  ),
17
).pipe(Stream.flatMap((file) => Stream.fromIterableEffect(file.getLines)))
18

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

Effect.acquireRelease registers file.close in the scope created by Stream.scoped. The file therefore remains open while Stream.fromIterableEffect emits its contents.

Finalization

Stream.ensuring runs a finalizer after the stream’s own finalizers, whether the stream succeeds, fails, or is interrupted.

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

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

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

7
const program = application.pipe(
8
  Stream.ensuring(
9
    deleteDir("tmp").pipe(
10
      Effect.andThen(Console.log("Temporary directory was deleted.")),
11
    ),
12
  ),
13
)
14

15
await Effect.runPromise(Stream.runCollect(program)) // => [undefined]