@@ -1,1809 +0,0 @@
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)