# 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](#constructor-reference) at the bottom enumerates every
public constructor with a runnable example.

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

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

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

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

```ts
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 `ReadableStream`s in
any environment (browser, Deno, Node, Bun). See the
[Platform](https://effect.plants.sh/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](https://effect.plants.sh/streaming/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.

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

Stream.DefaultChunkSize
// => 4096
```

### From values and synchronous sources

#### `Stream.make`

Creates a stream from a variadic list of values.

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

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

#### `Stream.succeed`

Creates a single-element stream from a pure value.

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

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

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

#### `Stream.suspend`

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

#### `Stream.failSync`

Like `fail`, but the error is computed lazily.

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

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

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

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

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

```ts
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 `Take`s terminate the stream.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

```ts
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](https://effect.plants.sh/streaming/consuming-streams/) and
[Transforming streams](https://effect.plants.sh/streaming/transforming-streams/).