Leftovers

In this section, we’ll look at handling elements left unconsumed by sinks. Sinks may process only a portion of the elements from an upstream source, leaving some elements as “leftovers.” Here’s how to collect or ignore these remaining elements.

Collecting Leftovers

If a sink doesn’t consume all elements from the upstream source, the remaining elements are called leftovers. Sink.mapEnd transforms both the sink result and its optional leftovers, so you can move the leftovers into the result returned by Stream.run.

Example (Collecting Leftover Elements)

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

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

5
// Take the first 3 elements and collect any leftovers
6
const sink1 = Sink.take<number>(3).pipe(
7
  Sink.mapEnd(([a, leftover]) => [[a, leftover ?? []] as const]),
8
)
9

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

12
// Take only the first element and collect the rest as leftovers
13
const sink2 = Sink.head<number>().pipe(
14
  Sink.mapEnd(([a, leftover]) => [[a, leftover ?? []] as const]),
15
)
16

17
await Effect.runPromise(Stream.run(stream, sink2)) // => [Option.some(1), [2, 3, 4, 5]]

Ignoring Leftovers

If leftover elements are not needed, you can ignore them using Sink.ignoreLeftover. This approach discards any unconsumed elements, so the sink operation focuses only on the elements it needs.

Example (Ignoring Leftover Elements)

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

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

5
// Take the first 3 elements and ignore any remaining elements
6
const sink = Sink.take<number>(3).pipe(
7
  Sink.ignoreLeftover,
8
  Sink.mapEnd(([a, leftover]) => [[a, leftover ?? []] as const]),
9
)
10

11
await Effect.runPromise(Stream.run(stream, sink)) // => [[1, 2, 3], []]