Creating Streams

In this guide, we’ll explore a variety of ways to create streams using built-in constructors, from single-value streams and ranges to streams created from asynchronous sources like iterables, effects, and schedules.

Common Constructors

make

The Stream.make constructor lets you create a stream from a list of values. Each value passed to Stream.make will be emitted by the stream in the order provided.

Example (Creating a Stream from Values)

import { Stream, Effect } from "effect"

const stream = Stream.make(1, 2, 3)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2, 3 ] }

empty

When you need a stream that doesn’t emit any values, use Stream.empty. This constructor creates a stream that remains empty throughout its lifecycle.

Example (Creating an Empty Stream)

import { Stream, Effect } from "effect"

const stream = Stream.empty

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [] }

void

If you need a stream that emits a single void value, use Stream.void. This is useful for creating a stream that simply represents a single event or signal.

Example (Creating a Stream with a Single Void Value)

import { Stream, Effect } from "effect"

const stream = Stream.void

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ undefined ] }

range

To create a stream of sequential integers within a specified range [min, max], including both endpoints, you can use Stream.range.

Example (Creating a Stream of Numbers in a Range)

import { Stream, Effect } from "effect"

// Stream that generates numbers from 1 to 5
const stream = Stream.range(1, 5)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2, 3, 4, 5 ] }

iterate

With Stream.iterate, you can generate a stream by applying a function iteratively to an initial value. The initial value becomes the first element produced by the stream, followed by subsequent values produced by f(init), f(f(init)), and so on.

The Stream.iterate constructor generates a stream by repeatedly applying a function to an initial value. The stream begins with the initial value, followed by f(init), f(f(init)), and so on.

Example (Creating an Iterative Stream of Incrementing Numbers)

import { Stream, Effect } from "effect"

// Stream of incrementing numbers starting from 1
const stream = Stream.iterate(1, (n) => n + 1) // Produces 1, 2, 3, ...

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// Output: { _id: 'Chunk', values: [ 1, 2, 3, 4, 5 ] }

From Success and Failure

Similar to the Effect data type, you can create a Stream that either succeeds with a value or fails with an error using the Stream.succeed and Stream.fail functions.

Example (Creating Streams from Success and Failure)

import { Stream, Effect } from "effect"

// Stream that successfully emits a single numeric value
const streamWithNumber: Stream.Stream<number> = Stream.succeed(5)

Effect.runPromise(Stream.runCollect(streamWithNumber)).then(console.log)
// Output: { _id: 'Chunk', values: [ 5 ] }

// Stream that fails with an error message
const streamWithError: Stream.Stream<never, string> =
  Stream.fail("Uh oh!")

// Attempt to run the stream, which will throw an error
Effect.runPromise(Stream.runCollect(streamWithError))
/*
throws:
[Error: Uh oh!] {
  name: '(FiberFailure) Error',
  [Symbol(effect/Runtime/FiberFailure/Cause)]: {
    _tag: 'Fail',
    error: 'Uh oh!'
  }
  ...
}
*/

From Chunks

You can create a stream from a Chunk with Stream.fromChunk.

Example (Creating a Stream from a Single Chunk)

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

// Create a stream with values from a single Chunk
const stream = Stream.fromChunk(Chunk.make(1, 2, 3))

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2, 3 ] }

You can also construct a stream from multiple chunks, combining their values into a continuous stream.

Example (Creating a Stream from Multiple Chunks)

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

// Create a stream with values from multiple Chunks
const stream = Stream.fromChunks(Chunk.make(1, 2, 3), Chunk.make(4, 5, 6))

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2, 3, 4, 5, 6 ] }

From Effect

To create a stream from an Effect, use the Stream.fromEffect constructor. This is useful when you want to convert a one-time asynchronous effect into a stream.

Example (Creating a Stream from a Single Random Number)

In this example, we generate a stream that produces a single random number:

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

// Create a stream from an Effect that generates a random integer
const stream = Stream.fromEffect(Random.nextInt)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Example Output: { _id: 'Chunk', values: [ 1042302242 ] }

From Asynchronous Callback

If you have an asynchronous function that uses callbacks, you can capture its emitted values as a stream with the Stream.async function. This method is ideal for adapting functions that call back multiple times, streaming their outputs as they arrive.

Example (Creating a Stream from an Asynchronous Callback)

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

const events = [1, 2, 3, 4]

const stream = Stream.async<number>((emit) => {
  events.forEach((n) => {
    setTimeout(() => {
      if (n === 3) {
        // Terminate the stream after the third event
        emit(Effect.fail(Option.none()))
      } else {
        // Emit the current item to the stream
        emit(Effect.succeed(Chunk.of(n)))
      }
    }, 100 * n)
  })
})

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2 ] }

The emit: StreamEmit.Emit<R, E, A, void> type represents an asynchronous callback that can emit values multiple times. This callback accepts values of type Effect<Chunk<A>, Option<E>, R> and provides the following behaviors:

• When the callback succeeds with a Chunk<A>, it emits the elements in the chunk to the stream.
• If the callback fails with Some<E>, it terminates the stream with the specified error.
• If the callback fails with None, it ends the stream, signaling that there are no further values.

In short, this type lets you define how the callback interacts with the stream, specifying when to emit values, terminate due to an error, or simply end.

From Iterables

fromIterable

To create a stream from any Iterable (such as an array, set, or custom iterable), use the Stream.fromIterable constructor. This method provides a simple way to turn a collection of values into a stream.

Example (Creating a Stream from an Iterable)

In this example, we convert a Set of numbers into a stream:

import { Stream, Effect } from "effect"

const numbers = new Set([1, 2, 3])

// Create a stream from the Set of numbers
const stream = Stream.fromIterable(numbers)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2, 3 ] }

fromIterableEffect

If you have an effect that returns an Iterable, you can use the Stream.fromIterableEffect constructor to turn the effect’s result into a stream.

Example (Generating a Stream from an Effect)

In this example, imagine a database operation that retrieves a list of users. Since fetching users is an effectful operation, we can transform its result into a stream using Stream.fromIterableEffect:

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

// Define a database service with an operation to get users
class Database extends Context.Tag("Database")<
  Database,
  { readonly getUsers: Effect.Effect<Array<string>> }
>() {}

const getUsers = Database.pipe(Effect.andThen((_) => _.getUsers))

const stream = Stream.fromIterableEffect(getUsers)

Effect.runPromise(
  Stream.runCollect(
    stream.pipe(
      Stream.provideService(Database, {
        getUsers: Effect.succeed(["user1", "user2"])
      })
    )
  )
).then(console.log)
// Output: { _id: 'Chunk', values: [ 'user1', 'user2' ] }

Working with Service-Dependent Streams

When a stream depends on an external service, like Database, use Stream.provideService to supply the required service before running the stream.

fromAsyncIterable

Async iterables are another type of data source that can be converted into a stream. With the Stream.fromAsyncIterable constructor, you can work with asynchronous data sources and handle potential errors gracefully.

Example (Streaming from an Async Iterable)

In this example, we define an async iterable and then create a stream from it. An error handler is also provided to capture and process any errors that might occur during conversion.

import { Stream, Effect } from "effect"

const delay = (n: number, delay: number) =>
  new Promise((resolve) => setTimeout(() => resolve(n), delay))

// Define an async iterable that yields values asynchronously
const myAsyncIterable = async function* () {
  yield delay(1, 100)
  yield delay(2, 150)
}

// Create a stream from the async iterable, including an error handler
const stream = Stream.fromAsyncIterable(
  myAsyncIterable(),
  (e) => new Error(String(e)) // Custom error handling
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2 ] }

From Repetition

Repeating a Single Value

To create a stream that continuously repeats a single value, use the Stream.repeatValue constructor.

Example (Creating a Stream of Repeated Values)

import { Stream, Effect } from "effect"

// Create a stream that endlessly repeats the value 0
const stream = Stream.repeatValue(0)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// Output: { _id: 'Chunk', values: [ 0, 0, 0, 0, 0 ] }

In this example, the stream emits the value 0 repeatedly. Using Stream.take(5), we limit the output to the first five occurrences.

Repeating a Stream’s Content

You can use Stream.repeat to create a stream that continuously repeats the contents of another stream according to a schedule. This approach is useful for generating recurring events or repeating values over time.

Example (Repeating a Stream Sequence on a Schedule)

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

// Create a stream that endlessly repeats the values 1, 2, and 3
const stream = Stream.repeat(Stream.make(1, 2, 3), Schedule.forever)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// Output: { _id: 'Chunk', values: [ 1, 2, 3, 1, 2 ] }

In this example, the stream repeats the sequence [1, 2, 3]. The Stream.take(5) operator limits the output to the first five values.

Repeating an Effect’s Result

If you have an effectful operation, such as an API call, and want to repeatedly use its result in a stream, you can do this by creating a stream from the effect and repeating it.

Example (Generating a Stream of Random Numbers)

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

// Create a stream that repeats a random number generation effect
const stream = Stream.repeatEffect(Random.nextInt)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
/*
Example Output:
{
  _id: 'Chunk',
  values: [ 1666935266, 604851965, 2194299958, 3393707011, 4090317618 ]
}
*/

In this example, Stream.repeatEffect continually invokes the random number generator, creating a stream of random numbers. Using Stream.take(5), we limit the output to the first five values.

Repeating an Effect with Termination

You can evaluate an effect repeatedly to create a stream, with the ability to terminate based on specific conditions.

Example (Creating a Stream from an Iterator with Termination)

In this example, we create a helper function, drainIterator, that converts an Iterator into a stream. The stream will continue emitting values from the iterator until it reaches the end.

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

const drainIterator = <A>(it: Iterator<A>): Stream.Stream<A> =>
  Stream.repeatEffectOption(
    Effect.gen(function* () {
      const res = it.next()
      if (res.done) {
        // End the stream when iterator completes
        return yield* Effect.fail(Option.none())
      }
      // Emit the next iterator value
      return res.value
    })
  )

Effect.runPromise(
  Stream.runCollect(drainIterator(new Set([1, 2, 3]).values()))
).then(console.log)
// Output: { _id: 'Chunk', values: [ 1, 2, 3 ] }

Generating Ticks

The Stream.tick constructor lets you create a stream that emits void values at specified intervals, useful for setting up periodic events or timed triggers.

Example (Emitting Values at Fixed Intervals)

import { Stream, Effect } from "effect"

// Stream that emits every 100 milliseconds
const stream = Stream.tick("100 millis")

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
/*
Output:
{
  _id: 'Chunk',
  values: [ undefined, undefined, undefined, undefined, undefined ]
}
*/

In this example, the stream emits a void value every 100 milliseconds. Using Stream.take(5), we limit the output to the first five values.

From Unfolding / Pagination

The unfold function is often thought of as the opposite of reduce.

In reduce, we take a data structure and summarize it into a single return value, like calculating the sum of an Array<number>. By contrast, unfold starts with an initial value and iteratively builds a data structure, adding one element at a time based on a specified state function. For example, starting from 1 and using an increment function to generate a sequence of natural numbers.

unfold

The Stream module provides an unfold function, defined as follows:

declare const unfold: <S, A>(
  initialState: S,
  step: (s: S) => Option.Option<readonly [A, S]>
) => Stream<A>

Here’s how it works:

• initialState: The starting value for the stream generation.
• step: A state function that takes the current state s as input. If it returns None, the stream ends; if it returns Some<[A, S]>, A is the next element, and S is updated for the next iteration.

Example (Creating a Stream of Natural Numbers)

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

const stream = Stream.unfold(1, (n) => Option.some([n, n + 1]))

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// Output: { _id: 'Chunk', values: [ 1, 2, 3, 4, 5 ] }

unfoldEffect

For cases where the state transformation itself is effectful, Stream.unfoldEffect enables working with effects during the unfolding process.

Example (Generating Random Values with Effectful State Transformations)

Here’s an example of creating an infinite stream of random 1 and -1 values:

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

const stream = Stream.unfoldEffect(1, (n) =>
  Random.nextBoolean.pipe(
    Effect.map((b) => (b ? Option.some([n, -n]) : Option.some([n, n])))
  )
)

Effect.runPromise(Stream.runCollect(stream.pipe(Stream.take(5)))).then(
  console.log
)
// Example Output: { _id: 'Chunk', values: [ 1, 1, 1, 1, -1 ] }

paginate

Stream.paginate is similar to Stream.unfold but allows emitting values one step further.

Example (Creating a Simple Paginated Stream)

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

const stream = Stream.paginate(0, (n) => [
  n,
  n < 3 ? Option.some(n + 1) : Option.none()
])

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
// { _id: 'Chunk', values: [ 0, 1, 2, 3 ] }

Here’s how it works:

1. Start with an initial value of 0.
2. The function takes the current value n and returns a tuple. The first item is the emitted value (n), and the second item decides if the stream continues (Option.some(n + 1)) or stops (Option.none()).

Unfolding vs. Pagination

The difference between the Stream.unfold and Stream.paginate combinators often comes down to how they handle state and control continuation. Let’s explore these distinctions with a practical example.

Imagine a paginated API that returns data in chunks. Each request yields a PageResult object containing the results for the current page and a flag indicating whether it’s the last page or if there’s more data. Here’s a simple mock of such an API:

import { Chunk, Effect } from "effect"

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

Our goal is to turn this paginated API into a stream of RawData events. Initially, we might consider using Stream.unfoldChunkEffect, a variant of Stream.unfoldEffect that handles chunks of values rather than single values:

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

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

const firstAttempt = Stream.unfoldChunkEffect(0, (pageNumber) =>
  listPaginated(pageNumber).pipe(
    Effect.map((page) => {
      if (page.isLast) {
        return Option.none()
      }
      return Option.some([page.results, pageNumber + 1] as const)
    })
  )
)

Effect.runPromise(Stream.runCollect(firstAttempt)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ "Result 0-1", "Result 0-2", "Result 1-1", "Result 1-2" ]
}
*/

However, this approach has a drawback, it doesn’t include the results from the last page.

We could resolve this with extra logic to capture the last page, but this adds complexity:

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

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

const secondAttempt = Stream.unfoldChunkEffect(
  Option.some(0),
  (pageNumber) =>
    Option.match(pageNumber, {
      // We already hit the last page
      onNone: () => Effect.succeed(Option.none()),
      // We did not hit the last page yet
      onSome: (pageNumber) =>
        listPaginated(pageNumber).pipe(
          Effect.map((page) =>
            Option.some([
              page.results,
              page.isLast ? Option.none() : Option.some(pageNumber + 1)
            ])
          )
        )
    })
)

Effect.runPromise(Stream.runCollect(secondAttempt)).then(console.log)
/*
Output:
{
  _id: 'Chunk',
  values: [
    'Result 0-1',
    'Result 0-2',
    'Result 1-1',
    'Result 1-2',
    'Result 2-1',
    'Result 2-2'
  ]
}
*/

While this approach works, it’s clear that Stream.unfoldChunkEffect isn’t the most friendly option for retrieving data from paginated APIs. It requires additional workarounds to include the results from the last page.

This is where Stream.paginate (and its variants) comes to the rescue. It provides a more ergonomic way to convert a paginated API into a stream. Let’s rewrite our solution using Stream.paginateChunkEffect:

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

type RawData = string

class PageResult {
  constructor(
    readonly results: Chunk.Chunk<RawData>,
    readonly isLast: boolean
  ) {}
}

const pageSize = 2

const listPaginated = (
  pageNumber: number
): Effect.Effect<PageResult, Error> => {
  return Effect.succeed(
    new PageResult(
      Chunk.map(
        Chunk.range(1, pageSize),
        (index) => `Result ${pageNumber}-${index}`
      ),
      pageNumber === 2 // Return 3 pages
    )
  )
}

const finalAttempt = Stream.paginateChunkEffect(0, (pageNumber) =>
  listPaginated(pageNumber).pipe(
    Effect.andThen((page) => {
      return [
        page.results,
        page.isLast ? Option.none<number>() : Option.some(pageNumber + 1)
      ]
    })
  )
)

Effect.runPromise(Stream.runCollect(finalAttempt)).then(console.log)
/*
Output:
{
  _id: 'Chunk',
  values: [
    'Result 0-1',
    'Result 0-2',
    'Result 1-1',
    'Result 1-2',
    'Result 2-1',
    'Result 2-2'
  ]
}
*/

From Queue and PubSub

Effect provides two core asynchronous messaging types: Queue and PubSub. You can convert these data types into Streams using Stream.fromQueue and Stream.fromPubSub, making it easy to process their emitted values in a stream-based workflow.

From Schedule

To create a stream from a Schedule, you can use Stream.fromSchedule, which will emit values as dictated by the schedule’s configuration. This stream will produce an element for each interval specified in the schedule, continuing for the duration of the schedule.

The values emitted from Stream.fromSchedule are integers, starting from 0 and incrementing by 1 for each tick of the schedule.

Example (Creating a Stream from a Schedule)

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

// Configures a schedule to emit a value every 1 second, repeating 5 times
const schedule = Schedule.spaced("1 second").pipe(
  Schedule.compose(Schedule.recurs(5))
)

const stream = Stream.fromSchedule(schedule)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 0, 1, 2, 3, 4 ] }
*/