Resourceful Streams

The Stream module provides tools for working with resources that require explicit management. With functions like Stream.scoped and Stream.acquireRelease, you can create streams that handle resource acquisition and release automatically. Additionally, the Stream.finalizer and Stream.ensuring operators allow you to add final cleanup steps, ensuring any necessary post-stream actions are executed.

scoped

Stream.scoped is used to create a single-valued stream from a scoped resource. It can be handy when dealing with resources that require explicit acquisition, usage, and release.

Example (Creating a Scoped Stream)

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

// Creating a single-valued stream from a scoped resource
const stream = Stream.scoped(
  Effect.acquireUseRelease(
    Console.log("acquire"), // Resource acquisition
    () => Console.log("use"), // Resource usage within stream
    () => Console.log("release") // Resource release after stream
  )
)

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

acquireRelease

Stream.acquireRelease is used to manage resources within a stream, ensuring safe acquisition and release. This is particularly useful when working with resources that require explicit opening and closing, such as file handling. Stream.acquireRelease ensures that resources are acquired at the start and released after the stream completes, even if an error occurs.

Example (Acquiring and Releasing a File Resource)

In this example, we simulate file operations with an open function. The Stream.acquireRelease function guarantees that the file is properly opened and closed, and we process its contents while the file is open.

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

// Simulate file operations
const open = (filename: string) =>
  Effect.gen(function* () {
    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 // Ensure file is closed after stream completes
).pipe(
  Stream.flatMap(
    (file) => file.getLines // Use the acquired resource
  )
)

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

finalizer

The Stream.finalizer function creates a single-element stream that never fails and executes the finalizer when it ends.

Example (Executing a Finalizer After Stream Completion)

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

const program = Stream.fromEffect(Console.log("Application Logic.")).pipe(
  Stream.concat(Stream.finalizer(Console.log("Finalizing the stream")))
)

Effect.runPromise(Stream.runCollect(program)).then(console.log)
/*
Output:
Application Logic.
Finalizing the stream
{ _id: 'Chunk', values: [ undefined, undefined ] }
*/

ensuring

The Stream.ensuring function allows for additional actions to be performed after a stream’s finalization. This is helpful for tasks you want to complete after the stream has fully shut down.

Example (Executing Post-Finalization Actions)

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)
/*
Output:
Application Logic.
Finalizing the stream
Doing some other works after stream's finalization
{ _id: 'Chunk', values: [ undefined, undefined ] }
*/

The effect passed to Stream.ensuring will execute even if the stream encounters an error, ensuring that the post-finalization actions run reliably.

Example (Executing Post-Finalization Actions After an Error)

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

const program = Stream.fromEffect(Effect.fail("Oh no!")).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))
/*
Output:
Doing some other works after stream's finalization
[Error: Oh no!] {
  name: '(FiberFailure) Error',
  [Symbol(effect/Runtime/FiberFailure/Cause)]: {
    _tag: 'Fail',
    error: 'Oh no!'
  }
  ...
}
*/