Yoga.Om.Strom

newtype Strom :: Type -> Row Type -> Type -> Typenewtype Strom ctx err a

A Strom is a stream of values that can:

• Emit zero or more values of type a
• Fail with errors tracked in the err row
• Require a context ctx
• Perform effects in Om

Think of it as: Om ctx err (Array a) but lazy and chunked

The Step type is interpreted as:

• Loop (chunk, next) => emit chunk (if present) and continue with next
• Done finalChunk => emit final chunk (if present) and stop

Instances

• Functor (Strom ctx err)
• Apply (Strom ctx err)
• Applicative (Strom ctx err)
• Bind (Strom ctx err)
• Monad (Strom ctx err)
• Semigroup (Strom ctx err a)
• Monoid (Strom ctx err a)
• Alt (Strom ctx err)
• Plus (Strom ctx err)
• Alternative (Strom ctx err)

empty :: forall ctx err a. Strom ctx err a

An empty stream

succeed :: forall ctx err a. a -> Strom ctx err a

A stream with a single element

fail :: forall ctx err a. Strom ctx err a

A stream that fails immediately with an empty result (no error thrown)

fromAff :: forall ctx err a. Aff a -> Strom ctx err a

Create a stream from an Aff computation

fromOm :: forall ctx err a. Om ctx err a -> Strom ctx err a

Create a stream from an Om computation

fromArray :: forall ctx err a. Array a -> Strom ctx err a

Create a stream from an array

fromFoldable :: forall ctx err f a. Foldable f => f a -> Strom ctx err a

Create a stream from any foldable

rangeStrom :: forall ctx err. Int -> Int -> Strom ctx err Int

Create a stream of integers in a range [start, end)

iterateStrom :: forall ctx err a. (a -> a) -> a -> Strom ctx err a

Iterate producing values - limited to ~10,000 elements for stack safety For truly infinite streams, use iterateStromInfinite (with Aff overhead)

iterateStromInfinite :: forall ctx err a. (a -> a) -> a -> Strom ctx err a

Truly infinite iteration - stack-safe via Aff async boundaries Adds tiny delay (0ms) at each chunk to reset the stack Use this when you need more than 10k elements

repeatStrom :: forall ctx err a. a -> Strom ctx err a

Repeat a value - limited to ~10,000 elements for stack safety

repeatStromInfinite :: forall ctx err a. a -> Strom ctx err a

Truly infinite repeat - stack-safe via Aff async boundaries Adds tiny delay (0ms) at each chunk to reset the stack

repeatOmStrom :: forall ctx err a. Om ctx err a -> Strom ctx err a

Repeat an Om computation - limited to 100 iterations for stack safety For truly infinite repetition, use repeatOmStromInfinite (with Aff overhead)

repeatOmStromInfinite :: forall ctx err a. Om ctx err a -> Strom ctx err a

Truly infinite Om repetition - stack-safe via Aff async boundaries Adds tiny delay (0ms) at each pull to reset the stack

unfoldStrom :: forall ctx err a b. (b -> Maybe (a /\ b)) -> b -> Strom ctx err a

Unfold a stream from a seed value - can be infinite if f never returns Nothing Emits chunks lazily, terminates when f returns Nothing

unfoldOmStrom :: forall ctx err a b. (b -> Om ctx err (Maybe (a /\ b))) -> b -> Strom ctx err a

Unfold a stream with an effectful function

runCollect :: forall ctx err a. Strom ctx err a -> Om ctx err (Array a)

Collect all elements into an array (O(n) using chunk accumulation)

runDrain :: forall ctx err a. Strom ctx err a -> Om ctx err Unit

Run the stream and discard all elements

runFold :: forall ctx err a b. b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b

Fold over all elements

traverseStrom_ :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit

Traverse the stream with effects, discarding results

forStrom_ :: forall ctx err a. Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit

Alias for traverseStrom_ with arguments flipped

forMStrom_ :: forall ctx err a. Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit

Alias for traverseStrom_ (FP convention)

traverseMStrom_ :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit

Alias for traverseStrom_ (explicit monadic naming)

mapStrom :: forall ctx err a b. (a -> b) -> Strom ctx err a -> Strom ctx err b

Map over elements

mapMStrom :: forall ctx err a b. (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b

Map with a monadic effect

bindStrom :: forall ctx err a b. (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b

Bind for streams (monadic flatMap/chain) Use >>= operator for cleaner syntax: stream >>= f

scanStrom :: forall ctx err a b. (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b

Scan with accumulator (like foldl but emits intermediate results)

mapAccumStrom :: forall ctx err a b s. (s -> a -> (s /\ b)) -> s -> Strom ctx err a -> Strom ctx err b

Map with accumulator (stateful map)

tapStrom :: forall ctx err a. (a -> Unit) -> Strom ctx err a -> Strom ctx err a

Tap (observe without modifying)

tapMStrom :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a

Tap with monadic effect (observe with effects)

filterStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a

Filter elements

takeStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a

Take n elements (with proper state tracking across chunks)

takeWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a

Take while predicate is true

takeUntilStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a

Take until predicate is true (includes the matching element)

dropStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a

Drop n elements (with proper state tracking across chunks)

dropWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a

Drop while predicate is true

filterMStrom :: forall ctx err a. (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a

Filter with monadic predicate

collectStrom :: forall ctx err a b. (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b

Collect with partial function (mapMaybe)

collectMStrom :: forall ctx err a b. (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b

Collect with monadic partial function

changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a

Remove consecutive duplicates

groupedStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err (Array a)

Group elements into fixed-size arrays

chunkedStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err (Array a)

Alias for groupedStrom

groupByStrom :: forall ctx err a k. Eq k => (a -> k) -> Strom ctx err a -> Strom ctx err (Array a)

Group consecutive elements by a key function Emits groups when the key changes

partition :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> ((Strom ctx err a) /\ (Strom ctx err a))

Partition a stream by a predicate into two separate operations Returns a tuple of (trues, falses) Note: This requires running the stream twice, so it's not lazy

partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> ((Strom ctx err b) /\ (Strom ctx err c))

Partition and map in one pass

debounce :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a

Debounce a stream - adds a delay after each element Useful for rate-limiting or giving downstream time to process

throttle :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a

Throttle a stream - emit at most once per duration period Tracks last emission time and skips elements that arrive too quickly Note: Uses wall-clock time, not processing time

delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a

Delay each element by a duration

appendStrom :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

Run a stream in the background, returning a Fiber that can be cancelled

To use cancellable streams, convert your stream to Om and use Om.launchOm:

Example:

fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)
-- ... later ...
killFiber (error "Cancelled") fiber

Or for subscription-style:

fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)
-- ... later ...  
killFiber (error "Unsubscribed") fiber

Note: You can use Om.launchOm directly with any Om-returning function like:

• runCollect - collect results in background
• runDrain - run for side effects
• traverseStrom_ - process each element with callback
• runFold - fold with accumulator Append two streams Now properly handles multi-chunk streams with new Step protocol Note: s2 is appended TO s1 (s1 comes first, then s2) This matches pipeline semantics: stream # appendStrom other means stream followed by other

concatStrom :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a

Concatenate an array of streams Uses foldr with flip to match appendStrom's pipeline-friendly parameter order

zipStrom :: forall ctx err a b. Strom ctx err a -> Strom ctx err b -> Strom ctx err (a /\ b)

Zip two streams together

zipWithStrom :: forall ctx err a b c. (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c

Zip two streams with a function

interleave :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

Interleave two streams - alternates between elements from each stream Similar to merge but deterministic - always alternates

intersperse :: forall ctx err a. a -> Strom ctx err a -> Strom ctx err a

Intersperse a separator element between each element of the stream

data HaltStrategy

Merge termination strategy (like ZStream HaltStrategy)

Constructors

• HaltBoth
• HaltEither
• HaltLeft
• HaltRight

Instances

• Eq HaltStrategy

merge :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

Merge two streams concurrently (nondeterministic like ZStream merge) Elements are emitted as soon as they're available from either stream

mergeAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a

Merge multiple streams concurrently

mergeWith :: forall ctx err a b c. HaltStrategy -> (a -> c) -> (b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c

Merge with halt strategy and mapping functions (ZStream-like)

mergeHaltEither :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

mergeHaltLeft :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

mergeHaltRight :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

mergeND :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

Non-deterministic merge: Merge two streams concurrently, emitting elements as soon as they're available Unlike merge, this runs both streams in parallel and elements may arrive in any order The stream that produces results faster will have its elements appear first Optimized: Spawns producer fibers once and uses a queue for communication

mergeNDWith :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a -> Strom ctx err a

Non-deterministic merge with configurable buffer capacity (chunk queue)

mergeAllND :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a

Non-deterministic merge of multiple streams All streams run concurrently and emit elements as soon as they're available

mergeAllNDWith :: forall ctx err a. Int -> Array (Strom ctx err a) -> Strom ctx err a

Non-deterministic merge of multiple streams with configurable buffer capacity

race :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a

Race two streams - returns elements from whichever produces first Uses Om.race for true concurrent racing

raceAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a

Race multiple streams - returns elements from whichever produces first

mapPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b

Map over elements in parallel with bounded concurrency Processes chunks in parallel groups up to the concurrency limit

mapMPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b

Alias for mapPar with explicit "M" naming

foreachPar :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit

For each element, execute an effect in parallel with bounded concurrency

mapParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b

Unordered parallel map (currently same as mapPar)

mapMParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b

Alias for unordered mapPar with explicit "M" naming

foreachParUnordered :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit

Unordered parallel foreach

buffer :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a

Buffer stream elements with backpressure (destroys chunking)

bufferChunks :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a

Buffer stream chunks with backpressure (preserves chunking)

catchAll :: forall ctx err a. (Variant (exception :: Error | err) -> Strom ctx err a) -> Strom ctx err a -> Strom ctx err a

Catch all errors and recover with a handler function Note: Uses handleErrors' for full error handling

retry :: forall ctx err a. Strom ctx err a -> Strom ctx err a

Retry a stream on failure up to 3 times

retryN :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a

Retry a stream N times on failure

timeout :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a

Add a timeout to stream operations If the stream doesn't complete within the duration, it terminates as empty

ensuring :: forall ctx err a. Om ctx err Unit -> Strom ctx err a -> Strom ctx err a

Run a finaliser after the stream completes (success or failure)

bracket :: forall ctx err a b. Om ctx err a -> (a -> Om ctx err Unit) -> (a -> Strom ctx err b) -> Strom ctx err b

Bracket pattern: acquire, use, release Guarantees release runs even if stream fails

bracketExit :: forall ctx err a b. Om ctx err a -> (a -> Maybe (Variant (exception :: Error | err)) -> Om ctx err Unit) -> (a -> Strom ctx err b) -> Strom ctx err b

Bracket with exit information (whether stream succeeded or failed)

acquireRelease :: forall ctx err a b. Om ctx err a -> (a -> Om ctx err Unit) -> (a -> Strom ctx err b) -> Strom ctx err b

Simplified acquire-release pattern

mkStrom :: forall ctx err a. Om ctx err (Step ((Maybe (Chunk a)) /\ (Strom ctx err a)) (Maybe (Chunk a))) -> Strom ctx err a

Helper to create a Strom from a pull function

runStrom :: forall ctx err a. Strom ctx err a -> Om ctx err (Step ((Maybe (Chunk a)) /\ (Strom ctx err a)) (Maybe (Chunk a)))

Helper to unwrap a Strom