1f1948c · rowtype-yoga/purescript-yoga-om-strom

65 , partitionMap

66 -- Timing

67 , debounce

68+ , debounceEach

68 , throttle

69 , throttle

69 , delayStrom

70 , delayStrom

70 -- Combining

71 -- Combining

138import Effect.Aff.AVar as AVar

139import Effect.Aff.AVar as AVar

139import Effect.Aff.Class (liftAff)

140import Effect.Aff.Class (liftAff)

140import Effect.Exception as Exception

141import Effect.Exception as Exception

141−import Effect.Now (now)

142import Effect.Ref as Ref

143import Effect.Class (liftEffect)

144import Data.CatQueue as CatQueue

145import Data.CatQueue (CatQueue)

146−import Data.DateTime.Instant (unInstant)

147import Yoga.Om (Om)

146import Yoga.Om (Om)

148import Yoga.Om as Om

147import Yoga.Om as Om

149

148

460bindStrom :: 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

461bindStrom f stream = stream >>= f

460bindStrom f stream = stream >>= f

462

461

463−-- | Scan with accumulator (like foldl but emits intermediate results)

462+-- | Scan with accumulator (like foldl but emits intermediate results)

464scanStrom :: 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

465scanStrom f initial stream = scanHelper initial stream

464scanStrom f initial stream = scanHelper initial stream

466 where

465 where

687 pure $ Done $ if Array.null collected then Nothing else Just collected

686 pure $ Done $ if Array.null collected then Nothing else Just collected

688 Loop ((maybeChunk /\ next)) ->

687 Loop ((maybeChunk /\ next)) ->

689 do

688 do

689+ let

690− let collected = case maybeChunk of

690+ collected = case maybeChunk of

691− Nothing -> Nothing

691+ Nothing -> Nothing

692− Just chunk -> do

692+ Just chunk -> do

693− let c = Array.mapMaybe f chunk

693+ let c = Array.mapMaybe f chunk

694− if Array.null c then Nothing else Just c

694+ if Array.null c then Nothing else Just c

695

696 pure $ Loop $ (collected /\ (collectStrom f next))

697

714 pure $ if Array.null c then Nothing else Just c

715 pure $ Loop $ (collected /\ (collectMStrom f next))

716

717−-- | Remove consecutive duplicates

717+-- | Remove consecutive duplicates

718changesStrom :: forall ctx err a. Eq a => Strom ctx err a -> Strom ctx err a

719changesStrom stream = changesHelper Nothing stream

720 where

796 | n <= 0 || Array.null arr = []

797 | otherwise =

798 do

799+ let

799− let group = Array.take n arr

800+ group = Array.take n arr

800− rest = Array.drop n arr

801+ rest = Array.drop n arr

801 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)

802

803

803-- | Alias for groupedStrom

804-- | Alias for groupedStrom

864partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> ((Strom ctx err b) /\ (Strom ctx err c))

865partitionMap :: forall ctx err a b c. (a -> Either b c) -> Strom ctx err a -> ((Strom ctx err b) /\ (Strom ctx err c))

865partitionMap f stream =

866partitionMap f stream =

866 do

867 do

868+ let

867− let leftOnly x = case f x of

869+ leftOnly x = case f x of

868− Left b -> Just b

870+ Left b -> Just b

869− Right _ -> Nothing

871+ Right _ -> Nothing

870− rightOnly x = case f x of

872+ rightOnly x = case f x of

871− Left _ -> Nothing

873+ Left _ -> Nothing

872− Right c -> Just c

874+ Right c -> Just c

873 (collectStrom leftOnly stream)

875 (collectStrom leftOnly stream)

874 /\ (collectStrom rightOnly stream)

876 /\ (collectStrom rightOnly stream)

875

877

887 Done (Just chunk) -> do

889 Done (Just chunk) -> do

888 Om.delay duration

890 Om.delay duration

889 pure $ Done $ Just chunk

891 pure $ Done $ Just chunk

890− Loop ((maybeChunk /\ next)) -> do

892+ Loop (maybeChunk /\ next) -> do

891 case maybeChunk of

893 case maybeChunk of

892 Nothing -> pure $ Loop $ (Nothing /\ (debounce duration next))

894 Nothing -> pure $ Loop $ (Nothing /\ (debounce duration next))

893 Just chunk -> do

895 Just chunk -> do

894 Om.delay duration

896 Om.delay duration

895 pure $ Loop $ ((Just chunk) /\ (debounce duration next))

897 pure $ Loop $ ((Just chunk) /\ (debounce duration next))

896

898

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+

897−-- | Throttle a stream - emit at most once per duration period

905+-- | Throttle a stream - emits at most one element per duration period

898−-- | Tracks last emission time and skips elements that arrive too quickly

899−-- | Note: Uses wall-clock time, not processing time

900throttle :: 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

901−throttle (Milliseconds duration) stream = throttleWithState 0.0 stream

907+throttle duration stream = stream # bindStrom \a -> mkStrom do

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

908+ Om.delay duration

926− pure $ (currentMillis /\ (Array.snoc filtered value))

909+ pure $ Done $ Just [ a ]

927− else

928− pure $ (currentLast /\ filtered)

929− )

930− (pure $ (lastEmit /\ []))

931− chunk

932− pure $ (newLastEmit /\ results)

933

910

934-- | Delay each element by a duration

911-- | Delay each element by a duration

935delayStrom :: 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

1010--------------------------------------------------------------------------------

987--------------------------------------------------------------------------------

1011

988

1012-- | 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

1013−-- |

990+-- |

1014-- | 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:

1015−-- |

992+-- |

1016-- | Example:

993-- | Example:

1017-- | ```purescript

994-- | ```purescript

1018-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)

995-- | fiber <- Om.launchOm ctx handlers (Strom.runCollect myStream)

1019-- | -- ... later ...

996-- | -- ... later ...

1020-- | killFiber (error "Cancelled") fiber

997-- | killFiber (error "Cancelled") fiber

1021-- | ```

998-- | ```

1022−-- |

999+-- |

1023-- | Or for subscription-style:

1000-- | Or for subscription-style:

1024-- | ```purescript

1001-- | ```purescript

1025-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)

1002-- | fiber <- Om.launchOm ctx handlers (Strom.traverseStrom_ callback myStream)

1026−-- | -- ... later ...

1003+-- | -- ... later ...

1027-- | killFiber (error "Unsubscribed") fiber

1004-- | killFiber (error "Unsubscribed") fiber

1028-- | ```

1005-- | ```

1029-- |

1006-- |

1568 | n <= 0 || Array.null arr = []

1545 | n <= 0 || Array.null arr = []

1569 | otherwise =

1546 | otherwise =

1570 do

1547 do

1548+ let

1571− let group = Array.take n arr

1549+ group = Array.take n arr

1572− rest = Array.drop n arr

1550+ rest = Array.drop n arr

1573 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)

1574

1552

1575-- | Unordered parallel map (currently same as mapPar)

1553-- | Unordered parallel map (currently same as mapPar)

1715 case stepF, stepA of

1693 case stepF, stepA of

1716 Done (Just funcs), Done (Just vals) ->

1694 Done (Just funcs), Done (Just vals) ->

1717 do

1695 do

1696+ let

1718− let results = do

1697+ results = do

1719− f <- funcs

1698+ f <- funcs

1720− a <- vals

1699+ a <- vals

1721− pure (f a)

1700+ pure (f a)

1722 pure $ Done $ Just results

1701 pure $ Done $ Just results

1723 Done Nothing, _ -> pure $ Done Nothing

1702 Done Nothing, _ -> pure $ Done Nothing

1724 _, Done Nothing -> pure $ Done Nothing

1703 _, Done Nothing -> pure $ Done Nothing

1727 case maybeChunkF, maybeChunkA of

1706 case maybeChunkF, maybeChunkA of

1728 Just chunkF, Just chunkA ->

1707 Just chunkF, Just chunkA ->

1729 do

1708 do

1709+ let

1730− let applied = do

1710+ applied = do

1731− f <- chunkF

1711+ f <- chunkF

1732− a <- chunkA

1712+ a <- chunkA

1733− pure (f a)

1713+ pure (f a)

1734 pure $ Loop $ ((Just applied) /\ (nextF <*> nextA))

1714 pure $ Loop $ ((Just applied) /\ (nextF <*> nextA))

1735 _, _ -> pure $ Loop $ (Nothing /\ (nextF <*> nextA))

1715 _, _ -> pure $ Loop $ (Nothing /\ (nextF <*> nextA))

1736 Loop ((_ /\ nextF)), _ -> pure $ Loop $ (Nothing /\ (nextF <*> as))

1716 Loop ((_ /\ nextF)), _ -> pure $ Loop $ (Nothing /\ (nextF <*> as))

1755 Nothing -> pure $ Loop $ (Nothing /\ (next >>= f))

1735 Nothing -> pure $ Loop $ (Nothing /\ (next >>= f))

1756 Just chunk ->

1736 Just chunk ->

1757 do

1737 do

1738+ let

1758− let streams = Functor.map f chunk

1739+ streams = Functor.map f chunk

1759− prefixStream = concatStrom streams

1740+ prefixStream = concatStrom streams

1760 runStrom $ appendStrom prefixStream (next >>= f)

1741 runStrom $ appendStrom prefixStream (next >>= f)

1761 )

1742 )

1762

1743