Streams

Streams in Effect

Overview

Effect Streams provide:

• Lazy evaluation - Elements produced on demand
• Resource safety - Automatic cleanup
• Backpressure - Producer/​consumer coordination
• Composition - Transform, filter, merge streams
• Error handling - Typed errors in stream pipeline

Stream<A, E, R>
// Produces values of type A
// May fail with error E
// Requires environment R

Creating Streams

From Values

import { Stream } from "effect"

// From known values
const numbers = Stream.make(1, 2, 3, 4, 5)

// From iterable
const fromArray = Stream.fromIterable([1, 2, 3])

// Empty stream
const empty = Stream.empty

// Single value
const single = Stream.succeed(42)

// Infinite stream
const infinite = Stream.iterate(1, (n) => n + 1)

From Effects

// Single effect as stream
const fromEffect = Stream.fromEffect(fetchData())

// Repeat effect
const polling = Stream.repeatEffect(checkStatus())

// Repeat with schedule
const scheduled = Stream.repeatEffectWithSchedule(
  checkStatus(),
  Schedule.spaced("5 seconds")
)

From Async Sources

// From async iterable
const fromAsyncIterable = Stream.fromAsyncIterable(
  asyncGenerator(),
  (error) => new StreamError({ cause: error })
)

// From callback/​event emitter
const fromCallback = Stream.async<number, never>((emit) => {
  const handler = (value: number) => emit.single(value)
  eventEmitter.on("data", handler)
  return Effect.sync(() => eventEmitter.off("data", handler))
})

// From queue
const fromQueue = Stream.fromQueue(queue)

Generating Streams

// Unfold from seed
const naturals = Stream.unfold(1, (n) => Option.some([n, n + 1]))

// Range
const range = Stream.range(1, 100)

// Repeat value
const repeated = Stream.repeat(Stream.succeed("ping")).pipe(
  Stream.take(5)
)

Transforming Streams

map - Transform Elements

const doubled = numbers.pipe(
  Stream.map((n) => n * 2)
)

// With Effect
const enriched = users.pipe(
  Stream.mapEffect((user) => fetchProfile(user.id))
)

// Parallel mapping
const parallel = items.pipe(
  Stream.mapEffect(process, { concurrency: 10 })
)

filter - Select Elements

const evens = numbers.pipe(
  Stream.filter((n) => n % 2 === 0)
)

// With Effect
const valid = items.pipe(
  Stream.filterEffect((item) => validate(item))
)

flatMap - Nested Streams

const expanded = numbers.pipe(
  Stream.flatMap((n) => Stream.make(n, n * 10, n * 100))
)
// 1, 10, 100, 2, 20, 200, ...

take/​drop

const first5 = numbers.pipe(Stream.take(5))
const skip5 = numbers.pipe(Stream.drop(5))
const firstWhile = numbers.pipe(Stream.takeWhile((n) => n < 10))
const dropWhile = numbers.pipe(Stream.dropWhile((n) => n < 10))

Combining Streams

concat - Sequential

const combined = Stream.concat(stream1, stream2)
// or
const combined = stream1.pipe(Stream.concat(stream2))

merge - Interleaved

// Interleave elements from both
const merged = Stream.merge(stream1, stream2)

// Merge multiple
const allMerged = Stream.mergeAll([s1, s2, s3], { concurrency: 3 })

zip - Pair Elements

const zipped = Stream.zip(names, ages)
// Stream<[string, number]>

// With function
const combined = Stream.zipWith(
  names,
  ages,
  (name, age) => ({ name, age })
)

interleave

const interleaved = Stream.interleave(stream1, stream2)
// a1, b1, a2, b2, ...

Consuming Streams

Running to Collection

// To array
const array = yield* Stream.runCollect(numbers)
// Returns Chunk<number>

// To single value (first)
const first = yield* Stream.runHead(numbers)
// Returns Option<number>

// Fold/​reduce
const sum = yield* Stream.runFold(
  numbers,
  0,
  (acc, n) => acc + n
)

Running for Effects

// Execute effect for each element
yield* numbers.pipe(
  Stream.runForEach((n) => Effect.log(`Got: ${n}`))
)

// Drain (consume, ignore values)
yield* numbers.pipe(Stream.runDrain)

Running to Sink

import { Sink } from "effect"

// Custom sink
const sum = yield* numbers.pipe(
  Stream.run(Sink.sum)
)

// Collect to array
const array = yield* numbers.pipe(
  Stream.run(Sink.collectAll())
)

Chunking

Streams process elements in chunks for efficiency:

// Chunk elements
const chunked = numbers.pipe(
  Stream.grouped(10)  // Stream of 10-element chunks
)

// Process by chunk
const processed = numbers.pipe(
  Stream.mapChunks((chunk) => Chunk.map(chunk, (n) => n * 2))
)

// Re-chunk
const rechunked = numbers.pipe(
  Stream.rechunk(100)
)

Error Handling

// Catch errors
const safe = stream.pipe(
  Stream.catchAll((error) => Stream.succeed(fallbackValue))
)

// Catch specific error
const handled = stream.pipe(
  Stream.catchTag("NetworkError", (error) =>
    Stream.succeed(cachedValue)
  )
)

// Retry on failure
const resilient = stream.pipe(
  Stream.retry(Schedule.exponential("1 second"))
)

// On error, switch to fallback stream
const withFallback = stream.pipe(
  Stream.orElse(() => fallbackStream)
)

Resource Management

// Stream with resource lifecycle
const fileStream = Stream.acquireRelease(
  Effect.sync(() => fs.openSync("data.txt", "r")),
  (fd) => Effect.sync(() => fs.closeSync(fd))
).pipe(
  Stream.flatMap((fd) =>
    Stream.repeatEffectOption(
      Effect.sync(() => {
        const buffer = Buffer.alloc(1024)
        const bytes = fs.readSync(fd, buffer)
        return bytes > 0 ? Option.some(buffer.slice(0, bytes)) : Option.none()
      })
    )
  )
)

// Scoped streams
const scoped = Stream.scoped(
  Effect.acquireRelease(openConnection, closeConnection)
)

Sinks

Sinks consume stream elements:

import { Sink } from "effect"

// Built-in sinks
Sink.sum          // Sum numbers
Sink.count        // Count elements
Sink.head         // First element
Sink.last         // Last element
Sink.collectAll() // Collect to Chunk
Sink.forEach(f)   // Run effect for each

// Custom sink
const maxSink = Sink.foldLeft(
  Number.NEGATIVE_INFINITY,
  (max, n: number) => Math.max(max, n)
)

Common Patterns

Batched Processing

const batchProcess = stream.pipe(
  Stream.grouped(100),
  Stream.mapEffect((batch) =>
    Effect.tryPromise(() => api.processBatch(Chunk.toArray(batch)))
  )
)

Rate Limiting

const rateLimited = stream.pipe(
  Stream.throttle({
    units: 1,
    duration: "100 millis",
    strategy: "shape"
  })
)

Debouncing

const debounced = stream.pipe(
  Stream.debounce("500 millis")
)

Windowing

// Time-based windows
const windows = stream.pipe(
  Stream.groupedWithin(1000, "1 second")
)

Best Practices

1. Use chunking for efficiency - Batch operations when possible
2. Handle backpressure - Use appropriate buffer strategies
3. Clean up resources - Use acquireRelease for external resources
4. Process in parallel - Use concurrency option in mapEffect
5. Handle errors early - Catch/​retry before final consumption

Additional Resources

For comprehensive stream documentation, consult ${CLAUDE_PLUGIN_ROOT}/​references/​llms-full.txt.

Search for these sections:

• "Creating Streams" for stream construction
• "Consuming Streams" for running streams
• "Operations" for transformations
• "Error Handling in Streams" for error patterns
• "Resourceful Streams" for resource management
• "Sink" for custom sinks