Transforming Streams

Streams compose with operators the same way Effect does: each operator takes a stream and returns a new stream, so you build a pipeline with .pipe(...). Pure transforms (map, filter) reshape elements without running effects; effectful transforms (mapEffect, flatMap) let each element trigger more work — with explicit, opt-in concurrency. The example below walks an order-processing pipeline from raw events to enriched, taxed orders.

import { Stream } from "effect"

interface Order {
  readonly id: string
  readonly status: "paid" | "refunded"
  readonly subtotalCents: number
  readonly shippingCents: number
  readonly country: "US" | "CA" | "NZ"
}

interface NormalizedOrder extends Order {
  readonly totalCents: number
}

const orderEvents: Stream.Stream<Order> = Stream.make(
  { id: "ord_1", status: "paid", subtotalCents: 4_500, shippingCents: 500, country: "US" },
  { id: "ord_2", status: "refunded", subtotalCents: 2_000, shippingCents: 0, country: "CA" }
)

// `Stream.map` applies a pure, total function to every element.
const normalized = orderEvents.pipe(
  Stream.map((order): NormalizedOrder => ({
    ...order,
    totalCents: order.subtotalCents + order.shippingCents
  }))
)

// `Stream.filter` drops elements that fail the predicate.
const paid = normalized.pipe(
  Stream.filter((order) => order.status === "paid")
)

Effectful transforms with concurrency

Stream.mapEffect is the workhorse for per-element side effects — a lookup, an HTTP call, a write. By default it processes elements one at a time, preserving order. Pass { concurrency: n } to run up to n effects in flight at once; the stream takes care of backpressure so you never overwhelm the downstream consumer.

import { Effect, Stream } from "effect"

interface NormalizedOrder {
  readonly id: string
  readonly totalCents: number
  readonly country: "US" | "CA" | "NZ"
}

interface EnrichedOrder extends NormalizedOrder {
  readonly taxCents: number
  readonly grandTotalCents: number
}

// Define the per-element effect with `Effect.fn` — never a plain function that
// returns `Effect.gen`. This one simulates an effectful tax lookup.
const enrichOrder = Effect.fn("enrichOrder")(function*(order: NormalizedOrder) {
  yield* Effect.sleep("5 millis") // stand-in for a tax/risk service call

  const taxRate = order.country === "US" ? 0.08 : 0.13
  const taxCents = Math.round(order.totalCents * taxRate)

  return {
    ...order,
    taxCents,
    grandTotalCents: order.totalCents + taxCents
  } satisfies EnrichedOrder
})

const paid: Stream.Stream<NormalizedOrder> = Stream.make(
  { id: "ord_1", totalCents: 5_000, country: "US" as const }
)

// Run up to 4 enrichment effects concurrently. Output order is still preserved.
export const enriched = paid.pipe(
  Stream.mapEffect(enrichOrder, { concurrency: 4 })
)

Stream.flatMap is the other effectful combinator: it maps each element to a stream and flattens the results, again with optional { concurrency }. Use it when one input fans out into many outputs (one customer → their order history, one URL → its paginated pages).

import { Stream } from "effect"

// Each country fans out into a stream of synthetic orders; the inner streams are
// flattened into one. `concurrency: 2` runs two country sub-streams at once.
export const allOrders = Stream.make("US", "CA", "NZ").pipe(
  Stream.flatMap(
    (country) =>
      Stream.range(1, 50).pipe(
        Stream.map((i) => ({ id: `ord_${country}_${i}`, country }))
      ),
    { concurrency: 2 }
  )
)

These operators only describe a transformation — nothing runs until you attach a destructor from Consuming streams. When an element or effect can fail, see Error handling. Operators that change the cardinality of a stream (batching with grouped, merging multiple sources with merge/zip) live alongside these; this page focuses on per-element and stateful transforms.

Reference

Every transform below takes a stream and returns a new stream (unless noted). All are data-last and dual, so you can either pipe (stream.pipe(Stream.map(f))) or call directly (Stream.map(stream, f)). The examples use Stream.runCollect to show the emitted values.

Mapping

map

Applies a pure function to every element. The function also receives the element’s zero-based index.

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.map((n) => n * 2),
  Stream.runCollect
)
// => [2, 4, 6]

mapBoth

Maps the success and failure channels at once via { onFailure, onSuccess }. Only one of the two functions runs per outcome.

import { Stream } from "effect"

Stream.make(1, 2).pipe(
  Stream.mapBoth({
    onFailure: (e: string) => `error: ${e}`,
    onSuccess: (n) => n * 2
  })
)
// success values => [2, 4]; a failure "boom" => "error: boom"

mapArray

Transforms each underlying chunk (a NonEmptyReadonlyArray) as a whole, with its index. Useful when a per-chunk transform is cheaper than per-element.

import { Array, Stream } from "effect"

Stream.make(1, 2, 3, 4).pipe(
  Stream.rechunk(2),
  Stream.mapArray((chunk, i) => Array.map(chunk, (n) => n + i)),
  Stream.runCollect
)
// => [1, 2, 4, 5]

mapEffect

Maps each element through an effectful function. Accepts { concurrency, unordered } — concurrency runs up to N effects at once, and unordered: true lets results emit as soon as they finish rather than in input order.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.mapEffect((n) => Effect.succeed(n * 2), { concurrency: 4 }),
  Stream.runCollect
)
// => [2, 4, 6]

mapArrayEffect

Like mapArray, but the per-chunk transform is an Effect returning a new non-empty array.

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

Stream.fromArray([1, 2, 3, 4]).pipe(
  Stream.rechunk(2),
  Stream.mapArrayEffect((chunk, i) => Effect.succeed(Array.map(chunk, (n) => n + i * 10))),
  Stream.runCollect
)
// => [1, 2, 13, 14]

flattenEffect

Turns a Stream<Effect<A>> into a Stream<A> by running each effect. Accepts { concurrency, unordered }.

import { Effect, Stream } from "effect"

Stream.make(Effect.succeed(1), Effect.succeed(2), Effect.succeed(3)).pipe(
  Stream.flattenEffect(),
  Stream.runCollect
)
// => [1, 2, 3]

Effectful expansion

flatMap

Maps each element to a stream and flattens the results. With sequential concurrency inner streams are concatenated in order; { concurrency } runs several inner streams at once and merges their outputs. { bufferSize } bounds the merge buffer.

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.flatMap((n) => Stream.make(n, n * 2)),
  Stream.runCollect
)
// => [1, 2, 2, 4, 3, 6]

switchMap

Maps each element to a stream but interrupts the previous inner stream when a new element arrives — only the latest mapping survives. Ideal for “cancel the in-flight request when a newer one comes in”.

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.switchMap((n) => (n === 3 ? Stream.make(n) : Stream.never)),
  Stream.runCollect
)
// => [3]

flatten

Flattens a Stream<Stream<A>> into a Stream<A>. Accepts { concurrency, bufferSize }.

import { Stream } from "effect"

Stream.make(Stream.make(1, 2), Stream.make(3, 4)).pipe(
  Stream.flatten(),
  Stream.runCollect
)
// => [1, 2, 3, 4]

flattenArray

Flattens a Stream<NonEmptyReadonlyArray<A>> into a Stream<A>.

import { Array, Stream } from "effect"

Stream.make(Array.make(1, 2), Array.make(3)).pipe(
  Stream.flattenArray,
  Stream.runCollect
)
// => [1, 2, 3]

flattenIterable

Flattens a Stream<Iterable<A>> (any iterable, not just arrays) into a Stream<A>.

import { Stream } from "effect"

Stream.make([1, 2], [3, 4]).pipe(
  Stream.flattenIterable,
  Stream.runCollect
)
// => [1, 2, 3, 4]

flattenTake

Unwraps a Stream<Take<A, E>> (see Take): non-empty arrays emit their elements, and the terminal Exit ends or fails the stream.

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

Stream.make(Array.make(1, 2), Array.make(3), Exit.succeed<void>(undefined)).pipe(
  Stream.flattenTake,
  Stream.runCollect
)
// => [1, 2, 3]

Filtering

filter

Keeps only elements that satisfy a predicate. Passing a type Refinement narrows the element type.

import { Stream } from "effect"

Stream.make(1, 2, 3, 4).pipe(
  Stream.filter((n) => n % 2 === 0),
  Stream.runCollect
)
// => [2, 4]

filterMap

Filters and maps in one pass using a Filter: Result.succeed values are emitted, Result.fail values are dropped.

import { Result, Stream } from "effect"

Stream.make(1, 2, 3, 4).pipe(
  Stream.filterMap((n) => (n % 2 === 0 ? Result.succeed(n * 10) : Result.fail(n))),
  Stream.runCollect
)
// => [20, 40]

filterEffect

Keeps elements for which an effectful predicate returns true.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3, 4).pipe(
  Stream.filterEffect((n) => Effect.succeed(n > 2)),
  Stream.runCollect
)
// => [3, 4]

filterMapEffect

Effectful filter-and-map using a FilterEffect: Result.succeed is emitted, Result.fail is skipped, and effect failures fail the stream.

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

Stream.make(1, 2, 3, 4).pipe(
  Stream.filterMapEffect((n) =>
    Effect.succeed(n % 2 === 0 ? Result.succeed(`even:${n}`) : Result.fail(n))
  ),
  Stream.runCollect
)
// => ["even:2", "even:4"]

collect

Collects the whole stream into a single Array and emits it as one element. (This is the stream-returning sibling of runCollect, handy mid-pipeline.)

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.collect,
  Stream.runCollect
)
// => [[1, 2, 3]]

Taking and dropping

take

Emits at most the first n elements, then stops pulling upstream.

import { Stream } from "effect"

Stream.range(1, 100).pipe(Stream.take(2), Stream.runCollect)
// => [1, 2]

takeRight

Keeps the last n elements (buffers up to n in memory).

import { Stream } from "effect"

Stream.range(1, 6).pipe(Stream.takeRight(3), Stream.runCollect)
// => [4, 5, 6]

takeWhile

Emits the longest leading run that satisfies a predicate, then stops. Refinements narrow the type.

import { Stream } from "effect"

Stream.range(1, 5).pipe(Stream.takeWhile((n) => n % 3 !== 0), Stream.runCollect)
// => [1, 2]

takeWhileFilter

Like takeWhile but driven by a Filter, emitting the filter’s success values and stopping at the first Result.fail.

import { Result, Stream } from "effect"

Stream.make(1, 2, 9, 3).pipe(
  Stream.takeWhileFilter((n) => (n < 5 ? Result.succeed(n * 10) : Result.fail(n))),
  Stream.runCollect
)
// => [10, 20]

takeWhileEffect

takeWhile with an effectful predicate.

import { Effect, Stream } from "effect"

Stream.range(1, 5).pipe(
  Stream.takeWhileEffect((n) => Effect.succeed(n % 3 !== 0)),
  Stream.runCollect
)
// => [1, 2]

takeUntil

Emits elements until the predicate first matches, including the matching element. Pass { excludeLast: true } to drop it.

import { Stream } from "effect"

Stream.range(1, 5).pipe(Stream.takeUntil((n) => n % 3 === 0), Stream.runCollect)
// => [1, 2, 3]

Stream.range(1, 5).pipe(
  Stream.takeUntil((n) => n % 3 === 0, { excludeLast: true }),
  Stream.runCollect
)
// => [1, 2]

takeUntilEffect

takeUntil with an effectful predicate. Also supports { excludeLast }.

import { Effect, Stream } from "effect"

Stream.range(1, 5).pipe(
  Stream.takeUntilEffect((n) => Effect.succeed(n % 3 === 0)),
  Stream.runCollect
)
// => [1, 2, 3]

drop

Skips the first n elements.

import { Stream } from "effect"

Stream.make(1, 2, 3, 4, 5).pipe(Stream.drop(2), Stream.runCollect)
// => [3, 4, 5]

dropRight

Drops the last n elements (keeps n elements buffered to drop on completion).

import { Stream } from "effect"

Stream.make(1, 2, 3, 4, 5).pipe(Stream.dropRight(2), Stream.runCollect)
// => [1, 2, 3]

dropWhile

Drops the leading run that satisfies a predicate, then emits the rest unchanged.

import { Stream } from "effect"

Stream.make(1, 2, 3, 4, 5).pipe(Stream.dropWhile((n) => n < 3), Stream.runCollect)
// => [3, 4, 5]

dropWhileFilter

Drops the leading prefix while a Filter succeeds; the first Result.fail stops dropping and the remaining original elements pass through unfiltered.

import { Result, Stream } from "effect"

Stream.make(1, 2, 9, 3).pipe(
  Stream.dropWhileFilter((n) => (n < 5 ? Result.succeed(n) : Result.fail(n))),
  Stream.runCollect
)
// => [9, 3]

dropWhileEffect

dropWhile with an effectful predicate.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3, 4, 5).pipe(
  Stream.dropWhileEffect((n) => Effect.succeed(n < 3)),
  Stream.runCollect
)
// => [3, 4, 5]

dropUntil

Drops elements until the predicate first matches, also dropping that matching element.

import { Stream } from "effect"

Stream.make(1, 2, 3, 4, 5).pipe(Stream.dropUntil((n) => n >= 3), Stream.runCollect)
// => [4, 5]

dropUntilEffect

dropUntil with an effectful predicate.

import { Effect, Stream } from "effect"

Stream.range(1, 5).pipe(
  Stream.dropUntilEffect((n) => Effect.succeed(n % 3 === 0)),
  Stream.runCollect
)
// => [4, 5]

when

Emits the whole stream only if an effectful boolean test passes; otherwise emits nothing.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.when(Effect.succeed(true)),
  Stream.runCollect
)
// => [1, 2, 3]  (an `Effect.succeed(false)` test yields [])

peel

Runs a Sink to “peel off” a prefix, returning that value plus a scoped stream of the remainder. Returns an Effect, so it must run inside a scope.

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

Effect.scoped(
  Effect.gen(function*() {
    const [peeled, rest] = yield* Stream.peel(
      Stream.fromArrays([1, 2, 3], [4, 5, 6]),
      Sink.take<number>(3)
    )
    const remaining = yield* Stream.runCollect(rest)
    return [peeled, remaining]
  })
)
// => [[1, 2, 3], [4, 5, 6]]

Stateful and scan

scan

Threads state through the stream, emitting the initial state and each updated state — a running fold that emits every intermediate.

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.scan(0, (acc, n) => acc + n),
  Stream.runCollect
)
// => [0, 1, 3, 6]

scanEffect

scan with an effectful step function.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.scanEffect(0, (sum, n) => Effect.succeed(sum + n)),
  Stream.runCollect
)
// => [0, 1, 3, 6]

mapAccum

Stateful map: each element returns [nextState, valuesToEmit], emitting zero or more outputs per input. The optional { onHalt } flushes remaining state at completion.

import { Stream } from "effect"

Stream.make(0, 1, 2, 3).pipe(
  Stream.mapAccum(() => 0, (total, n) => {
    const next = total + n
    return [next, [next]] as const
  }),
  Stream.runCollect
)
// => [0, 1, 3, 6]

mapAccumArray

mapAccum at chunk granularity — the step receives a whole non-empty chunk.

import { Stream } from "effect"

Stream.make(1, 2, 3, 4, 5, 6).pipe(
  Stream.rechunk(2),
  Stream.mapAccumArray(() => 0, (sum: number, chunk) => {
    const next = chunk.reduce((acc, n) => acc + n, sum)
    return [next, [next]]
  }),
  Stream.runCollect
)
// => [3, 10, 21]

mapAccumEffect

mapAccum with an effectful step returning Effect<[nextState, values]>.

import { Effect, Stream } from "effect"

Stream.make(1, 1, 1).pipe(
  Stream.mapAccumEffect(() => 0, (total, n) => Effect.succeed([total + n, [total + n]])),
  Stream.runCollect
)
// => [1, 2, 3]

mapAccumArrayEffect

Effectful, chunk-level stateful map.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3, 4).pipe(
  Stream.rechunk(2),
  Stream.mapAccumArrayEffect(() => 0, (total, chunk) =>
    Effect.succeed([chunk.reduce((s, v) => s + v, total), [chunk.reduce((s, v) => s + v, total)]] as const)
  ),
  Stream.runCollect
)
// => [3, 10]

accumulate

Emits the growing prefix array for each input — element i produces the array of the first i + 1 elements.

import { Stream } from "effect"

Stream.fromArray([1, 2, 3]).pipe(
  Stream.rechunk(1),
  Stream.accumulate,
  Stream.runCollect
)
// => [[1], [1, 2], [1, 2, 3]]

Tapping

tap

Runs an effect for each element and forwards the element unchanged. Accepts { concurrency }.

import { Effect, Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.tap((n) => Effect.log(`saw ${n}`)),
  Stream.runCollect
)
// logs "saw 1/2/3"; => [1, 2, 3]

tapBoth

Taps both elements and errors via { onElement, onError }, leaving the stream’s values and failures intact.

import { Effect, Stream } from "effect"

Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapBoth({
    onElement: (v) => Effect.log(`seen: ${v}`),
    onError: (e) => Effect.log(`error: ${e}`)
  }),
  Stream.catch(() => Stream.make(3)),
  Stream.runCollect
)
// logs seen/error; => [1, 2, 3]

tapSink

Feeds every element to a Sink (e.g. a metric or audit log) while still emitting them downstream.

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

Effect.gen(function*() {
  const seen = yield* Ref.make<Array<number>>([])
  const result = yield* Stream.make(1, 2, 3).pipe(
    Stream.tapSink(Sink.forEach((v: number) => Ref.update(seen, (xs) => [...xs, v]))),
    Stream.runCollect
  )
  return [yield* Ref.get(seen), result]
})
// => [[1, 2, 3], [1, 2, 3]]

tapError

Runs an effect when the stream fails, without recovering (unless the tap itself fails).

import { Effect, Stream } from "effect"

Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapError((e) => Effect.log(`tapError: ${e}`)),
  Stream.catch(() => Stream.make(999)),
  Stream.runCollect
)
// logs "tapError: boom"; => [1, 2, 999]

tapCause

Like tapError, but receives the full Cause (covering defects and interruption, not just typed errors).

import { Effect, Stream } from "effect"

Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapCause((cause) => Effect.log(`cause: ${cause}`)),
  Stream.catch(() => Stream.make(0)),
  Stream.runCollect
)
// logs the cause; => [1, 2, 0]

Indexing and adjacency

zipWithIndex

Pairs each element with its zero-based index.

import { Stream } from "effect"

Stream.make("a", "b", "c").pipe(Stream.zipWithIndex, Stream.runCollect)
// => [["a", 0], ["b", 1], ["c", 2]]

zipWithNext

Pairs each element with the following element as an Option; the last is paired with Option.none().

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(Stream.zipWithNext, Stream.runCollect)
// => [[1, Some(2)], [2, Some(3)], [3, None]]

zipWithPrevious

Pairs each element with the previous element as an Option; the first is paired with Option.none().

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(Stream.zipWithPrevious, Stream.runCollect)
// => [[None, 1], [Some(1), 2], [Some(2), 3]]

zipWithPreviousAndNext

Triples each element with its previous and next neighbours as Options.

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(Stream.zipWithPreviousAndNext, Stream.runCollect)
// => [[None, 1, Some(2)], [Some(1), 2, Some(3)], [Some(2), 3, None]]

changes

Drops consecutive duplicates using Equal.equals, emitting only elements that differ from the previous one.

import { Stream } from "effect"

Stream.fromIterable([1, 1, 2, 2, 3]).pipe(Stream.changes, Stream.runCollect)
// => [1, 2, 3]

changesWith

changes with a custom equality predicate.

import { Stream } from "effect"

Stream.make("A", "a", "B", "b", "b").pipe(
  Stream.changesWith((x, y) => x.toLowerCase() === y.toLowerCase()),
  Stream.runCollect
)
// => ["A", "B"]

changesWithEffect

changes with an effectful equality check (true means “equal”, so it is skipped).

import { Effect, Stream } from "effect"

Stream.make(1, 1, 2, 2, 3, 3).pipe(
  Stream.changesWithEffect((a, b) => Effect.succeed(a === b)),
  Stream.runCollect
)
// => [1, 2, 3]

Reordering and structure

concat

Emits all of the first stream, then all of the second.

import { Stream } from "effect"

Stream.concat(Stream.make(1, 2, 3), Stream.make(4, 5, 6)).pipe(Stream.runCollect)
// => [1, 2, 3, 4, 5, 6]

prepend

Emits the values of an iterable before the stream’s own elements.

import { Stream } from "effect"

Stream.make(3, 4).pipe(Stream.prepend([1, 2]), Stream.runCollect)
// => [1, 2, 3, 4]

intersperse

Inserts a separator value between consecutive elements.

import { Stream } from "effect"

Stream.make(1, 2, 3, 4).pipe(Stream.intersperse(0), Stream.runCollect)
// => [1, 0, 2, 0, 3, 0, 4]

intersperseAffixes

Wraps the stream with { start, middle, end }; start and end are always emitted, even for an empty stream.

import { Stream } from "effect"

Stream.make("a", "b", "c").pipe(
  Stream.intersperseAffixes({ start: "[", middle: ",", end: "]" }),
  Stream.runCollect
)
// => ["[", "a", ",", "b", ",", "c", "]"]

interleave

Alternates pulls from two streams one element at a time; when one ends, the rest of the other is emitted.

import { Stream } from "effect"

Stream.interleave(Stream.make(2, 3), Stream.make(5, 6, 7)).pipe(Stream.runCollect)
// => [2, 5, 3, 6, 7]

interleaveWith

Deterministic interleave driven by a boolean decider stream (true pulls left, false pulls right).

import { Stream } from "effect"

Stream.interleaveWith(
  Stream.make(1, 3, 5),
  Stream.make(2, 4, 6),
  Stream.make(true, false, false, true, true)
).pipe(Stream.runCollect)
// => [1, 2, 4, 3, 5]

splitLines

Splits a Stream<string> into lines, handling \n, \r, and \r\n across chunk boundaries.

import { Stream } from "effect"

Stream.make("a\nb\r\n", "c\n").pipe(Stream.splitLines, Stream.runCollect)
// => ["a", "b", "c"]

split

Splits a stream into non-empty chunks delimited by elements that match the predicate; the delimiters themselves are removed.

import { Stream } from "effect"

Stream.make(1, 2, 0, 3, 4).pipe(
  Stream.split((n) => n === 0),
  Stream.runCollect
)
// => [[1, 2], [3, 4]]

forever

Repeats the stream endlessly. Pair with take to bound it.

import { Stream } from "effect"

Stream.make("A", "B").pipe(Stream.forever, Stream.take(5), Stream.runCollect)
// => ["A", "B", "A", "B", "A"]

repeat

Repeats the whole stream according to a Schedule, re-emitting it once per schedule recurrence.

import { Schedule, Stream } from "effect"

Stream.make(1, 2).pipe(Stream.repeat(Schedule.recurs(1)), Stream.runCollect)
// => [1, 2, 1, 2]

repeatElements

Repeats each element according to a schedule, including the original emission.

import { Schedule, Stream } from "effect"

Stream.make("A", "B", "C").pipe(
  Stream.repeatElements(Schedule.recurs(1)),
  Stream.runCollect
)
// => ["A", "A", "B", "B", "C", "C"]

schedule

Paces emission of elements according to a schedule (e.g. one every 10 ms), forwarding each element unchanged.

import { Schedule, Stream } from "effect"

Stream.make(1, 2, 3).pipe(
  Stream.schedule(Schedule.spaced("10 millis")),
  Stream.runCollect
)
// => [1, 2, 3]  (spaced ~10ms apart)

Text

decodeText

Decodes a Stream<Uint8Array> into a Stream<string> with TextDecoder. Pass { encoding } (default UTF-8) to override.

import { Stream } from "effect"

const enc = new TextEncoder()
Stream.make(enc.encode("Hello"), enc.encode(" World")).pipe(
  Stream.decodeText,
  Stream.runCollect
)
// => ["Hello", " World"]

encodeText

The inverse: encodes a Stream<string> into UTF-8 Uint8Array chunks.

import { Stream } from "effect"

Stream.make("Hi", "!").pipe(Stream.encodeText, Stream.runCollect)
// => [Uint8Array [72, 105], Uint8Array [33]]

Do-notation

Stream supports the same Do/bind notation as Effect, so you can build a record of values across stream steps. Each bind over a multi-element stream multiplies the cartesian shape (like nested flatMap).

Do

The starting point: a stream emitting a single empty record {}.

import { Stream } from "effect"

Stream.Do.pipe(Stream.runCollect)
// => [{}]

bind

Binds the result of a stream to a named field, flattening it. Accepts the same { concurrency, bufferSize } as flatMap.

import { Stream } from "effect"

Stream.Do.pipe(
  Stream.bind("a", () => Stream.make(1, 2)),
  Stream.bind("b", ({ a }) => Stream.succeed(a + 1)),
  Stream.runCollect
)
// => [{ a: 1, b: 2 }, { a: 2, b: 3 }]

bindEffect

Binds an Effect result to a named field for each element. Accepts { concurrency, bufferSize, unordered }.

import { Effect, Stream } from "effect"

Stream.Do.pipe(
  Stream.bind("value", () => Stream.make(1, 2)),
  Stream.bindEffect("double", ({ value }) => Effect.succeed(value * 2)),
  Stream.runCollect
)
// => [{ value: 1, double: 2 }, { value: 2, double: 4 }]

bindTo

Wraps each element in a single-field record — the entry point when starting from a plain stream rather than Stream.Do.

import { Stream } from "effect"

Stream.make(1, 2, 3).pipe(Stream.bindTo("value"), Stream.runCollect)
// => [{ value: 1 }, { value: 2 }, { value: 3 }]