Creating Sinks

In stream processing, sinks are used to consume and handle elements from a stream. Here, we’ll explore various sink constructors that allow you to create sinks for specific tasks.

Common Constructors

head

The Sink.head sink retrieves only the first element from a stream, wrapping it in Some. If the stream has no elements, it returns None.

Example (Retrieving the First Element)

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

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

5
await Effect.runPromise(Stream.run(nonEmptyStream, Sink.head())) // => Option.some(1)
6

7
const emptyStream = Stream.empty
8

9
await Effect.runPromise(Stream.run(emptyStream, Sink.head())) // => Option.none()

last

The Sink.last sink retrieves only the last element from a stream, wrapping it in Some. If the stream has no elements, it returns None.

Example (Retrieving the Last Element)

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

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

5
await Effect.runPromise(Stream.run(nonEmptyStream, Sink.last())) // => Option.some(4)
6

7
const emptyStream = Stream.empty
8

9
await Effect.runPromise(Stream.run(emptyStream, Sink.last())) // => Option.none()

count

The Sink.count sink consumes all elements of the stream and counts the number of elements fed to it.

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

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

5
await Effect.runPromise(Stream.run(stream, Sink.count)) // => 4

sum

The Sink.sum sink consumes all elements of the stream and sums incoming numeric values.

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

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

5
await Effect.runPromise(Stream.run(stream, Sink.sum)) // => 10

take

The Sink.take sink takes the specified number of values from the stream and returns them in an array.

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

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

5
await Effect.runPromise(Stream.run(stream, Sink.take(3))) // => [1, 2, 3]

drain

The Sink.drain sink ignores its inputs, effectively discarding them.

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

3
const stream = Stream.make(1, 2, 3, 4).pipe(Stream.tap(Console.log))
4

5
await Effect.runPromise(Stream.run(stream, Sink.drain)) // => undefined

timed

The Sink.timed sink executes the stream and measures its execution time, providing the Duration.

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

3
const stream = Stream.make(1, 2, 3, 4).pipe(
4
  Stream.schedule(Schedule.spaced("100 millis")),
5
)
6

7
Effect.runPromise(Stream.run(stream, Sink.timed)).then(console.log)
8
/*
9
Output:
10
{ _id: 'Duration', _tag: 'Millis', millis: 408 }
11
*/

forEach

The Sink.forEach sink executes the provided effectful function for every element fed to it.

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

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

5
await Effect.runPromise(Stream.run(stream, Sink.forEach(Console.log))) // => undefined

Creating Sinks from Success and Failure

Just as you can define streams to hold or manipulate data, you can also create sinks with specific success or failure outcomes using the Sink.fail and Sink.succeed functions.

Succeeding Sink

This example creates a sink that doesn’t consume any elements from its upstream source but instead immediately succeeds with a specified numeric value:

Example (Sink that Always Succeeds with a Value)

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

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

5
await Effect.runPromise(Stream.run(stream, Sink.succeed(0))) // => 0

Failing Sink

In this example, the sink also doesn’t consume any elements from its upstream source. Instead, it fails with a specified error message of type string:

Example (Sink that Always Fails with an Error Message)

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

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

5
await Effect.runPromiseExit(Stream.run(stream, Sink.fail("fail!"))) // => Exit.fail("fail!")

Collecting

Collecting All Elements

To gather all elements from a data stream into an array, use the Sink.collect sink.

The final output contains all elements from the stream in the order they were emitted.

Example (Collecting All Stream Elements)

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

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

5
await Effect.runPromise(Stream.run(stream, Sink.collect())) // => [1, 2, 3, 4]

Collecting a Specified Number

To collect a fixed number of elements from a stream into an array, use Sink.take. This sink stops collecting once it reaches the specified limit.

Example (Collecting a Limited Number of Elements)

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

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

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    // Collect the first 3 elements into an array
9
    Sink.take(3),
10
  ),
11
) // => [1, 2, 3]

Collecting While Meeting a Condition

To gather elements from a stream while they satisfy a specific condition, use Sink.takeWhile. This sink collects elements until the provided predicate returns false.

Example (Collecting Elements Until a Condition Fails)

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

3
const stream = Stream.make(1, 2, 0, 4, 0, 6, 7)
4

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    // Collect elements while they are not equal to 0
9
    Sink.takeWhile((n) => n !== 0),
10
  ),
11
) // => [1, 2]

Collecting into a HashSet

To accumulate stream elements into a native Set, fold them with Sink.reduce. This ensures that each element appears only once in the final set.

Example (Collecting Unique Elements into a HashSet)

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

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

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    Sink.reduce(
9
      () => new Set<number>(),
10
      (s, n) => s.add(n),
11
    ),
12
  ),
13
) // => new Set([1, 2, 3, 4])

Collecting into HashSets of a Specific Size

For controlled collection into a Set with a specified maximum size, fold with Sink.reduceWhile, stopping once the set reaches the given limit.

Example (Collecting Unique Elements with a Set Size Limit)

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

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

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    // Collect unique elements, limiting the set size to 3
9
    Sink.reduceWhile(
10
      () => new Set<number>(),
11
      (s) => s.size < 3,
12
      (s, n) => s.add(n),
13
    ),
14
  ),
15
) // => new Set([1, 2, 3])

Collecting into a HashMap

For more complex collection scenarios, fold elements into a native Map<K, A> with Sink.reduce, using a key function to define each element’s grouping and a merge function to combine values sharing the same key.

Example (Grouping and Merging Stream Elements in a HashMap)

In this example, we use (n) => n % 3 to determine map keys and (a, b) => a + b to merge elements with the same key:

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

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

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    Sink.reduce(
9
      () => new Map<number, number>(),
10
      (m, n) => {
11
        const key = n % 3 // Key function to group by element value
12
        return m.set(key, m.has(key) ? m.get(key)! + n : n) // Merge function to sum values with the same key
13
      },
14
    ),
15
  ),
16
) // => new Map([[1, 3], [0, 6], [2, 7]])

Collecting into a HashMap with Limited Keys

To accumulate elements into a native Map with a maximum number of keys, fold with Sink.reduceWhile, stopping once the map reaches the specified key limit. This requires a key function to define the grouping of each element and a merge function to combine values with the same key.

Example (Limiting Collected Keys in a HashMap)

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

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

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    Sink.reduceWhile(
9
      () => new Map<number, number>(),
10
      (m) => m.size < 3, // Stop once the map has 3 keys
11
      (m, n) => {
12
        const key = n // Key function to group by element value
13
        return m.set(key, m.has(key) ? m.get(key)! + n : n) // Merge function to sum values with the same key
14
      },
15
    ),
16
  ),
17
) // => new Map([[1, 1], [3, 3], [2, 2]])

Folding

Reducing Elements

If you want to reduce a stream into a single cumulative value by applying an operation to each element in sequence, you can use the Sink.reduce function.

Example (Summing Elements in a Stream with Sink.reduce)

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

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

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    // Use reduce to sequentially add each element, starting with 0
9
    Sink.reduce(
10
      () => 0,
11
      (a, b) => a + b,
12
    ),
13
  ),
14
) // => 10

Folding with Termination

Sometimes, you may want to fold elements in a stream but stop the process once a specific condition is met. This is known as “short-circuiting.” You can accomplish this with the Sink.fold function, which lets you define a termination condition.

Example (Folding with a Condition to Stop Early)

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

3
const stream = Stream.iterate(0, (n) => n + 1)
4

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    Sink.fold(
9
      () => 0, // Initial value
10
      (sum) => sum <= 10, // Termination condition
11
      (a, b) => Effect.succeed(a + b), // Folding operation
12
    ),
13
  ),
14
) // => 15

Folding Until a Limit

To accumulate elements until a specific count is reached, use Sink.foldUntil. This sink folds elements up to the specified limit and then stops.

Example (Accumulating a Set Number of Elements)

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

3
const stream = Stream.make(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
4

5
await Effect.runPromise(
6
  Stream.run(
7
    stream,
8
    // Fold elements, stopping after accumulating 3 values
9
    Sink.foldUntil(
10
      () => 0,
11
      3,
12
      (a, b) => Effect.succeed(a + b),
13
    ),
14
  ),
15
) // => 6

Folding with Weighted Elements

In some scenarios, you may want to fold elements based on a defined “weight” or “cost,” accumulating elements until a specified maximum cost is reached. You can accomplish this by building a custom sink out of Sink.fold, whose termination condition checks the accumulated cost instead of a plain element count.

Example (Accumulating Elements Based on Weight)

In the example below, each element has a weight of 1, and the folding resets when the accumulated weight hits 3.

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

3
const foldWeighted = <A>(cost: (a: A) => number, maxCost: number) =>
4
  Sink.fold<{ readonly elements: Array<A>; readonly cost: number }, A>(
5
    () => ({ elements: [], cost: 0 }),
6
    (state) => state.cost < maxCost, // Keep accumulating while under the max cost
7
    (state, a) =>
8
      Effect.succeed({
9
        elements: [...state.elements, a],
10
        cost: state.cost + cost(a),
11
      }),
12
  ).pipe(Sink.map((state) => state.elements))
13

14
const stream = Stream.make(3, 2, 4, 1, 5, 6, 2, 1, 3, 5, 6).pipe(
15
  Stream.transduce(
16
    foldWeighted(
17
      () => 1, // Each element has a weight of 1
18
      3, // Maximum accumulated cost
19
    ),
20
  ),
21
)
22

23
await Effect.runPromise(Stream.runCollect(stream)) // => [[3, 2, 4], [1, 5, 6], [2, 1, 3], [5, 6]]