Sinks in Effect

Overview

A Sink is a consumer of stream elements that produces a result:

Sink<A, In, L, E, R>
// A  - Result type (what sink produces)
// In - Input element type (what sink consumes)
// L  - Leftover type (unconsumed elements)
// E  - Error type
// R  - Required environment

Sinks are the counterpart to Streams - while Streams produce data, Sinks consume it.

Built-in Sinks

Collecting Elements

import { Stream, Sink } from "effect"

// Collect all elements to Chunk
const all = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(Sink.collectAll())
) // Chunk(1, 2, 3, 4, 5)

// Collect to Array
const array = yield* Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.collectAllToArray())
)

// Collect N elements
const firstThree = yield* Stream.range(1, 100).pipe(
  Stream.run(Sink.collectAllN(3))
) // Chunk(1, 2, 3)

// Collect while condition
const whileSmall = yield* Stream.iterate(1, (n) => n + 1).pipe(
  Stream.run(Sink.collectAllWhile((n) => n < 5))
) // Chunk(1, 2, 3, 4)

Aggregation Sinks

// Sum numbers
const total = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(Sink.sum)
) // 15

// Count elements
const count = yield* Stream.make("a", "b", "c").pipe(
  Stream.run(Sink.count)
) // 3

// Find first element
const first = yield* Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.head)
) // Option.some(1)

// Find last element
const last = yield* Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.last)
) // Option.some(3)

// Take N elements
const taken = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(Sink.take(3))
) // Chunk(1, 2, 3)

Folding Sinks

// Fold left (reduce)
const product = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(Sink.foldLeft(1, (acc, n) => acc * n))
) // 120

// Fold with early termination
const sumUntil100 = yield* Stream.iterate(1, (n) => n + 1).pipe(
  Stream.run(
    Sink.fold(
      0,
      (sum) => sum < 100,  // Continue while sum < 100
      (sum, n) => sum + n
    )
  )
)

// Fold with effect
const foldWithLog = Sink.foldEffect(
  0,
  (sum) => sum < 100,
  (sum, n) =>
    Effect.gen(function* () {
      yield* Effect.log(`Adding ${n} to ${sum}`)
      return sum + n
    })
)

Side Effect Sinks

// Execute effect for each element
yield* Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.forEach((n) => Effect.log(`Got: ${n}`)))
)

// Drain (consume all, return nothing)
yield* Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.drain)
)

Creating Custom Sinks

Sink.make

const maxSink = Sink.make<number, number, never, never, never>(
  // Initial state
  Number.NEGATIVE_INFINITY,
  // Process each element
  (max, n) => n > max ? n : max,
  // Extract result
  (max) => max
)

const max = yield* Stream.make(3, 1, 4, 1, 5, 9).pipe(
  Stream.run(maxSink)
) // 9

Sink.fromEffect

const logAndReturn = <A>(label: string) =>
  Sink.fromEffect(
    Effect.gen(function* () {
      yield* Effect.log(`Starting ${label}`)
      return [] as A[]
    })
  )

Sink.fromPush

For more control over the sink lifecycle:

const customSink = Sink.fromPush<number, number, never, never>((input) =>
  Effect.sync(() =>
    Option.match(input, {
      onNone: () => Either.left(finalResult),  // Stream ended
      onSome: (chunk) => {
        // Process chunk
        // Return Either.right to continue, Either.left to finish
        return Either.right(undefined)
      }
    })
  )
)

Sink Operations

Transforming Sinks

// Map over result
const doubledSum = Sink.sum.pipe(
  Sink.map((sum) => sum * 2)
)

// Contramap over input
const lengthSum = Sink.sum.pipe(
  Sink.contramap((s: string) => s.length)
)

// Dimap both
const processStrings = Sink.sum.pipe(
  Sink.dimap(
    (s: string) => s.length,  // Transform input
    (sum) => `Total length: ${sum}`  // Transform output
  )
)

Combining Sinks

// Zip two sinks (run both, combine results)
const sumAndCount = Sink.zip(Sink.sum, Sink.count)

const [sum, count] = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(sumAndCount)
) // [15, 5]

// Race two sinks (first to complete wins)
const firstOrSum = Sink.race(
  Sink.head,
  Sink.sum.pipe(Sink.map(Option.some))
)

Filtering

// Filter input before processing
const sumPositive = Sink.sum.pipe(
  Sink.filterInput((n: number) => n > 0)
)

const result = yield* Stream.make(-1, 2, -3, 4, -5).pipe(
  Stream.run(sumPositive)
) // 6

Leftovers

Sinks can leave unconsumed elements:

// Take returns leftovers
const takeThree = Sink.take<number>(3)

// When used with splitSink, leftovers go to next sink
const [first, rest] = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(
    Sink.take<number>(3).pipe(
      Sink.collectLeftover
    )
  )
)
// first: Chunk(1, 2, 3)
// rest: Chunk(4, 5) - leftovers

Sink Concurrency

Parallel Sinks

// Run sinks in parallel
const parallelSinks = Sink.zipPar(
  Sink.sum,
  Sink.count,
  Sink.collectAll<number>()
)

const [sum, count, all] = yield* Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.run(parallelSinks)
)

Chunked Processing

// Process in chunks for efficiency
const chunkedSum = Sink.foldChunks(
  0,
  () => true,
  (sum, chunk: Chunk.Chunk<number>) =>
    sum + Chunk.reduce(chunk, 0, (a, b) => a + b)
)

Common Patterns

Batched Database Insert

const batchInsert = (batchSize: number) =>
  Sink.collectAllN<Record>(batchSize).pipe(
    Sink.mapEffect((batch) =>
      Effect.tryPromise(() => db.insertMany(Chunk.toArray(batch)))
    )
  )

yield* recordStream.pipe(
  Stream.run(batchInsert(100))
)

Aggregation Pipeline

const stats = Sink.zip(
  Sink.sum,
  Sink.zip(Sink.count, Sink.zip(Sink.head, Sink.last))
).pipe(
  Sink.map(([sum, [count, [first, last]]]) => ({
    sum,
    count,
    average: count > 0 ? sum / count : 0,
    first,
    last
  }))
)

Write to File

const writeToFile = (path: string) =>
  Sink.forEach((line: string) =>
    Effect.gen(function* () {
      const fs = yield* FileSystem
      yield* fs.appendFileString(path, line + "\n")
    })
  )

Best Practices

1. Use built-in sinks when possible - Optimized and tested
2. Combine sinks with zip - Run multiple aggregations in one pass
3. Use foldChunks for efficiency - Process chunks, not elements
4. Handle leftovers - Consider unconsumed elements
5. Prefer Sink over forEach - More composable

Additional Resources

For comprehensive sink documentation, consult ${CLAUDEPLUGINROOT}/references/llms-full.txt.

Search for these sections:

• "Creating Sinks" for sink construction
• "Sink Operations" for transformations
• "Sink Concurrency" for parallel processing
• "Leftovers" for handling unconsumed elements