Creating Streams

Most real streams originate from some external source — an array already in memory, an effect you want to repeat, a paginated HTTP endpoint, a callback-based event API, or a Node.js readable. Stream ships a constructor for each shape so you can lift the source into a typed, resource-safe pipeline. The constructors below cover the cases you will reach for most often; the constructor reference at the bottom enumerates every public constructor with a runnable example.

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

// `Stream.make` / `Stream.fromIterable` turn values already in memory into a
// stream. `make` takes varargs; `fromIterable` takes any iterable.
export const numbers = Stream.fromIterable<number>([1, 2, 3, 4, 5])
export const letters = Stream.make("a", "b", "c")

// `Stream.range` emits an inclusive range of integers.
export const oneToTen = Stream.range(1, 10)

// `Stream.fromEffect` runs an effect once and emits its single result. Useful
// for adapting an existing Effect into a (one-element) stream.
export const config = Stream.fromEffect(Effect.succeed({ retries: 3 }))

// `Stream.fromEffectSchedule` turns a single effect into a *polling* stream by
// re-running it on a Schedule. Great for metrics, health checks, and refresh
// loops. Here we sample every 30 seconds, but only keep the first 3 samples.
export const samples = Stream.fromEffectSchedule(
  Effect.succeed(3),
  Schedule.spaced("30 seconds")
).pipe(Stream.take(3))

Paginated sources

APIs that return one page at a time map cleanly onto Stream.paginate. You give it a starting cursor and a function that, for each cursor, returns the page of values plus an Option of the next cursor (Option.none() ends the stream).

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

// Read a paginated "jobs" API. Each call returns one page of results and the
// next cursor; returning `Option.none()` stops the stream.
export const fetchJobsPage = Stream.paginate(
  0, // start at page 0 (the cursor)
  Effect.fn(function*(page) {
    // Simulate network latency for the page request.
    yield* Effect.sleep("50 millis")

    const results = Array.range(0, 99).map((i) => `Job ${i + 1 + page * 100}`)

    // Only fetch the first 10 pages, then end the stream.
    const nextPage = page <= 10 ? Option.some(page + 1) : Option.none()

    return [results, nextPage] as const
  })
)

Stream.paginate flattens the per-page arrays for you: the resulting stream emits individual string jobs, not arrays of jobs.

Async iterables and callback APIs

JavaScript surfaces two common imperative shapes: async iterables (async function*) and callback registrations (event listeners, subscriptions). Both lift into a stream, and both let you convert thrown/raw errors into a typed error so the failure shows up in the stream’s E channel.

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

// A typed error for whatever the async iterable might throw. `Schema.Defect`
// captures an unknown cause without forcing you to model it.
class LetterError extends Schema.TaggedErrorClass<LetterError>()("LetterError", {
  cause: Schema.Defect
}) {}

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

// `Stream.fromAsyncIterable` drives the iterator and maps any thrown error
// into our typed `LetterError`.
export const letters = Stream.fromAsyncIterable(
  asyncLetters(),
  (cause) => new LetterError({ cause })
)

const button = document.getElementById("my-button")!

// `Stream.fromEventListener` is the dedicated constructor for DOM events.
export const clicks = Stream.fromEventListener<PointerEvent>(button, "click")

// `Stream.callback` adapts *any* callback-based API. You receive a `Queue` and
// push values into it; the stream emits whatever you offer. Register the
// listener with `Effect.acquireRelease` so it is torn down when the stream ends.
export const clicksViaCallback = Stream.callback<PointerEvent>(
  Effect.fn(function*(queue) {
    function onEvent(event: PointerEvent) {
      // `offerUnsafe` enqueues without an effect — safe inside a sync callback.
      Queue.offerUnsafe(queue, event)
    }

    yield* Effect.acquireRelease(
      Effect.sync(() => button.addEventListener("click", onEvent)),
      () => Effect.sync(() => button.removeEventListener("click", onEvent))
    )
  })
)

The acquireRelease finalizer is the key to leak-free streaming: when the consumer stops pulling — whether the stream completes, fails, or is interrupted — the listener is removed. See Resource Management for how acquire/release and Scope work in general.

Web and Node.js readable streams

Stream.fromReadableStream bridges a Web ReadableStream straight from core effect. Build the readable lazily so each subscription gets its own reader, and map low-level read failures into a typed error.

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

class WebStreamError extends Data.TaggedError("WebStreamError")<{
  readonly cause: unknown
}> {}

export const webChunks = Stream.fromReadableStream<number, WebStreamError>({
  evaluate: () =>
    new ReadableStream({
      start(controller) {
        controller.enqueue(1)
        controller.enqueue(2)
        controller.enqueue(3)
        controller.close()
      }
    }),
  onError: (cause) => new WebStreamError({ cause })
  // releaseLockOnEnd defaults to false → the reader is canceled on finalize.
})

On the Node platform, NodeStream.fromReadable does the same for a node:stream readable, converting emitter errors into a typed error.

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

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

// Build the readable lazily (per subscription) and map its errors. `closeOnDone`
// destroys the readable when the stream finishes (the default).
export const fileChunks = NodeStream.fromReadable<string>({
  evaluate: () => Readable.from(["Hello", " ", "world", "!"]),
  onError: (cause) => new NodeStreamError({ cause }),
  closeOnDone: true
})

NodeStream lives in the platform package

NodeStream is provided by the @effect/platform-node package, not core effect. Use the core Stream.fromReadableStream for Web ReadableStreams in any environment (browser, Deno, Node, Bun). See the Platform section for the rest of the Node bindings.

Constructor reference

Every public Stream constructor, grouped by where the data comes from. Each entry has a short example with inline // => results (assume the stream is run with Stream.runCollect, which gathers all emitted values into an array). The chunk-based internals are explained in Consuming streams; here we focus on what each constructor emits.

Stream.DefaultChunkSize

The default number of elements a stream pulls per chunk. Constructors such as range and fromIterable accept a chunkSize override; otherwise they use this value.

import { Stream } from "effect"

Stream.DefaultChunkSize
// => 4096

From values and synchronous sources

Stream.make

Creates a stream from a variadic list of values.

import { Stream } from "effect"

Stream.make(1, 2, 3)
// => emits: 1, 2, 3

Stream.succeed

Creates a single-element stream from a pure value.

import { Stream } from "effect"

Stream.succeed("hello")
// => emits: "hello"

Stream.sync

Emits the result of a thunk, re-evaluated each time the stream is run.

import { Stream } from "effect"

Stream.sync(() => 2 + 1)
// => emits: 3

Stream.suspend

Lazily constructs a stream, deferring the factory until the stream is run.

import { Stream } from "effect"

Stream.suspend(() => Stream.make(1, 2, 3))
// => emits: 1, 2, 3

Stream.empty

The stream that emits no values and ends immediately.

import { Stream } from "effect"

Stream.empty
// => emits: (nothing)

Stream.never

The stream that never emits and never ends — useful as a placeholder or to keep a merged pipeline alive.

import { Stream } from "effect"

Stream.never.pipe(Stream.take(0))
// => emits: (nothing, then completes via take(0))

Stream.range

Emits an inclusive range of integers; min > max yields an empty stream. Accepts an optional chunkSize third argument.

import { Stream } from "effect"

Stream.range(1, 5)
// => emits: 1, 2, 3, 4, 5

Stream.iterate

Creates an infinite stream by repeatedly applying a function to a seed value.

import { Stream } from "effect"

Stream.iterate(1, (n) => n + 1).pipe(Stream.take(3))
// => emits: 1, 2, 3

Stream.tick

Emits void immediately, then again after each interval — a bare clock you can map or zip against.

import { Stream } from "effect"

Stream.tick("200 millis").pipe(Stream.take(3))
// => emits: undefined, undefined, undefined (one every 200ms)

From iterables and arrays

Stream.fromIterable

Creates a stream from any iterable; accepts an optional { chunkSize }.

import { Stream } from "effect"

Stream.fromIterable([1, 2, 3])
// => emits: 1, 2, 3

Stream.fromIteratorSucceed

Consumes values from an IterableIterator (e.g. a generator), with an optional maxChunkSize.

import { Stream } from "effect"

function* gen() {
  yield 1
  yield 2
  yield 3
}

Stream.fromIteratorSucceed(gen())
// => emits: 1, 2, 3

Stream.fromArray

Creates a stream from a ReadonlyArray. The whole array is emitted as one chunk.

import { Stream } from "effect"

Stream.fromArray([1, 2, 3])
// => emits: 1, 2, 3

Stream.fromArrays

Creates a stream from a variadic list of arrays, flattening them in order.

import { Stream } from "effect"

Stream.fromArrays([1, 2], [3, 4])
// => emits: 1, 2, 3, 4

Stream.fromIterableEffect

Creates a stream from an effect that produces an iterable — handy for reading a collection out of a service.

import { Effect, Stream } from "effect"

Stream.fromIterableEffect(Effect.succeed(["user1", "user2"]))
// => emits: "user1", "user2"

Stream.fromIterableEffectRepeat

Repeatedly runs an effect that yields an iterable, concatenating each batch forever.

import { Effect, Stream } from "effect"

Stream.fromIterableEffectRepeat(Effect.succeed([1, 2])).pipe(Stream.take(5))
// => emits: 1, 2, 1, 2, 1

Stream.fromArrayEffect

Creates a stream from an effect that produces an array.

import { Effect, Stream } from "effect"

Stream.fromArrayEffect(Effect.succeed(["Ada", "Grace"]))
// => emits: "Ada", "Grace"

From effects

Stream.fromEffect

Runs an effect once and emits its single result.

import { Effect, Stream } from "effect"

Stream.fromEffect(Effect.succeed(42))
// => emits: 42

Stream.fromEffectDrain

Runs an effect for its side effects only and emits nothing.

import { Console, Stream } from "effect"

Stream.fromEffectDrain(Console.log("side effect"))
// => emits: (nothing); logs "side effect"

Stream.fromEffectRepeat

Repeatedly runs an effect forever, emitting each result.

import { Random, Stream } from "effect"

Stream.fromEffectRepeat(Random.nextInt).pipe(Stream.take(3))
// => emits: 3 random ints

Stream.fromEffectSchedule

Re-runs an effect on a Schedule, emitting the effect’s result on each tick. The first value is emitted immediately. Ideal for polling, metrics, and refresh loops.

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

Stream.fromEffectSchedule(Effect.succeed("ping"), Schedule.recurs(2))
// => emits: "ping", "ping", "ping"

Stream.fromSchedule

Emits each output of a schedule that requires no input, for as long as the schedule continues.

import { Schedule, Stream } from "effect"

const schedule = Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(2)))

Stream.fromSchedule(schedule)
// => emits: 0, 1, 2

From failures

Stream.fail

A stream that terminates with the given error in its E channel.

import { Stream } from "effect"

Stream.fail("Uh oh!")
// => fails with: "Uh oh!"

Stream.failSync

Like fail, but the error is computed lazily.

import { Stream } from "effect"

Stream.failSync(() => "Uh oh!")
// => fails with: "Uh oh!"

Stream.failCause

Fails with a full Cause (a fail, die, or interrupt), preserving cause structure.

import { Cause, Stream } from "effect"

Stream.failCause(Cause.fail("Database connection failed"))
// => fails with cause: Fail("Database connection failed")

Stream.failCauseSync

Like failCause, but the Cause is computed lazily.

import { Cause, Stream } from "effect"

Stream.failCauseSync(() => Cause.fail("Connection timeout"))
// => fails with cause: Fail("Connection timeout")

Stream.die

A stream that dies with an unrecoverable defect (an unexpected, untyped error).

import { Stream } from "effect"

Stream.die(new Error("Boom"))
// => dies with defect: Error("Boom")

From concurrency primitives

Stream.fromQueue

Drains a Queue.Dequeue, emitting batches as they arrive. The stream ends when the queue fails with Cause.Done; other queue failures propagate.

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

Effect.gen(function*() {
  const queue = yield* Queue.unbounded<number>()
  yield* Queue.offer(queue, 1)
  yield* Queue.offer(queue, 2)
  yield* Queue.shutdown(queue)
  return Stream.fromQueue(queue)
})
// => emits: 1, 2

Stream.fromPubSub

Subscribes to a PubSub and emits each published value.

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

Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<number>()
  yield* PubSub.publish(pubsub, 1)
  return Stream.fromPubSub(pubsub).pipe(Stream.take(1))
})
// => emits: 1

Stream.fromPubSubTake

Subscribes to a PubSub of Take values, so end and failure signals carried in the Takes terminate the stream.

import { Effect, Exit, PubSub, Stream, Take } from "effect"

Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<Take.Take<number, string>>({ replay: 3 })
  yield* PubSub.publish(pubsub, [1])
  yield* PubSub.publish(pubsub, [2])
  yield* PubSub.publish(pubsub, Exit.succeed<void>(undefined)) // end signal
  return Stream.fromPubSubTake(pubsub)
})
// => emits: 1, 2 (then ends)

Stream.fromSubscription

Emits values from an already-acquired PubSub.Subscription. Use Stream.take or cancellation to bound how much you consume.

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

Effect.scoped(Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<number>()
  const subscription = yield* PubSub.subscribe(pubsub)
  yield* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)
  return yield* Stream.fromSubscription(subscription).pipe(Stream.take(2), Stream.runCollect)
}))
// => [ 1, 2 ]

From external sources

Stream.fromReadableStream

Bridges a lazily-supplied Web ReadableStream, mapping read failures with onError. The reader is canceled on finalize (set releaseLockOnEnd: true to release the lock instead).

import { Data, Stream } from "effect"

class ReadError extends Data.TaggedError("ReadError")<{ readonly cause: unknown }> {}

Stream.fromReadableStream<number, ReadError>({
  evaluate: () =>
    new ReadableStream({
      start(c) {
        c.enqueue(1)
        c.enqueue(2)
        c.close()
      }
    }),
  onError: (cause) => new ReadError({ cause })
})
// => emits: 1, 2

Stream.fromAsyncIterable

Drives an AsyncIterable, mapping any thrown error into a typed E.

import { Data, Stream } from "effect"

class IterError extends Data.TaggedError("IterError")<{ readonly cause: unknown }> {}

const iterable = (async function*() {
  yield 1
  yield 2
  yield 3
})()

Stream.fromAsyncIterable(iterable, (cause) => new IterError({ cause }))
// => emits: 1, 2, 3

Stream.fromEventListener

The dedicated constructor for addEventListener/removeEventListener targets (DOM elements, or any Stream.EventListener<A>). The listener is registered on subscribe and removed on teardown.

import { Stream } from "effect"

const button = document.getElementById("my-button")!

Stream.fromEventListener<PointerEvent>(button, "click")
// => emits: a PointerEvent per click

Stream.callback

Adapts any callback-based API: you receive a Queue and push values into it with the Queue APIs. Supports bufferSize and strategy options.

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

Stream.callback<number>((queue) =>
  Effect.sync(() => {
    Queue.offerUnsafe(queue, 1)
    Queue.offerUnsafe(queue, 2)
    Queue.endUnsafe(queue) // signal completion
  })
)
// => emits: 1, 2

Generative

Stream.unfold

Repeatedly applies an effectful step to a state. Each [value, nextState] emits value; returning undefined ends the stream.

import { Effect, Stream } from "effect"

Stream.unfold(1, (n) => Effect.succeed([n, n + 1] as const)).pipe(Stream.take(5))
// => emits: 1, 2, 3, 4, 5

Stream.paginate

Like unfold, but each step emits an array of values plus an Option of the next state — built for paginated APIs. Option.none() ends the stream.

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

Stream.paginate(0, (n: number) =>
  Effect.succeed([[n], n < 3 ? Option.some(n + 1) : Option.none<number>()] as const))
// => emits: 0, 1, 2, 3

Service accessors

Stream.service

Accesses a service from the context (via its Context.Key) and emits it as a single element.

import { Context, Stream } from "effect"

class Greeter extends Context.Service<Greeter, {
  readonly greet: (name: string) => string
}>()("Greeter") {}

Stream.service(Greeter).pipe(Stream.map((g) => g.greet("World")))
// => emits: the Greeter service, mapped to "Hello, World!" once provided

Stream.serviceOption

Emits an Option of the service, so the dependency is optional rather than required.

import { Context, Option, Stream } from "effect"

class Greeter extends Context.Service<Greeter, {
  readonly greet: (name: string) => string
}>()("Greeter") {}

Stream.serviceOption(Greeter).pipe(
  Stream.map(Option.match({
    onNone: () => "No greeter",
    onSome: (g) => g.greet("World")
  }))
)
// => emits: "No greeter" or "Hello, World!" depending on context

Low-level constructors

These build streams from Channels or raw pull effects. Reach for them when writing your own primitive operators; everything above is built on top of them.

Stream.fromChannel

Creates a stream from a Channel that emits non-empty arrays.

import { Channel, Stream } from "effect"

Stream.fromChannel(Channel.succeed([1, 2, 3] as const))
// => emits: 1, 2, 3

Stream.fromPull

Creates a stream from a pull effect — an effect that yields the next chunk on demand and signals completion with Cause.done(). Pairs with Stream.toPull.

import { Effect, Stream } from "effect"

Effect.scoped(Effect.gen(function*() {
  const pull = yield* Stream.toPull(Stream.make(1, 2, 3))
  return Stream.fromPull(Effect.succeed(pull))
}))
// => emits: 1, 2, 3

Stream.transformPull

Derives a stream by transforming the upstream’s pull effect, with access to the stream’s Scope.

import { Effect, Stream } from "effect"

Stream.transformPull(Stream.make(1, 2, 3), (pull) => Effect.succeed(pull))
// => emits: 1, 2, 3

Stream.transformPullBracket

Like transformPull, but also provides a forked Scope that is closed once the resulting stream finishes processing — use it to attach finalizers.

import { Console, Effect, Scope, Stream } from "effect"

Stream.transformPullBracket(Stream.make(1, 2, 3), (pull, _scope, forkedScope) =>
  Effect.gen(function*() {
    yield* Scope.addFinalizer(forkedScope, Console.log("releasing"))
    return pull
  }))
// => emits: 1, 2, 3; logs "releasing" on completion

Stream.scoped

Runs a stream that requires a Scope inside a managed scope, ensuring its finalizers run when the stream completes.

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

Stream.scoped(
  Stream.fromEffect(
    Effect.acquireRelease(
      Console.log("acquire").pipe(Effect.as("resource")),
      () => Console.log("release")
    )
  )
)
// => emits: "resource"; logs "acquire" then "release"

Stream.unwrap

Flattens an Effect that produces a Stream into a stream. The effect runs once when the stream starts.

import { Effect, Stream } from "effect"

Stream.unwrap(Effect.succeed(Stream.make(1, 2, 3)))
// => emits: 1, 2, 3

With a stream in hand, the next step is to do something with it — see Consuming streams and Transforming streams.