Sink Operations

In previous sections, we learned how to create and use sinks. Now, let’s explore some operations that let you transform or filter sink behavior.

Adapting Sink Input

At times, you may have a sink that works with one type of input, but your current stream uses a different type. The Sink.mapInput function helps you adapt your sink to a new input type by transforming the input values. While Sink.map changes the sink’s output, Sink.mapInput changes the input it accepts.

Example (Converting String Input to Numeric for Summing)

Suppose you have a Sink.sum that calculates the sum of numbers. If your stream contains strings rather than numbers, Sink.mapInput can convert those strings into numbers, allowing Sink.sum to work with your stream:

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

3
// A stream of numeric strings
4
const stream = Stream.make("1", "2", "3", "4", "5")
5

6
// Define a sink for summing numeric values
7
const numericSum = Sink.sum
8

9
// Use mapInput to adapt the sink, converting strings to numbers
10
const stringSum = numericSum.pipe(
11
  Sink.mapInput((s: string) => Number.parseFloat(s)),
12
)
13

14
await Effect.runPromise(Stream.run(stream, stringSum)) // => 15

Transforming Both Input and Output

When you need to transform both the input and output of a sink, combine Sink.mapInput with Sink.map. Together they let you transform the input type, perform the operation, and then transform the output to a new type. This can be useful for complete conversions between input and output types.

Example (Converting Input to Integer, Summing, and Converting Output to String)

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

3
// A stream of numeric strings
4
const stream = Stream.make("1", "2", "3", "4", "5")
5

6
// Convert string inputs to numbers, sum them,
7
// then convert the result to a string
8
const sumSink = Sink.sum.pipe(
9
  // Transform input: string to number
10
  Sink.mapInput((s: string) => Number.parseFloat(s)),
11
  // Transform output: number to string
12
  Sink.map((n) => String(n)),
13
)
14

15
await Effect.runPromise(Stream.run(stream, sumSink)) // => "15"

Filtering Input

You can filter the elements a sink ends up processing by filtering the stream that feeds it, before handing it off with Stream.transduce. This lets you combine a filter with a sink like Sink.take that only cares about elements meeting certain criteria.

Example (Filtering Negative Numbers in Groups of Three)

In the example below, elements are collected into arrays of three, but only positive numbers are included:

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

3
// Define a stream with positive, negative, and zero values
4
const stream = Stream.fromIterable([
5
  1, -2, 0, 1, 3, -3, 4, 2, 0, 1, -3, 1, 1, 6,
6
]).pipe(
7
  // Filter out non-positive numbers before grouping
8
  Stream.filter((n) => n > 0),
9
  // Collect the remaining elements in groups of 3
10
  Stream.transduce(Sink.take(3)),
11
)
12

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