Stream Processing

Effect Streams are pull-based, effectful sequences of values over time. You can model finite or infinite data sources, transform them lazily, and consume them with controlled concurrency.

Creating Streams

From Iterables

import { Stream } from "effect"

const numbers = Stream.fromIterable([1, 2, 3, 4, 5])

Polling with Schedules

Stream.fromEffectSchedule turns a single effect into a polling stream.

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

const samples = Stream.fromEffectSchedule(
  Effect.succeed(3),
  Schedule.spaced("30 seconds")
).pipe(Stream.take(3))

Paginated APIs

Stream.paginate handles APIs that return one page at a time. The function returns the current page and optionally the next cursor.

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

const fetchJobsPage = Stream.paginate(
  0,
  Effect.fn(function*(page) {
    yield* Effect.sleep("50 millis")
    const results = Array.range(0, 100).map((i) => `Job ${i + 1 + page * 100}`)
    const nextPage = page <= 10 ? Option.some(page + 1) : Option.none()
    return [results, nextPage] as const
  })
)

From Async Iterables

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

class LetterError extends Schema.TaggedError<LetterError>()("LetterError", {
  cause: Schema.Defect()
}) {}

async function* asyncIterable() {
  yield "a"
  yield "b"
  yield "c"
}

const letters = Stream.fromAsyncIterable(
  asyncIterable(),
  (cause) => new LetterError({ cause })
)

From Callbacks

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

const callbackStream = Stream.callback<PointerEvent>(Effect.fn(function*(queue) {
  function onEvent(event: PointerEvent) {
    Queue.offerUnsafe(queue, event)
  }
  yield* Effect.acquireRelease(
    Effect.sync(() => button.addEventListener("click", onEvent)),
    () => Effect.sync(() => button.removeEventListener("click", onEvent))
  )
}))

From Node.js Readable Streams

import { NodeStream } from "@effect/platform-node"
import { Effect, Schema, Stream } from "effect"
import { Readable } from "node:stream"

class NodeStreamError extends Schema.TaggedError<NodeStreamError>()("NodeStreamError", {
  cause: Schema.Defect()
}) {}

const nodeStream = NodeStream.fromReadable({
  evaluate: () => Readable.from(["Hello", " ", "world", "!"]),
  onError: (cause) => new NodeStreamError({ cause }),
  closeOnDone: true
})

Transforming Streams

map and flatMap

import { Stream } from "effect"

const doubled = Stream.fromIterable([1, 2, 3]).pipe(
  Stream.map((n) => n * 2)
)

const expanded = Stream.make("US", "CA", "NZ").pipe(
  Stream.flatMap(
    (country) => Stream.range(1, 50).pipe(
      Stream.map((i) => ({ id: `ord_${country}_${i}`, country }))
    ),
    { concurrency: 2 }
  )
)

filter

const evens = Stream.fromIterable([1, 2, 3, 4, 5, 6]).pipe(
  Stream.filter((n) => n % 2 === 0)
)

mapEffect

Stream.mapEffect performs effectful per-element transforms with concurrency control.

import { Effect, Stream } from "effect"

const enrichOrder = Effect.fn(function*(order: { country: string; total: number }) {
  yield* Effect.sleep("5 millis")
  const taxRate = order.country === "US" ? 0.08 : 0.13
  return { ...order, tax: Math.round(order.total * taxRate) }
})

const enriched = Stream.succeed({ country: "US", total: 4500 }).pipe(
  Stream.mapEffect(enrichOrder, { concurrency: 4 })
)

Consuming Streams

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

const stream = Stream.fromIterable([1, 2, 3, 4, 5])

const collected = Stream.runCollect(stream)
const drained = Stream.runDrain(stream)
const logged = stream.pipe(
  Stream.runForEach((n) => Effect.logInfo(`Got ${n}`))
)
const summed = stream.pipe(Stream.run(Sink.sum))
const first = stream.pipe(Stream.runHead)
const last = stream.pipe(Stream.runLast)
const total = stream.pipe(
  Stream.runFold(() => 0, (acc: number, n: number) => acc + n)
)

Windowing Operators

const firstTwo = stream.pipe(Stream.take(2), Stream.runCollect)
const afterFirst = stream.pipe(Stream.drop(1), Stream.runCollect)
const untilFive = stream.pipe(
  Stream.takeWhile((n) => n < 5),
  Stream.runCollect
)

Resourceful Streams

Use Effect.acquireRelease inside stream constructors to manage resource lifetimes. The finalizer runs when the stream ends or is interrupted.

import { Effect, Stream } from "effect"

const fileLines = Stream.fromIterable([1, 2, 3]).pipe(
  Stream.flatMap((n) =>
    Stream.fromEffect(
      Effect.acquireRelease(
        Effect.sync(() => console.log(`opening resource ${n}`)),
        () => Effect.sync(() => console.log(`closing resource ${n}`))
      )
    )
  )
)

Encoding and Decoding Streams

Use Stream.pipeThroughChannel with Ndjson and Msgpack modules to decode and encode streams of structured data.

import { Schema, Stream } from "effect"
import { Ndjson } from "effect/unstable/encoding"

class LogEntry extends Schema.Class<LogEntry>("LogEntry")({
  timestamp: Schema.DateTimeUtcFromString,
  level: Schema.Literals(["info", "warn", "error"]),
  message: Schema.String
}) {}

const ndjsonInput =
  '{"timestamp":"2025-06-01T00:00:00Z","level":"info","message":"start"}\n' +
  '{"timestamp":"2025-06-01T00:00:01Z","level":"error","message":"oops"}\n'

const decoded = Stream.make(ndjsonInput).pipe(
  Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
  Stream.filter((entry) => entry.level === "error"),
  Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
  Stream.runCollect
)

Tip: Use Ndjson.decodeString() for raw JSON without schema validation. Use Ndjson.decodeSchemaString(Schema)() to validate each line against a schema in one pass. For binary I/O (TCP sockets), use Ndjson.decode() and Ndjson.encode() which work with Uint8Array chunks.

Handling Empty Lines

const tolerant = Stream.make('{"ok":true}\n\n{"ok":false}\n').pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString({ ignoreEmptyLines: true })),
  Stream.runCollect
)

Error Handling in Streams

Ndjson.NdjsonError is raised when encoding or decoding fails. Catch it with Stream.catchTag.

const safe = Stream.make("not-valid-json\n").pipe(
  Stream.pipeThroughChannel(Ndjson.decodeString()),
  Stream.catchTag("NdjsonError", (err) =>
    Stream.succeed({ recovered: true, kind: err.kind })
  ),
  Stream.runCollect
)

Gotcha: Ndjson imports moved in v4. Use effect/unstable/encoding/Ndjson instead of @effect/platform/Ndjson. The same applies to Msgpack: use effect/unstable/encoding/Msgpack.