@@ -134,7 +134,7 @@ import Data.Time.Duration (Milliseconds(..))
134134import Data.Traversable (traverse, traverse_)
135135import Data.Tuple.Nested (type (/\), (/\))
136136import Data.Variant (Variant)
137−import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
137+import Effect.Aff (Aff, Error, delay, forkAff, killFiber)
138138import Effect.Aff.AVar as AVar
139139import Effect.Aff.Class (liftAff)
140140import Effect.Exception as Exception
@@ -506,7 +506,7 @@ mapAccumStrom f initial stream = mapAccumHelper initial stream
506506 Done Nothing -> pure $ Done Nothing
507507 Done (Just chunk) -> do
508508 let
509− (finalState /\ results) = Array.foldl
509+ (_ /\ results) = Array.foldl
510510 ( \((st /\ acc)) a ->
511511 do
512512 let (newSt /\ b) = f st a
@@ -575,7 +575,6 @@ takeStrom n stream = takeHelper n stream
575575 Nothing -> pure $ Loop $ (Nothing /\ (takeHelper remaining next))
576576 Just chunk -> do
577577 let
578− chunkLen = Array.length chunk
579578 taken = Array.take remaining chunk
580579 numTaken = Array.length taken
581580 newRemaining = remaining - numTaken
@@ -758,16 +757,16 @@ groupedStrom size stream
758757 | size <= 0 = empty
759758 | otherwise = groupedHelper [] stream
760759 where
761− groupedHelper buffer s = mkStrom do
760+ groupedHelper buf s = mkStrom do
762761 step <- runStrom s
763762 case step of
764763 Done Nothing ->
765764 -- End of stream, emit remaining buffer if non-empty
766− pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
765+ pure $ Done $ if Array.null buf then Nothing else Just [ buf ]
767766 Done (Just chunk) -> do
768767 -- Process final chunk with buffer
769768 let
770− combined = buffer <> chunk
769+ combined = buf <> chunk
771770 groups = groupsOf size combined
772771 completeGroups = Array.dropEnd 1 groups
773772 lastGroup = case Array.last groups of
@@ -780,10 +779,10 @@ groupedStrom size stream
780779 else pure $ Done $ if Array.null (completeGroups <> [ remainder ]) then Nothing else Just (completeGroups <> [ remainder ])
781780 Loop ((maybeChunk /\ next)) ->
782781 case maybeChunk of
783− Nothing -> pure $ Loop $ (Nothing /\ (groupedHelper buffer next))
782+ Nothing -> pure $ Loop $ (Nothing /\ (groupedHelper buf next))
784783 Just chunk -> do
785784 let
786− combined = buffer <> chunk
785+ combined = buf <> chunk
787786 groups = groupsOf size combined
788787 completeGroups = if Array.null groups then [] else Array.dropEnd 1 groups
789788 newBuffer = case Array.last groups of
@@ -810,11 +809,11 @@ chunkedStrom = groupedStrom
810809groupByStrom :: forall ctx err a k. Eq k => (a -> k) -> Strom ctx err a -> Strom ctx err (Array a)
811810groupByStrom keyFn stream = groupByHelper Nothing [] stream
812811 where
813− groupByHelper lastKey buffer s = mkStrom do
812+ groupByHelper lastKey buf s = mkStrom do
814813 step <- runStrom s
815814 case step of
816815 Done Nothing ->
817− pure $ Done $ if Array.null buffer then Nothing else Just [ buffer ]
816+ pure $ Done $ if Array.null buf then Nothing else Just [ buf ]
818817 Done (Just chunk) -> do
819818 let
820819 (_ /\ result) = Array.foldl
@@ -827,7 +826,7 @@ groupByStrom keyFn stream = groupByHelper Nothing [] stream
827826 if k == itemKey then ((Just itemKey) /\ { groups: acc.groups, buffer: Array.snoc acc.buffer item })
828827 else ((Just itemKey) /\ { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] })
829828 )
830− (lastKey /\ { groups: [], buffer })
829+ (lastKey /\ { groups: [], buffer: buf })
831830 chunk
832831 allGroups =
833832 if Array.null result.buffer then result.groups
@@ -835,7 +834,7 @@ groupByStrom keyFn stream = groupByHelper Nothing [] stream
835834 pure $ Done $ if Array.null allGroups then Nothing else Just allGroups
836835 Loop ((maybeChunk /\ next)) ->
837836 case maybeChunk of
838− Nothing -> pure $ Loop $ (Nothing /\ (groupByHelper lastKey buffer next))
837+ Nothing -> pure $ Loop $ (Nothing /\ (groupByHelper lastKey buf next))
839838 Just chunk -> do
840839 let
841840 (newKey /\ result) = Array.foldl
@@ -848,7 +847,7 @@ groupByStrom keyFn stream = groupByHelper Nothing [] stream
848847 if k == itemKey then ((Just itemKey) /\ { groups: acc.groups, buffer: Array.snoc acc.buffer item })
849848 else ((Just itemKey) /\ { groups: Array.snoc acc.groups acc.buffer, buffer: [ item ] })
850849 )
851− (lastKey /\ { groups: [], buffer })
850+ (lastKey /\ { groups: [], buffer: buf })
852851 chunk
853852 if Array.null result.groups then pure $ Loop $ (Nothing /\ (groupByHelper newKey result.buffer next))
854853 else pure $ Loop $ ((Just result.groups) /\ (groupByHelper newKey result.buffer next))
@@ -906,7 +905,7 @@ throttle (Milliseconds duration) stream = throttleWithState 0.0 stream
906905 case step of
907906 Done Nothing -> pure $ Done Nothing
908907 Done (Just chunk) -> do
909− (newLastEmit /\ filtered) <- filterByTime lastEmit chunk
908+ (_ /\ filtered) <- filterByTime lastEmit chunk
910909 pure $ Done $ if Array.null filtered then Nothing else Just filtered
911910 Loop ((maybeChunk /\ next)) -> do
912911 case maybeChunk of
@@ -1394,7 +1393,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
13941393 let
13951394 producer s = do
13961395 stepResult <- Om.runReader ctx (runStrom s)
1397− step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1396+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
13981397 case step of
13991398 Done Nothing ->
14001399 enqueue (StreamDone StreamId1)
@@ -1413,7 +1412,7 @@ mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
14131412 let
14141413 producer s = do
14151414 stepResult <- Om.runReader ctx (runStrom s)
1416− step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1415+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
14171416 case step of
14181417 Done Nothing ->
14191418 enqueue (StreamDone StreamId2)