# 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.

```ts
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.

```ts
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).

```ts
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](https://effect.plants.sh/streaming/consuming-streams/). When an
element or effect can fail, see [Error handling](https://effect.plants.sh/streaming/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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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 }`.

```ts
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.

```ts
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".

```ts
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 }`.

```ts
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>`.

```ts
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>`.

```ts
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`](https://effect.plants.sh/streaming/pull-and-take/)):
non-empty arrays emit their elements, and the terminal `Exit` ends or fails the
stream.

```ts
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.

```ts
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`](https://effect.plants.sh/pattern-matching/filter/):
`Result.succeed` values are emitted, `Result.fail` values are dropped.

```ts
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`.

```ts
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.

```ts
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.)

```ts
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.

```ts
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).

```ts
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.

```ts
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`.

```ts
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.

```ts
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.

```ts
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 }`.

```ts
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.

```ts
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).

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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`](https://effect.plants.sh/streaming/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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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]>`.

```ts
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.

```ts
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.

```ts
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 }`.

```ts
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.

```ts
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`](https://effect.plants.sh/streaming/sink/) (e.g. a metric or audit log)
while still emitting them downstream.

```ts
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).

```ts
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).

```ts
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.

```ts
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()`.

```ts
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()`.

```ts
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 `Option`s.

```ts
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.

```ts
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.

```ts
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).

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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).

```ts
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.

```ts
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.

```ts
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.

```ts
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`](https://effect.plants.sh/scheduling/schedule/),
re-emitting it once per schedule recurrence.

```ts
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.

```ts
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.

```ts
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.

```ts
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.

```ts
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 `{}`.

```ts
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`.

```ts
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 }`.

```ts
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`.

```ts
import { Stream } from "effect"

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