streams_skill

Overview

This skill explains how Effect handles streaming data processing, including creating, transforming, composing, and consuming streams with resource safety and backpressure. It summarizes core primitives like Stream, Sink, Channel, and common operations such as map, filter, flatMap, concat, and merge. The guidance highlights chunking, error handling, and patterns for real-time and batched processing.

How this skill works

Streams produce elements lazily and on demand, carrying types for values, errors, and required environment (Stream<A, E, R>). They coordinate producer and consumer via backpressure and process elements in chunks for efficiency. Streams can be built from values, effects, async iterables, callbacks, or queues, and are consumed by running to collections, running for side effects, or piping into Sinks.

When to use it

• When processing continuous or real-time data sources (events, sockets, sensors).
• When you need resource-safe reading of files, network connections, or DB cursors.
• When composing multiple asynchronous producers (merge, concat, zip).
• When batching or chunked processing improves throughput or reduces latency.
• When you must control concurrency, backpressure, or retry and error recovery.

Best practices

• Batch work with grouped/rechunk to reduce per-element overhead and improve throughput.
• Use acquireRelease or scoped streams to ensure external resources are opened and closed safely.
• Apply backpressure and buffer strategies; prefer bounded queues to avoid unbounded memory growth.
• Use mapEffect with concurrency options for parallel processing when side-effects are independent.
• Catch, retry, or fallback earlier in the pipeline rather than at final consumption.

Example use cases

• Consume a remote event stream, filter and enrich events, then write to a database sink with batching.
• Poll an API on a schedule and stream status updates into a real-time dashboard with throttle or debounce.
• Stream file contents safely using acquireRelease and process in fixed-size chunks for upload or parsing.
• Merge telemetry streams from multiple sources and aggregate metrics with Sink.foldLeft or Sink.sum.
• Implement backpressure-aware webhook handling using a queue-backed Stream and bounded concurrency.