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 a single Chunk, 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
Effect.runPromise(collectedData).then(console.log)
8
/*
9
Output:
10
{
11
  _id: "Chunk",
12
  values: [ 1, 2, 3, 4, 5 ]
13
}
14
*/

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(Stream.runForEach((n) => Console.log(n)))
4

5
Effect.runPromise(effect).then(console.log)
6
/*
7
Output:
8
1
9
2
10
3
11
undefined
12
*/

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

Using a Fold Operation

The Stream.fold function is another way to consume a stream by performing a fold operation over the stream of values and returning an effect containing the result. Here are a couple of examples:

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

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

5
Effect.runPromise(foldedStream).then(console.log)
6
// Output: 15
7

8
const foldedWhileStream = Stream.make(1, 2, 3, 4, 5).pipe(
9
  Stream.runFoldWhile(
10
    0,
11
    (n) => n <= 3,
12
    (a, b) => a + b,
13
  ),
14
)
15

16
Effect.runPromise(foldedWhileStream).then(console.log)
17
// Output: 6

In the first example (foldedStream), we use Stream.runFold to calculate the sum of all elements. In the second example (foldedWhileStream), we use Stream.runFoldWhile to calculate the sum but only until a certain condition is met.

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
Effect.runPromise(effect).then(console.log)
6
// Output: 6

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