Sink

A Sink<A, In, L, E, R> is the consumer side of streaming. Where a Stream describes how values are produced, a Sink describes how they are consumed: it pulls In elements, may fail with E, requires services R, and produces a result A once it is done — optionally returning L leftover elements it did not use. You run a stream through a sink with Stream.run, and every run* destructor you have already seen (runCollect, runFold, …) is really a specialised sink underneath.

     ┌─── result produced by the Sink
     │  ┌─── elements consumed by the Sink
     │  │   ┌─── leftover elements
     │  │   │  ┌─── possible errors
     │  │   │  │  ┌─── required services
     ▼  ▼   ▼  ▼  ▼
  Sink<A, In, L, E, R>

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

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

// `Sink.take(2)` consumes exactly two elements and returns them as an array.
// Its type is `Sink<Array<number>, number, number>` — it produces an array,
// consumes numbers, and reports the unused elements ([3, 4, 5]) as leftovers.
const firstTwo = Sink.take<number>(2)

// `Stream.run` feeds the stream into the sink and yields the sink's result.
export const result: Effect.Effect<Array<number>> = numbers.pipe(
  Stream.run(firstTwo)
)
// => [1, 2]

Built-in sinks

The Sink module ships the consumers you reach for most often. Each is a value you can pass to Stream.run:

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

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

// `Sink.collect()` gathers every element into an array.
export const all = numbers.pipe(Stream.run(Sink.collect<number>()))
// => [1, 2, 3, 4, 5]

// `Sink.sum` and `Sink.count` aggregate without retaining elements.
export const total = numbers.pipe(Stream.run(Sink.sum)) // => 15
export const howMany = numbers.pipe(Stream.run(Sink.count)) // => 5

// `Sink.head()` / `Sink.last()` capture an edge element as an Option.
export const first = numbers.pipe(Stream.run(Sink.head<number>()))

// `Sink.forEach` runs an effect per element and produces void — a streaming
// equivalent of `Stream.runForEach`.
export const logged = numbers.pipe(
  Stream.run(Sink.forEach((n) => Effect.logInfo(`got ${n}`)))
)

// `Sink.drain` consumes and discards everything, producing void.
export const drained = numbers.pipe(Stream.run(Sink.drain))

Folding and early termination

Like a stream, a sink can fold elements into a single accumulated value — and, crucially, it can stop early. Sink.fold takes an initial state, a while predicate, and a step function; it keeps consuming only while the predicate holds. Elements pulled past the stopping point come back as leftovers.

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

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

// Sum elements while the running total stays below 10. As soon as it reaches 10
// the sink completes; remaining input becomes leftover.
const sumUntilTen = Sink.fold(
  () => 0, // initial state (a thunk so it can be re-evaluated safely)
  (total) => total < 10, // keep going while this holds
  (total, n: number) => Effect.succeed(total + n) // effectful step
)

export const partialSum = numbers.pipe(Stream.run(sumUntilTen))

Sink.reduce is the simpler, non-effectful cousin for folding without a continuation predicate, and Sink.takeWhile collects the longest prefix matching a predicate. For batch-level folding over each pulled array there is Sink.foldArray.

Transforming sinks

Sinks compose. You can map their result, adapt their input, recover from failures, and even sequence one after another.

import { Sink, Stream } from "effect"

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

// `Sink.map` transforms the *result* once the sink completes.
const average = Sink.collect<number>().pipe(
  Sink.map((arr) => arr.length === 0 ? 0 : arr.reduce((a, b) => a + b, 0) / arr.length)
)

// `Sink.mapInput` adapts the *input* side, letting a `Sink<_, number>` consume
// a stream of strings by parsing first.
const sumOfParsed = Sink.sum.pipe(
  Sink.mapInput((s: string) => Number(s))
)

// `Sink.flatMap` runs a second sink after the first, using the first result to
// decide what to do next. Here we take 2 elements, then drain the rest.
const takeThenDrain = Sink.take<number>(2).pipe(
  Sink.flatMap((firstTwo) =>
    Sink.drain.pipe(Sink.map(() => firstTwo))
  )
)

export const avg = numbers.pipe(Stream.run(average))
export const head2 = numbers.pipe(Stream.run(takeThenDrain))

Sink.orElse switches to a fallback sink if the first one fails, and Sink.catchCause recovers based on the full Cause — the same error-handling shape you use for streams and effects.

Building a custom sink

When the built-ins do not fit, Sink.make<In>() lets you describe a consumer as a pipeline over a Stream<In>. The final step must return an Effect; its success value becomes the sink’s result. This is how you turn any stream-shaped computation into a reusable consumer.

import { Stream, Sink } from "effect"

// A custom sink that collects elements and returns both the count and the sum,
// expressed as a small stream pipeline ending in an Effect.
const countAndSum = Sink.make<number>()((stream: Stream.Stream<number>) =>
  stream.pipe(
    Stream.runFold(
      () => ({ count: 0, sum: 0 }),
      (acc, n) => ({ count: acc.count + 1, sum: acc.sum + n })
    )
  )
)

export const summary = Stream.make(10, 20, 30).pipe(Stream.run(countAndSum))
// => { count: 3, sum: 60 }

Because a sink is a first-class value, you can name it, test it in isolation, and reuse it across many streams — the same way you reuse an Effect. Reach for an explicit sink whenever a consumption strategy is worth a name; for one-off consumption, the run* destructors are the shorter path.

---

API reference

Everything below is run with Stream.run(stream, sink) (or stream.pipe(Stream.run(sink))). Sinks pair naturally with the grouping combinators on the transforming streams page (Stream.transduce / Stream.aggregate repeatedly feed a stream through a sink) and with the run* destructors.

Leftovers and the End tuple

When a sink finishes it produces an End<A, L> — a tuple [value, leftover?]. The optional second element is a non-empty array of input that was pulled from upstream but not consumed. Stream.run returns only the value; leftovers matter when one sink hands off to another (flatMap, orElse) or when you build parsers. The In type parameter is contravariant, so a sink accepting broader input can be used where narrower input is expected.

Constructors

make

Creates a pipe-style sink builder over input type In. The first pipeline step receives the input as a Stream<In>; the final step must return an Effect whose success becomes the result.

import { Sink, Stream } from "effect"

const lastTwo = Sink.make<number>()((s: Stream.Stream<number>) =>
  s.pipe(Stream.takeRight(2), Stream.runCollect)
)
// Stream.run(Stream.make(1, 2, 3), lastTwo) // => [2, 3]

fromChannel

Builds a sink from a Channel that processes non-empty arrays of input and emits an End<A, L>. Low-level; pairs with toChannel.

import { Sink } from "effect"

// const sink = Sink.fromChannel(channel)
// channel: Channel<never, E, End<A, L>, NonEmptyReadonlyArray<In>, never, void, R>

fromTransform

The lowest-level constructor: you receive the upstream pull and the active scope and return an effect completing with the sink’s End value. Every other constructor is built on this.

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

// Always succeed with 0, consuming nothing.
const zero = Sink.fromTransform(() => Effect.succeed([0] as const))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(zero))
// => 0

fromEffect

Ignores all upstream input and completes with the success value of an effect.

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

const sink = Sink.fromEffect(Effect.succeed("done"))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => "done"

fromEffectEnd

Like fromEffect, but the effect returns an End<A, L> so it can supply both a result and explicit leftovers.

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

const sink = Sink.fromEffectEnd(Effect.succeed([42] as const))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => 42

fromQueue

Offers every consumed element to a Queue, then ends the queue when upstream finishes. Completes with void.

import { Cause, Effect, Queue, Sink, Stream } from "effect"

export const program = Effect.gen(function* () {
  // `fromQueue` ends the queue when upstream finishes, so the queue's error
  // channel must include `Cause.Done`.
  const queue = yield* Queue.make<number, Cause.Done>()
  yield* Stream.make(1, 2, 3).pipe(Stream.run(Sink.fromQueue(queue)))
  return yield* Queue.takeAll(queue)
})
// => [1, 2, 3]

fromPubSub

Publishes every consumed element to a PubSub. Completes with void when upstream ends.

import { PubSub, Sink, Stream } from "effect"

declare const pubsub: PubSub.PubSub<number>
const sink = Sink.fromPubSub(pubsub)
// Stream.make(1, 2, 3).pipe(Stream.run(sink)) // => void, publishing each element

succeed

Immediately ends with a constant value (and optional leftovers), ignoring input.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.succeed(42)))
// => 42

sync

Immediately ends with a lazily evaluated value.

import { Sink } from "effect"

const sink = Sink.sync(() => 42)
// Stream.run(stream, sink) // => 42, computed when the sink runs

suspend

Defers construction of a sink to when it is run — useful for per-run mutable state.

import { Sink, Stream } from "effect"

const counter = Sink.suspend(() => Sink.count)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(counter))
// => 3

fail

A sink that always fails with the given typed error.

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

const sink = Sink.fail("boom" as const)
Effect.runPromiseExit(Stream.make(1).pipe(Stream.run(sink)))
// => Exit.fail("boom")

failSync

A sink that fails with a lazily evaluated error.

import { Sink } from "effect"

const sink = Sink.failSync(() => new Error("late"))
// Stream.run(stream, sink) // => fails with Error: late

failCause

A sink that halts with a specified Cause (e.g. a typed failure or a defect).

import { Cause, Sink } from "effect"

const sink = Sink.failCause(Cause.fail("oops"))
// Stream.run(stream, sink) // => fails with "oops"

failCauseSync

A sink that halts with a lazily evaluated Cause.

import { Cause, Sink } from "effect"

const sink = Sink.failCauseSync(() => Cause.die("defect"))
// Stream.run(stream, sink) // => dies with "defect"

die

A sink that halts with an unrecoverable defect.

import { Sink } from "effect"

const sink = Sink.die(new Error("unexpected"))
// Stream.run(stream, sink) // => dies with Error: unexpected

never

A sink that never completes. Useful as a placeholder or to keep a fiber alive.

import { Sink } from "effect"

const sink = Sink.never
// Stream.run(stream, sink) // => never resolves

drain

Consumes and discards all input, completing with void.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.drain))
// => undefined

Folding

fold

Folds input element by element with an effectful step, continuing while the while predicate holds. Stopping mid-array yields the remaining elements as leftovers.

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

const sumUnder10 = Sink.fold(
  () => 0,
  (total) => total < 10,
  (total, n: number) => Effect.succeed(total + n)
)
export const r = Stream.make(4, 5, 6, 7).pipe(Stream.run(sumUnder10))
// => 15 (4 + 5 + 6, stops once >= 10)

foldArray

Folds each pulled non-empty array at once with an effectful step, continuing while the predicate holds. Batch-level; produces no leftovers.

import { Effect, Sink } from "effect"

const totalLengths = Sink.foldArray(
  () => 0,
  () => true,
  (acc, chunk: ReadonlyArray<string>) => Effect.succeed(acc + chunk.length)
)
// Stream.run(stream, totalLengths) // => total number of elements seen

foldUntil

Folds elements until a fixed maximum number have been consumed (or the stream ends). Extra elements from the final array become leftovers.

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

// Sum at most 3 elements.
const sumFirst3 = Sink.foldUntil(
  () => 0,
  3,
  (acc, n: number) => Effect.succeed(acc + n)
)
export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(sumFirst3))
// => 6 (1 + 2 + 3)

Reducing

reduce

Reduces every input element into state with a pure step. Consumes the whole stream.

import { Sink, Stream } from "effect"

const product = Sink.reduce(() => 1, (acc, n: number) => acc * n)
export const r = Stream.make(1, 2, 3, 4).pipe(Stream.run(product))
// => 24

reduceArray

Reduces each pulled non-empty array into state with a pure step.

import { Sink } from "effect"

const concat = Sink.reduceArray(
  () => "",
  (acc, chunk: ReadonlyArray<string>) => acc + chunk.join("")
)
// Stream.run(stream, concat) // => all chunks concatenated

reduceEffect

Reduces every element into state with an effectful step (can fail, can require services).

import { Effect, Sink } from "effect"

const sumLogged = Sink.reduceEffect(
  () => 0,
  (acc, n: number) => Effect.as(Effect.logInfo(`+${n}`), acc + n)
)
// Stream.run(stream, sumLogged) // => the sum, logging each element

reduceWhile

Reduces with a pure step while a predicate on the state holds; stops otherwise, returning unconsumed elements as leftovers.

import { Sink } from "effect"

const untilNegative = Sink.reduceWhile(
  () => 0,
  (sum) => sum >= 0,
  (sum, n: number) => sum + n
)
// Stream.run(stream, untilNegative) // => sum, stopping once it goes negative

reduceWhileEffect

Effectful variant of reduceWhile: the step is an Effect.

import { Effect, Sink } from "effect"

const sink = Sink.reduceWhileEffect(
  () => 0,
  (sum) => sum < 100,
  (sum, n: number) => Effect.succeed(sum + n)
)
// Stream.run(stream, sink) // => running sum, stops at >= 100

reduceWhileArray

Reduces non-empty input arrays with a pure step while a predicate holds. No leftovers.

import { Sink } from "effect"

const sink = Sink.reduceWhileArray(
  () => 0,
  (n) => n < 10,
  (n, chunk: ReadonlyArray<number>) => n + chunk.length
)
// Stream.run(stream, sink) // => count, stopping once it reaches 10

reduceWhileArrayEffect

Effectful array-level reduce that checks the predicate before consuming the next array. No leftovers.

import { Effect, Sink } from "effect"

const sink = Sink.reduceWhileArrayEffect(
  () => 0,
  (n) => n < 10,
  (n, chunk: ReadonlyArray<number>) => Effect.succeed(n + chunk.length)
)
// Stream.run(stream, sink) // => count, stopping once it reaches 10

Aggregates

sum

Sums all numeric input.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3, 4).pipe(Stream.run(Sink.sum))
// => 10

count

Counts the number of elements consumed.

import { Sink, Stream } from "effect"

export const r = Stream.make("a", "b", "c").pipe(Stream.run(Sink.count))
// => 3

collect

Accumulates every element into an array.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.collect<number>()))
// => [1, 2, 3]

head

The first element as Option, or Option.none() if the stream is empty. Later elements from the same pulled array become leftovers.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.head<number>()))
// => Option.some(1)

last

The final element as Option, or Option.none() if empty. Only completes when upstream ends.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.last<number>()))
// => Option.some(3)

find

The first element matching a predicate (or Refinement, which narrows the result type), as Option. Returns Option.none() if none match.

import { Sink, Stream } from "effect"

const firstEven = Sink.find((n: number) => n % 2 === 0)
export const r = Stream.make(1, 3, 4, 5).pipe(Stream.run(firstEven))
// => Option.some(4)

findEffect

Like find but the predicate is an Effect (can fail / require services).

import { Effect, Sink } from "effect"

const sink = Sink.findEffect((n: number) => Effect.succeed(n > 10))
// Stream.run(stream, sink) // => Option of the first element > 10

every

true only if all elements satisfy the predicate.

import { Sink, Stream } from "effect"

const allPositive = Sink.every((n: number) => n > 0)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(allPositive))
// => true

some

true if any element satisfies the predicate.

import { Sink, Stream } from "effect"

const anyNegative = Sink.some((n: number) => n < 0)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(anyNegative))
// => false

Taking

take

Collects exactly up to n elements into an array. Extra elements from the final array become leftovers; n <= 0 yields an empty array.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(Sink.take<number>(3)))
// => [1, 2, 3]

takeWhile

Collects the longest prefix satisfying a predicate (or Refinement). The first failing element is consumed and excluded; later elements become leftovers.

import { Sink, Stream } from "effect"

const ascendingPrefix = Sink.takeWhile((n: number) => n < 4)
export const r = Stream.make(1, 2, 3, 4, 1).pipe(Stream.run(ascendingPrefix))
// => [1, 2, 3]

takeWhileFilter

Applies a Filter while it succeeds, collecting each transformed output. The first failure stops collection.

import { Filter, Result, Sink, Stream } from "effect"

// Keep numbers below 5, mapping each to its double; stop at the first failure.
const doubleWhileSmall = Sink.takeWhileFilter(
  Filter.make((n: number) => (n < 5 ? Result.succeed(n * 2) : Result.fail(n)))
)
export const r = Stream.make(1, 2, 5, 3).pipe(Stream.run(doubleWhileSmall))
// => [2, 4]

takeWhileEffect

Collects elements while an effectful predicate returns true. The first false stops collection; later elements become leftovers.

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

const sink = Sink.takeWhileEffect((n: number) => Effect.succeed(n < 4))
export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(sink))
// => [1, 2, 3]

takeWhileFilterEffect

Applies a FilterEffect while it succeeds, collecting each transformed output.

import { Effect, Filter, Result, Sink } from "effect"

const sink = Sink.takeWhileFilterEffect(
  Filter.makeEffect((n: number) =>
    Effect.succeed(n < 5 ? Result.succeed(n) : Result.fail(n))
  )
)
// Stream.run(Stream.make(1, 2, 9), sink) // => [1, 2]

takeUntil

Collects elements until the predicate returns true, including the matching element.

import { Sink, Stream } from "effect"

const upToFirstEven = Sink.takeUntil((n: number) => n % 2 === 0)
export const r = Stream.make(1, 3, 4, 5).pipe(Stream.run(upToFirstEven))
// => [1, 3, 4]

takeUntilEffect

Like takeUntil but the predicate is an Effect.

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

const sink = Sink.takeUntilEffect((n: number) => Effect.succeed(n > 3))
export const r = Stream.make(1, 2, 4, 5).pipe(Stream.run(sink))
// => [1, 2, 4]

ForEach

forEach

Runs an effect for every element, discarding the results, producing void.

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

export const program = Stream.make(1, 2, 3).pipe(
  Stream.run(Sink.forEach((n) => Effect.logInfo(`item ${n}`)))
)
// => void, logging "item 1", "item 2", "item 3"

forEachArray

Runs an effect once per pulled non-empty array, producing void.

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

export const program = Stream.make(1, 2, 3).pipe(
  Stream.run(
    Sink.forEachArray((chunk) => Effect.logInfo(`chunk of ${chunk.length}`))
  )
)
// => void

forEachWhile

Runs an effect per element while it returns true; stops when it returns false or upstream ends.

import { Effect, Sink } from "effect"

const sink = Sink.forEachWhile((n: number) =>
  Effect.as(Effect.logInfo(`saw ${n}`), n < 3)
)
// Stream.run(Stream.make(1, 2, 3, 4), sink) // => void, logs 1, 2, 3 then stops

forEachWhileArray

Runs an effect per non-empty array while it returns true.

import { Effect, Sink } from "effect"

const sink = Sink.forEachWhileArray((chunk: ReadonlyArray<number>) =>
  Effect.succeed(chunk.length > 0)
)
// Stream.run(stream, sink) // => void

Transforming a sink

map

Transforms the result once the sink completes.

import { Sink, Stream } from "effect"

const countAsString = Sink.count.pipe(Sink.map((n) => `count=${n}`))
export const r = Stream.make(1, 2).pipe(Stream.run(countAsString))
// => "count=2"

as

Replaces the result with a constant, preserving input consumption.

import { Sink, Stream } from "effect"

const done = Sink.drain.pipe(Sink.as("done" as const))
export const r = Stream.make(1, 2, 3).pipe(Stream.run(done))
// => "done"

mapInput

Adapts the input side with a pure function so the sink can consume a different element type.

import { Sink, Stream } from "effect"

const sumOfLengths = Sink.sum.pipe(Sink.mapInput((s: string) => s.length))
export const r = Stream.make("ab", "cde").pipe(Stream.run(sumOfLengths))
// => 5

mapInputEffect

Adapts the input side with an effectful function (can fail / require services).

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

const sink = Sink.sum.pipe(
  Sink.mapInputEffect((s: string) => Effect.succeed(Number(s)))
)
export const r = Stream.make("1", "2", "3").pipe(Stream.run(sink))
// => 6

mapInputArray

Transforms each pulled non-empty input array before it reaches the sink.

import { Array, Sink, Stream } from "effect"

const sink = Sink.sum.pipe(
  Sink.mapInputArray((arr) => Array.map(arr, (s: string) => s.length))
)
export const r = Stream.make("a", "bb").pipe(Stream.run(sink))
// => 3

mapInputArrayEffect

Effectful variant of mapInputArray, transforming each input array via an Effect.

import { Array, Effect, Sink } from "effect"

const sink = Sink.sum.pipe(
  Sink.mapInputArrayEffect((arr) =>
    Effect.succeed(Array.map(arr, (s: string) => Number(s)))
  )
)
// Stream.run(Stream.make("1", "2"), sink) // => 3

mapEnd

Transforms the full End tuple, so you can change both the result and the leftovers.

import { Sink } from "effect"

// Take 2 elements but drop the reported leftovers.
const noLeftover = Sink.take<number>(2).pipe(Sink.mapEnd(([value]) => [value]))
// Stream.run(stream, noLeftover) // => [1, 2], leftover discarded

mapEffectEnd

Effectfully transforms the full End tuple (result + leftovers).

import { Effect, Sink } from "effect"

const sink = Sink.take<number>(2).pipe(
  Sink.mapEffectEnd(([value, leftover]) =>
    Effect.succeed([value.length, leftover] as const)
  )
)
// Stream.run(stream, sink) // => 2

mapEffect

Transforms the result with an Effect (can fail / require services).

import { Effect, Sink } from "effect"

const parsed = Sink.head<string>().pipe(
  Sink.mapEffect((opt) => Effect.succeed(opt))
)
// Stream.run(stream, parsed) // => Option of the first element

mapError

Transforms the error channel.

import { Sink } from "effect"

const sink = Sink.fail("boom" as const).pipe(
  Sink.mapError((e) => new Error(e))
)
// Stream.run(stream, sink) // => fails with Error: boom

mapLeftover

Transforms each leftover element.

import { Sink } from "effect"

const sink = Sink.take<number>(2).pipe(Sink.mapLeftover((n) => n * 10))
// leftovers (e.g. [3, 4, 5]) become [30, 40, 50]

flatMap

Runs a second sink after the first, choosing it from the first result. Leftovers of the first sink are fed to the second before pulling more upstream input.

import { Sink, Stream } from "effect"

const takeThenCount = Sink.take<number>(2).pipe(
  Sink.flatMap((first) => Sink.count.pipe(Sink.map((rest) => ({ first, rest }))))
)
export const r = Stream.make(1, 2, 3, 4, 5).pipe(Stream.run(takeThenCount))
// => { first: [1, 2], rest: 3 }

ignoreLeftover

Drops any leftovers a sink would produce, keeping the result. Does not pull additional input.

import { Sink, Stream } from "effect"

const firstTwo = Sink.take<number>(2).pipe(Sink.ignoreLeftover)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(firstTwo))
// => [1, 2], leftover [3] discarded

Combinators and error handling

orElse

Switches to a fallback sink (built from the error) if the first sink fails, continuing to consume from the same upstream.

import { Sink, Stream } from "effect"

const sink = Sink.fail("nope" as const).pipe(
  Sink.orElse(() => Sink.succeed("recovered" as const))
)
export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => "recovered"

catch

Handles typed errors with an effectful fallback value (exported as Sink.catch).

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

const sink = Sink.fail("nope" as const).pipe(
  Sink.catch((e) => Effect.succeed(`handled: ${e}`))
)
export const r = Stream.make(1).pipe(Stream.run(sink))
// => "handled: nope"

catchCause

Recovers based on the full Cause (defects, interruption, typed errors).

import { Cause, Effect, Sink, Stream } from "effect"

const sink = Sink.fail("nope" as const).pipe(
  Sink.catchCause((cause) => Effect.succeed(Cause.squash(cause)))
)
export const r = Stream.make(1).pipe(Stream.run(sink))
// => "nope"

Timing

summarized

Runs a summary effect before and after the sink, pairing the result with a value computed from the two summaries.

import { Clock, Duration, Sink } from "effect"

const timedDrain = Sink.drain.pipe(
  Sink.summarized(
    Clock.currentTimeMillis,
    (start, end) => Duration.millis(end - start)
  )
)
// Stream.run(stream, timedDrain) // => [void, Duration]

withDuration

Pairs the sink’s result with the time it took to run.

import { Sink } from "effect"

const sink = Sink.withDuration(Sink.collect<number>())
// Stream.run(stream, sink) // => [Array<number>, Duration]

timed

Drains all input and returns the elapsed duration.

import { Sink, Stream } from "effect"

export const r = Stream.make(1, 2, 3).pipe(Stream.run(Sink.timed))
// => Duration (elapsed while draining)

Lifecycle and context

onExit

Runs an effect after the sink completes, fails, or is interrupted, receiving the result Exit. The original result and leftovers are preserved.

import { Effect, Sink } from "effect"

const sink = Sink.collect<number>().pipe(
  Sink.onExit((exit) => Effect.logInfo(`done: ${exit._tag}`))
)
// Stream.run(stream, sink) // => the collected array, logging on exit

ensuring

Runs a finalizer effect after the sink completes, fails, or is interrupted.

import { Effect, Sink } from "effect"

const sink = Sink.drain.pipe(
  Sink.ensuring(Effect.logInfo("sink finished"))
)
// Stream.run(stream, sink) // => void, always logs at the end

provideContext

Provides a Context to the sink, removing those services from its requirements.

import { Context, Sink } from "effect"

declare const context: Context.Context<never>
const sink = Sink.drain.pipe(Sink.provideContext(context))

provideService

Provides a single service implementation, removing it from the requirements.

import { Context, Effect, Sink, Stream } from "effect"

class Config extends Context.Service<Config, { readonly factor: number }>()(
  "Config"
) {}

const sink = Sink.reduceEffect(
  () => 0,
  (acc, n: number) => Config.pipe(Effect.map((c) => acc + n * c.factor))
).pipe(Sink.provideService(Config, { factor: 2 }))

export const r = Stream.make(1, 2, 3).pipe(Stream.run(sink))
// => 12

unwrap

Builds a sink from a (possibly scoped) effect that produces a sink.

import { Effect, Sink } from "effect"

const sink = Sink.unwrap(
  Effect.succeed(Sink.forEach((n: number) => Effect.logInfo(`${n}`)))
)
// Stream.run(stream, sink) // => void

Interop and guards

toChannel

Converts a sink back into the underlying Channel.

import { Sink } from "effect"

const channel = Sink.toChannel(Sink.succeed(42))

isSink

Type guard that checks whether a value is a Sink.

import { Sink } from "effect"

console.log(Sink.isSink(Sink.never)) // => true
console.log(Sink.isSink({ data: [1, 2, 3] })) // => false