@@ -65,6 +65,7 @@ module Yoga.Om.Strom
6565 , partitionMap
6666 -- Timing
6767 , debounce
68+ , debounceEach
6869 , throttle
6970 , delayStrom
7071 -- Combining
@@ -138,12 +139,10 @@ import Effect.Aff (Aff, Error, delay, forkAff, killFiber)
138139import Effect.Aff.AVar as AVar
139140import Effect.Aff.Class (liftAff)
140141import Effect.Exception as Exception
141−import Effect.Now (now)
142142import Effect.Ref as Ref
143143import Effect.Class (liftEffect)
144144import Data.CatQueue as CatQueue
145145import Data.CatQueue (CatQueue)
146−import Data.DateTime.Instant (unInstant)
147146import Yoga.Om (Om)
148147import Yoga.Om as Om
149148
@@ -460,7 +459,7 @@ tapMStrom f stream = mkStrom do
460459bindStrom :: forall ctx err a b. (a -> Strom ctx err b) -> Strom ctx err a -> Strom ctx err b
461460bindStrom f stream = stream >>= f
462461
463−-- | Scan with accumulator (like foldl but emits intermediate results)
462+-- | Scan with accumulator (like foldl but emits intermediate results)
464463scanStrom :: forall ctx err a b. (b -> a -> b) -> b -> Strom ctx err a -> Strom ctx err b
465464scanStrom f initial stream = scanHelper initial stream
466465 where
@@ -687,11 +686,12 @@ collectStrom f stream = mkStrom do
687686 pure $ Done $ if Array.null collected then Nothing else Just collected
688687 Loop ((maybeChunk /\ next)) ->
689688 do
690− let collected = case maybeChunk of
691− Nothing -> Nothing
692− Just chunk -> do
693− let c = Array.mapMaybe f chunk
694− if Array.null c then Nothing else Just c
689+ let
690+ collected = case maybeChunk of
691+ Nothing -> Nothing
692+ Just chunk -> do
693+ let c = Array.mapMaybe f chunk
694+ if Array.null c then Nothing else Just c
695695
696696 pure $ Loop $ (collected /\ (collectStrom f next))
697697
@@ -714,7 +714,7 @@ collectMStrom f stream = mkStrom do
714714 pure $ if Array.null c then Nothing else Just c
715715 pure $ Loop $ (collected /\ (collectMStrom f next))
716716
717−-- | Remove consecutive duplicates
717+-- | Remove consecutive duplicates
718718changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a
719719changesStrom stream = changesHelper Nothing stream
720720 where
@@ -796,8 +796,9 @@ groupedStrom size stream
796796 | n <= 0 || Array.null arr = []
797797 | otherwise =
798798 do
799− let group = Array.take n arr
800− rest = Array.drop n arr
799+ let
800+ group = Array.take n arr
801+ rest = Array.drop n arr
801802 if Array.null group then [] else Array.cons group (groupsOf n rest)
802803
803804-- | Alias for groupedStrom
@@ -864,12 +865,13 @@ partition predicate stream =
864865partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> ((Strom ctx err b) /\ (Strom ctx err c))
865866partitionMap f stream =
866867 do
867− let leftOnly x = case f x of
868− Left b -> Just b
869− Right _ -> Nothing
870− rightOnly x = case f x of
871− Left _ -> Nothing
872− Right c -> Just c
868+ let
869+ leftOnly x = case f x of
870+ Left b -> Just b
871+ Right _ -> Nothing
872+ rightOnly x = case f x of
873+ Left _ -> Nothing
874+ Right c -> Just c
873875 (collectStrom leftOnly stream)
874876 /\ (collectStrom rightOnly stream)
875877
@@ -887,49 +889,24 @@ debounce duration stream = mkStrom do
887889 Done (Just chunk) -> do
888890 Om.delay duration
889891 pure $ Done $ Just chunk
890− Loop ((maybeChunk /\ next)) -> do
892+ Loop (maybeChunk /\ next) -> do
891893 case maybeChunk of
892894 Nothing -> pure $ Loop $ (Nothing /\ (debounce duration next))
893895 Just chunk -> do
894896 Om.delay duration
895897 pure $ Loop $ ((Just chunk) /\ (debounce duration next))
896898
897−-- | Throttle a stream - emit at most once per duration period
898−-- | Tracks last emission time and skips elements that arrive too quickly
899−-- | Note: Uses wall-clock time, not processing time
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
902+ Om.delay duration
903+ pure $ Done $ Just [ a ]
904+
905+-- | Throttle a stream - emits at most one element per duration period
900906throttle :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
901−throttle (Milliseconds duration) stream = throttleWithState 0.0 stream
902− where
903− throttleWithState lastEmit s = mkStrom do
904− step <- runStrom s
905− case step of
906− Done Nothing -> pure $ Done Nothing
907− Done (Just chunk) -> do
908− (_ /\ filtered) <- filterByTime lastEmit chunk
909− pure $ Done $ if Array.null filtered then Nothing else Just filtered
910− Loop ((maybeChunk /\ next)) -> do
911− case maybeChunk of
912− Nothing -> pure $ Loop $ (Nothing /\ (throttleWithState lastEmit next))
913− Just chunk -> do
914− (newLastEmit /\ filtered) <- filterByTime lastEmit chunk
915− if Array.null filtered then pure $ Loop $ (Nothing /\ (throttleWithState newLastEmit next))
916− else pure $ Loop $ ((Just filtered) /\ (throttleWithState newLastEmit next))
917−
918− filterByTime lastEmit chunk = do
919− (newLastEmit /\ results) <- Array.foldl
920− ( \acc value -> do
921− (currentLast /\ filtered) <- acc
922− currentTime <- liftEffect now
923− let Milliseconds currentMillis = unInstant currentTime
924− let elapsed = currentMillis - currentLast
925− if elapsed >= duration then
926− pure $ (currentMillis /\ (Array.snoc filtered value))
927− else
928− pure $ (currentLast /\ filtered)
929− )
930− (pure $ (lastEmit /\ []))
931− chunk
932− pure $ (newLastEmit /\ results)
907+throttle duration stream = stream # bindStrom \a -> mkStrom do
908+ Om.delay duration
909+ pure $ Done $ Just [ a ]
933910
934911-- | Delay each element by a duration
935912delayStrom :: forall ctx err a. Milliseconds -> Strom ctx err a -> Strom ctx err a
@@ -1010,20 +987,20 @@ traverseMStrom_ = traverseStrom_
1010987--------------------------------------------------------------------------------
1011988
1012989-- | Run a stream in the background, returning a Fiber that can be cancelled
1013−-- |
990+-- |
1014991-- | To use cancellable streams, convert your stream to Om and use Om.launchOm:
1015−-- |
992+-- |
1016993-- | Example:
1017994-- | ```purescript
1018995-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)
1019996-- | -- ... later ...
1020997-- | killFiber (error "Cancelled") fiber
1021998-- | ```
1022−-- |
999+-- |
10231000-- | Or for subscription-style:
10241001-- | ```purescript
10251002-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)
1026−-- | -- ... later ...
1003+-- | -- ... later ...
10271004-- | killFiber (error "Unsubscribed") fiber
10281005-- | ```
10291006-- |
@@ -1568,8 +1545,9 @@ mapPar concurrency f stream
15681545 | n <= 0 || Array.null arr = []
15691546 | otherwise =
15701547 do
1571− let group = Array.take n arr
1572− rest = Array.drop n arr
1548+ let
1549+ group = Array.take n arr
1550+ rest = Array.drop n arr
15731551 if Array.null group then [] else Array.cons group (chunksOf n rest)
15741552
15751553-- | Unordered parallel map (currently same as mapPar)
@@ -1715,10 +1693,11 @@ instance Apply (Strom ctx err) where
17151693 case stepF, stepA of
17161694 Done (Just funcs), Done (Just vals) ->
17171695 do
1718− let results = do
1719− f <- funcs
1720− a <- vals
1721− pure (f a)
1696+ let
1697+ results = do
1698+ f <- funcs
1699+ a <- vals
1700+ pure (f a)
17221701 pure $ Done $ Just results
17231702 Done Nothing, _ -> pure $ Done Nothing
17241703 _, Done Nothing -> pure $ Done Nothing
@@ -1727,10 +1706,11 @@ instance Apply (Strom ctx err) where
17271706 case maybeChunkF, maybeChunkA of
17281707 Just chunkF, Just chunkA ->
17291708 do
1730− let applied = do
1731− f <- chunkF
1732− a <- chunkA
1733− pure (f a)
1709+ let
1710+ applied = do
1711+ f <- chunkF
1712+ a <- chunkA
1713+ pure (f a)
17341714 pure $ Loop $ ((Just applied) /\ (nextF <*> nextA))
17351715 _, _ -> pure $ Loop $ (Nothing /\ (nextF <*> nextA))
17361716 Loop ((_ /\ nextF)), _ -> pure $ Loop $ (Nothing /\ (nextF <*> as))
@@ -1755,8 +1735,9 @@ instance Bind (Strom ctx err) where
17551735 Nothing -> pure $ Loop $ (Nothing /\ (next >>= f))
17561736 Just chunk ->
17571737 do
1758− let streams = Functor.map f chunk
1759− prefixStream = concatStrom streams
1738+ let
1739+ streams = Functor.map f chunk
1740+ prefixStream = concatStrom streams
17601741 runStrom $ appendStrom prefixStream (next >>= f)
17611742 )
17621743