Introduction
In stream processing, a Sink is a construct designed to consume elements generated by a Stream.
┌─── Type of the result produced by the Sink | ┌─── Type of elements consumed by the Sink | | ┌─── Type of any leftover elements │ | | ┌─── Type of possible errors │ │ | | ┌─── Type of required dependencies ▼ ▼ ▼ ▼ ▼Sink<A, In, L, E, R>Here’s an overview of what a Sink does:
- It consumes a varying number of
Inelements, which may include zero, one, or multiple elements. - It can encounter errors of type
Eduring processing. - It produces a result of type
Aonce processing completes. - It can also return a remainder of type
L, representing any leftover elements.
To process a stream using a Sink, you can pass it directly to the Stream.run function:
Example (Using a Sink to Collect Stream Elements)
import { Stream, Sink, Effect } from "effect"
// ┌─── Stream<number, never, never>// ▼const stream = Stream.make(1, 2, 3)
// Create a sink to take the first 2 elements of the stream//// ┌─── Sink<Chunk<number>, number, number, never, never>// ▼const sink = Sink.take<number>(2)
// Run the stream through the sink to collect the elements//// ┌─── Effect<number, never, never>// ▼const sum = Stream.run(stream, sink)
Effect.runPromise(sum).then(console.log)/*Output:{ _id: 'Chunk', values: [ 1, 2 ] }*/The type of sink is as follows:
┌─── result | ┌─── consumed elements | | ┌─── leftover elements │ | | ┌─── no errors │ │ | | ┌─── no dependencies ▼ ▼ ▼ ▼ ▼Sink<Chunk<number>, number, number, never, never>Here’s the breakdown:
Chunk<number>: The final result produced by the sink after processing elements (in this case, a Chunk of numbers).number(first occurrence): The type of elements that the sink will consume from the stream.number(second occurrence): The type of leftover elements, if any, that are not consumed.never(first occurrence): Indicates that this sink does not produce any errors.never(second occurrence): Shows that no dependencies are required to operate this sink.