@@ -65,6 +65,7 @@ module Yoga.Om.Strom
65 , partitionMap
65 , partitionMap
67 , debounce
67 , debounce
68 , throttle
69 , throttle
69 , delayStrom
70 , delayStrom
70 -- Combining
71 -- Combining
@@ -108,6 +109,9 @@ module Yoga.Om.Strom
108 , bracket
109 , bracket
109 , bracketExit
110 , bracketExit
110 , acquireRelease
111 , acquireRelease
112+ -- Internal (needed by WebStream)
113import Prelude
117import Prelude
@@ -129,18 +133,16 @@ import Data.List as List
129import Data.Maybe (Maybe(..))
133import Data.Maybe (Maybe(..))
130import Data.Time.Duration (Milliseconds(..))
134import Data.Time.Duration (Milliseconds(..))
131import Data.Traversable (traverse, traverse_)
135import Data.Traversable (traverse, traverse_)
132−import Data.Tuple (Tuple(..))
136+import Data.Tuple.Nested (type (/\), (/\))
133import Data.Variant (Variant)
137import Data.Variant (Variant)
134−import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
138+import Effect.Aff (Aff, Error, delay, forkAff, killFiber)
135import Effect.Aff.AVar as AVar
139import Effect.Aff.AVar as AVar
136import Effect.Aff.Class (liftAff)
140import Effect.Aff.Class (liftAff)
137import Effect.Exception as Exception
141import Effect.Exception as Exception
138−import Effect.Now (now)
139import Effect.Ref as Ref
142import Effect.Ref as Ref
140import Effect.Class (liftEffect)
143import Effect.Class (liftEffect)
141import Data.CatQueue as CatQueue
144import Data.CatQueue as CatQueue
142import Data.CatQueue (CatQueue)
145import Data.CatQueue (CatQueue)
143−import Data.DateTime.Instant (unInstant)
144import Yoga.Om (Om)
146import Yoga.Om (Om)
145import Yoga.Om as Om
147import Yoga.Om as Om
@@ -160,18 +162,18 @@ import Yoga.Om as Om
160-- | - Loop (chunk, next) => emit chunk (if present) and continue with next
162-- | - Loop (chunk, next) => emit chunk (if present) and continue with next
161-- | - Done finalChunk => emit final chunk (if present) and stop
163-- | - Done finalChunk => emit final chunk (if present) and stop
162newtype Strom ctx err a = Strom
164newtype Strom ctx err a = Strom
163− { pull :: Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a)))
165+ { pull :: Om ctx err (Step (((Maybe (Chunk a)) /\ (Strom ctx err a))) (Maybe (Chunk a)))
166-- | Chunks for efficient batch processing
168-- | Chunks for efficient batch processing
167type Chunk a = Array a
169type Chunk a = Array a
169-- | Helper to create a Strom from a pull function
171-- | Helper to create a Strom from a pull function
170−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 :: forall ctx err a. Om ctx err (Step (((Maybe (Chunk a)) /\ (Strom ctx err a))) (Maybe (Chunk a))) -> Strom ctx err a
171mkStrom pull = Strom { pull }
173mkStrom pull = Strom { pull }
173-- | Helper to unwrap a Strom
175-- | Helper to unwrap a Strom
174−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 :: forall ctx err a. Strom ctx err a -> Om ctx err (Step (((Maybe (Chunk a)) /\ (Strom ctx err a))) (Maybe (Chunk a)))
175runStrom (Strom s) = s.pull
177runStrom (Strom s) = s.pull
177--------------------------------------------------------------------------------
179--------------------------------------------------------------------------------
@@ -181,22 +183,22 @@ runStrom (Strom s) = s.pull
181-- | Stream identifier for tracking which stream produced data
183-- | Stream identifier for tracking which stream produced data
182data StreamId = StreamId1 | StreamId2
184data StreamId = StreamId1 | StreamId2
184−derive instance eqStreamId :: Eq StreamId
186+derive instance Eq StreamId
185−derive instance ordStreamId :: Ord StreamId
187+derive instance Ord StreamId
187-- | Messages passed through the coordination queue
189-- | Messages passed through the coordination queue
188data QueueItem a
190data QueueItem a
189 = DataChunk (Array a) StreamId
191 = DataChunk (Array a) StreamId
190 | StreamDone StreamId
192 | StreamDone StreamId
192−derive instance functorQueueItem :: Functor QueueItem
194+derive instance Functor QueueItem
194-- | Messages for mergeAllND buffered queue
196-- | Messages for mergeAllND buffered queue
195data MergeAllItem a
197data MergeAllItem a
196 = MergeChunk (Array a)
198 = MergeChunk (Array a)
197 | MergeDone
199 | MergeDone
199−derive instance functorMergeAllItem :: Functor MergeAllItem
201+derive instance Functor MergeAllItem
201-- | Merge termination strategy (like ZStream HaltStrategy)
203-- | Merge termination strategy (like ZStream HaltStrategy)
202data HaltStrategy
204data HaltStrategy
@@ -205,14 +207,14 @@ data HaltStrategy
205 | HaltLeft
207 | HaltLeft
206 | HaltRight
208 | HaltRight
208−derive instance eqHaltStrategy :: Eq HaltStrategy
210+derive instance Eq HaltStrategy
210-- | Messages for buffering queues
212-- | Messages for buffering queues
211data BufferItem a
213data BufferItem a
212 = BufferChunk (Array a)
214 = BufferChunk (Array a)
213 | BufferDone
215 | BufferDone
215−derive instance functorBufferItem :: Functor BufferItem
217+derive instance Functor BufferItem
217--------------------------------------------------------------------------------
219--------------------------------------------------------------------------------
218-- Construction
220-- Construction
@@ -268,7 +270,7 @@ rangeStrom start end = rangeHelper start end
268 chunkEnd = min (current + chunkSize) limit
270 chunkEnd = min (current + chunkSize) limit
269 chunk = Array.range current (chunkEnd - 1)
271 chunk = Array.range current (chunkEnd - 1)
270 if chunkEnd >= limit then pure $ Done $ Just chunk -- Final chunk
272 if chunkEnd >= limit then pure $ Done $ Just chunk -- Final chunk
271− else pure $ Loop $ Tuple (Just chunk) (rangeHelper chunkEnd limit) -- Emit chunk and continue
273+ else pure $ Loop $ ((Just chunk) /\ (rangeHelper chunkEnd limit)) -- Emit chunk and continue
273-- | Iterate producing values - limited to ~10,000 elements for stack safety
275-- | Iterate producing values - limited to ~10,000 elements for stack safety
274-- | For truly infinite streams, use `iterateStromInfinite` (with Aff overhead)
276-- | For truly infinite streams, use `iterateStromInfinite` (with Aff overhead)
@@ -306,7 +308,7 @@ iterateStromInfinite f initial = iterateHelper initial
306 finalVal <- STRef.read valRef
308 finalVal <- STRef.read valRef
307 frozen <- STArray.unsafeFreeze arr
309 frozen <- STArray.unsafeFreeze arr
308 pure { arr: frozen, nextVal: finalVal }
310 pure { arr: frozen, nextVal: finalVal }
309− pure $ Loop $ Tuple (Just result.arr) (iterateHelper result.nextVal)
311+ pure $ Loop $ ((Just result.arr) /\ (iterateHelper result.nextVal))
311-- | Repeat a value - limited to ~10,000 elements for stack safety
313-- | Repeat a value - limited to ~10,000 elements for stack safety
312repeatStrom :: forall ctx err a. a -> Strom ctx err a
314repeatStrom :: forall ctx err a. a -> Strom ctx err a
@@ -321,7 +323,7 @@ repeatStromInfinite a = repeatHelper unit
321 -- Async boundary resets the stack!
323 -- Async boundary resets the stack!
322 liftAff $ delay (Milliseconds 0.0)
324 liftAff $ delay (Milliseconds 0.0)
323 let chunk = Array.replicate 10000 a
325 let chunk = Array.replicate 10000 a
324− pure $ Loop $ Tuple (Just chunk) (repeatHelper unit)
326+ pure $ Loop $ ((Just chunk) /\ (repeatHelper unit))
326-- | Repeat an Om computation - limited to 100 iterations for stack safety
328-- | Repeat an Om computation - limited to 100 iterations for stack safety
327-- | For truly infinite repetition, use `repeatOmStromInfinite` (with Aff overhead)
329-- | For truly infinite repetition, use `repeatOmStromInfinite` (with Aff overhead)
@@ -333,7 +335,7 @@ repeatOmStrom om = repeatOmHelper 0
333 | count >= maxCount = empty
335 | count >= maxCount = empty
334 | otherwise = mkStrom do
336 | otherwise = mkStrom do
335 value <- om
337 value <- om
336− pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper (count + 1))
338+ pure $ Loop $ ((Just [ value ]) /\ (repeatOmHelper (count + 1)))
338-- | Truly infinite Om repetition - stack-safe via Aff async boundaries
340-- | Truly infinite Om repetition - stack-safe via Aff async boundaries
339-- | Adds tiny delay (0ms) at each pull to reset the stack
341-- | Adds tiny delay (0ms) at each pull to reset the stack
@@ -344,11 +346,11 @@ repeatOmStromInfinite om = repeatOmHelper unit
344 -- Async boundary resets the stack!
346 -- Async boundary resets the stack!
345 liftAff $ delay (Milliseconds 0.0)
347 liftAff $ delay (Milliseconds 0.0)
346 value <- om
348 value <- om
347− pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper unit)
349+ pure $ Loop $ ((Just [ value ]) /\ (repeatOmHelper unit))
349-- | Unfold a stream from a seed value - can be infinite if f never returns Nothing
351-- | Unfold a stream from a seed value - can be infinite if f never returns Nothing
350-- | Emits chunks lazily, terminates when f returns Nothing
352-- | Emits chunks lazily, terminates when f returns Nothing
351−unfoldStrom :: forall ctx err a b. (b -> Maybe (Tuple a b)) -> b -> Strom ctx err a
353+unfoldStrom :: forall ctx err a b. (b -> Maybe ((a /\ b))) -> b -> Strom ctx err a
352unfoldStrom f seed = unfoldHelper seed
354unfoldStrom f seed = unfoldHelper seed
354 unfoldHelper currentSeed = mkStrom do
356 unfoldHelper currentSeed = mkStrom do
@@ -357,7 +359,7 @@ unfoldStrom f seed = unfoldHelper seed
357 Nothing -> pure $ Done $ if Array.null chunk.values then Nothing else Just chunk.values
359 Nothing -> pure $ Done $ if Array.null chunk.values then Nothing else Just chunk.values
358 Just nextSeed ->
360 Just nextSeed ->
359 if Array.null chunk.values then runStrom (unfoldHelper nextSeed)
361 if Array.null chunk.values then runStrom (unfoldHelper nextSeed)
360− else pure $ Loop $ Tuple (Just chunk.values) (unfoldHelper nextSeed)
362+ else pure $ Loop $ ((Just chunk.values) /\ (unfoldHelper nextSeed))
362 buildChunk seed' size = ST.run do
364 buildChunk seed' size = ST.run do
363 arr <- STArray.unsafeThaw []
365 arr <- STArray.unsafeThaw []
@@ -370,7 +372,7 @@ unfoldStrom f seed = unfoldHelper seed
370 s <- STRef.read seedRef
372 s <- STRef.read seedRef
371 case f s of
373 case f s of
372 Nothing -> void $ STRef.write true stoppedRef
374 Nothing -> void $ STRef.write true stoppedRef
373− Just (Tuple value nextS) -> do
375+ Just ((value /\ nextS)) -> do
374 void $ STArray.push value arr
376 void $ STArray.push value arr
375 void $ STRef.write nextS seedRef
377 void $ STRef.write nextS seedRef
376 frozen <- STArray.unsafeFreeze arr
378 frozen <- STArray.unsafeFreeze arr
@@ -379,14 +381,14 @@ unfoldStrom f seed = unfoldHelper seed
379 pure { values: frozen, finalSeed: if stopped then Nothing else Just finalSeed }
381 pure { values: frozen, finalSeed: if stopped then Nothing else Just finalSeed }
381-- | Unfold a stream with an effectful function
383-- | Unfold a stream with an effectful function
382−unfoldOmStrom :: forall ctx err a b. (b -> Om ctx err (Maybe (Tuple a b))) -> b -> Strom ctx err a
384+unfoldOmStrom :: forall ctx err a b. (b -> Om ctx err (Maybe ((a /\ b)))) -> b -> Strom ctx err a
383unfoldOmStrom f seed = mkStrom go
385unfoldOmStrom f seed = mkStrom go
386 result <- f seed
388 result <- f seed
387 case result of
389 case result of
388 Nothing -> pure $ Done Nothing
390 Nothing -> pure $ Done Nothing
389− Just (Tuple value nextSeed) -> pure $ Loop $ Tuple (Just [ value ]) (unfoldOmStrom f nextSeed)
391+ Just ((value /\ nextSeed)) -> pure $ Loop $ ((Just [ value ]) /\ (unfoldOmStrom f nextSeed))
391--------------------------------------------------------------------------------
393--------------------------------------------------------------------------------
392-- Transformations
394-- Transformations
@@ -399,7 +401,7 @@ mapStrom f stream = mkStrom do
399 case step of
401 case step of
400 Done Nothing -> pure $ Done Nothing
402 Done Nothing -> pure $ Done Nothing
401 Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
403 Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
402− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple (Functor.map (Functor.map f) maybeChunk) (mapStrom f next)
404+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ ((Functor.map (Functor.map f) maybeChunk) /\ (mapStrom f next))
404-- | Map with a monadic effect
406-- | Map with a monadic effect
405mapMStrom :: forall ctx err a b. (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
407mapMStrom :: forall ctx err a b. (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
@@ -410,11 +412,11 @@ mapMStrom f stream = mkStrom do
410 Done (Just chunk) -> do
412 Done (Just chunk) -> do
411 mapped <- traverse f chunk
413 mapped <- traverse f chunk
412 pure $ Done $ Just mapped
414 pure $ Done $ Just mapped
413− Loop (Tuple maybeChunk next) -> do
415+ Loop ((maybeChunk /\ next)) -> do
414 mappedChunk <- case maybeChunk of
416 mappedChunk <- case maybeChunk of
415 Nothing -> pure Nothing
417 Nothing -> pure Nothing
416 Just chunk -> Just <$> traverse f chunk
418 Just chunk -> Just <$> traverse f chunk
417− pure $ Loop $ Tuple mappedChunk (mapMStrom f next)
419+ pure $ Loop $ (mappedChunk /\ (mapMStrom f next))
419-- | Tap (observe without modifying)
421-- | Tap (observe without modifying)
420tapStrom :: forall ctx err a. (a -> Unit) -> Strom ctx err a -> Strom ctx err a
422tapStrom :: forall ctx err a. (a -> Unit) -> Strom ctx err a -> Strom ctx err a
@@ -425,16 +427,15 @@ tapStrom f stream = mkStrom do
425 Done (Just chunk) -> do
427 Done (Just chunk) -> do
426 let _ = Functor.map f chunk
428 let _ = Functor.map f chunk
427 pure $ Done $ Just chunk
429 pure $ Done $ Just chunk
428− Loop (Tuple maybeChunk next) -> do
430+ Loop ((maybeChunk /\ next)) -> do
430 tappedChunk = case maybeChunk of
432 tappedChunk = case maybeChunk of
431 Nothing -> Nothing
433 Nothing -> Nothing
432 Just chunk ->
434 Just chunk ->
434− _ = Functor.map f chunk
436+ let _ = Functor.map f chunk
436 Just chunk
437 Just chunk
437− pure $ Loop $ Tuple tappedChunk (tapStrom f next)
438+ pure $ Loop $ (tappedChunk /\ (tapStrom f next))
439-- | Tap with monadic effect (observe with effects)
440-- | Tap with monadic effect (observe with effects)
440tapMStrom :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a
441tapMStrom :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a
@@ -445,20 +446,20 @@ tapMStrom f stream = mkStrom do
445 Done (Just chunk) -> do
446 Done (Just chunk) -> do
446 _ <- traverse f chunk
447 _ <- traverse f chunk
447 pure $ Done $ Just chunk
448 pure $ Done $ Just chunk
448− Loop (Tuple maybeChunk next) -> do
449+ Loop ((maybeChunk /\ next)) -> do
449 tappedChunk <- case maybeChunk of
450 tappedChunk <- case maybeChunk of
450 Nothing -> pure Nothing
451 Nothing -> pure Nothing
451 Just chunk -> do
452 Just chunk -> do
452 _ <- traverse f chunk
453 _ <- traverse f chunk
453 pure $ Just chunk
454 pure $ Just chunk
454− pure $ Loop $ Tuple tappedChunk (tapMStrom f next)
455+ pure $ Loop $ (tappedChunk /\ (tapMStrom f next))
456-- | Bind for streams (monadic flatMap/chain)
457-- | Bind for streams (monadic flatMap/chain)
457-- | Use >>= operator for cleaner syntax: `stream >>= f`
458-- | Use >>= operator for cleaner syntax: `stream >>= f`
458bindStrom :: forall ctx err a b. (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b
459bindStrom :: forall ctx err a b. (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b
459bindStrom f stream = stream >>= f
460bindStrom f stream = stream >>= f
461−-- | Scan with accumulator (like foldl but emits intermediate results)
462+-- | Scan with accumulator (like foldl but emits intermediate results)
462scanStrom :: forall ctx err a b. (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b
463scanStrom :: forall ctx err a b. (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b
463scanStrom f initial stream = scanHelper initial stream
464scanStrom f initial stream = scanHelper initial stream
@@ -469,35 +470,33 @@ scanStrom f initial stream = scanHelper initial stream
469 Done (Just chunk) -> do
470 Done (Just chunk) -> do
470 -- Build results by folding and collecting intermediate values
471 -- Build results by folding and collecting intermediate values
472− Tuple _ results = Array.foldl
473+ (_ /\ results) = Array.foldl
473− ( \(Tuple prevAcc accList) item ->
474+ ( \((prevAcc /\ accList)) item ->
475− newAcc = f prevAcc item
476+ let newAcc = f prevAcc item
477− Tuple newAcc (Array.snoc accList newAcc)
477+ (newAcc /\ (Array.snoc accList newAcc))
479− (Tuple acc [])
479+ (acc /\ [])
481 pure $ Done $ if Array.null results then Nothing else Just results
481 pure $ Done $ if Array.null results then Nothing else Just results
482− Loop (Tuple maybeChunk next) ->
482+ Loop ((maybeChunk /\ next)) ->
483 case maybeChunk of
483 case maybeChunk of
484− Nothing -> pure $ Loop $ Tuple Nothing (scanHelper acc next)
484+ Nothing -> pure $ Loop $ (Nothing /\ (scanHelper acc next))
485 Just chunk -> do
485 Just chunk -> do
487− Tuple newAcc results = Array.foldl
487+ (newAcc /\ results) = Array.foldl
488− ( \(Tuple prevAcc accList) item ->
488+ ( \((prevAcc /\ accList)) item ->
490− currAcc = f prevAcc item
490+ let currAcc = f prevAcc item
492− Tuple currAcc (Array.snoc accList currAcc)
491+ (currAcc /\ (Array.snoc accList currAcc))
494− (Tuple acc [])
493+ (acc /\ [])
496− if Array.null results then pure $ Loop $ Tuple Nothing (scanHelper newAcc next)
495+ if Array.null results then pure $ Loop $ (Nothing /\ (scanHelper newAcc next))
497− else pure $ Loop $ Tuple (Just results) (scanHelper newAcc next)
496+ else pure $ Loop $ ((Just results) /\ (scanHelper newAcc next))
499-- | Map with accumulator (stateful map)
498-- | Map with accumulator (stateful map)
500−mapAccumStrom :: forall ctx err a b s. (s -> a -> Tuple s b) -> s -> Strom ctx err a -> Strom ctx err b
499+mapAccumStrom :: forall ctx err a b s. (s -> a -> (s /\ b)) -> s -> Strom ctx err a -> Strom ctx err b
501mapAccumStrom f initial stream = mapAccumHelper initial stream
500mapAccumStrom f initial stream = mapAccumHelper initial stream
503 mapAccumHelper state s = mkStrom do
502 mapAccumHelper state s = mkStrom do
@@ -506,32 +505,30 @@ mapAccumStrom f initial stream = mapAccumHelper initial stream
506 Done Nothing -> pure $ Done Nothing
505 Done Nothing -> pure $ Done Nothing
507 Done (Just chunk) -> do
506 Done (Just chunk) -> do
509− Tuple finalState results = Array.foldl
508+ (_ /\ results) = Array.foldl
510− ( \(Tuple st acc) a ->
509+ ( \((st /\ acc)) a ->
512− Tuple newSt b = f st a
511+ let (newSt /\ b) = f st a
514− Tuple newSt (Array.snoc acc b)
512+ (newSt /\ (Array.snoc acc b))
516− (Tuple state [])
514+ (state /\ [])
518 pure $ Done $ if Array.null results then Nothing else Just results
516 pure $ Done $ if Array.null results then Nothing else Just results
519− Loop (Tuple maybeChunk next) ->
517+ Loop ((maybeChunk /\ next)) ->
520 case maybeChunk of
518 case maybeChunk of
521− Nothing -> pure $ Loop $ Tuple Nothing (mapAccumHelper state next)
519+ Nothing -> pure $ Loop $ (Nothing /\ (mapAccumHelper state next))
522 Just chunk -> do
520 Just chunk -> do
524− Tuple newState results = Array.foldl
522+ (newState /\ results) = Array.foldl
525− ( \(Tuple st acc) a ->
523+ ( \((st /\ acc)) a ->
527− Tuple newSt b = f st a
525+ let (newSt /\ b) = f st a
529− Tuple newSt (Array.snoc acc b)
526+ (newSt /\ (Array.snoc acc b))
531− (Tuple state [])
528+ (state /\ [])
533− if Array.null results then pure $ Loop $ Tuple Nothing (mapAccumHelper newState next)
530+ if Array.null results then pure $ Loop $ (Nothing /\ (mapAccumHelper newState next))
534− else pure $ Loop $ Tuple (Just results) (mapAccumHelper newState next)
531+ else pure $ Loop $ ((Just results) /\ (mapAccumHelper newState next))
536-- | Filter elements
533-- | Filter elements
537filterStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
534filterStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -542,16 +539,15 @@ filterStrom predicate stream = mkStrom do
542 Done (Just chunk) -> do
539 Done (Just chunk) -> do
543 let filtered = Array.filter predicate chunk
540 let filtered = Array.filter predicate chunk
544 pure $ Done $ if Array.null filtered then Nothing else Just filtered
541 pure $ Done $ if Array.null filtered then Nothing else Just filtered
545− Loop (Tuple maybeChunk next) -> do
542+ Loop ((maybeChunk /\ next)) -> do
547 filteredChunk = case maybeChunk of
544 filteredChunk = case maybeChunk of
548 Nothing -> Nothing
545 Nothing -> Nothing
549 Just chunk ->
546 Just chunk ->
551− filtered = Array.filter predicate chunk
548+ let filtered = Array.filter predicate chunk
553 if Array.null filtered then Nothing else Just filtered
549 if Array.null filtered then Nothing else Just filtered
554− pure $ Loop $ Tuple filteredChunk (filterStrom predicate next)
550+ pure $ Loop $ (filteredChunk /\ (filterStrom predicate next))
556--------------------------------------------------------------------------------
552--------------------------------------------------------------------------------
557-- Selection
553-- Selection
@@ -573,18 +569,17 @@ takeStrom n stream = takeHelper n stream
573 numTaken = Array.length taken
569 numTaken = Array.length taken
574 if numTaken == 0 then pure $ Done Nothing
570 if numTaken == 0 then pure $ Done Nothing
575 else pure $ Done $ Just taken
571 else pure $ Done $ Just taken
576− Loop (Tuple maybeChunk next) ->
572+ Loop ((maybeChunk /\ next)) ->
577 case maybeChunk of
573 case maybeChunk of
578− Nothing -> pure $ Loop $ Tuple Nothing (takeHelper remaining next)
574+ Nothing -> pure $ Loop $ (Nothing /\ (takeHelper remaining next))
579 Just chunk -> do
575 Just chunk -> do
581− chunkLen = Array.length chunk
582 taken = Array.take remaining chunk
577 taken = Array.take remaining chunk
583 numTaken = Array.length taken
578 numTaken = Array.length taken
584 newRemaining = remaining - numTaken
579 newRemaining = remaining - numTaken
585 if numTaken == 0 then pure $ Done Nothing
580 if numTaken == 0 then pure $ Done Nothing
586 else if newRemaining <= 0 then pure $ Done $ Just taken
581 else if newRemaining <= 0 then pure $ Done $ Just taken
587− else pure $ Loop $ Tuple (Just taken) (takeHelper newRemaining next)
582+ else pure $ Loop $ ((Just taken) /\ (takeHelper newRemaining next))
589-- | Take while predicate is true
584-- | Take while predicate is true
590takeWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
585takeWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -596,7 +591,7 @@ takeWhileStrom predicate stream = mkStrom do
596 let taken = Array.takeWhile predicate chunk
591 let taken = Array.takeWhile predicate chunk
597 if Array.length taken < Array.length chunk then pure $ Done $ Just taken
592 if Array.length taken < Array.length chunk then pure $ Done $ Just taken
598 else pure $ Done $ Just taken
593 else pure $ Done $ Just taken
599− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (takeWhileStrom predicate next)
594+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (takeWhileStrom predicate next))
601-- | Drop n elements (with proper state tracking across chunks)
596-- | Drop n elements (with proper state tracking across chunks)
602dropStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
597dropStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
@@ -613,19 +608,17 @@ dropStrom n stream
613 Done (Just chunk) -> do
608 Done (Just chunk) -> do
614 let dropped = Array.drop remaining chunk
609 let dropped = Array.drop remaining chunk
615 pure $ Done $ if Array.null dropped then Nothing else Just dropped
610 pure $ Done $ if Array.null dropped then Nothing else Just dropped
616− Loop (Tuple maybeChunk next) ->
611+ Loop ((maybeChunk /\ next)) ->
617 case maybeChunk of
612 case maybeChunk of
618− Nothing -> pure $ Loop $ Tuple Nothing (dropHelper remaining next)
613+ Nothing -> pure $ Loop $ (Nothing /\ (dropHelper remaining next))
619 Just chunk ->
614 Just chunk ->
621− chunkLen = Array.length chunk
616+ let chunkLen = Array.length chunk
623− if remaining >= chunkLen then pure $ Loop $ Tuple Nothing (dropHelper (remaining - chunkLen) next)
617+ if remaining >= chunkLen then pure $ Loop $ (Nothing /\ (dropHelper (remaining - chunkLen) next))
626− dropped = Array.drop remaining chunk
620+ let dropped = Array.drop remaining chunk
628− pure $ Loop $ Tuple (Just dropped) next
621+ pure $ Loop $ ((Just dropped) /\ next)
630-- | Take until predicate is true (includes the matching element)
623-- | Take until predicate is true (includes the matching element)
631takeUntilStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
624takeUntilStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -637,12 +630,12 @@ takeUntilStrom predicate stream = mkStrom do
637 case Array.findIndex predicate chunk of
630 case Array.findIndex predicate chunk of
638 Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Include the matching element
631 Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Include the matching element
639 Nothing -> pure $ Done $ Just chunk
632 Nothing -> pure $ Done $ Just chunk
640− Loop (Tuple maybeChunk next) ->
633+ Loop ((maybeChunk /\ next)) ->
641 case maybeChunk of
634 case maybeChunk of
642− Nothing -> pure $ Loop $ Tuple Nothing (takeUntilStrom predicate next)
635+ Nothing -> pure $ Loop $ (Nothing /\ (takeUntilStrom predicate next))
643 Just chunk -> case Array.findIndex predicate chunk of
636 Just chunk -> case Array.findIndex predicate chunk of
644 Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Found match, stop here
637 Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Found match, stop here
645− Nothing -> pure $ Loop $ Tuple (Just chunk) (takeUntilStrom predicate next) -- Keep going
638+ Nothing -> pure $ Loop $ ((Just chunk) /\ (takeUntilStrom predicate next)) -- Keep going
647-- | Drop while predicate is true
640-- | Drop while predicate is true
648dropWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
641dropWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -657,13 +650,13 @@ dropWhileStrom predicate stream = dropWhileHelper true stream
657 Done (Just chunk) -> do
650 Done (Just chunk) -> do
658 let dropped = Array.dropWhile predicate chunk
651 let dropped = Array.dropWhile predicate chunk
659 pure $ Done $ if Array.null dropped then Nothing else Just dropped
652 pure $ Done $ if Array.null dropped then Nothing else Just dropped
660− Loop (Tuple maybeChunk next) ->
653+ Loop ((maybeChunk /\ next)) ->
661 case maybeChunk of
654 case maybeChunk of
662− Nothing -> pure $ Loop $ Tuple Nothing (dropWhileHelper stillDropping next)
655+ Nothing -> pure $ Loop $ (Nothing /\ (dropWhileHelper stillDropping next))
663 Just chunk -> do
656 Just chunk -> do
664 let dropped = Array.dropWhile predicate chunk
657 let dropped = Array.dropWhile predicate chunk
665− if Array.null dropped then pure $ Loop $ Tuple Nothing (dropWhileHelper true next)
658+ if Array.null dropped then pure $ Loop $ (Nothing /\ (dropWhileHelper true next))
666− else pure $ Loop $ Tuple (Just dropped) next -- Found elements to keep, stop dropping
659+ else pure $ Loop $ ((Just dropped) /\ next) -- Found elements to keep, stop dropping
668-- | Filter with monadic predicate
661-- | Filter with monadic predicate
669filterMStrom :: forall ctx err a. (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a
662filterMStrom :: forall ctx err a. (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -674,13 +667,13 @@ filterMStrom predicate stream = mkStrom do
674 Done (Just chunk) -> do
667 Done (Just chunk) -> do
675 filtered <- Array.filterA predicate chunk
668 filtered <- Array.filterA predicate chunk
676 pure $ Done $ if Array.null filtered then Nothing else Just filtered
669 pure $ Done $ if Array.null filtered then Nothing else Just filtered
677− Loop (Tuple maybeChunk next) -> do
670+ Loop ((maybeChunk /\ next)) -> do
678 filteredChunk <- case maybeChunk of
671 filteredChunk <- case maybeChunk of
679 Nothing -> pure Nothing
672 Nothing -> pure Nothing
680 Just chunk -> do
673 Just chunk -> do
681 filtered <- Array.filterA predicate chunk
674 filtered <- Array.filterA predicate chunk
682 pure $ if Array.null filtered then Nothing else Just filtered
675 pure $ if Array.null filtered then Nothing else Just filtered
683− pure $ Loop $ Tuple filteredChunk (filterMStrom predicate next)
676+ pure $ Loop $ (filteredChunk /\ (filterMStrom predicate next))
685-- | Collect with partial function (mapMaybe)
678-- | Collect with partial function (mapMaybe)
686collectStrom :: forall ctx err a b. (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b
679collectStrom :: forall ctx err a b. (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b
@@ -691,17 +684,16 @@ collectStrom f stream = mkStrom do
691 Done (Just chunk) -> do
684 Done (Just chunk) -> do
692 let collected = Array.mapMaybe f chunk
685 let collected = Array.mapMaybe f chunk
693 pure $ Done $ if Array.null collected then Nothing else Just collected
686 pure $ Done $ if Array.null collected then Nothing else Just collected
694− Loop (Tuple maybeChunk next) ->
687+ Loop ((maybeChunk /\ next)) ->
696− collected = case maybeChunk of
690+ collected = case maybeChunk of
697− Nothing -> Nothing
691+ Nothing -> Nothing
698− Just chunk ->
692+ Just chunk -> do
700− c = Array.mapMaybe f chunk
693+ let c = Array.mapMaybe f chunk
702 if Array.null c then Nothing else Just c
694 if Array.null c then Nothing else Just c
704− pure $ Loop $ Tuple collected (collectStrom f next)
696+ pure $ Loop $ (collected /\ (collectStrom f next))
706-- | Collect with monadic partial function
698-- | Collect with monadic partial function
707collectMStrom :: forall ctx err a b. (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b
699collectMStrom :: forall ctx err a b. (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b
@@ -713,16 +705,16 @@ collectMStrom f stream = mkStrom do
713 maybes <- traverse f chunk
705 maybes <- traverse f chunk
714 let collected = Array.mapMaybe identity maybes
706 let collected = Array.mapMaybe identity maybes
715 pure $ Done $ if Array.null collected then Nothing else Just collected
707 pure $ Done $ if Array.null collected then Nothing else Just collected
716− Loop (Tuple maybeChunk next) -> do
708+ Loop ((maybeChunk /\ next)) -> do
717 collected <- case maybeChunk of
709 collected <- case maybeChunk of
718 Nothing -> pure Nothing
710 Nothing -> pure Nothing
719 Just chunk -> do
711 Just chunk -> do
720 maybes <- traverse f chunk
712 maybes <- traverse f chunk
721 let c = Array.mapMaybe identity maybes
713 let c = Array.mapMaybe identity maybes
722 pure $ if Array.null c then Nothing else Just c
714 pure $ if Array.null c then Nothing else Just c
723− pure $ Loop $ Tuple collected (collectMStrom f next)
715+ pure $ Loop $ (collected /\ (collectMStrom f next))
725−-- | Remove consecutive duplicates
717+-- | Remove consecutive duplicates
726changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a
718changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a
727changesStrom stream = changesHelper Nothing stream
719changesStrom stream = changesHelper Nothing stream
@@ -732,28 +724,28 @@ changesStrom stream = changesHelper Nothing stream
732 Done Nothing -> pure $ Done Nothing
724 Done Nothing -> pure $ Done Nothing
733 Done (Just chunk) -> do
725 Done (Just chunk) -> do
735− Tuple _ result = Array.foldl
727+ (_ /\ result) = Array.foldl
736− ( \(Tuple prev acc) curr ->
728+ ( \((prev /\ acc)) curr ->
737− if Just curr == prev then Tuple (Just curr) acc
729+ if Just curr == prev then ((Just curr) /\ acc)
738− else Tuple (Just curr) (Array.snoc acc curr)
730+ else ((Just curr) /\ (Array.snoc acc curr))
740− (Tuple lastSeen [])
732+ (lastSeen /\ [])
742 pure $ Done $ if Array.null result then Nothing else Just result
734 pure $ Done $ if Array.null result then Nothing else Just result
743− Loop (Tuple maybeChunk next) ->
735+ Loop ((maybeChunk /\ next)) ->
744 case maybeChunk of
736 case maybeChunk of
745− Nothing -> pure $ Loop $ Tuple Nothing (changesHelper lastSeen next)
737+ Nothing -> pure $ Loop $ (Nothing /\ (changesHelper lastSeen next))
746 Just chunk -> do
738 Just chunk -> do
748− Tuple newLast result = Array.foldl
740+ (newLast /\ result) = Array.foldl
749− ( \(Tuple prev acc) curr ->
741+ ( \((prev /\ acc)) curr ->
750− if Just curr == prev then Tuple (Just curr) acc
742+ if Just curr == prev then ((Just curr) /\ acc)
751− else Tuple (Just curr) (Array.snoc acc curr)
743+ else ((Just curr) /\ (Array.snoc acc curr))
753− (Tuple lastSeen [])
745+ (lastSeen /\ [])
755− if Array.null result then pure $ Loop $ Tuple Nothing (changesHelper newLast next)
747+ if Array.null result then pure $ Loop $ (Nothing /\ (changesHelper newLast next))
756− else pure $ Loop $ Tuple (Just result) (changesHelper newLast next)
748+ else pure $ Loop $ ((Just result) /\ (changesHelper newLast next))
758--------------------------------------------------------------------------------
750--------------------------------------------------------------------------------
759-- Grouping
751-- Grouping
@@ -765,16 +757,16 @@ groupedStrom size stream
765 | size <= 0 = empty
757 | size <= 0 = empty
766 | otherwise = groupedHelper [] stream
758 | otherwise = groupedHelper [] stream
768− groupedHelper buffer s = mkStrom do
760+ groupedHelper buf s = mkStrom do
769 step <- runStrom s
761 step <- runStrom s
770 case step of
762 case step of
771 Done Nothing ->
763 Done Nothing ->
772 -- End of stream, emit remaining buffer if non-empty
764 -- End of stream, emit remaining buffer if non-empty
773− pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
765+ pure $ Done $ if Array.null buf then Nothing else Just [ buf ]
774 Done (Just chunk) -> do
766 Done (Just chunk) -> do
775 -- Process final chunk with buffer
767 -- Process final chunk with buffer
777− combined = buffer <> chunk
769+ combined = buf <> chunk
778 groups = groupsOf size combined
770 groups = groupsOf size combined
779 completeGroups = Array.dropEnd 1 groups
771 completeGroups = Array.dropEnd 1 groups
780 lastGroup = case Array.last groups of
772 lastGroup = case Array.last groups of
@@ -785,28 +777,28 @@ groupedStrom size stream
785 Nothing -> []
777 Nothing -> []
786 if Array.null remainder then pure $ Done $ if Array.null (completeGroups <> lastGroup) then Nothing else Just (completeGroups <> lastGroup)
778 if Array.null remainder then pure $ Done $ if Array.null (completeGroups <> lastGroup) then Nothing else Just (completeGroups <> lastGroup)
787 else pure $ Done $ if Array.null (completeGroups <> [ remainder ]) then Nothing else Just (completeGroups <> [ remainder ])
779 else pure $ Done $ if Array.null (completeGroups <> [ remainder ]) then Nothing else Just (completeGroups <> [ remainder ])
788− Loop (Tuple maybeChunk next) ->
780+ Loop ((maybeChunk /\ next)) ->
789 case maybeChunk of
781 case maybeChunk of
790− Nothing -> pure $ Loop $ Tuple Nothing (groupedHelper buffer next)
782+ Nothing -> pure $ Loop $ (Nothing /\ (groupedHelper buf next))
791 Just chunk -> do
783 Just chunk -> do
793− combined = buffer <> chunk
785+ combined = buf <> chunk
794 groups = groupsOf size combined
786 groups = groupsOf size combined
795 completeGroups = if Array.null groups then [] else Array.dropEnd 1 groups
787 completeGroups = if Array.null groups then [] else Array.dropEnd 1 groups
796 newBuffer = case Array.last groups of
788 newBuffer = case Array.last groups of
797 Just g -> if Array.length g < size then g else []
789 Just g -> if Array.length g < size then g else []
798 Nothing -> []
790 Nothing -> []
799− if Array.null completeGroups then pure $ Loop $ Tuple Nothing (groupedHelper newBuffer next)
791+ if Array.null completeGroups then pure $ Loop $ (Nothing /\ (groupedHelper newBuffer next))
800− else pure $ Loop $ Tuple (Just completeGroups) (groupedHelper newBuffer next)
792+ else pure $ Loop $ ((Just completeGroups) /\ (groupedHelper newBuffer next))
802 groupsOf :: forall b. Int -> Array b -> Array (Array b)
794 groupsOf :: forall b. Int -> Array b -> Array (Array b)
803 groupsOf n arr
795 groupsOf n arr
804 | n <= 0 || Array.null arr = []
796 | n <= 0 || Array.null arr = []
805 | otherwise =
797 | otherwise =
807− group = Array.take n arr
800+ group = Array.take n arr
808− rest = Array.drop n arr
801+ rest = Array.drop n arr
810 if Array.null group then [] else Array.cons group (groupsOf n rest)
802 if Array.null group then [] else Array.cons group (groupsOf n rest)
812-- | Alias for groupedStrom
804-- | Alias for groupedStrom
@@ -818,74 +810,70 @@ chunkedStrom = groupedStrom
818groupByStrom :: forall ctx err a k. Eq k => (a -> k) -> Strom ctx err a -> Strom ctx err (Array a)
810groupByStrom :: forall ctx err a k. Eq k => (a -> k) -> Strom ctx err a -> Strom ctx err (Array a)
819groupByStrom keyFn stream = groupByHelper Nothing [] stream
811groupByStrom keyFn stream = groupByHelper Nothing [] stream
821− groupByHelper lastKey buffer s = mkStrom do
813+ groupByHelper lastKey buf s = mkStrom do
822 step <- runStrom s
814 step <- runStrom s
823 case step of
815 case step of
824 Done Nothing ->
816 Done Nothing ->
825− pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
817+ pure $ Done $ if Array.null buf then Nothing else Just [ buf ]
826 Done (Just chunk) -> do
818 Done (Just chunk) -> do
828− Tuple _ result = Array.foldl
820+ (_ /\ result) = Array.foldl
829− ( \(Tuple currentKey acc) item ->
821+ ( \((currentKey /\ acc)) item ->
831− itemKey = keyFn item
823+ let itemKey = keyFn item
833 case currentKey of
824 case currentKey of
834− Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
825+ Nothing -> ((Just itemKey) /\ { groups: [], buffer: [ item ] })
835 Just k ->
826 Just k ->
836− if k == itemKey then Tuple (Just itemKey) { groups: acc.groups, buffer: Array.snoc acc.buffer item }
827+ if k == itemKey then ((Just itemKey) /\ { groups: acc.groups, buffer: Array.snoc acc.buffer item })
837− else Tuple (Just itemKey) { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] }
828+ else ((Just itemKey) /\ { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] })
839− (Tuple lastKey { groups: [], buffer })
830+ (lastKey /\ { groups: [], buffer: buf })
841 allGroups =
832 allGroups =
842 if Array.null result.buffer then result.groups
833 if Array.null result.buffer then result.groups
843 else Array.snoc result.groups result.buffer
834 else Array.snoc result.groups result.buffer
844 pure $ Done $ if Array.null allGroups then Nothing else Just allGroups
835 pure $ Done $ if Array.null allGroups then Nothing else Just allGroups
845− Loop (Tuple maybeChunk next) ->
836+ Loop ((maybeChunk /\ next)) ->
846 case maybeChunk of
837 case maybeChunk of
847− Nothing -> pure $ Loop $ Tuple Nothing (groupByHelper lastKey buffer next)
838+ Nothing -> pure $ Loop $ (Nothing /\ (groupByHelper lastKey buf next))
848 Just chunk -> do
839 Just chunk -> do
850− Tuple newKey result = Array.foldl
841+ (newKey /\ result) = Array.foldl
851− ( \(Tuple currentKey acc) item ->
842+ ( \((currentKey /\ acc)) item ->
853− itemKey = keyFn item
844+ let itemKey = keyFn item
855 case currentKey of
845 case currentKey of
856− Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
846+ Nothing -> ((Just itemKey) /\ { groups: [], buffer: [ item ] })
857 Just k ->
847 Just k ->
858− if k == itemKey then Tuple (Just itemKey) { groups: acc.groups, buffer: Array.snoc acc.buffer item }
848+ if k == itemKey then ((Just itemKey) /\ { groups: acc.groups, buffer: Array.snoc acc.buffer item })
859− else Tuple (Just itemKey) { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] }
849+ else ((Just itemKey) /\ { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] })
861− (Tuple lastKey { groups: [], buffer })
851+ (lastKey /\ { groups: [], buffer: buf })
863− if Array.null result.groups then pure $ Loop $ Tuple Nothing (groupByHelper newKey result.buffer next)
853+ if Array.null result.groups then pure $ Loop $ (Nothing /\ (groupByHelper newKey result.buffer next))
864− else pure $ Loop $ Tuple (Just result.groups) (groupByHelper newKey result.buffer next)
854+ else pure $ Loop $ ((Just result.groups) /\ (groupByHelper newKey result.buffer next))
866-- | Partition a stream by a predicate into two separate operations
856-- | Partition a stream by a predicate into two separate operations
867-- | Returns a tuple of (trues, falses)
857-- | Returns a tuple of (trues, falses)
868-- | Note: This requires running the stream twice, so it's not lazy
858-- | Note: This requires running the stream twice, so it's not lazy
869−partition :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Tuple (Strom ctx err a) (Strom ctx err a)
859+partition :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> ((Strom ctx err a) /\ (Strom ctx err a))
870partition predicate stream =
860partition predicate stream =
872− (filterStrom predicate stream)
861+ (filterStrom predicate stream)
873− (filterStrom (not <<< predicate) stream)
862+ /\ (filterStrom (not <<< predicate) stream)
875-- | Partition and map in one pass
864-- | Partition and map in one pass
876−partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> Tuple (Strom ctx err b) (Strom ctx err c)
865+partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> ((Strom ctx err b) /\ (Strom ctx err c))
877partitionMap f stream =
866partitionMap f stream =
879− leftOnly x = case f x of
869+ leftOnly x = case f x of
880− Left b -> Just b
870+ Left b -> Just b
881− Right _ -> Nothing
871+ Right _ -> Nothing
882− rightOnly x = case f x of
872+ rightOnly x = case f x of
883− Left _ -> Nothing
873+ Left _ -> Nothing
884− Right c -> Just c
874+ Right c -> Just c
887− (collectStrom leftOnly stream)
875+ (collectStrom leftOnly stream)
888− (collectStrom rightOnly stream)
876+ /\ (collectStrom rightOnly stream)
890--------------------------------------------------------------------------------
878--------------------------------------------------------------------------------
@@ -901,49 +889,24 @@ debounce duration stream = mkStrom do
901 Done (Just chunk) -> do
889 Done (Just chunk) -> do
902 Om.delay duration
890 Om.delay duration
903 pure $ Done $ Just chunk
891 pure $ Done $ Just chunk
904− Loop (Tuple maybeChunk next) -> do
892+ Loop (maybeChunk /\ next) -> do
905 case maybeChunk of
893 case maybeChunk of
906− Nothing -> pure $ Loop $ Tuple Nothing (debounce duration next)
894+ Nothing -> pure $ Loop $ (Nothing /\ (debounce duration next))
907 Just chunk -> do
895 Just chunk -> do
908 Om.delay duration
896 Om.delay duration
909− pure $ Loop $ Tuple (Just chunk) (debounce duration next)
897+ pure $ Loop $ ((Just chunk) /\ (debounce duration next))
899+-- | Debounce a stream per element - adds a delay after each individual element
900+debounceEach :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
901+debounceEach duration stream = stream # bindStrom \a -> mkStrom do
903+ pure $ Done $ Just [ a ]
911−-- | Throttle a stream - emit at most once per duration period
905+-- | Throttle a stream - emits at most one element per duration period
912−-- | Tracks last emission time and skips elements that arrive too quickly
913−-- | Note: Uses wall-clock time, not processing time
914throttle :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
906throttle :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
915−throttle (Milliseconds duration) stream = throttleWithState 0.0 stream
907+throttle duration stream = stream # bindStrom \a -> mkStrom do
917− throttleWithState lastEmit s = mkStrom do
920− Done Nothing -> pure $ Done Nothing
921− Done (Just chunk) -> do
922− Tuple newLastEmit filtered <- filterByTime lastEmit chunk
923− pure $ Done $ if Array.null filtered then Nothing else Just filtered
924− Loop (Tuple maybeChunk next) -> do
926− Nothing -> pure $ Loop $ Tuple Nothing (throttleWithState lastEmit next)
928− Tuple newLastEmit filtered <- filterByTime lastEmit chunk
929− if Array.null filtered then pure $ Loop $ Tuple Nothing (throttleWithState newLastEmit next)
930− else pure $ Loop $ Tuple (Just filtered) (throttleWithState newLastEmit next)
932− filterByTime lastEmit chunk = do
933− Tuple newLastEmit results <- Array.foldl
935− Tuple currentLast filtered <- acc
936− currentTime <- liftEffect now
937− let Milliseconds currentMillis = unInstant currentTime
938− let elapsed = currentMillis - currentLast
939− if elapsed >= duration then
908+ Om.delay duration
940− pure $ Tuple currentMillis (Array.snoc filtered value)
909+ pure $ Done $ Just [ a ]
942− pure $ Tuple currentLast filtered
944− (pure $ Tuple lastEmit [])
946− pure $ Tuple newLastEmit results
948-- | Delay each element by a duration
911-- | Delay each element by a duration
949delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
912delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
@@ -957,33 +920,33 @@ delayStrom duration stream = mkStrom do
958-- | Fold over all elements
921-- | Fold over all elements
959runFold :: forall ctx err a b. b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b
922runFold :: forall ctx err a b. b -> (b -> a -> b) -> Strom ctx err a -> Om ctx err b
960−runFold initial f stream = tailRecM go (Tuple initial stream)
923+runFold initial f stream = tailRecM go ((initial /\ stream))
962− go (Tuple acc s) = do
925+ go ((acc /\ s)) = do
963 step <- runStrom s
926 step <- runStrom s
964 case step of
927 case step of
965 Done Nothing -> pure $ Done acc
928 Done Nothing -> pure $ Done acc
966 Done (Just chunk) -> pure $ Done $ foldl f acc chunk
929 Done (Just chunk) -> pure $ Done $ foldl f acc chunk
967− Loop (Tuple maybeChunk next) ->
930+ Loop ((maybeChunk /\ next)) ->
968 -- Process chunk if present and continue
931 -- Process chunk if present and continue
969 case maybeChunk of
932 case maybeChunk of
970− Nothing -> pure $ Loop $ Tuple acc next
933+ Nothing -> pure $ Loop $ (acc /\ next)
971− Just chunk -> pure $ Loop $ Tuple (foldl f acc chunk) next
934+ Just chunk -> pure $ Loop $ ((foldl f acc chunk) /\ next)
973-- | Collect all elements into an array (O(n) using chunk accumulation)
936-- | Collect all elements into an array (O(n) using chunk accumulation)
974runCollect :: forall ctx err a. Strom ctx err a -> Om ctx err (Array a)
937runCollect :: forall ctx err a. Strom ctx err a -> Om ctx err (Array a)
975−runCollect stream = tailRecM go (Tuple List.Nil stream)
938+runCollect stream = tailRecM go (List.Nil /\ stream)
977− go (Tuple chunkList s) = do
940+ go ((chunkList /\ s)) = do
978 step <- runStrom s
941 step <- runStrom s
979 case step of
942 case step of
980 Done Nothing -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse chunkList
943 Done Nothing -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse chunkList
981 Done (Just chunk) -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse $ List.Cons chunk chunkList
944 Done (Just chunk) -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse $ List.Cons chunk chunkList
982− Loop (Tuple maybeChunk next) ->
945+ Loop ((maybeChunk /\ next)) ->
983 -- Emit the chunk (if present) and continue
946 -- Emit the chunk (if present) and continue
984 case maybeChunk of
947 case maybeChunk of
985− Nothing -> pure $ Loop $ Tuple chunkList next
948+ Nothing -> pure $ Loop $ (chunkList /\ next)
986− Just chunk -> pure $ Loop $ Tuple (List.Cons chunk chunkList) next
949+ Just chunk -> pure $ Loop $ ((List.Cons chunk chunkList) /\ next)
988-- | Run the stream and discard all elements
951-- | Run the stream and discard all elements
989runDrain :: forall ctx err a. Strom ctx err a -> Om ctx err Unit
952runDrain :: forall ctx err a. Strom ctx err a -> Om ctx err Unit
@@ -1000,7 +963,7 @@ traverseStrom_ f stream = tailRecM go stream
1000 Done (Just chunk) -> do
963 Done (Just chunk) -> do
1001 _ <- traverse f chunk
964 _ <- traverse f chunk
1002 pure $ Done unit
965 pure $ Done unit
1003− Loop (Tuple maybeChunk next) -> do
966+ Loop ((maybeChunk /\ next)) -> do
1004 -- Process chunk if present, then continue
967 -- Process chunk if present, then continue
1005 case maybeChunk of
968 case maybeChunk of
1006 Nothing -> pure unit
969 Nothing -> pure unit
@@ -1024,20 +987,20 @@ traverseMStrom_ = traverseStrom_
1024--------------------------------------------------------------------------------
987--------------------------------------------------------------------------------
1026-- | Run a stream in the background, returning a Fiber that can be cancelled
989-- | Run a stream in the background, returning a Fiber that can be cancelled
1028-- | To use cancellable streams, convert your stream to Om and use Om.launchOm:
991-- | To use cancellable streams, convert your stream to Om and use Om.launchOm:
1030-- | Example:
993-- | Example:
1031-- | ```purescript
994-- | ```purescript
1032-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)
995-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)
1033-- | -- ... later ...
996-- | -- ... later ...
1034-- | killFiber (error "Cancelled") fiber
997-- | killFiber (error "Cancelled") fiber
1037-- | Or for subscription-style:
1000-- | Or for subscription-style:
1038-- | ```purescript
1001-- | ```purescript
1039-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)
1002-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)
1040−-- | -- ... later ...
1003+-- | -- ... later ...
1041-- | killFiber (error "Unsubscribed") fiber
1004-- | killFiber (error "Unsubscribed") fiber
@@ -1060,8 +1023,8 @@ appendStrom toAppend baseStream = mkStrom do
1060 step <- runStrom baseStream
1023 step <- runStrom baseStream
1061 case step of
1024 case step of
1062 Done Nothing -> runStrom toAppend
1025 Done Nothing -> runStrom toAppend
1063− Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) toAppend -- Emit chunk and continue with toAppend
1026+ Done (Just chunk) -> pure $ Loop $ ((Just chunk) /\ toAppend) -- Emit chunk and continue with toAppend
1064− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (appendStrom toAppend next)
1027+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (appendStrom toAppend next))
1066-- | Concatenate an array of streams
1029-- | Concatenate an array of streams
1067-- | Uses foldr with flip to match appendStrom's pipeline-friendly parameter order
1030-- | Uses foldr with flip to match appendStrom's pipeline-friendly parameter order
@@ -1069,8 +1032,8 @@ concatStrom :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1069concatStrom streams = Array.foldr (flip appendStrom) empty streams
1032concatStrom streams = Array.foldr (flip appendStrom) empty streams
1071-- | Zip two streams together
1034-- | Zip two streams together
1072−zipStrom :: forall ctx err a b. Strom ctx err a -> Strom ctx err b -> Strom ctx err (Tuple a b)
1035+zipStrom :: forall ctx err a b. Strom ctx err a -> Strom ctx err b -> Strom ctx err ((a /\ b))
1073−zipStrom = zipWithStrom Tuple
1036+zipStrom = zipWithStrom (/\)
1075-- | Zip two streams with a function
1038-- | Zip two streams with a function
1076zipWithStrom :: forall ctx err a b c. (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c
1039zipWithStrom :: forall ctx err a b c. (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c
@@ -1083,13 +1046,13 @@ zipWithStrom f s1 s2 = mkStrom do
1083 pure $ Done $ Just zipped
1046 pure $ Done $ Just zipped
1084 Done Nothing, _ -> pure $ Done Nothing
1047 Done Nothing, _ -> pure $ Done Nothing
1085 _, Done Nothing -> pure $ Done Nothing
1048 _, Done Nothing -> pure $ Done Nothing
1086− Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1049+ Loop ((maybeChunk1 /\ next1)), Loop ((maybeChunk2 /\ next2)) ->
1087 -- Both looping: zip chunks if both present, continue with both nexts
1050 -- Both looping: zip chunks if both present, continue with both nexts
1088 case maybeChunk1, maybeChunk2 of
1051 case maybeChunk1, maybeChunk2 of
1089− Just chunk1, Just chunk2 -> pure $ Loop $ Tuple (Just $ Array.zipWith f chunk1 chunk2) (zipWithStrom f next1 next2)
1052+ Just chunk1, Just chunk2 -> pure $ Loop $ ((Just $ Array.zipWith f chunk1 chunk2) /\ (zipWithStrom f next1 next2))
1090− _, _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 next2)
1053+ _, _ -> pure $ Loop $ (Nothing /\ (zipWithStrom f next1 next2))
1091− Loop (Tuple _ next1), _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 s2)
1054+ Loop ((_ /\ next1)), _ -> pure $ Loop $ (Nothing /\ (zipWithStrom f next1 s2))
1092− _, Loop (Tuple _ next2) -> pure $ Loop $ Tuple Nothing (zipWithStrom f s1 next2)
1055+ _, Loop ((_ /\ next2)) -> pure $ Loop $ (Nothing /\ (zipWithStrom f s1 next2))
1094-- | Interleave two streams - alternates between elements from each stream
1057-- | Interleave two streams - alternates between elements from each stream
1095-- | Similar to merge but deterministic - always alternates
1058-- | Similar to merge but deterministic - always alternates
@@ -1101,16 +1064,16 @@ interleave s1 s2 = interleaveHelper true s1 s2
1101 step1 <- runStrom first
1064 step1 <- runStrom first
1102 case step1 of
1065 case step1 of
1103 Done Nothing -> runStrom second
1066 Done Nothing -> runStrom second
1104− Done (Just chunk1) -> pure $ Loop $ Tuple (Just chunk1) (interleaveHelper false second first)
1067+ Done (Just chunk1) -> pure $ Loop $ ((Just chunk1) /\ (interleaveHelper false second first))
1105− Loop (Tuple maybeChunk1 next1) ->
1068+ Loop ((maybeChunk1 /\ next1)) ->
1106− pure $ Loop $ Tuple maybeChunk1 (interleaveHelper false second next1)
1069+ pure $ Loop $ (maybeChunk1 /\ (interleaveHelper false second next1))
1108 step2 <- runStrom second
1071 step2 <- runStrom second
1109 case step2 of
1072 case step2 of
1110 Done Nothing -> runStrom first
1073 Done Nothing -> runStrom first
1111− Done (Just chunk2) -> pure $ Loop $ Tuple (Just chunk2) (interleaveHelper true first second)
1074+ Done (Just chunk2) -> pure $ Loop $ ((Just chunk2) /\ (interleaveHelper true first second))
1112− Loop (Tuple maybeChunk2 next2) ->
1075+ Loop ((maybeChunk2 /\ next2)) ->
1113− pure $ Loop $ Tuple maybeChunk2 (interleaveHelper true first next2)
1076+ pure $ Loop $ (maybeChunk2 /\ (interleaveHelper true first next2))
1115-- | Intersperse a separator element between each element of the stream
1078-- | Intersperse a separator element between each element of the stream
1116intersperse :: forall ctx err a. a -> Strom ctx err a -> Strom ctx err a
1079intersperse :: forall ctx err a. a -> Strom ctx err a -> Strom ctx err a
@@ -1126,15 +1089,15 @@ intersperse separator stream = intersperseHelper true stream
1126 if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1089 if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1127 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1090 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1128 pure $ Done $ Just withSeparators
1091 pure $ Done $ Just withSeparators
1129− Loop (Tuple maybeChunk next) ->
1092+ Loop ((maybeChunk /\ next)) ->
1130 case maybeChunk of
1093 case maybeChunk of
1131− Nothing -> pure $ Loop $ Tuple Nothing (intersperseHelper isFirst next)
1094+ Nothing -> pure $ Loop $ (Nothing /\ (intersperseHelper isFirst next))
1132 Just chunk -> do
1095 Just chunk -> do
1134 withSeparators =
1097 withSeparators =
1135 if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1098 if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
1136 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1099 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1137− pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
1100+ pure $ Loop $ ((Just withSeparators) /\ (intersperseHelper false next))
1139--------------------------------------------------------------------------------
1102--------------------------------------------------------------------------------
1140-- Buffering
1103-- Buffering
@@ -1175,7 +1138,7 @@ bufferChunks capacity stream
1175 Nothing -> do
1138 Nothing -> do
1176 AVar.put unit mutex
1139 AVar.put unit mutex
1178− Just (Tuple head tail) -> do
1141+ Just ((head /\ tail)) -> do
1179 liftEffect $ Ref.write tail bufferRef
1142 liftEffect $ Ref.write tail bufferRef
1180 liftEffect $ Ref.write (size - 1) sizeRef
1143 liftEffect $ Ref.write (size - 1) sizeRef
1181 let hasMore = (size - 1) > 0
1144 let hasMore = (size - 1) > 0
@@ -1193,7 +1156,7 @@ bufferChunks capacity stream
1193 Done (Just chunk) -> do
1156 Done (Just chunk) -> do
1194 enqueue (BufferChunk chunk)
1157 enqueue (BufferChunk chunk)
1195 enqueue BufferDone
1158 enqueue BufferDone
1196− Loop (Tuple maybeChunk next) -> do
1159+ Loop ((maybeChunk /\ next)) -> do
1197 case maybeChunk of
1160 case maybeChunk of
1198 Just chunk -> enqueue (BufferChunk chunk)
1161 Just chunk -> enqueue (BufferChunk chunk)
1199 Nothing -> pure unit
1162 Nothing -> pure unit
@@ -1209,7 +1172,7 @@ bufferChunks capacity stream
1209 killFiber (Exception.error "done") fiber
1172 killFiber (Exception.error "done") fiber
1210 pure $ Done Nothing
1173 pure $ Done Nothing
1211 BufferChunk chunk ->
1174 BufferChunk chunk ->
1212− pure $ Loop $ Tuple (Just chunk) consumer
1175+ pure $ Loop $ ((Just chunk) /\ consumer)
1214 stepResult <- Om.runReader ctx (runStrom consumer)
1177 stepResult <- Om.runReader ctx (runStrom consumer)
1215 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1178 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1249,7 +1212,7 @@ buffer capacity stream
1249 Nothing -> do
1212 Nothing -> do
1250 AVar.put unit mutex
1213 AVar.put unit mutex
1252− Just (Tuple head tail) -> do
1215+ Just ((head /\ tail)) -> do
1253 liftEffect $ Ref.write tail bufferRef
1216 liftEffect $ Ref.write tail bufferRef
1254 liftEffect $ Ref.write (size - 1) sizeRef
1217 liftEffect $ Ref.write (size - 1) sizeRef
1255 let hasMore = (size - 1) > 0
1218 let hasMore = (size - 1) > 0
@@ -1267,7 +1230,7 @@ buffer capacity stream
1267 Done (Just chunk) -> do
1230 Done (Just chunk) -> do
1268 traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1231 traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1269 enqueue BufferDone
1232 enqueue BufferDone
1270− Loop (Tuple maybeChunk next) -> do
1233+ Loop ((maybeChunk /\ next)) -> do
1271 case maybeChunk of
1234 case maybeChunk of
1272 Just chunk -> traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1235 Just chunk -> traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1273 Nothing -> pure unit
1236 Nothing -> pure unit
@@ -1283,7 +1246,7 @@ buffer capacity stream
1283 killFiber (Exception.error "done") fiber
1246 killFiber (Exception.error "done") fiber
1284 pure $ Done Nothing
1247 pure $ Done Nothing
1285 BufferChunk chunk ->
1248 BufferChunk chunk ->
1286− pure $ Loop $ Tuple (Just chunk) consumer
1249+ pure $ Loop $ ((Just chunk) /\ consumer)
1288 stepResult <- Om.runReader ctx (runStrom consumer)
1251 stepResult <- Om.runReader ctx (runStrom consumer)
1289 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1252 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1392,7 +1355,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
1392 Nothing -> do
1355 Nothing -> do
1393 AVar.put unit mutex
1356 AVar.put unit mutex
1395− Just (Tuple head tail) -> do
1358+ Just ((head /\ tail)) -> do
1396 liftEffect $ Ref.write tail bufferRef
1359 liftEffect $ Ref.write tail bufferRef
1397 liftEffect $ Ref.write (size - 1) sizeRef
1360 liftEffect $ Ref.write (size - 1) sizeRef
1398 let hasMore = (size - 1) > 0
1361 let hasMore = (size - 1) > 0
@@ -1407,14 +1370,14 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
1408 producer s = do
1371 producer s = do
1409 stepResult <- Om.runReader ctx (runStrom s)
1372 stepResult <- Om.runReader ctx (runStrom s)
1410− step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1373+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1411 case step of
1374 case step of
1412 Done Nothing ->
1375 Done Nothing ->
1413 enqueue (StreamDone StreamId1)
1376 enqueue (StreamDone StreamId1)
1414 Done (Just chunk) -> do
1377 Done (Just chunk) -> do
1415 enqueue (DataChunk chunk StreamId1)
1378 enqueue (DataChunk chunk StreamId1)
1416 enqueue (StreamDone StreamId1)
1379 enqueue (StreamDone StreamId1)
1417− Loop (Tuple maybeChunk next) -> do
1380+ Loop ((maybeChunk /\ next)) -> do
1418 case maybeChunk of
1381 case maybeChunk of
1419 Just chunk -> enqueue (DataChunk chunk StreamId1)
1382 Just chunk -> enqueue (DataChunk chunk StreamId1)
1420 Nothing -> pure unit
1383 Nothing -> pure unit
@@ -1426,14 +1389,14 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
1427 producer s = do
1390 producer s = do
1428 stepResult <- Om.runReader ctx (runStrom s)
1391 stepResult <- Om.runReader ctx (runStrom s)
1429− step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1392+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1430 case step of
1393 case step of
1431 Done Nothing ->
1394 Done Nothing ->
1432 enqueue (StreamDone StreamId2)
1395 enqueue (StreamDone StreamId2)
1433 Done (Just chunk) -> do
1396 Done (Just chunk) -> do
1434 enqueue (DataChunk chunk StreamId2)
1397 enqueue (DataChunk chunk StreamId2)
1435 enqueue (StreamDone StreamId2)
1398 enqueue (StreamDone StreamId2)
1436− Loop (Tuple maybeChunk next) -> do
1399+ Loop ((maybeChunk /\ next)) -> do
1437 case maybeChunk of
1400 case maybeChunk of
1438 Just chunk -> enqueue (DataChunk chunk StreamId2)
1401 Just chunk -> enqueue (DataChunk chunk StreamId2)
1439 Nothing -> pure unit
1402 Nothing -> pure unit
@@ -1460,7 +1423,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
1460 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1423 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1461 DataChunk chunk _streamId ->
1424 DataChunk chunk _streamId ->
1462 -- Got a chunk
1425 -- Got a chunk
1463− pure $ Loop $ Tuple (Just chunk) (consumer doneLeft doneRight)
1426+ pure $ Loop $ ((Just chunk) /\ (consumer doneLeft doneRight))
1465 stepResult <- Om.runReader ctx (runStrom $ consumer false false)
1428 stepResult <- Om.runReader ctx (runStrom $ consumer false false)
1466 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1429 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1505,7 +1468,7 @@ mergeAllNDWith bufferCapacity streams
1505 Nothing -> do
1468 Nothing -> do
1506 AVar.put unit mutex
1469 AVar.put unit mutex
1508− Just (Tuple head tail) -> do
1471+ Just ((head /\ tail)) -> do
1509 liftEffect $ Ref.write tail bufferRef
1472 liftEffect $ Ref.write tail bufferRef
1510 liftEffect $ Ref.write (size - 1) sizeRef
1473 liftEffect $ Ref.write (size - 1) sizeRef
1511 let hasMore = (size - 1) > 0
1474 let hasMore = (size - 1) > 0
@@ -1523,7 +1486,7 @@ mergeAllNDWith bufferCapacity streams
1523 Done (Just chunk) -> do
1486 Done (Just chunk) -> do
1524 enqueue (MergeChunk chunk)
1487 enqueue (MergeChunk chunk)
1525 enqueue MergeDone
1488 enqueue MergeDone
1526− Loop (Tuple maybeChunk next) -> do
1489+ Loop ((maybeChunk /\ next)) -> do
1527 case maybeChunk of
1490 case maybeChunk of
1528 Just chunk -> enqueue (MergeChunk chunk)
1491 Just chunk -> enqueue (MergeChunk chunk)
1529 Nothing -> pure unit
1492 Nothing -> pure unit
@@ -1544,7 +1507,7 @@ mergeAllNDWith bufferCapacity streams
1544 stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1507 stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1545 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1508 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1546 MergeChunk chunk ->
1509 MergeChunk chunk ->
1547− pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1510+ pure $ Loop $ ((Just chunk) /\ (consumer doneCount))
1549 stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1512 stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1550 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1513 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1568,23 +1531,23 @@ mapPar concurrency f stream
1568 let groups = chunksOf concurrency chunk
1531 let groups = chunksOf concurrency chunk
1569 results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1532 results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1570 pure $ Done $ Just $ Array.concat results
1533 pure $ Done $ Just $ Array.concat results
1571− Loop (Tuple maybeChunk next) -> do
1534+ Loop ((maybeChunk /\ next)) -> do
1572 mappedChunk <- case maybeChunk of
1535 mappedChunk <- case maybeChunk of
1573 Nothing -> pure Nothing
1536 Nothing -> pure Nothing
1574 Just chunk -> do
1537 Just chunk -> do
1575 let groups = chunksOf concurrency chunk
1538 let groups = chunksOf concurrency chunk
1576 results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1539 results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
1577 pure $ Just $ Array.concat results
1540 pure $ Just $ Array.concat results
1578− pure $ Loop $ Tuple mappedChunk (mapPar concurrency f next)
1541+ pure $ Loop $ (mappedChunk /\ (mapPar concurrency f next))
1580 chunksOf :: forall x. Int -> Array x -> Array (Array x)
1543 chunksOf :: forall x. Int -> Array x -> Array (Array x)
1581 chunksOf n arr
1544 chunksOf n arr
1582 | n <= 0 || Array.null arr = []
1545 | n <= 0 || Array.null arr = []
1583 | otherwise =
1546 | otherwise =
1585− group = Array.take n arr
1549+ group = Array.take n arr
1586− rest = Array.drop n arr
1550+ rest = Array.drop n arr
1588 if Array.null group then [] else Array.cons group (chunksOf n rest)
1551 if Array.null group then [] else Array.cons group (chunksOf n rest)
1590-- | Unordered parallel map (currently same as mapPar)
1553-- | Unordered parallel map (currently same as mapPar)
@@ -1720,7 +1683,7 @@ instance Functor (Strom ctx err) where
1720 case step of
1683 case step of
1721 Done Nothing -> pure $ Done Nothing
1684 Done Nothing -> pure $ Done Nothing
1722 Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
1685 Done (Just chunk) -> pure $ Done $ Just $ Functor.map f chunk
1723− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple (Functor.map (Functor.map f) maybeChunk) (Functor.map f next)
1686+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ ((Functor.map (Functor.map f) maybeChunk) /\ (Functor.map f next))
1726instance Apply (Strom ctx err) where
1689instance Apply (Strom ctx err) where
@@ -1729,29 +1692,29 @@ instance Apply (Strom ctx err) where
1729 runStrom as >>= \stepA ->
1692 runStrom as >>= \stepA ->
1730 case stepF, stepA of
1693 case stepF, stepA of
1731 Done (Just funcs), Done (Just vals) ->
1694 Done (Just funcs), Done (Just vals) ->
1733− results = do
1697+ results = do
1734− f <- funcs
1698+ f <- funcs
1735− a <- vals
1699+ a <- vals
1736− pure (f a)
1700+ pure (f a)
1738 pure $ Done $ Just results
1701 pure $ Done $ Just results
1739 Done Nothing, _ -> pure $ Done Nothing
1702 Done Nothing, _ -> pure $ Done Nothing
1740 _, Done Nothing -> pure $ Done Nothing
1703 _, Done Nothing -> pure $ Done Nothing
1741− Loop (Tuple maybeChunkF nextF), Loop (Tuple maybeChunkA nextA) ->
1704+ Loop ((maybeChunkF /\ nextF)), Loop ((maybeChunkA /\ nextA)) ->
1742 -- Apply chunks if both present
1705 -- Apply chunks if both present
1743 case maybeChunkF, maybeChunkA of
1706 case maybeChunkF, maybeChunkA of
1744 Just chunkF, Just chunkA ->
1707 Just chunkF, Just chunkA ->
1746− applied = do
1710+ applied = do
1747− f <- chunkF
1711+ f <- chunkF
1748− a <- chunkA
1712+ a <- chunkA
1749− pure (f a)
1713+ pure (f a)
1751− pure $ Loop $ Tuple (Just applied) (nextF <*> nextA)
1714+ pure $ Loop $ ((Just applied) /\ (nextF <*> nextA))
1752− _, _ -> pure $ Loop $ Tuple Nothing (nextF <*> nextA)
1715+ _, _ -> pure $ Loop $ (Nothing /\ (nextF <*> nextA))
1753− Loop (Tuple _ nextF), _ -> pure $ Loop $ Tuple Nothing (nextF <*> as)
1716+ Loop ((_ /\ nextF)), _ -> pure $ Loop $ (Nothing /\ (nextF <*> as))
1754− _, Loop (Tuple _ nextA) -> pure $ Loop $ Tuple Nothing (fs <*> nextA)
1717+ _, Loop ((_ /\ nextA)) -> pure $ Loop $ (Nothing /\ (fs <*> nextA))
1757instance Applicative (Strom ctx err) where
1720instance Applicative (Strom ctx err) where
@@ -1763,19 +1726,18 @@ instance Bind (Strom ctx err) where
1763 case step of
1726 case step of
1764 Done Nothing -> pure $ Done Nothing
1727 Done Nothing -> pure $ Done Nothing
1765 Done (Just chunk) ->
1728 Done (Just chunk) ->
1767− streams = Functor.map f chunk
1730+ let streams = Functor.map f chunk
1769 runStrom $ concatStrom streams
1731 runStrom $ concatStrom streams
1770− Loop (Tuple maybeChunk next) ->
1732+ Loop ((maybeChunk /\ next)) ->
1771 -- When looping, if we have a chunk, we need to process it through f and concat
1733 -- When looping, if we have a chunk, we need to process it through f and concat
1772 case maybeChunk of
1734 case maybeChunk of
1773− Nothing -> pure $ Loop $ Tuple Nothing (next >>= f)
1735+ Nothing -> pure $ Loop $ (Nothing /\ (next >>= f))
1774 Just chunk ->
1736 Just chunk ->
1776− streams = Functor.map f chunk
1739+ streams = Functor.map f chunk
1777− prefixStream = concatStrom streams
1740+ prefixStream = concatStrom streams
1779 runStrom $ appendStrom prefixStream (next >>= f)
1741 runStrom $ appendStrom prefixStream (next >>= f)
@@ -1786,8 +1748,8 @@ instance Semigroup (Strom ctx err a) where
1786 ( runStrom s1 >>= \step ->
1748 ( runStrom s1 >>= \step ->
1787 case step of
1749 case step of
1788 Done Nothing -> runStrom s2
1750 Done Nothing -> runStrom s2
1789− Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1751+ Done (Just chunk) -> pure $ Loop $ ((Just chunk) /\ s2)
1790− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <> s2)
1752+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (next <> s2))
1793instance Monoid (Strom ctx err a) where
1755instance Monoid (Strom ctx err a) where
@@ -1798,8 +1760,8 @@ instance Alt (Strom ctx err) where
1798 ( runStrom s1 >>= \step ->
1760 ( runStrom s1 >>= \step ->
1799 case step of
1761 case step of
1800 Done Nothing -> runStrom s2
1762 Done Nothing -> runStrom s2
1801− Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1763+ Done (Just chunk) -> pure $ Loop $ ((Just chunk) /\ s2)
1802− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <|> s2)
1764+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (next <|> s2))
1805instance Plus (Strom ctx err) where
1767instance Plus (Strom ctx err) where