@@ -0,0 +1,1809 @@
1+-- @inline export mkStrom arity=1
2+-- @inline export runStrom arity=1
3+-- @inline export mapStrom arity=2
4+-- @inline export filterStrom arity=2
5+-- @inline export takeStrom arity=2
6+-- @inline export runFold arity=3
7+-- @inline export runDrain arity=1
8+-- @inline export rangeStrom arity=2
9+-- @inline export scanStrom arity=3
10+-- @inline export zipWithStrom arity=3
11+-- @inline export mapPar arity=3
12+-- @inline export bindStrom arity=2
13+module Yoga.Om.Strom
14+ ( Strom
15+ -- Construction
16+ , empty
17+ , succeed
18+ , fail
19+ -- , die -- TODO: Fix type constraints
20+ , fromAff
21+ , fromOm
22+ , fromArray
23+ , fromFoldable
24+ , rangeStrom
25+ , iterateStrom
26+ , iterateStromInfinite
27+ , repeatStrom
28+ , repeatStromInfinite
29+ , repeatOmStrom
30+ , repeatOmStromInfinite
31+ , unfoldStrom
32+ , unfoldOmStrom
33+ -- Running
34+ , runCollect
35+ , runDrain
36+ , runFold
37+ , traverseStrom_
38+ , forStrom_
39+ , forMStrom_
40+ , traverseMStrom_
41+ -- Transformations
42+ , mapStrom
43+ , mapMStrom
44+ , bindStrom
45+ , scanStrom
46+ , mapAccumStrom
47+ , tapStrom
48+ , tapMStrom
49+ , filterStrom
50+ -- Selection
51+ , takeStrom
52+ , takeWhileStrom
53+ , takeUntilStrom
54+ , dropStrom
55+ , dropWhileStrom
56+ , filterMStrom
57+ , collectStrom
58+ , collectMStrom
59+ , changesStrom
60+ -- Grouping
61+ , groupedStrom
62+ , chunkedStrom
63+ , groupByStrom
64+ , partition
65+ , partitionMap
66+ -- Timing
67+ , debounce
68+ , throttle
69+ , delayStrom
70+ -- Combining
71+ , appendStrom
72+ , concatStrom
73+ , zipStrom
74+ , zipWithStrom
75+ , interleave
76+ , intersperse
77+ , HaltStrategy(..)
78+ , merge
79+ , mergeAll
80+ , mergeWith
81+ , mergeHaltEither
82+ , mergeHaltLeft
83+ , mergeHaltRight
84+ , mergeND
85+ , mergeNDWith
86+ , mergeAllND
87+ , mergeAllNDWith
88+ , race
89+ , raceAll
90+ -- Parallel Processing
91+ , mapPar
92+ , mapMPar
93+ , foreachPar
94+ , mapParUnordered
95+ , mapMParUnordered
96+ , foreachParUnordered
97+ -- Buffering
98+ , buffer
99+ , bufferChunks
100+ -- Error Handling
101+ , catchAll
102+ -- , catchSome -- TODO: Fix type constraints
103+ , retry
104+ , retryN
105+ , timeout
106+ , ensuring
107+ -- Resource Safety
108+ , bracket
109+ , bracketExit
110+ , acquireRelease
111+ ) where
112+
113+import Prelude
114+
115+import Control.Alt (class Alt, (<|>))
116+import Control.Alternative (class Alternative)
117+import Control.Monad.Error.Class (throwError)
118+import Control.Monad.Rec.Class (Step(..), tailRecM)
119+import Control.Monad.ST as ST
120+import Control.Monad.ST.Ref as STRef
121+import Control.Plus (class Plus)
122+import Data.Array as Array
123+import Data.Array.ST as STArray
124+import Data.Array.ST.Iterator as Iterator
125+import Data.Either (Either(..), either)
126+import Data.Foldable (class Foldable, foldl)
127+import Data.Functor (map) as Functor
128+import Data.List as List
129+import Data.Maybe (Maybe(..))
130+import Data.Time.Duration (Milliseconds(..))
131+import Data.Traversable (traverse, traverse_)
132+import Data.Tuple (Tuple(..))
133+import Data.Tuple.Nested (type (/\), (/\))
134+import Data.Variant (Variant)
135+import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
136+import Effect.Aff.AVar as AVar
137+import Effect.Aff.Class (liftAff)
138+import Effect.Exception as Exception
139+import Effect.Now (now)
140+import Effect.Ref as Ref
141+import Effect.Class (liftEffect)
142+import Data.CatQueue as CatQueue
143+import Data.CatQueue (CatQueue)
144+import Data.DateTime.Instant (unInstant)
145+import Yoga.Om (Om)
146+import Yoga.Om as Om
147+
148+--------------------------------------------------------------------------------
149+-- Core Types
150+--------------------------------------------------------------------------------
151+
152+-- | A Strom is a stream of values that can:
153+-- | - Emit zero or more values of type `a`
154+-- | - Fail with errors tracked in the `err` row
155+-- | - Require a context `ctx`
156+-- | - Perform effects in Om
157+-- |
158+-- | Think of it as: `Om ctx err (Array a)` but lazy and chunked
159+-- |
160+-- | The Step type is interpreted as:
161+-- | - Loop (chunk, next) => emit chunk (if present) and continue with next
162+-- | - Done finalChunk => emit final chunk (if present) and stop
163+newtype Strom ctx err a = Strom
164+ { pull :: Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a)))
165+ }
166+
167+-- | Chunks for efficient batch processing
168+type Chunk a = Array a
169+
170+-- | Helper to create a Strom from a pull function
171+mkStrom :: forall ctx err a. Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a))) -> Strom ctx err a
172+mkStrom pull = Strom { pull }
173+
174+-- | Helper to unwrap a Strom
175+runStrom :: forall ctx err a. Strom ctx err a -> Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a)))
176+runStrom (Strom s) = s.pull
177+
178+--------------------------------------------------------------------------------
179+-- Non-Deterministic Merge Types
180+--------------------------------------------------------------------------------
181+
182+-- | Stream identifier for tracking which stream produced data
183+data StreamId = StreamId1 | StreamId2
184+
185+derive instance eqStreamId :: Eq StreamId
186+derive instance ordStreamId :: Ord StreamId
187+
188+-- | Messages passed through the coordination queue
189+data QueueItem a
190+ = DataChunk (Array a) StreamId
191+ | StreamDone StreamId
192+
193+derive instance functorQueueItem :: Functor QueueItem
194+
195+-- | Messages for mergeAllND buffered queue
196+data MergeAllItem a
197+ = MergeChunk (Array a)
198+ | MergeDone
199+
200+derive instance functorMergeAllItem :: Functor MergeAllItem
201+
202+-- | Merge termination strategy (like ZStream HaltStrategy)
203+data HaltStrategy
204+ = HaltBoth
205+ | HaltEither
206+ | HaltLeft
207+ | HaltRight
208+
209+derive instance eqHaltStrategy :: Eq HaltStrategy
210+
211+-- | Messages for buffering queues
212+data BufferItem a
213+ = BufferChunk (Array a)
214+ | BufferDone
215+
216+derive instance functorBufferItem :: Functor BufferItem
217+
218+--------------------------------------------------------------------------------
219+-- Construction
220+--------------------------------------------------------------------------------
221+
222+-- | An empty stream
223+empty :: forall ctx err a. Strom ctx err a
224+empty = mkStrom $ pure $ Done Nothing
225+
226+-- | A stream with a single element
227+succeed :: forall ctx err a. a -> Strom ctx err a
228+succeed a = mkStrom $ pure $ Done $ Just [ a ]
229+
230+-- | A stream that fails immediately with an empty result (no error thrown)
231+fail :: forall ctx err a. Strom ctx err a
232+fail = empty
233+
234+-- TODO: Add die function with proper type constraints
235+-- die :: forall ctx err errors a. SingletonVariantRecord err (exception :: Error | errors) => Record err -> Strom ctx (exception :: Error | errors) a
236+-- die error = mkStrom $ Om.throw error
237+
238+-- | Create a stream from an Aff computation
239+fromAff :: forall ctx err a. Aff a -> Strom ctx err a
240+fromAff aff = mkStrom do
241+ value <- liftAff aff
242+ pure $ Done $ Just [ value ]
243+
244+-- | Create a stream from an Om computation
245+fromOm :: forall ctx err a. Om ctx err a -> Strom ctx err a
246+fromOm om = mkStrom do
247+ value <- om
248+ pure $ Done $ Just [ value ]
249+
250+-- | Create a stream from an array
251+fromArray :: forall ctx err a. Array a -> Strom ctx err a
252+fromArray arr =
253+ if Array.null arr then mkStrom $ pure $ Done Nothing
254+ else mkStrom $ pure $ Done $ Just arr
255+
256+-- | Create a stream from any foldable
257+fromFoldable :: forall ctx err f a. Foldable f => f a -> Strom ctx err a
258+fromFoldable = fromArray <<< Array.fromFoldable
259+
260+-- | Create a stream of integers in a range [start, end)
261+rangeStrom :: forall ctx err. Int -> Int -> Strom ctx err Int
262+rangeStrom start end = rangeHelper start end
263+ where
264+ rangeHelper current limit
265+ | current >= limit = empty
266+ | otherwise = mkStrom do
267+ let
268+ chunkSize = 1000
269+ chunkEnd = min (current + chunkSize) limit
270+ chunk = Array.range current (chunkEnd - 1)
271+ if chunkEnd >= limit then pure $ Done $ Just chunk -- Final chunk
272+ else pure $ Loop $ Tuple (Just chunk) (rangeHelper chunkEnd limit) -- Emit chunk and continue
273+
274+-- | Iterate producing values - limited to ~10,000 elements for stack safety
275+-- | For truly infinite streams, use `iterateStromInfinite` (with Aff overhead)
276+iterateStrom :: forall ctx err a. (a -> a) -> a -> Strom ctx err a
277+iterateStrom f initial = fromArray $ ST.run do
278+ arr <- STArray.unsafeThaw []
279+ valRef <- STRef.new initial
280+ iter <- Iterator.iterator (\i -> if i < 10000 then Just i else Nothing)
281+ Iterator.iterate iter \_ -> do
282+ v <- STRef.read valRef
283+ void $ STArray.push v arr
284+ void $ STRef.modify f valRef
285+ STArray.unsafeFreeze arr
286+
287+-- | Truly infinite iteration - stack-safe via Aff async boundaries
288+-- | Adds tiny delay (0ms) at each chunk to reset the stack
289+-- | Use this when you need more than 10k elements
290+iterateStromInfinite :: forall ctx err a. (a -> a) -> a -> Strom ctx err a
291+iterateStromInfinite f initial = iterateHelper initial
292+ where
293+ iterateHelper current = mkStrom do
294+ -- Async boundary resets the stack!
295+ liftAff $ delay (Milliseconds 0.0)
296+ let
297+ chunkSize = 10000
298+ -- Build chunk using STArray with Iterator (stack-safe!)
299+ result = ST.run do
300+ arr <- STArray.unsafeThaw []
301+ valRef <- STRef.new current
302+ iter <- Iterator.iterator (\i -> if i < chunkSize then Just i else Nothing)
303+ Iterator.iterate iter \_ -> do
304+ val <- STRef.read valRef
305+ void $ STArray.push val arr
306+ void $ STRef.modify f valRef
307+ finalVal <- STRef.read valRef
308+ frozen <- STArray.unsafeFreeze arr
309+ pure { arr: frozen, nextVal: finalVal }
310+ pure $ Loop $ Tuple (Just result.arr) (iterateHelper result.nextVal)
311+
312+-- | Repeat a value - limited to ~10,000 elements for stack safety
313+repeatStrom :: forall ctx err a. a -> Strom ctx err a
314+repeatStrom a = fromArray (Array.replicate 10000 a)
315+
316+-- | Truly infinite repeat - stack-safe via Aff async boundaries
317+-- | Adds tiny delay (0ms) at each chunk to reset the stack
318+repeatStromInfinite :: forall ctx err a. a -> Strom ctx err a
319+repeatStromInfinite a = repeatHelper unit
320+ where
321+ repeatHelper _ = mkStrom do
322+ -- Async boundary resets the stack!
323+ liftAff $ delay (Milliseconds 0.0)
324+ let chunk = Array.replicate 10000 a
325+ pure $ Loop $ Tuple (Just chunk) (repeatHelper unit)
326+
327+-- | Repeat an Om computation - limited to 100 iterations for stack safety
328+-- | For truly infinite repetition, use `repeatOmStromInfinite` (with Aff overhead)
329+repeatOmStrom :: forall ctx err a. Om ctx err a -> Strom ctx err a
330+repeatOmStrom om = repeatOmHelper 0
331+ where
332+ maxCount = 100
333+ repeatOmHelper count
334+ | count >= maxCount = empty
335+ | otherwise = mkStrom do
336+ value <- om
337+ pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper (count + 1))
338+
339+-- | Truly infinite Om repetition - stack-safe via Aff async boundaries
340+-- | Adds tiny delay (0ms) at each pull to reset the stack
341+repeatOmStromInfinite :: forall ctx err a. Om ctx err a -> Strom ctx err a
342+repeatOmStromInfinite om = repeatOmHelper unit
343+ where
344+ repeatOmHelper _ = mkStrom do
345+ -- Async boundary resets the stack!
346+ liftAff $ delay (Milliseconds 0.0)
347+ value <- om
348+ pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper unit)
349+
350+-- | Unfold a stream from a seed value - can be infinite if f never returns Nothing
351+-- | Emits chunks lazily, terminates when f returns Nothing
352+unfoldStrom :: forall ctx err a b. (b -> Maybe (Tuple a b)) -> b -> Strom ctx err a
353+unfoldStrom f seed = unfoldHelper seed
354+ where
355+ unfoldHelper currentSeed = mkStrom do
356+ let chunk = buildChunk currentSeed 10000
357+ case chunk.finalSeed of
358+ Nothing -> pure $ Done $ if Array.null chunk.values then Nothing else Just chunk.values
359+ Just nextSeed ->
360+ if Array.null chunk.values then runStrom (unfoldHelper nextSeed)
361+ else pure $ Loop $ Tuple (Just chunk.values) (unfoldHelper nextSeed)
362+
363+ buildChunk seed' size = ST.run do
364+ arr <- STArray.unsafeThaw []
365+ seedRef <- STRef.new seed'
366+ stoppedRef <- STRef.new false
367+ iter <- Iterator.iterator (\i -> if i < size then Just i else Nothing)
368+ Iterator.iterate iter \_ -> do
369+ stopped <- STRef.read stoppedRef
370+ when (not stopped) do
371+ s <- STRef.read seedRef
372+ case f s of
373+ Nothing -> void $ STRef.write true stoppedRef
374+ Just (Tuple value nextS) -> do
375+ void $ STArray.push value arr
376+ void $ STRef.write nextS seedRef
377+ frozen <- STArray.unsafeFreeze arr
378+ stopped <- STRef.read stoppedRef
379+ finalSeed <- STRef.read seedRef
380+ pure { values: frozen, finalSeed: if stopped then Nothing else Just finalSeed }
381+
382+-- | Unfold a stream with an effectful function
383+unfoldOmStrom :: forall ctx err a b. (b -> Om ctx err (Maybe (Tuple a b))) -> b -> Strom ctx err a
384+unfoldOmStrom f seed = mkStrom go
385+ where
386+ go = do
387+ result <- f seed
388+ case result of
389+ Nothing -> pure $ Done Nothing
390+ Just (Tuple value nextSeed) -> pure $ Loop $ Tuple (Just [ value ]) (unfoldOmStrom f nextSeed)
391+
392+--------------------------------------------------------------------------------
393+-- Transformations
394+--------------------------------------------------------------------------------
395+
396+-- | Map over elements
397+mapStrom :: forall ctx err a b. (a -> b) -> Strom ctx err a -> Strom ctx err b
398+mapStrom f stream = mkStrom do
399+ step <- runStrom stream
400+ case step of
401+ Done Nothing -> pure $ Done Nothing
402+ Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
403+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple (Functor.map (Functor.map f) maybeChunk) (mapStrom f next)
404+
405+-- | Map with a monadic effect
406+mapMStrom :: forall ctx err a b. (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
407+mapMStrom f stream = mkStrom do
408+ step <- runStrom stream
409+ case step of
410+ Done Nothing -> pure $ Done Nothing
411+ Done (Just chunk) -> do
412+ mapped <- traverse f chunk
413+ pure $ Done $ Just mapped
414+ Loop (Tuple maybeChunk next) -> do
415+ mappedChunk <- case maybeChunk of
416+ Nothing -> pure Nothing
417+ Just chunk -> Just <$> traverse f chunk
418+ pure $ Loop $ Tuple mappedChunk (mapMStrom f next)
419+
420+-- | Tap (observe without modifying)
421+tapStrom :: forall ctx err a. (a -> Unit) -> Strom ctx err a -> Strom ctx err a
422+tapStrom f stream = mkStrom do
423+ step <- runStrom stream
424+ case step of
425+ Done Nothing -> pure $ Done Nothing
426+ Done (Just chunk) -> do
427+ let _ = Functor.map f chunk
428+ pure $ Done $ Just chunk
429+ Loop (Tuple maybeChunk next) -> do
430+ let
431+ tappedChunk = case maybeChunk of
432+ Nothing -> Nothing
433+ Just chunk ->
434+ let
435+ _ = Functor.map f chunk
436+ in
437+ Just chunk
438+ pure $ Loop $ Tuple tappedChunk (tapStrom f next)
439+
440+-- | Tap with monadic effect (observe with effects)
441+tapMStrom :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a
442+tapMStrom f stream = mkStrom do
443+ step <- runStrom stream
444+ case step of
445+ Done Nothing -> pure $ Done Nothing
446+ Done (Just chunk) -> do
447+ _ <- traverse f chunk
448+ pure $ Done $ Just chunk
449+ Loop (Tuple maybeChunk next) -> do
450+ tappedChunk <- case maybeChunk of
451+ Nothing -> pure Nothing
452+ Just chunk -> do
453+ _ <- traverse f chunk
454+ pure $ Just chunk
455+ pure $ Loop $ Tuple tappedChunk (tapMStrom f next)
456+
457+-- | Bind for streams (monadic flatMap/chain)
458+-- | Use >>= operator for cleaner syntax: `stream >>= f`
459+bindStrom :: forall ctx err a b. (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b
460+bindStrom f stream = stream >>= f
461+
462+-- | Scan with accumulator (like foldl but emits intermediate results)
463+scanStrom :: forall ctx err a b. (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b
464+scanStrom f initial stream = scanHelper initial stream
465+ where
466+ scanHelper acc s = mkStrom do
467+ step <- runStrom s
468+ case step of
469+ Done Nothing -> pure $ Done Nothing
470+ Done (Just chunk) -> do
471+ -- Build results by folding and collecting intermediate values
472+ let
473+ Tuple _ results = Array.foldl
474+ ( \(Tuple prevAcc accList) item ->
475+ let
476+ newAcc = f prevAcc item
477+ in
478+ Tuple newAcc (Array.snoc accList newAcc)
479+ )
480+ (Tuple acc [])
481+ chunk
482+ pure $ Done $ if Array.null results then Nothing else Just results
483+ Loop (Tuple maybeChunk next) ->
484+ case maybeChunk of
485+ Nothing -> pure $ Loop $ Tuple Nothing (scanHelper acc next)
486+ Just chunk -> do
487+ let
488+ Tuple newAcc results = Array.foldl
489+ ( \(Tuple prevAcc accList) item ->
490+ let
491+ currAcc = f prevAcc item
492+ in
493+ Tuple currAcc (Array.snoc accList currAcc)
494+ )
495+ (Tuple acc [])
496+ chunk
497+ if Array.null results then pure $ Loop $ Tuple Nothing (scanHelper newAcc next)
498+ else pure $ Loop $ Tuple (Just results) (scanHelper newAcc next)
499+
500+-- | Map with accumulator (stateful map)
501+mapAccumStrom :: forall ctx err a b s. (s -> a -> Tuple s b) -> s -> Strom ctx err a -> Strom ctx err b
502+mapAccumStrom f initial stream = mapAccumHelper initial stream
503+ where
504+ mapAccumHelper state s = mkStrom do
505+ step <- runStrom s
506+ case step of
507+ Done Nothing -> pure $ Done Nothing
508+ Done (Just chunk) -> do
509+ let
510+ Tuple finalState results = Array.foldl
511+ ( \(Tuple st acc) a ->
512+ let
513+ Tuple newSt b = f st a
514+ in
515+ Tuple newSt (Array.snoc acc b)
516+ )
517+ (Tuple state [])
518+ chunk
519+ pure $ Done $ if Array.null results then Nothing else Just results
520+ Loop (Tuple maybeChunk next) ->
521+ case maybeChunk of
522+ Nothing -> pure $ Loop $ Tuple Nothing (mapAccumHelper state next)
523+ Just chunk -> do
524+ let
525+ Tuple newState results = Array.foldl
526+ ( \(Tuple st acc) a ->
527+ let
528+ Tuple newSt b = f st a
529+ in
530+ Tuple newSt (Array.snoc acc b)
531+ )
532+ (Tuple state [])
533+ chunk
534+ if Array.null results then pure $ Loop $ Tuple Nothing (mapAccumHelper newState next)
535+ else pure $ Loop $ Tuple (Just results) (mapAccumHelper newState next)
536+
537+-- | Filter elements
538+filterStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
539+filterStrom predicate stream = mkStrom do
540+ step <- runStrom stream
541+ case step of
542+ Done Nothing -> pure $ Done Nothing
543+ Done (Just chunk) -> do
544+ let filtered = Array.filter predicate chunk
545+ pure $ Done $ if Array.null filtered then Nothing else Just filtered
546+ Loop (Tuple maybeChunk next) -> do
547+ let
548+ filteredChunk = case maybeChunk of
549+ Nothing -> Nothing
550+ Just chunk ->
551+ let
552+ filtered = Array.filter predicate chunk
553+ in
554+ if Array.null filtered then Nothing else Just filtered
555+ pure $ Loop $ Tuple filteredChunk (filterStrom predicate next)
556+
557+--------------------------------------------------------------------------------
558+-- Selection
559+--------------------------------------------------------------------------------
560+
561+-- | Take n elements (with proper state tracking across chunks)
562+takeStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
563+takeStrom n stream = takeHelper n stream
564+ where
565+ takeHelper remaining s
566+ | remaining <= 0 = empty
567+ | otherwise = mkStrom do
568+ step <- runStrom s
569+ case step of
570+ Done Nothing -> pure $ Done Nothing
571+ Done (Just chunk) -> do
572+ let
573+ taken = Array.take remaining chunk
574+ numTaken = Array.length taken
575+ if numTaken == 0 then pure $ Done Nothing
576+ else pure $ Done $ Just taken
577+ Loop (Tuple maybeChunk next) ->
578+ case maybeChunk of
579+ Nothing -> pure $ Loop $ Tuple Nothing (takeHelper remaining next)
580+ Just chunk -> do
581+ let
582+ chunkLen = Array.length chunk
583+ taken = Array.take remaining chunk
584+ numTaken = Array.length taken
585+ newRemaining = remaining - numTaken
586+ if numTaken == 0 then pure $ Done Nothing
587+ else if newRemaining <= 0 then pure $ Done $ Just taken
588+ else pure $ Loop $ Tuple (Just taken) (takeHelper newRemaining next)
589+
590+-- | Take while predicate is true
591+takeWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
592+takeWhileStrom predicate stream = mkStrom do
593+ step <- runStrom stream
594+ case step of
595+ Done Nothing -> pure $ Done Nothing
596+ Done (Just chunk) -> do
597+ let taken = Array.takeWhile predicate chunk
598+ if Array.length taken < Array.length chunk then pure $ Done $ Just taken
599+ else pure $ Done $ Just taken
600+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (takeWhileStrom predicate next)
601+
602+-- | Drop n elements (with proper state tracking across chunks)
603+dropStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
604+dropStrom n stream
605+ | n <= 0 = stream
606+ | otherwise = dropHelper n stream
607+ where
608+ dropHelper remaining s
609+ | remaining <= 0 = s
610+ | otherwise = mkStrom do
611+ step <- runStrom s
612+ case step of
613+ Done Nothing -> pure $ Done Nothing
614+ Done (Just chunk) -> do
615+ let dropped = Array.drop remaining chunk
616+ pure $ Done $ if Array.null dropped then Nothing else Just dropped
617+ Loop (Tuple maybeChunk next) ->
618+ case maybeChunk of
619+ Nothing -> pure $ Loop $ Tuple Nothing (dropHelper remaining next)
620+ Just chunk ->
621+ let
622+ chunkLen = Array.length chunk
623+ in
624+ if remaining >= chunkLen then pure $ Loop $ Tuple Nothing (dropHelper (remaining - chunkLen) next)
625+ else
626+ let
627+ dropped = Array.drop remaining chunk
628+ in
629+ pure $ Loop $ Tuple (Just dropped) next
630+
631+-- | Take until predicate is true (includes the matching element)
632+takeUntilStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
633+takeUntilStrom predicate stream = mkStrom do
634+ step <- runStrom stream
635+ case step of
636+ Done Nothing -> pure $ Done Nothing
637+ Done (Just chunk) -> do
638+ case Array.findIndex predicate chunk of
639+ Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Include the matching element
640+ Nothing -> pure $ Done $ Just chunk
641+ Loop (Tuple maybeChunk next) ->
642+ case maybeChunk of
643+ Nothing -> pure $ Loop $ Tuple Nothing (takeUntilStrom predicate next)
644+ Just chunk -> case Array.findIndex predicate chunk of
645+ Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Found match, stop here
646+ Nothing -> pure $ Loop $ Tuple (Just chunk) (takeUntilStrom predicate next) -- Keep going
647+
648+-- | Drop while predicate is true
649+dropWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
650+dropWhileStrom predicate stream = dropWhileHelper true stream
651+ where
652+ dropWhileHelper stillDropping s
653+ | not stillDropping = s
654+ | otherwise = mkStrom do
655+ step <- runStrom s
656+ case step of
657+ Done Nothing -> pure $ Done Nothing
658+ Done (Just chunk) -> do
659+ let dropped = Array.dropWhile predicate chunk
660+ pure $ Done $ if Array.null dropped then Nothing else Just dropped
661+ Loop (Tuple maybeChunk next) ->
662+ case maybeChunk of
663+ Nothing -> pure $ Loop $ Tuple Nothing (dropWhileHelper stillDropping next)
664+ Just chunk -> do
665+ let dropped = Array.dropWhile predicate chunk
666+ if Array.null dropped then pure $ Loop $ Tuple Nothing (dropWhileHelper true next)
667+ else pure $ Loop $ Tuple (Just dropped) next -- Found elements to keep, stop dropping
668+
669+-- | Filter with monadic predicate
670+filterMStrom :: forall ctx err a. (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a
671+filterMStrom predicate stream = mkStrom do
672+ step <- runStrom stream
673+ case step of
674+ Done Nothing -> pure $ Done Nothing
675+ Done (Just chunk) -> do
676+ filtered <- Array.filterA predicate chunk
677+ pure $ Done $ if Array.null filtered then Nothing else Just filtered
678+ Loop (Tuple maybeChunk next) -> do
679+ filteredChunk <- case maybeChunk of
680+ Nothing -> pure Nothing
681+ Just chunk -> do
682+ filtered <- Array.filterA predicate chunk
683+ pure $ if Array.null filtered then Nothing else Just filtered
684+ pure $ Loop $ Tuple filteredChunk (filterMStrom predicate next)
685+
686+-- | Collect with partial function (mapMaybe)
687+collectStrom :: forall ctx err a b. (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b
688+collectStrom f stream = mkStrom do
689+ step <- runStrom stream
690+ case step of
691+ Done Nothing -> pure $ Done Nothing
692+ Done (Just chunk) -> do
693+ let collected = Array.mapMaybe f chunk
694+ pure $ Done $ if Array.null collected then Nothing else Just collected
695+ Loop (Tuple maybeChunk next) ->
696+ let
697+ collected = case maybeChunk of
698+ Nothing -> Nothing
699+ Just chunk ->
700+ let
701+ c = Array.mapMaybe f chunk
702+ in
703+ if Array.null c then Nothing else Just c
704+ in
705+ pure $ Loop $ Tuple collected (collectStrom f next)
706+
707+-- | Collect with monadic partial function
708+collectMStrom :: forall ctx err a b. (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b
709+collectMStrom f stream = mkStrom do
710+ step <- runStrom stream
711+ case step of
712+ Done Nothing -> pure $ Done Nothing
713+ Done (Just chunk) -> do
714+ maybes <- traverse f chunk
715+ let collected = Array.mapMaybe identity maybes
716+ pure $ Done $ if Array.null collected then Nothing else Just collected
717+ Loop (Tuple maybeChunk next) -> do
718+ collected <- case maybeChunk of
719+ Nothing -> pure Nothing
720+ Just chunk -> do
721+ maybes <- traverse f chunk
722+ let c = Array.mapMaybe identity maybes
723+ pure $ if Array.null c then Nothing else Just c
724+ pure $ Loop $ Tuple collected (collectMStrom f next)
725+
726+-- | Remove consecutive duplicates
727+changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a
728+changesStrom stream = changesHelper Nothing stream
729+ where
730+ changesHelper lastSeen s = mkStrom do
731+ step <- runStrom s
732+ case step of
733+ Done Nothing -> pure $ Done Nothing
734+ Done (Just chunk) -> do
735+ let
736+ Tuple _ result = Array.foldl
737+ ( \(Tuple prev acc) curr ->
738+ if Just curr == prev then Tuple (Just curr) acc
739+ else Tuple (Just curr) (Array.snoc acc curr)
740+ )
741+ (Tuple lastSeen [])
742+ chunk
743+ pure $ Done $ if Array.null result then Nothing else Just result
744+ Loop (Tuple maybeChunk next) ->
745+ case maybeChunk of
746+ Nothing -> pure $ Loop $ Tuple Nothing (changesHelper lastSeen next)
747+ Just chunk -> do
748+ let
749+ Tuple newLast result = Array.foldl
750+ ( \(Tuple prev acc) curr ->
751+ if Just curr == prev then Tuple (Just curr) acc
752+ else Tuple (Just curr) (Array.snoc acc curr)
753+ )
754+ (Tuple lastSeen [])
755+ chunk
756+ if Array.null result then pure $ Loop $ Tuple Nothing (changesHelper newLast next)
757+ else pure $ Loop $ Tuple (Just result) (changesHelper newLast next)
758+
759+--------------------------------------------------------------------------------
760+-- Grouping
761+--------------------------------------------------------------------------------
762+
763+-- | Group elements into fixed-size arrays
764+groupedStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err (Array a)
765+groupedStrom size stream
766+ | size <= 0 = empty
767+ | otherwise = groupedHelper [] stream
768+ where
769+ groupedHelper buffer s = mkStrom do
770+ step <- runStrom s
771+ case step of
772+ Done Nothing ->
773+ -- End of stream, emit remaining buffer if non-empty
774+ pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
775+ Done (Just chunk) -> do
776+ -- Process final chunk with buffer
777+ let
778+ combined = buffer <> chunk
779+ groups = groupsOf size combined
780+ completeGroups = Array.dropEnd 1 groups
781+ lastGroup = case Array.last groups of
782+ Just g -> if Array.length g == size then [ g ] else []
783+ Nothing -> []
784+ remainder = case Array.last groups of
785+ Just g -> if Array.length g < size then g else []
786+ Nothing -> []
787+ if Array.null remainder then pure $ Done $ if Array.null (completeGroups <> lastGroup) then Nothing else Just (completeGroups <> lastGroup)
788+ else pure $ Done $ if Array.null (completeGroups <> [ remainder ]) then Nothing else Just (completeGroups <> [ remainder ])
789+ Loop (Tuple maybeChunk next) ->
790+ case maybeChunk of
791+ Nothing -> pure $ Loop $ Tuple Nothing (groupedHelper buffer next)
792+ Just chunk -> do
793+ let
794+ combined = buffer <> chunk
795+ groups = groupsOf size combined
796+ completeGroups = if Array.null groups then [] else Array.dropEnd 1 groups
797+ newBuffer = case Array.last groups of
798+ Just g -> if Array.length g < size then g else []
799+ Nothing -> []
800+ if Array.null completeGroups then pure $ Loop $ Tuple Nothing (groupedHelper newBuffer next)
801+ else pure $ Loop $ Tuple (Just completeGroups) (groupedHelper newBuffer next)
802+
803+ groupsOf :: forall b. Int -> Array b -> Array (Array b)
804+ groupsOf n arr
805+ | n <= 0 || Array.null arr = []
806+ | otherwise =
807+ let
808+ group = Array.take n arr
809+ rest = Array.drop n arr
810+ in
811+ if Array.null group then [] else Array.cons group (groupsOf n rest)
812+
813+-- | Alias for groupedStrom
814+chunkedStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err (Array a)
815+chunkedStrom = groupedStrom
816+
817+-- | Group consecutive elements by a key function
818+-- | Emits groups when the key changes
819+groupByStrom :: forall ctx err a k. Eq k => (a -> k) -> Strom ctx err a -> Strom ctx err (Array a)
820+groupByStrom keyFn stream = groupByHelper Nothing [] stream
821+ where
822+ groupByHelper lastKey buffer s = mkStrom do
823+ step <- runStrom s
824+ case step of
825+ Done Nothing ->
826+ pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
827+ Done (Just chunk) -> do
828+ let
829+ Tuple _ result = Array.foldl
830+ ( \(Tuple currentKey acc) item ->
831+ let
832+ itemKey = keyFn item
833+ in
834+ case currentKey of
835+ Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
836+ Just k ->
837+ if k == itemKey then Tuple (Just itemKey) { groups: acc.groups, buffer: Array.snoc acc.buffer item }
838+ else Tuple (Just itemKey) { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] }
839+ )
840+ (Tuple lastKey { groups: [], buffer })
841+ chunk
842+ allGroups =
843+ if Array.null result.buffer then result.groups
844+ else Array.snoc result.groups result.buffer
845+ pure $ Done $ if Array.null allGroups then Nothing else Just allGroups
846+ Loop (Tuple maybeChunk next) ->
847+ case maybeChunk of
848+ Nothing -> pure $ Loop $ Tuple Nothing (groupByHelper lastKey buffer next)
849+ Just chunk -> do
850+ let
851+ Tuple newKey result = Array.foldl
852+ ( \(Tuple currentKey acc) item ->
853+ let
854+ itemKey = keyFn item
855+ in
856+ case currentKey of
857+ Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
858+ Just k ->
859+ if k == itemKey then Tuple (Just itemKey) { groups: acc.groups, buffer: Array.snoc acc.buffer item }
860+ else Tuple (Just itemKey) { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] }
861+ )
862+ (Tuple lastKey { groups: [], buffer })
863+ chunk
864+ if Array.null result.groups then pure $ Loop $ Tuple Nothing (groupByHelper newKey result.buffer next)
865+ else pure $ Loop $ Tuple (Just result.groups) (groupByHelper newKey result.buffer next)
866+
867+-- | Partition a stream by a predicate into two separate operations
868+-- | Returns a tuple of (trues, falses)
869+-- | Note: This requires running the stream twice, so it's not lazy
870+partition :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Tuple (Strom ctx err a) (Strom ctx err a)
871+partition predicate stream =
872+ Tuple
873+ (filterStrom predicate stream)
874+ (filterStrom (not <<< predicate) stream)
875+
876+-- | Partition and map in one pass
877+partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> Tuple (Strom ctx err b) (Strom ctx err c)
878+partitionMap f stream =
879+ let
880+ leftOnly x = case f x of
881+ Left b -> Just b
882+ Right _ -> Nothing
883+ rightOnly x = case f x of
884+ Left _ -> Nothing
885+ Right c -> Just c
886+ in
887+ Tuple
888+ (collectStrom leftOnly stream)
889+ (collectStrom rightOnly stream)
890+
891+--------------------------------------------------------------------------------
892+-- Timing
893+--------------------------------------------------------------------------------
894+
895+-- | Debounce a stream - adds a delay after each element
896+-- | Useful for rate-limiting or giving downstream time to process
897+debounce :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
898+debounce duration stream = mkStrom do
899+ step <- runStrom stream
900+ case step of
901+ Done Nothing -> pure $ Done Nothing
902+ Done (Just chunk) -> do
903+ Om.delay duration
904+ pure $ Done $ Just chunk
905+ Loop (Tuple maybeChunk next) -> do
906+ case maybeChunk of
907+ Nothing -> pure $ Loop $ Tuple Nothing (debounce duration next)
908+ Just chunk -> do
909+ Om.delay duration
910+ pure $ Loop $ Tuple (Just chunk) (debounce duration next)
911+
912+-- | Throttle a stream - emit at most once per duration period
913+-- | Tracks last emission time and skips elements that arrive too quickly
914+-- | Note: Uses wall-clock time, not processing time
915+throttle :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
916+throttle (Milliseconds duration) stream = throttleWithState 0.0 stream
917+ where
918+ throttleWithState lastEmit s = mkStrom do
919+ step <- runStrom s
920+ case step of
921+ Done Nothing -> pure $ Done Nothing
922+ Done (Just chunk) -> do
923+ Tuple newLastEmit filtered <- filterByTime lastEmit chunk
924+ pure $ Done $ if Array.null filtered then Nothing else Just filtered
925+ Loop (Tuple maybeChunk next) -> do
926+ case maybeChunk of
927+ Nothing -> pure $ Loop $ Tuple Nothing (throttleWithState lastEmit next)
928+ Just chunk -> do
929+ Tuple newLastEmit filtered <- filterByTime lastEmit chunk
930+ if Array.null filtered then pure $ Loop $ Tuple Nothing (throttleWithState newLastEmit next)
931+ else pure $ Loop $ Tuple (Just filtered) (throttleWithState newLastEmit next)
932+
933+ filterByTime lastEmit chunk = do
934+ Tuple newLastEmit results <- Array.foldl
935+ ( \acc value -> do
936+ Tuple currentLast filtered <- acc
937+ currentTime <- liftEffect now
938+ let Milliseconds currentMillis = unInstant currentTime
939+ let elapsed = currentMillis - currentLast
940+ if elapsed >= duration then
941+ pure $ Tuple currentMillis (Array.snoc filtered value)
942+ else
943+ pure $ Tuple currentLast filtered
944+ )
945+ (pure $ Tuple lastEmit [])
946+ chunk
947+ pure $ Tuple newLastEmit results
948+
949+-- | Delay each element by a duration
950+delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
951+delayStrom duration stream = mkStrom do
952+ Om.delay duration
953+ runStrom stream
954+
955+--------------------------------------------------------------------------------
956+-- Running
957+--------------------------------------------------------------------------------
958+
959+-- | Fold over all elements
960+runFold :: forall ctx err a b. b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b
961+runFold initial f stream = tailRecM go (Tuple initial stream)
962+ where
963+ go (Tuple acc s) = do
964+ step <- runStrom s
965+ case step of
966+ Done Nothing -> pure $ Done acc
967+ Done (Just chunk) -> pure $ Done $ foldl f acc chunk
968+ Loop (Tuple maybeChunk next) ->
969+ -- Process chunk if present and continue
970+ case maybeChunk of
971+ Nothing -> pure $ Loop $ Tuple acc next
972+ Just chunk -> pure $ Loop $ Tuple (foldl f acc chunk) next
973+
974+-- | Collect all elements into an array (O(n) using chunk accumulation)
975+runCollect :: forall ctx err a. Strom ctx err a -> Om ctx err (Array a)
976+runCollect stream = tailRecM go (Tuple List.Nil stream)
977+ where
978+ go (Tuple chunkList s) = do
979+ step <- runStrom s
980+ case step of
981+ Done Nothing -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse chunkList
982+ Done (Just chunk) -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse $ List.Cons chunk chunkList
983+ Loop (Tuple maybeChunk next) ->
984+ -- Emit the chunk (if present) and continue
985+ case maybeChunk of
986+ Nothing -> pure $ Loop $ Tuple chunkList next
987+ Just chunk -> pure $ Loop $ Tuple (List.Cons chunk chunkList) next
988+
989+-- | Run the stream and discard all elements
990+runDrain :: forall ctx err a. Strom ctx err a -> Om ctx err Unit
991+runDrain = runFold unit (\_ _ -> unit)
992+
993+-- | Traverse the stream with effects, discarding results
994+traverseStrom_ :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
995+traverseStrom_ f stream = tailRecM go stream
996+ where
997+ go s = do
998+ step <- runStrom s
999+ case step of
1000+ Done Nothing -> pure $ Done unit
1001+ Done (Just chunk) -> do
1002+ _ <- traverse f chunk
1003+ pure $ Done unit
1004+ Loop (Tuple maybeChunk next) -> do
1005+ -- Process chunk if present, then continue
1006+ case maybeChunk of
1007+ Nothing -> pure unit
1008+ Just chunk -> void $ traverse f chunk
1009+ pure $ Loop next
1010+
1011+-- | Alias for traverseStrom_ with arguments flipped
1012+forStrom_ :: forall ctx err a. Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit
1013+forStrom_ = flip traverseStrom_
1014+
1015+-- | Alias for traverseStrom_ (FP convention)
1016+forMStrom_ :: forall ctx err a. Strom ctx err a -> (a -> Om ctx err Unit) -> Om ctx err Unit
1017+forMStrom_ = forStrom_
1018+
1019+-- | Alias for traverseStrom_ (explicit monadic naming)
1020+traverseMStrom_ :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1021+traverseMStrom_ = traverseStrom_
1022+
1023+--------------------------------------------------------------------------------
1024+-- Cancellable Execution
1025+--------------------------------------------------------------------------------
1026+
1027+-- | Run a stream in the background, returning a Fiber that can be cancelled
1028+-- |
1029+-- | To use cancellable streams, convert your stream to Om and use Om.launchOm:
1030+-- |
1031+-- | Example:
1032+-- | ```purescript
1033+-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)
1034+-- | -- ... later ...
1035+-- | killFiber (error "Cancelled") fiber
1036+-- | ```
1037+-- |
1038+-- | Or for subscription-style:
1039+-- | ```purescript
1040+-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)
1041+-- | -- ... later ...
1042+-- | killFiber (error "Unsubscribed") fiber
1043+-- | ```
1044+-- |
1045+-- | Note: You can use Om.launchOm directly with any Om-returning function like:
1046+-- | - `runCollect` - collect results in background
1047+-- | - `runDrain` - run for side effects
1048+-- | - `traverseStrom_` - process each element with callback
1049+-- | - `runFold` - fold with accumulator
1050+
1051+--------------------------------------------------------------------------------
1052+-- Combining
1053+--------------------------------------------------------------------------------
1054+
1055+-- | Append two streams
1056+-- | Now properly handles multi-chunk streams with new Step protocol
1057+-- | Note: s2 is appended TO s1 (s1 comes first, then s2)
1058+-- | This matches pipeline semantics: stream # appendStrom other means stream followed by other
1059+appendStrom :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1060+appendStrom toAppend baseStream = mkStrom do
1061+ step <- runStrom baseStream
1062+ case step of
1063+ Done Nothing -> runStrom toAppend
1064+ Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) toAppend -- Emit chunk and continue with toAppend
1065+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (appendStrom toAppend next)
1066+
1067+-- | Concatenate an array of streams
1068+-- | Uses foldr with flip to match appendStrom's pipeline-friendly parameter order
1069+concatStrom :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1070+concatStrom streams = Array.foldr (flip appendStrom) empty streams
1071+
1072+-- | Zip two streams together
1073+zipStrom :: forall ctx err a b. Strom ctx err a -> Strom ctx err b -> Strom ctx err (Tuple a b)
1074+zipStrom = zipWithStrom Tuple
1075+
1076+-- | Zip two streams with a function
1077+zipWithStrom :: forall ctx err a b c. (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c
1078+zipWithStrom f s1 s2 = mkStrom do
1079+ step1 <- runStrom s1
1080+ step2 <- runStrom s2
1081+ case step1, step2 of
1082+ Done (Just arr1), Done (Just arr2) -> do
1083+ let zipped = Array.zipWith f arr1 arr2
1084+ pure $ Done $ Just zipped
1085+ Done Nothing, _ -> pure $ Done Nothing
1086+ _, Done Nothing -> pure $ Done Nothing
1087+ Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1088+ -- Both looping: zip chunks if both present, continue with both nexts
1089+ case maybeChunk1, maybeChunk2 of
1090+ Just chunk1, Just chunk2 -> pure $ Loop $ Tuple (Just $ Array.zipWith f chunk1 chunk2) (zipWithStrom f next1 next2)
1091+ _, _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 next2)
1092+ Loop (Tuple _ next1), _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 s2)
1093+ _, Loop (Tuple _ next2) -> pure $ Loop $ Tuple Nothing (zipWithStrom f s1 next2)
1094+
1095+-- | Interleave two streams - alternates between elements from each stream
1096+-- | Similar to merge but deterministic - always alternates
1097+interleave :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1098+interleave s1 s2 = interleaveHelper true s1 s2
1099+ where
1100+ interleaveHelper preferFirst first second = mkStrom do
1101+ if preferFirst then do
1102+ step1 <- runStrom first
1103+ case step1 of
1104+ Done Nothing -> runStrom second
1105+ Done (Just chunk1) -> pure $ Loop $ Tuple (Just chunk1) (interleaveHelper false second first)
1106+ Loop (Tuple maybeChunk1 next1) ->
1107+ pure $ Loop $ Tuple maybeChunk1 (interleaveHelper false second next1)
1108+ else do
1109+ step2 <- runStrom second
1110+ case step2 of
1111+ Done Nothing -> runStrom first
1112+ Done (Just chunk2) -> pure $ Loop $ Tuple (Just chunk2) (interleaveHelper true first second)
1113+ Loop (Tuple maybeChunk2 next2) ->
1114+ pure $ Loop $ Tuple maybeChunk2 (interleaveHelper true first next2)
1115+
1116+-- | Intersperse a separator element between each element of the stream
1117+intersperse :: forall ctx err a. a -> Strom ctx err a -> Strom ctx err a
1118+intersperse separator stream = intersperseHelper true stream
1119+ where
1120+ intersperseHelper isFirst s = mkStrom do
1121+ step <- runStrom s
1122+ case step of
1123+ Done Nothing -> pure $ Done Nothing
1124+ Done (Just chunk) -> do
1125+ let
1126+ withSeparators =
1127+ if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1128+ else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1129+ pure $ Done $ Just withSeparators
1130+ Loop (Tuple maybeChunk next) ->
1131+ case maybeChunk of
1132+ Nothing -> pure $ Loop $ Tuple Nothing (intersperseHelper isFirst next)
1133+ Just chunk -> do
1134+ let
1135+ withSeparators =
1136+ if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1137+ else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1138+ pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
1139+
1140+--------------------------------------------------------------------------------
1141+-- Buffering
1142+--------------------------------------------------------------------------------
1143+
1144+-- | Buffer stream chunks with backpressure (preserves chunking)
1145+bufferChunks :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1146+bufferChunks capacity stream
1147+ | capacity <= 0 = stream
1148+ | otherwise = mkStrom do
1149+ ctx <- Om.ask
1150+ liftAff do
1151+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (BufferItem a))
1152+ sizeRef <- liftEffect $ Ref.new 0
1153+ mutex <- AVar.new unit
1154+ notEmpty <- AVar.empty
1155+ notFull <- AVar.new unit
1156+
1157+ let
1158+ enqueue item = do
1159+ AVar.take notFull
1160+ AVar.take mutex
1161+ size <- liftEffect $ Ref.read sizeRef
1162+ let wasEmpty = size == 0
1163+ buf <- liftEffect $ Ref.read bufferRef
1164+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1165+ liftEffect $ Ref.write (size + 1) sizeRef
1166+ AVar.put unit mutex
1167+ when wasEmpty $ AVar.put unit notEmpty
1168+ when (size + 1 < capacity) $ AVar.put unit notFull
1169+
1170+ dequeue = do
1171+ AVar.take notEmpty
1172+ AVar.take mutex
1173+ buf <- liftEffect $ Ref.read bufferRef
1174+ size <- liftEffect $ Ref.read sizeRef
1175+ case CatQueue.uncons buf of
1176+ Nothing -> do
1177+ AVar.put unit mutex
1178+ dequeue
1179+ Just (Tuple head tail) -> do
1180+ liftEffect $ Ref.write tail bufferRef
1181+ liftEffect $ Ref.write (size - 1) sizeRef
1182+ let hasMore = (size - 1) > 0
1183+ let wasFull = size == capacity
1184+ AVar.put unit mutex
1185+ when hasMore $ AVar.put unit notEmpty
1186+ when wasFull $ AVar.put unit notFull
1187+ pure head
1188+
1189+ producer s = do
1190+ stepResult <- Om.runReader ctx (runStrom s)
1191+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1192+ case step of
1193+ Done Nothing -> enqueue BufferDone
1194+ Done (Just chunk) -> do
1195+ enqueue (BufferChunk chunk)
1196+ enqueue BufferDone
1197+ Loop (Tuple maybeChunk next) -> do
1198+ case maybeChunk of
1199+ Just chunk -> enqueue (BufferChunk chunk)
1200+ Nothing -> pure unit
1201+ producer next
1202+
1203+ fiber <- forkAff (producer stream)
1204+
1205+ let
1206+ consumer = mkStrom $ liftAff do
1207+ msg <- dequeue
1208+ case msg of
1209+ BufferDone -> do
1210+ killFiber (Exception.error "done") fiber
1211+ pure $ Done Nothing
1212+ BufferChunk chunk ->
1213+ pure $ Loop $ Tuple (Just chunk) consumer
1214+
1215+ stepResult <- Om.runReader ctx (runStrom consumer)
1216+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1217+
1218+-- | Buffer stream elements with backpressure (destroys chunking)
1219+buffer :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1220+buffer capacity stream
1221+ | capacity <= 0 = stream
1222+ | otherwise = mkStrom do
1223+ ctx <- Om.ask
1224+ liftAff do
1225+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (BufferItem a))
1226+ sizeRef <- liftEffect $ Ref.new 0
1227+ mutex <- AVar.new unit
1228+ notEmpty <- AVar.empty
1229+ notFull <- AVar.new unit
1230+
1231+ let
1232+ enqueue item = do
1233+ AVar.take notFull
1234+ AVar.take mutex
1235+ size <- liftEffect $ Ref.read sizeRef
1236+ let wasEmpty = size == 0
1237+ buf <- liftEffect $ Ref.read bufferRef
1238+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1239+ liftEffect $ Ref.write (size + 1) sizeRef
1240+ AVar.put unit mutex
1241+ when wasEmpty $ AVar.put unit notEmpty
1242+ when (size + 1 < capacity) $ AVar.put unit notFull
1243+
1244+ dequeue = do
1245+ AVar.take notEmpty
1246+ AVar.take mutex
1247+ buf <- liftEffect $ Ref.read bufferRef
1248+ size <- liftEffect $ Ref.read sizeRef
1249+ case CatQueue.uncons buf of
1250+ Nothing -> do
1251+ AVar.put unit mutex
1252+ dequeue
1253+ Just (Tuple head tail) -> do
1254+ liftEffect $ Ref.write tail bufferRef
1255+ liftEffect $ Ref.write (size - 1) sizeRef
1256+ let hasMore = (size - 1) > 0
1257+ let wasFull = size == capacity
1258+ AVar.put unit mutex
1259+ when hasMore $ AVar.put unit notEmpty
1260+ when wasFull $ AVar.put unit notFull
1261+ pure head
1262+
1263+ producer s = do
1264+ stepResult <- Om.runReader ctx (runStrom s)
1265+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1266+ case step of
1267+ Done Nothing -> enqueue BufferDone
1268+ Done (Just chunk) -> do
1269+ traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1270+ enqueue BufferDone
1271+ Loop (Tuple maybeChunk next) -> do
1272+ case maybeChunk of
1273+ Just chunk -> traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1274+ Nothing -> pure unit
1275+ producer next
1276+
1277+ fiber <- forkAff (producer stream)
1278+
1279+ let
1280+ consumer = mkStrom $ liftAff do
1281+ msg <- dequeue
1282+ case msg of
1283+ BufferDone -> do
1284+ killFiber (Exception.error "done") fiber
1285+ pure $ Done Nothing
1286+ BufferChunk chunk ->
1287+ pure $ Loop $ Tuple (Just chunk) consumer
1288+
1289+ stepResult <- Om.runReader ctx (runStrom consumer)
1290+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1291+
1292+--------------------------------------------------------------------------------
1293+-- Concurrency
1294+--------------------------------------------------------------------------------
1295+
1296+-- | Merge two streams concurrently (nondeterministic like ZStream merge)
1297+-- | Elements are emitted as soon as they're available from either stream
1298+merge :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1299+merge = mergeNDWith 1024
1300+
1301+-- | Merge with halt strategy and mapping functions (ZStream-like)
1302+mergeWith
1303+ :: forall ctx err a b c
1304+ . HaltStrategy
1305+ -> (a -> c)
1306+ -> (b -> c)
1307+ -> Strom ctx err a
1308+ -> Strom ctx err b
1309+ -> Strom ctx err c
1310+mergeWith strategy f g s1 s2 =
1311+ mergeNDWithStrategy 1024 strategy (mapStrom f s1) (mapStrom g s2)
1312+
1313+mergeHaltEither :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1314+mergeHaltEither = mergeNDWithStrategy 1024 HaltEither
1315+
1316+mergeHaltLeft :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1317+mergeHaltLeft = mergeNDWithStrategy 1024 HaltLeft
1318+
1319+mergeHaltRight :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1320+mergeHaltRight = mergeNDWithStrategy 1024 HaltRight
1321+
1322+-- | Merge multiple streams concurrently
1323+mergeAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1324+mergeAll streams = Array.foldl merge empty streams
1325+
1326+-- | Race two streams - returns elements from whichever produces first
1327+-- | Uses Om.race for true concurrent racing
1328+race :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1329+race = mergeHaltEither
1330+
1331+-- | Race multiple streams - returns elements from whichever produces first
1332+raceAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1333+raceAll streams =
1334+ case Array.uncons streams of
1335+ Nothing -> empty
1336+ Just { head, tail } -> Array.foldl mergeHaltEither head tail
1337+
1338+-- | Non-deterministic merge: Merge two streams concurrently, emitting elements as soon as they're available
1339+-- | Unlike `merge`, this runs both streams in parallel and elements may arrive in any order
1340+-- | The stream that produces results faster will have its elements appear first
1341+-- | Optimized: Spawns producer fibers once and uses a queue for communication
1342+mergeND :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1343+mergeND = mergeNDWith 1024
1344+
1345+-- | Non-deterministic merge with configurable buffer capacity (chunk queue)
1346+mergeNDWith :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1347+mergeNDWith bufferCapacity = mergeNDWithStrategy bufferCapacity HaltBoth
1348+
1349+mergeNDWithStrategy
1350+ :: forall ctx err a
1351+ . Int
1352+ -> HaltStrategy
1353+ -> Strom ctx err a
1354+ -> Strom ctx err a
1355+ -> Strom ctx err a
1356+mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
1357+ -- Capture context
1358+ ctx <- Om.ask
1359+
1360+ liftAff do
1361+ -- Create buffered queue with backpressure
1362+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (QueueItem a))
1363+ sizeRef <- liftEffect $ Ref.new 0
1364+ mutex <- AVar.new unit
1365+ notEmpty <- AVar.empty
1366+ notFull <- AVar.new unit
1367+
1368+ let
1369+ shouldStop doneLeft doneRight =
1370+ case strategy of
1371+ HaltBoth -> doneLeft && doneRight
1372+ HaltEither -> doneLeft || doneRight
1373+ HaltLeft -> doneLeft
1374+ HaltRight -> doneRight
1375+ enqueue item = do
1376+ AVar.take notFull
1377+ AVar.take mutex
1378+ size <- liftEffect $ Ref.read sizeRef
1379+ let wasEmpty = size == 0
1380+ buf <- liftEffect $ Ref.read bufferRef
1381+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1382+ liftEffect $ Ref.write (size + 1) sizeRef
1383+ AVar.put unit mutex
1384+ when wasEmpty $ AVar.put unit notEmpty
1385+ when (size + 1 < bufferCapacity) $ AVar.put unit notFull
1386+
1387+ dequeue = do
1388+ AVar.take notEmpty
1389+ AVar.take mutex
1390+ buf <- liftEffect $ Ref.read bufferRef
1391+ size <- liftEffect $ Ref.read sizeRef
1392+ case CatQueue.uncons buf of
1393+ Nothing -> do
1394+ AVar.put unit mutex
1395+ dequeue
1396+ Just (Tuple head tail) -> do
1397+ liftEffect $ Ref.write tail bufferRef
1398+ liftEffect $ Ref.write (size - 1) sizeRef
1399+ let hasMore = (size - 1) > 0
1400+ let wasFull = size == bufferCapacity
1401+ AVar.put unit mutex
1402+ when hasMore $ AVar.put unit notEmpty
1403+ when wasFull $ AVar.put unit notFull
1404+ pure head
1405+
1406+ -- Spawn producer fiber for stream1
1407+ fiber1 <- forkAff do
1408+ let
1409+ producer s = do
1410+ stepResult <- Om.runReader ctx (runStrom s)
1411+ step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1412+ case step of
1413+ Done Nothing ->
1414+ enqueue (StreamDone StreamId1)
1415+ Done (Just chunk) -> do
1416+ enqueue (DataChunk chunk StreamId1)
1417+ enqueue (StreamDone StreamId1)
1418+ Loop (Tuple maybeChunk next) -> do
1419+ case maybeChunk of
1420+ Just chunk -> enqueue (DataChunk chunk StreamId1)
1421+ Nothing -> pure unit
1422+ producer next
1423+ producer stream1
1424+
1425+ -- Spawn producer fiber for stream2
1426+ fiber2 <- forkAff do
1427+ let
1428+ producer s = do
1429+ stepResult <- Om.runReader ctx (runStrom s)
1430+ step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1431+ case step of
1432+ Done Nothing ->
1433+ enqueue (StreamDone StreamId2)
1434+ Done (Just chunk) -> do
1435+ enqueue (DataChunk chunk StreamId2)
1436+ enqueue (StreamDone StreamId2)
1437+ Loop (Tuple maybeChunk next) -> do
1438+ case maybeChunk of
1439+ Just chunk -> enqueue (DataChunk chunk StreamId2)
1440+ Nothing -> pure unit
1441+ producer next
1442+ producer stream2
1443+
1444+ -- Consumer: pull from queue
1445+ let
1446+ consumer doneLeft doneRight = mkStrom $ liftAff do
1447+ if shouldStop doneLeft doneRight then do
1448+ -- Both streams done, cleanup
1449+ killFiber (Exception.error "done") fiber1
1450+ killFiber (Exception.error "done") fiber2
1451+ pure $ Done Nothing
1452+ else do
1453+ msg <- dequeue
1454+ case msg of
1455+ StreamDone streamId -> do
1456+ -- One stream finished, recurse
1457+ let
1458+ doneLeft' = doneLeft || streamId == StreamId1
1459+ doneRight' = doneRight || streamId == StreamId2
1460+ stepResult <- Om.runReader ctx (runStrom $ consumer doneLeft' doneRight')
1461+ either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1462+ DataChunk chunk _streamId ->
1463+ -- Got a chunk
1464+ pure $ Loop $ Tuple (Just chunk) (consumer doneLeft doneRight)
1465+
1466+ stepResult <- Om.runReader ctx (runStrom $ consumer false false)
1467+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1468+
1469+-- | Non-deterministic merge of multiple streams
1470+-- | All streams run concurrently and emit elements as soon as they're available
1471+mergeAllND :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1472+mergeAllND = mergeAllNDWith 1024
1473+
1474+-- | Non-deterministic merge of multiple streams with configurable buffer capacity
1475+mergeAllNDWith :: forall ctx err a. Int -> Array (Strom ctx err a) -> Strom ctx err a
1476+mergeAllNDWith bufferCapacity streams
1477+ | Array.null streams = empty
1478+ | otherwise = mkStrom do
1479+ ctx <- Om.ask
1480+ liftAff do
1481+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (MergeAllItem a))
1482+ sizeRef <- liftEffect $ Ref.new 0
1483+ mutex <- AVar.new unit
1484+ notEmpty <- AVar.empty
1485+ notFull <- AVar.new unit
1486+
1487+ let
1488+ enqueue item = do
1489+ AVar.take notFull
1490+ AVar.take mutex
1491+ size <- liftEffect $ Ref.read sizeRef
1492+ let wasEmpty = size == 0
1493+ buf <- liftEffect $ Ref.read bufferRef
1494+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1495+ liftEffect $ Ref.write (size + 1) sizeRef
1496+ AVar.put unit mutex
1497+ when wasEmpty $ AVar.put unit notEmpty
1498+ when (size + 1 < bufferCapacity) $ AVar.put unit notFull
1499+
1500+ dequeue = do
1501+ AVar.take notEmpty
1502+ AVar.take mutex
1503+ buf <- liftEffect $ Ref.read bufferRef
1504+ size <- liftEffect $ Ref.read sizeRef
1505+ case CatQueue.uncons buf of
1506+ Nothing -> do
1507+ AVar.put unit mutex
1508+ dequeue
1509+ Just (Tuple head tail) -> do
1510+ liftEffect $ Ref.write tail bufferRef
1511+ liftEffect $ Ref.write (size - 1) sizeRef
1512+ let hasMore = (size - 1) > 0
1513+ let wasFull = size == bufferCapacity
1514+ AVar.put unit mutex
1515+ when hasMore $ AVar.put unit notEmpty
1516+ when wasFull $ AVar.put unit notFull
1517+ pure head
1518+
1519+ producer s = do
1520+ stepResult <- Om.runReader ctx (runStrom s)
1521+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1522+ case step of
1523+ Done Nothing -> enqueue MergeDone
1524+ Done (Just chunk) -> do
1525+ enqueue (MergeChunk chunk)
1526+ enqueue MergeDone
1527+ Loop (Tuple maybeChunk next) -> do
1528+ case maybeChunk of
1529+ Just chunk -> enqueue (MergeChunk chunk)
1530+ Nothing -> pure unit
1531+ producer next
1532+
1533+ fibers <- traverse (\s -> forkAff (producer s)) streams
1534+
1535+ let
1536+ totalStreams = Array.length streams
1537+ consumer doneCount = mkStrom $ liftAff do
1538+ if doneCount >= totalStreams then do
1539+ traverse_ (\fiber -> killFiber (Exception.error "done") fiber) fibers
1540+ pure $ Done Nothing
1541+ else do
1542+ msg <- dequeue
1543+ case msg of
1544+ MergeDone -> do
1545+ stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1546+ either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1547+ MergeChunk chunk ->
1548+ pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1549+
1550+ stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1551+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1552+
1553+--------------------------------------------------------------------------------
1554+-- Parallel Processing
1555+--------------------------------------------------------------------------------
1556+
1557+-- | Map over elements in parallel with bounded concurrency
1558+-- | Processes chunks in parallel groups up to the concurrency limit
1559+mapPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1560+mapPar concurrency f stream
1561+ | concurrency <= 1 = mapMStrom f stream -- Sequential fallback
1562+ | otherwise =
1563+ mkStrom do
1564+ step <- runStrom stream
1565+ case step of
1566+ Done Nothing -> pure $ Done Nothing
1567+ Done (Just chunk) -> do
1568+ -- Split chunk into groups of `concurrency` size and process each group in parallel
1569+ let groups = chunksOf concurrency chunk
1570+ results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1571+ pure $ Done $ Just $ Array.concat results
1572+ Loop (Tuple maybeChunk next) -> do
1573+ mappedChunk <- case maybeChunk of
1574+ Nothing -> pure Nothing
1575+ Just chunk -> do
1576+ let groups = chunksOf concurrency chunk
1577+ results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1578+ pure $ Just $ Array.concat results
1579+ pure $ Loop $ Tuple mappedChunk (mapPar concurrency f next)
1580+ where
1581+ chunksOf :: forall x. Int -> Array x -> Array (Array x)
1582+ chunksOf n arr
1583+ | n <= 0 || Array.null arr = []
1584+ | otherwise =
1585+ let
1586+ group = Array.take n arr
1587+ rest = Array.drop n arr
1588+ in
1589+ if Array.null group then [] else Array.cons group (chunksOf n rest)
1590+
1591+-- | Unordered parallel map (currently same as mapPar)
1592+mapParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1593+mapParUnordered = mapPar
1594+
1595+-- | Alias for mapPar with explicit "M" naming
1596+mapMPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1597+mapMPar = mapPar
1598+
1599+-- | Alias for unordered mapPar with explicit "M" naming
1600+mapMParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1601+mapMParUnordered = mapParUnordered
1602+
1603+-- | For each element, execute an effect in parallel with bounded concurrency
1604+foreachPar :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1605+foreachPar concurrency f stream = runDrain (mapPar concurrency f stream)
1606+
1607+-- | Unordered parallel foreach
1608+foreachParUnordered :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1609+foreachParUnordered concurrency f stream = runDrain (mapParUnordered concurrency f stream)
1610+
1611+--------------------------------------------------------------------------------
1612+-- Error Handling
1613+--------------------------------------------------------------------------------
1614+
1615+-- | Catch all errors and recover with a handler function
1616+-- | Note: Uses handleErrors' for full error handling
1617+catchAll :: forall ctx err a. (Variant (exception :: Error | err) -> Strom ctx err a) -> Strom ctx err a -> Strom ctx err a
1618+catchAll handler stream = mkStrom do
1619+ Om.handleErrors'
1620+ (\err -> runStrom (handler err))
1621+ (runStrom stream)
1622+
1623+-- TODO: Fix catchSome type constraints
1624+-- catchSome :: forall ctx err handlers a. (Record handlers) -> Strom ctx err a -> Strom ctx err a
1625+-- catchSome handlers stream = mkStrom do
1626+-- Om.handleErrors handlers (runStrom stream)
1627+
1628+-- | Retry a stream on failure up to 3 times
1629+retry :: forall ctx err a. Strom ctx err a -> Strom ctx err a
1630+retry = retryN 3
1631+
1632+-- | Retry a stream N times on failure
1633+retryN :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1634+retryN n stream
1635+ | n <= 0 = stream
1636+ | otherwise = catchAll (\_ -> retryN (n - 1) stream) stream
1637+
1638+-- | Run a finaliser after the stream completes (success or failure)
1639+ensuring :: forall ctx err a. Om ctx err Unit -> Strom ctx err a -> Strom ctx err a
1640+ensuring finaliser stream = mkStrom do
1641+ result <- Om.handleErrors'
1642+ ( \err -> do
1643+ finaliser
1644+ throwError err
1645+ )
1646+ ( runStrom stream >>= \step -> do
1647+ finaliser
1648+ pure step
1649+ )
1650+ pure result
1651+
1652+-- | Add a timeout to stream operations
1653+-- | If the stream doesn't complete within the duration, it terminates as empty
1654+timeout :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
1655+timeout duration stream = mkStrom do
1656+ Om.race [ runStrom stream, Om.delay duration *> runStrom empty ]
1657+
1658+--------------------------------------------------------------------------------
1659+-- Resource Safety
1660+--------------------------------------------------------------------------------
1661+
1662+-- | Bracket pattern: acquire, use, release
1663+-- | Guarantees release runs even if stream fails
1664+bracket
1665+ :: forall ctx err a b
1666+ . Om ctx err a -- acquire
1667+ -> (a -> Om ctx err Unit) -- release
1668+ -> (a -> Strom ctx err b) -- use
1669+ -> Strom ctx err b
1670+bracket acquire release use = mkStrom do
1671+ resource <- acquire
1672+ Om.handleErrors'
1673+ ( \err -> do
1674+ release resource
1675+ throwError err
1676+ )
1677+ (runStrom (use resource))
1678+ >>= \step -> do
1679+ case step of
1680+ Done _ -> release resource
1681+ Loop _ -> pure unit
1682+ pure step
1683+
1684+-- | Bracket with exit information (whether stream succeeded or failed)
1685+bracketExit
1686+ :: forall ctx err a b
1687+ . Om ctx err a -- acquire
1688+ -> (a -> Maybe (Variant (exception :: Error | err)) -> Om ctx err Unit) -- release with exit info
1689+ -> (a -> Strom ctx err b) -- use
1690+ -> Strom ctx err b
1691+bracketExit acquire release use = mkStrom do
1692+ resource <- acquire
1693+ Om.handleErrors'
1694+ ( \err -> do
1695+ release resource (Just err)
1696+ throwError err
1697+ )
1698+ (runStrom (use resource))
1699+ >>= \step -> do
1700+ case step of
1701+ Done _ -> release resource Nothing
1702+ Loop _ -> pure unit
1703+ pure step
1704+
1705+-- | Simplified acquire-release pattern
1706+acquireRelease
1707+ :: forall ctx err a b
1708+ . Om ctx err a -- acquire
1709+ -> (a -> Om ctx err Unit) -- release
1710+ -> (a -> Strom ctx err b) -- use
1711+ -> Strom ctx err b
1712+acquireRelease = bracket
1713+
1714+--------------------------------------------------------------------------------
1715+-- Instances
1716+--------------------------------------------------------------------------------
1717+
1718+instance Functor (Strom ctx err) where
1719+ map f stream = mkStrom
1720+ ( runStrom stream >>= \step ->
1721+ case step of
1722+ Done Nothing -> pure $ Done Nothing
1723+ Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
1724+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple (Functor.map (Functor.map f) maybeChunk) (Functor.map f next)
1725+ )
1726+
1727+instance Apply (Strom ctx err) where
1728+ apply fs as = mkStrom
1729+ ( runStrom fs >>= \stepF ->
1730+ runStrom as >>= \stepA ->
1731+ case stepF, stepA of
1732+ Done (Just funcs), Done (Just vals) ->
1733+ let
1734+ results = do
1735+ f <- funcs
1736+ a <- vals
1737+ pure (f a)
1738+ in
1739+ pure $ Done $ Just results
1740+ Done Nothing, _ -> pure $ Done Nothing
1741+ _, Done Nothing -> pure $ Done Nothing
1742+ Loop (Tuple maybeChunkF nextF), Loop (Tuple maybeChunkA nextA) ->
1743+ -- Apply chunks if both present
1744+ case maybeChunkF, maybeChunkA of
1745+ Just chunkF, Just chunkA ->
1746+ let
1747+ applied = do
1748+ f <- chunkF
1749+ a <- chunkA
1750+ pure (f a)
1751+ in
1752+ pure $ Loop $ Tuple (Just applied) (nextF <*> nextA)
1753+ _, _ -> pure $ Loop $ Tuple Nothing (nextF <*> nextA)
1754+ Loop (Tuple _ nextF), _ -> pure $ Loop $ Tuple Nothing (nextF <*> as)
1755+ _, Loop (Tuple _ nextA) -> pure $ Loop $ Tuple Nothing (fs <*> nextA)
1756+ )
1757+
1758+instance Applicative (Strom ctx err) where
1759+ pure = succeed
1760+
1761+instance Bind (Strom ctx err) where
1762+ bind stream f = mkStrom
1763+ ( runStrom stream >>= \step ->
1764+ case step of
1765+ Done Nothing -> pure $ Done Nothing
1766+ Done (Just chunk) ->
1767+ let
1768+ streams = Functor.map f chunk
1769+ in
1770+ runStrom $ concatStrom streams
1771+ Loop (Tuple maybeChunk next) ->
1772+ -- When looping, if we have a chunk, we need to process it through f and concat
1773+ case maybeChunk of
1774+ Nothing -> pure $ Loop $ Tuple Nothing (next >>= f)
1775+ Just chunk ->
1776+ let
1777+ streams = Functor.map f chunk
1778+ prefixStream = concatStrom streams
1779+ in
1780+ runStrom $ appendStrom prefixStream (next >>= f)
1781+ )
1782+
1783+instance Monad (Strom ctx err)
1784+
1785+instance Semigroup (Strom ctx err a) where
1786+ append s1 s2 = mkStrom
1787+ ( runStrom s1 >>= \step ->
1788+ case step of
1789+ Done Nothing -> runStrom s2
1790+ Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1791+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <> s2)
1792+ )
1793+
1794+instance Monoid (Strom ctx err a) where
1795+ mempty = empty
1796+
1797+instance Alt (Strom ctx err) where
1798+ alt s1 s2 = mkStrom
1799+ ( runStrom s1 >>= \step ->
1800+ case step of
1801+ Done Nothing -> runStrom s2
1802+ Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1803+ Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <|> s2)
1804+ )
1805+
1806+instance Plus (Strom ctx err) where
1807+ empty = mkStrom $ pure $ Done Nothing
1808+
1809+instance Alternative (Strom ctx err)