Consuming Streams

When working with streams, it’s essential to understand how to consume the data they produce. In this guide, we’ll walk through several common methods for consuming streams.

Using runCollect

To gather all the elements from a stream into an array, you can use the Stream.runCollect function.

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

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

5
const collectedData = Stream.runCollect(stream)
6

7
await Effect.runPromise(collectedData) // => [1, 2, 3, 4, 5]

Using runForEach

Another way to consume elements of a stream is by using Stream.runForEach. It takes a callback function that receives each element of the stream. Here’s an example:

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

3
const effect = Stream.make(1, 2, 3).pipe(
4
  Stream.runForEach((n) => Console.log(n)),
5
)
6

7
await Effect.runPromise(effect) // => undefined

In this example, we use Stream.runForEach to log each element to the console.

Using runFold

Stream.runFold consumes a stream by reducing its values and returns an effect containing the result. For early termination, use Stream.runForEachWhile and keep the accumulator local to an Effect.suspend block.

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

3
const foldedStream = Stream.make(1, 2, 3, 4, 5).pipe(
4
  Stream.runFold(
5
    () => 0,
6
    (a, b) => a + b,
7
  ),
8
)
9

10
await Effect.runPromise(foldedStream) // => 15
11

12
const foldedWhileStream = Effect.suspend(() => {
13
  let acc = 0
14
  return Stream.make(1, 2, 3, 4, 5)
15
    .pipe(
16
      Stream.runForEachWhile((n) => {
17
        acc = acc + n
18
        return Effect.succeed(acc <= 3)
19
      }),
20
    )
21
    .pipe(Effect.map(() => acc))
22
})
23

24
await Effect.runPromise(foldedWhileStream) // => 6

In the first example, Stream.runFold calculates the sum of all elements. In the second, Stream.runForEachWhile stops after the accumulator exceeds 3; the element that makes the predicate false has already been consumed, so the result is 6.

Using a Sink

To consume a stream using a Sink, you can pass the Sink to the Stream.run function. Here’s an example:

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

3
const effect = Stream.make(1, 2, 3).pipe(Stream.run(Sink.sum))
4

5
await Effect.runPromise(effect) // => 6

In this example, we use a Sink to calculate the sum of the elements in the stream.