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. To capture these leftovers, use Sink.collectLeftover, which returns a tuple containing the result of the sink operation and any unconsumed elements.

Example (Collecting 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 collect any leftovers
6
const sink1 = Sink.take<number>(3).pipe(Sink.collectLeftover)
7

8
Effect.runPromise(Stream.run(stream, sink1)).then(console.log)
9
/*
10
Output:
11
[
12
  { _id: 'Chunk', values: [ 1, 2, 3 ] },
13
  { _id: 'Chunk', values: [ 4, 5 ] }
14
]
15
*/
16

17
// Take only the first element and collect the rest as leftovers
18
const sink2 = Sink.head<number>().pipe(Sink.collectLeftover)
19

20
Effect.runPromise(Stream.run(stream, sink2)).then(console.log)
21
/*
22
Output:
23
[
24
  { _id: 'Option', _tag: 'Some', value: 1 },
25
  { _id: 'Chunk', values: [ 2, 3, 4, 5 ] }
26
]
27
*/

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(Sink.ignoreLeftover, Sink.collectLeftover)
7

8
Effect.runPromise(Stream.run(stream, sink)).then(console.log)
9
/*
10
Output:
11
[ { _id: 'Chunk', values: [ 1, 2, 3 ] }, { _id: 'Chunk', values: [] } ]
12
*/