@@ -108,6 +108,9 @@ module Yoga.Om.Strom
108108 , bracket
109109 , bracketExit
110110 , acquireRelease
111+ -- Internal (needed by WebStream)
112+ , mkStrom
113+ , runStrom
111114 ) where
112115
113116import Prelude
@@ -129,7 +132,7 @@ import Data.List as List
129132import Data.Maybe (Maybe(..))
130133import Data.Time.Duration (Milliseconds(..))
131134import Data.Traversable (traverse, traverse_)
132−import Data.Tuple (Tuple(..))
135+import Data.Tuple.Nested (type (/\), (/\))
133136import Data.Variant (Variant)
134137import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
135138import Effect.Aff.AVar as AVar
@@ -160,18 +163,18 @@ import Yoga.Om as Om
160163-- | - Loop (chunk, next) => emit chunk (if present) and continue with next
161164-- | - Done finalChunk => emit final chunk (if present) and stop
162165newtype Strom ctx err a = Strom
163− { pull :: Om ctx err (Step (Tuple (Maybe (Chunk a)) (Strom ctx err a)) (Maybe (Chunk a)))
166+ { pull :: Om ctx err (Step (((Maybe (Chunk a)) /\ (Strom ctx err a))) (Maybe (Chunk a)))
164167 }
165168
166169-- | Chunks for efficient batch processing
167170type Chunk a = Array a
168171
169172-- | 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
173+mkStrom :: forall ctx err a. Om ctx err (Step (((Maybe (Chunk a)) /\ (Strom ctx err a))) (Maybe (Chunk a))) -> Strom ctx err a
171174mkStrom pull = Strom { pull }
172175
173176-- | 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)))
177+runStrom :: forall ctx err a. Strom ctx err a -> Om ctx err (Step (((Maybe (Chunk a)) /\ (Strom ctx err a))) (Maybe (Chunk a)))
175178runStrom (Strom s) = s.pull
176179
177180--------------------------------------------------------------------------------
@@ -181,22 +184,22 @@ runStrom (Strom s) = s.pull
181184-- | Stream identifier for tracking which stream produced data
182185data StreamId = StreamId1 | StreamId2
183186
184−derive instance eqStreamId :: Eq StreamId
185−derive instance ordStreamId :: Ord StreamId
187+derive instance Eq StreamId
188+derive instance Ord StreamId
186189
187190-- | Messages passed through the coordination queue
188191data QueueItem a
189192 = DataChunk (Array a) StreamId
190193 | StreamDone StreamId
191194
192−derive instance functorQueueItem :: Functor QueueItem
195+derive instance Functor QueueItem
193196
194197-- | Messages for mergeAllND buffered queue
195198data MergeAllItem a
196199 = MergeChunk (Array a)
197200 | MergeDone
198201
199−derive instance functorMergeAllItem :: Functor MergeAllItem
202+derive instance Functor MergeAllItem
200203
201204-- | Merge termination strategy (like ZStream HaltStrategy)
202205data HaltStrategy
@@ -205,14 +208,14 @@ data HaltStrategy
205208 | HaltLeft
206209 | HaltRight
207210
208−derive instance eqHaltStrategy :: Eq HaltStrategy
211+derive instance Eq HaltStrategy
209212
210213-- | Messages for buffering queues
211214data BufferItem a
212215 = BufferChunk (Array a)
213216 | BufferDone
214217
215−derive instance functorBufferItem :: Functor BufferItem
218+derive instance Functor BufferItem
216219
217220--------------------------------------------------------------------------------
218221-- Construction
@@ -268,7 +271,7 @@ rangeStrom start end = rangeHelper start end
268271 chunkEnd = min (current + chunkSize) limit
269272 chunk = Array.range current (chunkEnd - 1)
270273 if chunkEnd >= limit then pure $ Done $ Just chunk -- Final chunk
271− else pure $ Loop $ Tuple (Just chunk) (rangeHelper chunkEnd limit) -- Emit chunk and continue
274+ else pure $ Loop $ ((Just chunk) /\ (rangeHelper chunkEnd limit)) -- Emit chunk and continue
272275
273276-- | Iterate producing values - limited to ~10,000 elements for stack safety
274277-- | For truly infinite streams, use `iterateStromInfinite` (with Aff overhead)
@@ -306,7 +309,7 @@ iterateStromInfinite f initial = iterateHelper initial
306309 finalVal <- STRef.read valRef
307310 frozen <- STArray.unsafeFreeze arr
308311 pure { arr: frozen, nextVal: finalVal }
309− pure $ Loop $ Tuple (Just result.arr) (iterateHelper result.nextVal)
312+ pure $ Loop $ ((Just result.arr) /\ (iterateHelper result.nextVal))
310313
311314-- | Repeat a value - limited to ~10,000 elements for stack safety
312315repeatStrom :: forall ctx err a. a -> Strom ctx err a
@@ -321,7 +324,7 @@ repeatStromInfinite a = repeatHelper unit
321324 -- Async boundary resets the stack!
322325 liftAff $ delay (Milliseconds 0.0)
323326 let chunk = Array.replicate 10000 a
324− pure $ Loop $ Tuple (Just chunk) (repeatHelper unit)
327+ pure $ Loop $ ((Just chunk) /\ (repeatHelper unit))
325328
326329-- | Repeat an Om computation - limited to 100 iterations for stack safety
327330-- | For truly infinite repetition, use `repeatOmStromInfinite` (with Aff overhead)
@@ -333,7 +336,7 @@ repeatOmStrom om = repeatOmHelper 0
333336 | count >= maxCount = empty
334337 | otherwise = mkStrom do
335338 value <- om
336− pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper (count + 1))
339+ pure $ Loop $ ((Just [ value ]) /\ (repeatOmHelper (count + 1)))
337340
338341-- | Truly infinite Om repetition - stack-safe via Aff async boundaries
339342-- | Adds tiny delay (0ms) at each pull to reset the stack
@@ -344,11 +347,11 @@ repeatOmStromInfinite om = repeatOmHelper unit
344347 -- Async boundary resets the stack!
345348 liftAff $ delay (Milliseconds 0.0)
346349 value <- om
347− pure $ Loop $ Tuple (Just [ value ]) (repeatOmHelper unit)
350+ pure $ Loop $ ((Just [ value ]) /\ (repeatOmHelper unit))
348351
349352-- | Unfold a stream from a seed value - can be infinite if f never returns Nothing
350353-- | 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
354+unfoldStrom :: forall ctx err a b. (b -> Maybe ((a /\ b))) -> b -> Strom ctx err a
352355unfoldStrom f seed = unfoldHelper seed
353356 where
354357 unfoldHelper currentSeed = mkStrom do
@@ -357,7 +360,7 @@ unfoldStrom f seed = unfoldHelper seed
357360 Nothing -> pure $ Done $ if Array.null chunk.values then Nothing else Just chunk.values
358361 Just nextSeed ->
359362 if Array.null chunk.values then runStrom (unfoldHelper nextSeed)
360− else pure $ Loop $ Tuple (Just chunk.values) (unfoldHelper nextSeed)
363+ else pure $ Loop $ ((Just chunk.values) /\ (unfoldHelper nextSeed))
361364
362365 buildChunk seed' size = ST.run do
363366 arr <- STArray.unsafeThaw []
@@ -370,7 +373,7 @@ unfoldStrom f seed = unfoldHelper seed
370373 s <- STRef.read seedRef
371374 case f s of
372375 Nothing -> void $ STRef.write true stoppedRef
373− Just (Tuple value nextS) -> do
376+ Just ((value /\ nextS)) -> do
374377 void $ STArray.push value arr
375378 void $ STRef.write nextS seedRef
376379 frozen <- STArray.unsafeFreeze arr
@@ -379,14 +382,14 @@ unfoldStrom f seed = unfoldHelper seed
379382 pure { values: frozen, finalSeed: if stopped then Nothing else Just finalSeed }
380383
381384-- | 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
385+unfoldOmStrom :: forall ctx err a b. (b -> Om ctx err (Maybe ((a /\ b)))) -> b -> Strom ctx err a
383386unfoldOmStrom f seed = mkStrom go
384387 where
385388 go = do
386389 result <- f seed
387390 case result of
388391 Nothing -> pure $ Done Nothing
389− Just (Tuple value nextSeed) -> pure $ Loop $ Tuple (Just [ value ]) (unfoldOmStrom f nextSeed)
392+ Just ((value /\ nextSeed)) -> pure $ Loop $ ((Just [ value ]) /\ (unfoldOmStrom f nextSeed))
390393
391394--------------------------------------------------------------------------------
392395-- Transformations
@@ -399,7 +402,7 @@ mapStrom f stream = mkStrom do
399402 case step of
400403 Done Nothing -> pure $ Done Nothing
401404 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)
405+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ ((Functor.map (Functor.map f) maybeChunk) /\ (mapStrom f next))
403406
404407-- | Map with a monadic effect
405408mapMStrom :: forall ctx err a b. (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
@@ -410,11 +413,11 @@ mapMStrom f stream = mkStrom do
410413 Done (Just chunk) -> do
411414 mapped <- traverse f chunk
412415 pure $ Done $ Just mapped
413− Loop (Tuple maybeChunk next) -> do
416+ Loop ((maybeChunk /\ next)) -> do
414417 mappedChunk <- case maybeChunk of
415418 Nothing -> pure Nothing
416419 Just chunk -> Just <$> traverse f chunk
417− pure $ Loop $ Tuple mappedChunk (mapMStrom f next)
420+ pure $ Loop $ (mappedChunk /\ (mapMStrom f next))
418421
419422-- | Tap (observe without modifying)
420423tapStrom :: forall ctx err a. (a -> Unit) -> Strom ctx err a -> Strom ctx err a
@@ -425,16 +428,15 @@ tapStrom f stream = mkStrom do
425428 Done (Just chunk) -> do
426429 let _ = Functor.map f chunk
427430 pure $ Done $ Just chunk
428− Loop (Tuple maybeChunk next) -> do
431+ Loop ((maybeChunk /\ next)) -> do
429432 let
430433 tappedChunk = case maybeChunk of
431434 Nothing -> Nothing
432435 Just chunk ->
433− let
434− _ = Functor.map f chunk
435− in
436+ do
437+ let _ = Functor.map f chunk
436438 Just chunk
437− pure $ Loop $ Tuple tappedChunk (tapStrom f next)
439+ pure $ Loop $ (tappedChunk /\ (tapStrom f next))
438440
439441-- | Tap with monadic effect (observe with effects)
440442tapMStrom :: forall ctx err a. (a -> Om ctx err Unit) -> Strom ctx err a -> Strom ctx err a
@@ -445,13 +447,13 @@ tapMStrom f stream = mkStrom do
445447 Done (Just chunk) -> do
446448 _ <- traverse f chunk
447449 pure $ Done $ Just chunk
448− Loop (Tuple maybeChunk next) -> do
450+ Loop ((maybeChunk /\ next)) -> do
449451 tappedChunk <- case maybeChunk of
450452 Nothing -> pure Nothing
451453 Just chunk -> do
452454 _ <- traverse f chunk
453455 pure $ Just chunk
454− pure $ Loop $ Tuple tappedChunk (tapMStrom f next)
456+ pure $ Loop $ (tappedChunk /\ (tapMStrom f next))
455457
456458-- | Bind for streams (monadic flatMap/chain)
457459-- | Use >>= operator for cleaner syntax: `stream >>= f`
@@ -469,35 +471,33 @@ scanStrom f initial stream = scanHelper initial stream
469471 Done (Just chunk) -> do
470472 -- Build results by folding and collecting intermediate values
471473 let
472− Tuple _ results = Array.foldl
473− ( \(Tuple prevAcc accList) item ->
474− let
475− newAcc = f prevAcc item
476− in
477− Tuple newAcc (Array.snoc accList newAcc)
474+ (_ /\ results) = Array.foldl
475+ ( \((prevAcc /\ accList)) item ->
476+ do
477+ let newAcc = f prevAcc item
478+ (newAcc /\ (Array.snoc accList newAcc))
478479 )
479− (Tuple acc [])
480+ (acc /\ [])
480481 chunk
481482 pure $ Done $ if Array.null results then Nothing else Just results
482− Loop (Tuple maybeChunk next) ->
483+ Loop ((maybeChunk /\ next)) ->
483484 case maybeChunk of
484− Nothing -> pure $ Loop $ Tuple Nothing (scanHelper acc next)
485+ Nothing -> pure $ Loop $ (Nothing /\ (scanHelper acc next))
485486 Just chunk -> do
486487 let
487− Tuple newAcc results = Array.foldl
488− ( \(Tuple prevAcc accList) item ->
489− let
490− currAcc = f prevAcc item
491− in
492− Tuple currAcc (Array.snoc accList currAcc)
488+ (newAcc /\ results) = Array.foldl
489+ ( \((prevAcc /\ accList)) item ->
490+ do
491+ let currAcc = f prevAcc item
492+ (currAcc /\ (Array.snoc accList currAcc))
493493 )
494− (Tuple acc [])
494+ (acc /\ [])
495495 chunk
496− if Array.null results then pure $ Loop $ Tuple Nothing (scanHelper newAcc next)
497− else pure $ Loop $ Tuple (Just results) (scanHelper newAcc next)
496+ if Array.null results then pure $ Loop $ (Nothing /\ (scanHelper newAcc next))
497+ else pure $ Loop $ ((Just results) /\ (scanHelper newAcc next))
498498
499499-- | 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
500+mapAccumStrom :: forall ctx err a b s. (s -> a -> (s /\ b)) -> s -> Strom ctx err a -> Strom ctx err b
501501mapAccumStrom f initial stream = mapAccumHelper initial stream
502502 where
503503 mapAccumHelper state s = mkStrom do
@@ -506,32 +506,30 @@ mapAccumStrom f initial stream = mapAccumHelper initial stream
506506 Done Nothing -> pure $ Done Nothing
507507 Done (Just chunk) -> do
508508 let
509− Tuple finalState results = Array.foldl
510− ( \(Tuple st acc) a ->
511− let
512− Tuple newSt b = f st a
513− in
514− Tuple newSt (Array.snoc acc b)
509+ (finalState /\ results) = Array.foldl
510+ ( \((st /\ acc)) a ->
511+ do
512+ let (newSt /\ b) = f st a
513+ (newSt /\ (Array.snoc acc b))
515514 )
516− (Tuple state [])
515+ (state /\ [])
517516 chunk
518517 pure $ Done $ if Array.null results then Nothing else Just results
519− Loop (Tuple maybeChunk next) ->
518+ Loop ((maybeChunk /\ next)) ->
520519 case maybeChunk of
521− Nothing -> pure $ Loop $ Tuple Nothing (mapAccumHelper state next)
520+ Nothing -> pure $ Loop $ (Nothing /\ (mapAccumHelper state next))
522521 Just chunk -> do
523522 let
524− Tuple newState results = Array.foldl
525− ( \(Tuple st acc) a ->
526− let
527− Tuple newSt b = f st a
528− in
529− Tuple newSt (Array.snoc acc b)
523+ (newState /\ results) = Array.foldl
524+ ( \((st /\ acc)) a ->
525+ do
526+ let (newSt /\ b) = f st a
527+ (newSt /\ (Array.snoc acc b))
530528 )
531− (Tuple state [])
529+ (state /\ [])
532530 chunk
533− if Array.null results then pure $ Loop $ Tuple Nothing (mapAccumHelper newState next)
534− else pure $ Loop $ Tuple (Just results) (mapAccumHelper newState next)
531+ if Array.null results then pure $ Loop $ (Nothing /\ (mapAccumHelper newState next))
532+ else pure $ Loop $ ((Just results) /\ (mapAccumHelper newState next))
535533
536534-- | Filter elements
537535filterStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -542,16 +540,15 @@ filterStrom predicate stream = mkStrom do
542540 Done (Just chunk) -> do
543541 let filtered = Array.filter predicate chunk
544542 pure $ Done $ if Array.null filtered then Nothing else Just filtered
545− Loop (Tuple maybeChunk next) -> do
543+ Loop ((maybeChunk /\ next)) -> do
546544 let
547545 filteredChunk = case maybeChunk of
548546 Nothing -> Nothing
549547 Just chunk ->
550− let
551− filtered = Array.filter predicate chunk
552− in
548+ do
549+ let filtered = Array.filter predicate chunk
553550 if Array.null filtered then Nothing else Just filtered
554− pure $ Loop $ Tuple filteredChunk (filterStrom predicate next)
551+ pure $ Loop $ (filteredChunk /\ (filterStrom predicate next))
555552
556553--------------------------------------------------------------------------------
557554-- Selection
@@ -573,9 +570,9 @@ takeStrom n stream = takeHelper n stream
573570 numTaken = Array.length taken
574571 if numTaken == 0 then pure $ Done Nothing
575572 else pure $ Done $ Just taken
576− Loop (Tuple maybeChunk next) ->
573+ Loop ((maybeChunk /\ next)) ->
577574 case maybeChunk of
578− Nothing -> pure $ Loop $ Tuple Nothing (takeHelper remaining next)
575+ Nothing -> pure $ Loop $ (Nothing /\ (takeHelper remaining next))
579576 Just chunk -> do
580577 let
581578 chunkLen = Array.length chunk
@@ -584,7 +581,7 @@ takeStrom n stream = takeHelper n stream
584581 newRemaining = remaining - numTaken
585582 if numTaken == 0 then pure $ Done Nothing
586583 else if newRemaining <= 0 then pure $ Done $ Just taken
587− else pure $ Loop $ Tuple (Just taken) (takeHelper newRemaining next)
584+ else pure $ Loop $ ((Just taken) /\ (takeHelper newRemaining next))
588585
589586-- | Take while predicate is true
590587takeWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -596,7 +593,7 @@ takeWhileStrom predicate stream = mkStrom do
596593 let taken = Array.takeWhile predicate chunk
597594 if Array.length taken < Array.length chunk then pure $ Done $ Just taken
598595 else pure $ Done $ Just taken
599− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (takeWhileStrom predicate next)
596+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (takeWhileStrom predicate next))
600597
601598-- | Drop n elements (with proper state tracking across chunks)
602599dropStrom :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
@@ -613,19 +610,17 @@ dropStrom n stream
613610 Done (Just chunk) -> do
614611 let dropped = Array.drop remaining chunk
615612 pure $ Done $ if Array.null dropped then Nothing else Just dropped
616− Loop (Tuple maybeChunk next) ->
613+ Loop ((maybeChunk /\ next)) ->
617614 case maybeChunk of
618− Nothing -> pure $ Loop $ Tuple Nothing (dropHelper remaining next)
615+ Nothing -> pure $ Loop $ (Nothing /\ (dropHelper remaining next))
619616 Just chunk ->
620− let
621− chunkLen = Array.length chunk
622− in
623− if remaining >= chunkLen then pure $ Loop $ Tuple Nothing (dropHelper (remaining - chunkLen) next)
617+ do
618+ let chunkLen = Array.length chunk
619+ if remaining >= chunkLen then pure $ Loop $ (Nothing /\ (dropHelper (remaining - chunkLen) next))
624620 else
625− let
626− dropped = Array.drop remaining chunk
627− in
628− pure $ Loop $ Tuple (Just dropped) next
621+ do
622+ let dropped = Array.drop remaining chunk
623+ pure $ Loop $ ((Just dropped) /\ next)
629624
630625-- | Take until predicate is true (includes the matching element)
631626takeUntilStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -637,12 +632,12 @@ takeUntilStrom predicate stream = mkStrom do
637632 case Array.findIndex predicate chunk of
638633 Just idx -> pure $ Done $ Just $ Array.take (idx + 1) chunk -- Include the matching element
639634 Nothing -> pure $ Done $ Just chunk
640− Loop (Tuple maybeChunk next) ->
635+ Loop ((maybeChunk /\ next)) ->
641636 case maybeChunk of
642− Nothing -> pure $ Loop $ Tuple Nothing (takeUntilStrom predicate next)
637+ Nothing -> pure $ Loop $ (Nothing /\ (takeUntilStrom predicate next))
643638 Just chunk -> case Array.findIndex predicate chunk of
644639 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
640+ Nothing -> pure $ Loop $ ((Just chunk) /\ (takeUntilStrom predicate next)) -- Keep going
646641
647642-- | Drop while predicate is true
648643dropWhileStrom :: forall ctx err a. (a -> Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -657,13 +652,13 @@ dropWhileStrom predicate stream = dropWhileHelper true stream
657652 Done (Just chunk) -> do
658653 let dropped = Array.dropWhile predicate chunk
659654 pure $ Done $ if Array.null dropped then Nothing else Just dropped
660− Loop (Tuple maybeChunk next) ->
655+ Loop ((maybeChunk /\ next)) ->
661656 case maybeChunk of
662− Nothing -> pure $ Loop $ Tuple Nothing (dropWhileHelper stillDropping next)
657+ Nothing -> pure $ Loop $ (Nothing /\ (dropWhileHelper stillDropping next))
663658 Just chunk -> do
664659 let dropped = Array.dropWhile predicate chunk
665− if Array.null dropped then pure $ Loop $ Tuple Nothing (dropWhileHelper true next)
666− else pure $ Loop $ Tuple (Just dropped) next -- Found elements to keep, stop dropping
660+ if Array.null dropped then pure $ Loop $ (Nothing /\ (dropWhileHelper true next))
661+ else pure $ Loop $ ((Just dropped) /\ next) -- Found elements to keep, stop dropping
667662
668663-- | Filter with monadic predicate
669664filterMStrom :: forall ctx err a. (a -> Om ctx err Boolean) -> Strom ctx err a -> Strom ctx err a
@@ -674,13 +669,13 @@ filterMStrom predicate stream = mkStrom do
674669 Done (Just chunk) -> do
675670 filtered <- Array.filterA predicate chunk
676671 pure $ Done $ if Array.null filtered then Nothing else Just filtered
677− Loop (Tuple maybeChunk next) -> do
672+ Loop ((maybeChunk /\ next)) -> do
678673 filteredChunk <- case maybeChunk of
679674 Nothing -> pure Nothing
680675 Just chunk -> do
681676 filtered <- Array.filterA predicate chunk
682677 pure $ if Array.null filtered then Nothing else Just filtered
683− pure $ Loop $ Tuple filteredChunk (filterMStrom predicate next)
678+ pure $ Loop $ (filteredChunk /\ (filterMStrom predicate next))
684679
685680-- | Collect with partial function (mapMaybe)
686681collectStrom :: forall ctx err a b. (a -> Maybe b) -> Strom ctx err a -> Strom ctx err b
@@ -691,17 +686,15 @@ collectStrom f stream = mkStrom do
691686 Done (Just chunk) -> do
692687 let collected = Array.mapMaybe f chunk
693688 pure $ Done $ if Array.null collected then Nothing else Just collected
694− Loop (Tuple maybeChunk next) ->
695− let
696− collected = case maybeChunk of
697− Nothing -> Nothing
698− Just chunk ->
699− let
700− c = Array.mapMaybe f chunk
701− in
702− if Array.null c then Nothing else Just c
703− in
704− pure $ Loop $ Tuple collected (collectStrom f next)
689+ Loop ((maybeChunk /\ next)) ->
690+ do
691+ let collected = case maybeChunk of
692+ Nothing -> Nothing
693+ Just chunk -> do
694+ let c = Array.mapMaybe f chunk
695+ if Array.null c then Nothing else Just c
696+
697+ pure $ Loop $ (collected /\ (collectStrom f next))
705698
706699-- | Collect with monadic partial function
707700collectMStrom :: forall ctx err a b. (a -> Om ctx err (Maybe b)) -> Strom ctx err a -> Strom ctx err b
@@ -713,14 +706,14 @@ collectMStrom f stream = mkStrom do
713706 maybes <- traverse f chunk
714707 let collected = Array.mapMaybe identity maybes
715708 pure $ Done $ if Array.null collected then Nothing else Just collected
716− Loop (Tuple maybeChunk next) -> do
709+ Loop ((maybeChunk /\ next)) -> do
717710 collected <- case maybeChunk of
718711 Nothing -> pure Nothing
719712 Just chunk -> do
720713 maybes <- traverse f chunk
721714 let c = Array.mapMaybe identity maybes
722715 pure $ if Array.null c then Nothing else Just c
723− pure $ Loop $ Tuple collected (collectMStrom f next)
716+ pure $ Loop $ (collected /\ (collectMStrom f next))
724717
725718-- | Remove consecutive duplicates
726719changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a
@@ -732,28 +725,28 @@ changesStrom stream = changesHelper Nothing stream
732725 Done Nothing -> pure $ Done Nothing
733726 Done (Just chunk) -> do
734727 let
735− Tuple _ result = Array.foldl
736− ( \(Tuple prev acc) curr ->
737− if Just curr == prev then Tuple (Just curr) acc
738− else Tuple (Just curr) (Array.snoc acc curr)
728+ (_ /\ result) = Array.foldl
729+ ( \((prev /\ acc)) curr ->
730+ if Just curr == prev then ((Just curr) /\ acc)
731+ else ((Just curr) /\ (Array.snoc acc curr))
739732 )
740− (Tuple lastSeen [])
733+ (lastSeen /\ [])
741734 chunk
742735 pure $ Done $ if Array.null result then Nothing else Just result
743− Loop (Tuple maybeChunk next) ->
736+ Loop ((maybeChunk /\ next)) ->
744737 case maybeChunk of
745− Nothing -> pure $ Loop $ Tuple Nothing (changesHelper lastSeen next)
738+ Nothing -> pure $ Loop $ (Nothing /\ (changesHelper lastSeen next))
746739 Just chunk -> do
747740 let
748− Tuple newLast result = Array.foldl
749− ( \(Tuple prev acc) curr ->
750− if Just curr == prev then Tuple (Just curr) acc
751− else Tuple (Just curr) (Array.snoc acc curr)
741+ (newLast /\ result) = Array.foldl
742+ ( \((prev /\ acc)) curr ->
743+ if Just curr == prev then ((Just curr) /\ acc)
744+ else ((Just curr) /\ (Array.snoc acc curr))
752745 )
753− (Tuple lastSeen [])
746+ (lastSeen /\ [])
754747 chunk
755− if Array.null result then pure $ Loop $ Tuple Nothing (changesHelper newLast next)
756− else pure $ Loop $ Tuple (Just result) (changesHelper newLast next)
748+ if Array.null result then pure $ Loop $ (Nothing /\ (changesHelper newLast next))
749+ else pure $ Loop $ ((Just result) /\ (changesHelper newLast next))
757750
758751--------------------------------------------------------------------------------
759752-- Grouping
@@ -785,9 +778,9 @@ groupedStrom size stream
785778 Nothing -> []
786779 if Array.null remainder then pure $ Done $ if Array.null (completeGroups <> lastGroup) then Nothing else Just (completeGroups <> lastGroup)
787780 else pure $ Done $ if Array.null (completeGroups <> [ remainder ]) then Nothing else Just (completeGroups <> [ remainder ])
788− Loop (Tuple maybeChunk next) ->
781+ Loop ((maybeChunk /\ next)) ->
789782 case maybeChunk of
790− Nothing -> pure $ Loop $ Tuple Nothing (groupedHelper buffer next)
783+ Nothing -> pure $ Loop $ (Nothing /\ (groupedHelper buffer next))
791784 Just chunk -> do
792785 let
793786 combined = buffer <> chunk
@@ -796,17 +789,16 @@ groupedStrom size stream
796789 newBuffer = case Array.last groups of
797790 Just g -> if Array.length g < size then g else []
798791 Nothing -> []
799− if Array.null completeGroups then pure $ Loop $ Tuple Nothing (groupedHelper newBuffer next)
800− else pure $ Loop $ Tuple (Just completeGroups) (groupedHelper newBuffer next)
792+ if Array.null completeGroups then pure $ Loop $ (Nothing /\ (groupedHelper newBuffer next))
793+ else pure $ Loop $ ((Just completeGroups) /\ (groupedHelper newBuffer next))
801794
802795 groupsOf :: forall b. Int -> Array b -> Array (Array b)
803796 groupsOf n arr
804797 | n <= 0 || Array.null arr = []
805798 | otherwise =
806− let
807− group = Array.take n arr
808− rest = Array.drop n arr
809− in
799+ do
800+ let group = Array.take n arr
801+ rest = Array.drop n arr
810802 if Array.null group then [] else Array.cons group (groupsOf n rest)
811803
812804-- | Alias for groupedStrom
@@ -825,67 +817,62 @@ groupByStrom keyFn stream = groupByHelper Nothing [] stream
825817 pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
826818 Done (Just chunk) -> do
827819 let
828− Tuple _ result = Array.foldl
829− ( \(Tuple currentKey acc) item ->
830− let
831− itemKey = keyFn item
832− in
820+ (_ /\ result) = Array.foldl
821+ ( \((currentKey /\ acc)) item ->
822+ do
823+ let itemKey = keyFn item
833824 case currentKey of
834− Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
825+ Nothing -> ((Just itemKey) /\ { groups: [], buffer: [ item ] })
835826 Just k ->
836− if k == itemKey then Tuple (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 ] }
827+ if k == itemKey then ((Just itemKey) /\ { groups: acc.groups, buffer: Array.snoc acc.buffer item })
828+ else ((Just itemKey) /\ { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] })
838829 )
839− (Tuple lastKey { groups: [], buffer })
830+ (lastKey /\ { groups: [], buffer })
840831 chunk
841832 allGroups =
842833 if Array.null result.buffer then result.groups
843834 else Array.snoc result.groups result.buffer
844835 pure $ Done $ if Array.null allGroups then Nothing else Just allGroups
845− Loop (Tuple maybeChunk next) ->
836+ Loop ((maybeChunk /\ next)) ->
846837 case maybeChunk of
847− Nothing -> pure $ Loop $ Tuple Nothing (groupByHelper lastKey buffer next)
838+ Nothing -> pure $ Loop $ (Nothing /\ (groupByHelper lastKey buffer next))
848839 Just chunk -> do
849840 let
850− Tuple newKey result = Array.foldl
851− ( \(Tuple currentKey acc) item ->
852− let
853− itemKey = keyFn item
854− in
841+ (newKey /\ result) = Array.foldl
842+ ( \((currentKey /\ acc)) item ->
843+ do
844+ let itemKey = keyFn item
855845 case currentKey of
856− Nothing -> Tuple (Just itemKey) { groups: [], buffer: [ item ] }
846+ Nothing -> ((Just itemKey) /\ { groups: [], buffer: [ item ] })
857847 Just k ->
858− if k == itemKey then Tuple (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 ] }
848+ if k == itemKey then ((Just itemKey) /\ { groups: acc.groups, buffer: Array.snoc acc.buffer item })
849+ else ((Just itemKey) /\ { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] })
860850 )
861− (Tuple lastKey { groups: [], buffer })
851+ (lastKey /\ { groups: [], buffer })
862852 chunk
863− if Array.null result.groups then pure $ Loop $ Tuple Nothing (groupByHelper newKey result.buffer next)
864− else pure $ Loop $ Tuple (Just result.groups) (groupByHelper newKey result.buffer next)
853+ if Array.null result.groups then pure $ Loop $ (Nothing /\ (groupByHelper newKey result.buffer next))
854+ else pure $ Loop $ ((Just result.groups) /\ (groupByHelper newKey result.buffer next))
865855
866856-- | Partition a stream by a predicate into two separate operations
867857-- | Returns a tuple of (trues, falses)
868858-- | 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))
870860partition predicate stream =
871− Tuple
872− (filterStrom predicate stream)
873− (filterStrom (not <<< predicate) stream)
861+ (filterStrom predicate stream)
862+ /\ (filterStrom (not <<< predicate) stream)
874863
875864-- | 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))
877866partitionMap f stream =
878− let
879− leftOnly x = case f x of
880− Left b -> Just b
881− Right _ -> Nothing
882− rightOnly x = case f x of
883− Left _ -> Nothing
884− Right c -> Just c
885− in
886− Tuple
887− (collectStrom leftOnly stream)
888− (collectStrom rightOnly stream)
867+ do
868+ let leftOnly x = case f x of
869+ Left b -> Just b
870+ Right _ -> Nothing
871+ rightOnly x = case f x of
872+ Left _ -> Nothing
873+ Right c -> Just c
874+ (collectStrom leftOnly stream)
875+ /\ (collectStrom rightOnly stream)
889876
890877--------------------------------------------------------------------------------
891878-- Timing
@@ -901,12 +888,12 @@ debounce duration stream = mkStrom do
901888 Done (Just chunk) -> do
902889 Om.delay duration
903890 pure $ Done $ Just chunk
904− Loop (Tuple maybeChunk next) -> do
891+ Loop ((maybeChunk /\ next)) -> do
905892 case maybeChunk of
906− Nothing -> pure $ Loop $ Tuple Nothing (debounce duration next)
893+ Nothing -> pure $ Loop $ (Nothing /\ (debounce duration next))
907894 Just chunk -> do
908895 Om.delay duration
909− pure $ Loop $ Tuple (Just chunk) (debounce duration next)
896+ pure $ Loop $ ((Just chunk) /\ (debounce duration next))
910897
911898-- | Throttle a stream - emit at most once per duration period
912899-- | Tracks last emission time and skips elements that arrive too quickly
@@ -919,31 +906,31 @@ throttle (Milliseconds duration) stream = throttleWithState 0.0 stream
919906 case step of
920907 Done Nothing -> pure $ Done Nothing
921908 Done (Just chunk) -> do
922− Tuple newLastEmit filtered <- filterByTime lastEmit chunk
909+ (newLastEmit /\ filtered) <- filterByTime lastEmit chunk
923910 pure $ Done $ if Array.null filtered then Nothing else Just filtered
924− Loop (Tuple maybeChunk next) -> do
911+ Loop ((maybeChunk /\ next)) -> do
925912 case maybeChunk of
926− Nothing -> pure $ Loop $ Tuple Nothing (throttleWithState lastEmit next)
913+ Nothing -> pure $ Loop $ (Nothing /\ (throttleWithState lastEmit next))
927914 Just chunk -> do
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)
915+ (newLastEmit /\ filtered) <- filterByTime lastEmit chunk
916+ if Array.null filtered then pure $ Loop $ (Nothing /\ (throttleWithState newLastEmit next))
917+ else pure $ Loop $ ((Just filtered) /\ (throttleWithState newLastEmit next))
931918
932919 filterByTime lastEmit chunk = do
933− Tuple newLastEmit results <- Array.foldl
920+ (newLastEmit /\ results) <- Array.foldl
934921 ( \acc value -> do
935− Tuple currentLast filtered <- acc
922+ (currentLast /\ filtered) <- acc
936923 currentTime <- liftEffect now
937924 let Milliseconds currentMillis = unInstant currentTime
938925 let elapsed = currentMillis - currentLast
939926 if elapsed >= duration then
940− pure $ Tuple currentMillis (Array.snoc filtered value)
927+ pure $ (currentMillis /\ (Array.snoc filtered value))
941928 else
942− pure $ Tuple currentLast filtered
929+ pure $ (currentLast /\ filtered)
943930 )
944− (pure $ Tuple lastEmit [])
931+ (pure $ (lastEmit /\ []))
945932 chunk
946− pure $ Tuple newLastEmit results
933+ pure $ (newLastEmit /\ results)
947934
948935-- | Delay each element by a duration
949936delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
@@ -957,33 +944,33 @@ delayStrom duration stream = mkStrom do
957944
958945-- | Fold over all elements
959946runFold :: 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)
947+runFold initial f stream = tailRecM go ((initial /\ stream))
961948 where
962− go (Tuple acc s) = do
949+ go ((acc /\ s)) = do
963950 step <- runStrom s
964951 case step of
965952 Done Nothing -> pure $ Done acc
966953 Done (Just chunk) -> pure $ Done $ foldl f acc chunk
967− Loop (Tuple maybeChunk next) ->
954+ Loop ((maybeChunk /\ next)) ->
968955 -- Process chunk if present and continue
969956 case maybeChunk of
970− Nothing -> pure $ Loop $ Tuple acc next
971− Just chunk -> pure $ Loop $ Tuple (foldl f acc chunk) next
957+ Nothing -> pure $ Loop $ (acc /\ next)
958+ Just chunk -> pure $ Loop $ ((foldl f acc chunk) /\ next)
972959
973960-- | Collect all elements into an array (O(n) using chunk accumulation)
974961runCollect :: forall ctx err a. Strom ctx err a -> Om ctx err (Array a)
975−runCollect stream = tailRecM go (Tuple List.Nil stream)
962+runCollect stream = tailRecM go (List.Nil /\ stream)
976963 where
977− go (Tuple chunkList s) = do
964+ go ((chunkList /\ s)) = do
978965 step <- runStrom s
979966 case step of
980967 Done Nothing -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse chunkList
981968 Done (Just chunk) -> pure $ Done $ Array.concat $ Array.fromFoldable $ List.reverse $ List.Cons chunk chunkList
982− Loop (Tuple maybeChunk next) ->
969+ Loop ((maybeChunk /\ next)) ->
983970 -- Emit the chunk (if present) and continue
984971 case maybeChunk of
985− Nothing -> pure $ Loop $ Tuple chunkList next
986− Just chunk -> pure $ Loop $ Tuple (List.Cons chunk chunkList) next
972+ Nothing -> pure $ Loop $ (chunkList /\ next)
973+ Just chunk -> pure $ Loop $ ((List.Cons chunk chunkList) /\ next)
987974
988975-- | Run the stream and discard all elements
989976runDrain :: forall ctx err a. Strom ctx err a -> Om ctx err Unit
@@ -1000,7 +987,7 @@ traverseStrom_ f stream = tailRecM go stream
1000987 Done (Just chunk) -> do
1001988 _ <- traverse f chunk
1002989 pure $ Done unit
1003− Loop (Tuple maybeChunk next) -> do
990+ Loop ((maybeChunk /\ next)) -> do
1004991 -- Process chunk if present, then continue
1005992 case maybeChunk of
1006993 Nothing -> pure unit
@@ -1060,8 +1047,8 @@ appendStrom toAppend baseStream = mkStrom do
10601047 step <- runStrom baseStream
10611048 case step of
10621049 Done Nothing -> runStrom toAppend
1063− Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) toAppend -- Emit chunk and continue with toAppend
1064− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (appendStrom toAppend next)
1050+ Done (Just chunk) -> pure $ Loop $ ((Just chunk) /\ toAppend) -- Emit chunk and continue with toAppend
1051+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (appendStrom toAppend next))
10651052
10661053-- | Concatenate an array of streams
10671054-- | Uses foldr with flip to match appendStrom's pipeline-friendly parameter order
@@ -1069,8 +1056,8 @@ concatStrom :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
10691056concatStrom streams = Array.foldr (flip appendStrom) empty streams
10701057
10711058-- | 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)
1073−zipStrom = zipWithStrom Tuple
1059+zipStrom :: forall ctx err a b. Strom ctx err a -> Strom ctx err b -> Strom ctx err ((a /\ b))
1060+zipStrom = zipWithStrom (/\)
10741061
10751062-- | Zip two streams with a function
10761063zipWithStrom :: forall ctx err a b c. (a -> b -> c) -> Strom ctx err a -> Strom ctx err b -> Strom ctx err c
@@ -1083,13 +1070,13 @@ zipWithStrom f s1 s2 = mkStrom do
10831070 pure $ Done $ Just zipped
10841071 Done Nothing, _ -> pure $ Done Nothing
10851072 _, Done Nothing -> pure $ Done Nothing
1086− Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1073+ Loop ((maybeChunk1 /\ next1)), Loop ((maybeChunk2 /\ next2)) ->
10871074 -- Both looping: zip chunks if both present, continue with both nexts
10881075 case maybeChunk1, maybeChunk2 of
1089− Just chunk1, Just chunk2 -> pure $ Loop $ Tuple (Just $ Array.zipWith f chunk1 chunk2) (zipWithStrom f next1 next2)
1090− _, _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 next2)
1091− Loop (Tuple _ next1), _ -> pure $ Loop $ Tuple Nothing (zipWithStrom f next1 s2)
1092− _, Loop (Tuple _ next2) -> pure $ Loop $ Tuple Nothing (zipWithStrom f s1 next2)
1076+ Just chunk1, Just chunk2 -> pure $ Loop $ ((Just $ Array.zipWith f chunk1 chunk2) /\ (zipWithStrom f next1 next2))
1077+ _, _ -> pure $ Loop $ (Nothing /\ (zipWithStrom f next1 next2))
1078+ Loop ((_ /\ next1)), _ -> pure $ Loop $ (Nothing /\ (zipWithStrom f next1 s2))
1079+ _, Loop ((_ /\ next2)) -> pure $ Loop $ (Nothing /\ (zipWithStrom f s1 next2))
10931080
10941081-- | Interleave two streams - alternates between elements from each stream
10951082-- | Similar to merge but deterministic - always alternates
@@ -1101,16 +1088,16 @@ interleave s1 s2 = interleaveHelper true s1 s2
11011088 step1 <- runStrom first
11021089 case step1 of
11031090 Done Nothing -> runStrom second
1104− Done (Just chunk1) -> pure $ Loop $ Tuple (Just chunk1) (interleaveHelper false second first)
1105− Loop (Tuple maybeChunk1 next1) ->
1106− pure $ Loop $ Tuple maybeChunk1 (interleaveHelper false second next1)
1091+ Done (Just chunk1) -> pure $ Loop $ ((Just chunk1) /\ (interleaveHelper false second first))
1092+ Loop ((maybeChunk1 /\ next1)) ->
1093+ pure $ Loop $ (maybeChunk1 /\ (interleaveHelper false second next1))
11071094 else do
11081095 step2 <- runStrom second
11091096 case step2 of
11101097 Done Nothing -> runStrom first
1111− Done (Just chunk2) -> pure $ Loop $ Tuple (Just chunk2) (interleaveHelper true first second)
1112− Loop (Tuple maybeChunk2 next2) ->
1113− pure $ Loop $ Tuple maybeChunk2 (interleaveHelper true first next2)
1098+ Done (Just chunk2) -> pure $ Loop $ ((Just chunk2) /\ (interleaveHelper true first second))
1099+ Loop ((maybeChunk2 /\ next2)) ->
1100+ pure $ Loop $ (maybeChunk2 /\ (interleaveHelper true first next2))
11141101
11151102-- | Intersperse a separator element between each element of the stream
11161103intersperse :: forall ctx err a. a -> Strom ctx err a -> Strom ctx err a
@@ -1126,15 +1113,15 @@ intersperse separator stream = intersperseHelper true stream
11261113 if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
11271114 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
11281115 pure $ Done $ Just withSeparators
1129− Loop (Tuple maybeChunk next) ->
1116+ Loop ((maybeChunk /\ next)) ->
11301117 case maybeChunk of
1131− Nothing -> pure $ Loop $ Tuple Nothing (intersperseHelper isFirst next)
1118+ Nothing -> pure $ Loop $ (Nothing /\ (intersperseHelper isFirst next))
11321119 Just chunk -> do
11331120 let
11341121 withSeparators =
11351122 if isFirst then Array.intercalate [ separator ] (Array.singleton <$> chunk)
11361123 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1137− pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
1124+ pure $ Loop $ ((Just withSeparators) /\ (intersperseHelper false next))
11381125
11391126--------------------------------------------------------------------------------
11401127-- Buffering
@@ -1175,7 +1162,7 @@ bufferChunks capacity stream
11751162 Nothing -> do
11761163 AVar.put unit mutex
11771164 dequeue
1178− Just (Tuple head tail) -> do
1165+ Just ((head /\ tail)) -> do
11791166 liftEffect $ Ref.write tail bufferRef
11801167 liftEffect $ Ref.write (size - 1) sizeRef
11811168 let hasMore = (size - 1) > 0
@@ -1193,7 +1180,7 @@ bufferChunks capacity stream
11931180 Done (Just chunk) -> do
11941181 enqueue (BufferChunk chunk)
11951182 enqueue BufferDone
1196− Loop (Tuple maybeChunk next) -> do
1183+ Loop ((maybeChunk /\ next)) -> do
11971184 case maybeChunk of
11981185 Just chunk -> enqueue (BufferChunk chunk)
11991186 Nothing -> pure unit
@@ -1209,7 +1196,7 @@ bufferChunks capacity stream
12091196 killFiber (Exception.error "done") fiber
12101197 pure $ Done Nothing
12111198 BufferChunk chunk ->
1212− pure $ Loop $ Tuple (Just chunk) consumer
1199+ pure $ Loop $ ((Just chunk) /\ consumer)
12131200
12141201 stepResult <- Om.runReader ctx (runStrom consumer)
12151202 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1249,7 +1236,7 @@ buffer capacity stream
12491236 Nothing -> do
12501237 AVar.put unit mutex
12511238 dequeue
1252− Just (Tuple head tail) -> do
1239+ Just ((head /\ tail)) -> do
12531240 liftEffect $ Ref.write tail bufferRef
12541241 liftEffect $ Ref.write (size - 1) sizeRef
12551242 let hasMore = (size - 1) > 0
@@ -1267,7 +1254,7 @@ buffer capacity stream
12671254 Done (Just chunk) -> do
12681255 traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
12691256 enqueue BufferDone
1270− Loop (Tuple maybeChunk next) -> do
1257+ Loop ((maybeChunk /\ next)) -> do
12711258 case maybeChunk of
12721259 Just chunk -> traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
12731260 Nothing -> pure unit
@@ -1283,7 +1270,7 @@ buffer capacity stream
12831270 killFiber (Exception.error "done") fiber
12841271 pure $ Done Nothing
12851272 BufferChunk chunk ->
1286− pure $ Loop $ Tuple (Just chunk) consumer
1273+ pure $ Loop $ ((Just chunk) /\ consumer)
12871274
12881275 stepResult <- Om.runReader ctx (runStrom consumer)
12891276 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1392,7 +1379,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
13921379 Nothing -> do
13931380 AVar.put unit mutex
13941381 dequeue
1395− Just (Tuple head tail) -> do
1382+ Just ((head /\ tail)) -> do
13961383 liftEffect $ Ref.write tail bufferRef
13971384 liftEffect $ Ref.write (size - 1) sizeRef
13981385 let hasMore = (size - 1) > 0
@@ -1414,7 +1401,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
14141401 Done (Just chunk) -> do
14151402 enqueue (DataChunk chunk StreamId1)
14161403 enqueue (StreamDone StreamId1)
1417− Loop (Tuple maybeChunk next) -> do
1404+ Loop ((maybeChunk /\ next)) -> do
14181405 case maybeChunk of
14191406 Just chunk -> enqueue (DataChunk chunk StreamId1)
14201407 Nothing -> pure unit
@@ -1433,7 +1420,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
14331420 Done (Just chunk) -> do
14341421 enqueue (DataChunk chunk StreamId2)
14351422 enqueue (StreamDone StreamId2)
1436− Loop (Tuple maybeChunk next) -> do
1423+ Loop ((maybeChunk /\ next)) -> do
14371424 case maybeChunk of
14381425 Just chunk -> enqueue (DataChunk chunk StreamId2)
14391426 Nothing -> pure unit
@@ -1460,7 +1447,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
14601447 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
14611448 DataChunk chunk _streamId ->
14621449 -- Got a chunk
1463− pure $ Loop $ Tuple (Just chunk) (consumer doneLeft doneRight)
1450+ pure $ Loop $ ((Just chunk) /\ (consumer doneLeft doneRight))
14641451
14651452 stepResult <- Om.runReader ctx (runStrom $ consumer false false)
14661453 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1505,7 +1492,7 @@ mergeAllNDWith bufferCapacity streams
15051492 Nothing -> do
15061493 AVar.put unit mutex
15071494 dequeue
1508− Just (Tuple head tail) -> do
1495+ Just ((head /\ tail)) -> do
15091496 liftEffect $ Ref.write tail bufferRef
15101497 liftEffect $ Ref.write (size - 1) sizeRef
15111498 let hasMore = (size - 1) > 0
@@ -1523,7 +1510,7 @@ mergeAllNDWith bufferCapacity streams
15231510 Done (Just chunk) -> do
15241511 enqueue (MergeChunk chunk)
15251512 enqueue MergeDone
1526− Loop (Tuple maybeChunk next) -> do
1513+ Loop ((maybeChunk /\ next)) -> do
15271514 case maybeChunk of
15281515 Just chunk -> enqueue (MergeChunk chunk)
15291516 Nothing -> pure unit
@@ -1544,7 +1531,7 @@ mergeAllNDWith bufferCapacity streams
15441531 stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
15451532 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
15461533 MergeChunk chunk ->
1547− pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1534+ pure $ Loop $ ((Just chunk) /\ (consumer doneCount))
15481535
15491536 stepResult <- Om.runReader ctx (runStrom $ consumer 0)
15501537 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
@@ -1568,23 +1555,22 @@ mapPar concurrency f stream
15681555 let groups = chunksOf concurrency chunk
15691556 results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
15701557 pure $ Done $ Just $ Array.concat results
1571− Loop (Tuple maybeChunk next) -> do
1558+ Loop ((maybeChunk /\ next)) -> do
15721559 mappedChunk <- case maybeChunk of
15731560 Nothing -> pure Nothing
15741561 Just chunk -> do
15751562 let groups = chunksOf concurrency chunk
15761563 results <- traverse (\group -> Om.inParallel (Functor.map f group)) groups
15771564 pure $ Just $ Array.concat results
1578− pure $ Loop $ Tuple mappedChunk (mapPar concurrency f next)
1565+ pure $ Loop $ (mappedChunk /\ (mapPar concurrency f next))
15791566 where
15801567 chunksOf :: forall x. Int -> Array x -> Array (Array x)
15811568 chunksOf n arr
15821569 | n <= 0 || Array.null arr = []
15831570 | otherwise =
1584− let
1585− group = Array.take n arr
1586− rest = Array.drop n arr
1587− in
1571+ do
1572+ let group = Array.take n arr
1573+ rest = Array.drop n arr
15881574 if Array.null group then [] else Array.cons group (chunksOf n rest)
15891575
15901576-- | Unordered parallel map (currently same as mapPar)
@@ -1720,7 +1706,7 @@ instance Functor (Strom ctx err) where
17201706 case step of
17211707 Done Nothing -> pure $ Done Nothing
17221708 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)
1709+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ ((Functor.map (Functor.map f) maybeChunk) /\ (Functor.map f next))
17241710 )
17251711
17261712instance Apply (Strom ctx err) where
@@ -1729,29 +1715,27 @@ instance Apply (Strom ctx err) where
17291715 runStrom as >>= \stepA ->
17301716 case stepF, stepA of
17311717 Done (Just funcs), Done (Just vals) ->
1732− let
1733− results = do
1734− f <- funcs
1735− a <- vals
1736− pure (f a)
1737− in
1718+ do
1719+ let results = do
1720+ f <- funcs
1721+ a <- vals
1722+ pure (f a)
17381723 pure $ Done $ Just results
17391724 Done Nothing, _ -> pure $ Done Nothing
17401725 _, Done Nothing -> pure $ Done Nothing
1741− Loop (Tuple maybeChunkF nextF), Loop (Tuple maybeChunkA nextA) ->
1726+ Loop ((maybeChunkF /\ nextF)), Loop ((maybeChunkA /\ nextA)) ->
17421727 -- Apply chunks if both present
17431728 case maybeChunkF, maybeChunkA of
17441729 Just chunkF, Just chunkA ->
1745− let
1746− applied = do
1747− f <- chunkF
1748− a <- chunkA
1749− pure (f a)
1750− in
1751− pure $ Loop $ Tuple (Just applied) (nextF <*> nextA)
1752− _, _ -> pure $ Loop $ Tuple Nothing (nextF <*> nextA)
1753− Loop (Tuple _ nextF), _ -> pure $ Loop $ Tuple Nothing (nextF <*> as)
1754− _, Loop (Tuple _ nextA) -> pure $ Loop $ Tuple Nothing (fs <*> nextA)
1730+ do
1731+ let applied = do
1732+ f <- chunkF
1733+ a <- chunkA
1734+ pure (f a)
1735+ pure $ Loop $ ((Just applied) /\ (nextF <*> nextA))
1736+ _, _ -> pure $ Loop $ (Nothing /\ (nextF <*> nextA))
1737+ Loop ((_ /\ nextF)), _ -> pure $ Loop $ (Nothing /\ (nextF <*> as))
1738+ _, Loop ((_ /\ nextA)) -> pure $ Loop $ (Nothing /\ (fs <*> nextA))
17551739 )
17561740
17571741instance Applicative (Strom ctx err) where
@@ -1763,19 +1747,17 @@ instance Bind (Strom ctx err) where
17631747 case step of
17641748 Done Nothing -> pure $ Done Nothing
17651749 Done (Just chunk) ->
1766− let
1767− streams = Functor.map f chunk
1768− in
1750+ do
1751+ let streams = Functor.map f chunk
17691752 runStrom $ concatStrom streams
1770− Loop (Tuple maybeChunk next) ->
1753+ Loop ((maybeChunk /\ next)) ->
17711754 -- When looping, if we have a chunk, we need to process it through f and concat
17721755 case maybeChunk of
1773− Nothing -> pure $ Loop $ Tuple Nothing (next >>= f)
1756+ Nothing -> pure $ Loop $ (Nothing /\ (next >>= f))
17741757 Just chunk ->
1775− let
1776− streams = Functor.map f chunk
1777− prefixStream = concatStrom streams
1778− in
1758+ do
1759+ let streams = Functor.map f chunk
1760+ prefixStream = concatStrom streams
17791761 runStrom $ appendStrom prefixStream (next >>= f)
17801762 )
17811763
@@ -1786,8 +1768,8 @@ instance Semigroup (Strom ctx err a) where
17861768 ( runStrom s1 >>= \step ->
17871769 case step of
17881770 Done Nothing -> runStrom s2
1789− Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1790− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <> s2)
1771+ Done (Just chunk) -> pure $ Loop $ ((Just chunk) /\ s2)
1772+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (next <> s2))
17911773 )
17921774
17931775instance Monoid (Strom ctx err a) where
@@ -1798,8 +1780,8 @@ instance Alt (Strom ctx err) where
17981780 ( runStrom s1 >>= \step ->
17991781 case step of
18001782 Done Nothing -> runStrom s2
1801− Done (Just chunk) -> pure $ Loop $ Tuple (Just chunk) s2
1802− Loop (Tuple maybeChunk next) -> pure $ Loop $ Tuple maybeChunk (next <|> s2)
1783+ Done (Just chunk) -> pure $ Loop $ ((Just chunk) /\ s2)
1784+ Loop ((maybeChunk /\ next)) -> pure $ Loop $ (maybeChunk /\ (next <|> s2))
18031785 )
18041786
18051787instance Plus (Strom ctx err) where