@@ -256,6 +256,7 @@ module Streamly.Internal.Prelude
256256 , intersperseM_
257257 , intersperse
258258 , intersperseSuffix
259+ , intersperseSuffix_
259260 , interspersePrefix_
260261 , intersperseSuffixBySpan
261262 -- , intersperseBySpan
@@ -265,6 +266,7 @@ module Streamly.Internal.Prelude
265266 -- , intersperseByTime
266267 -- , intersperseByEvent
267268 , interjectSuffix
269+ , delay
268270 , delayPost
269271 , delayPre
270272
@@ -272,6 +274,7 @@ module Streamly.Internal.Prelude
272274 , indexed
273275 , indexedR
274276 , timestamped
277+ , timeIndexed
275278 -- , timestampedR -- timer
276279
277280 -- ** Reordering
@@ -1007,57 +1010,123 @@ fromCallback setCallback = concatM $ do
10071010-- Time related
10081011------------------------------------------------------------------------------
10091012
1010−-- | @times g@ returns a stream of time value tuples. The first component of
1011−-- the tuple is an absolute time reference denoting the start of the stream
1012−-- and the second component is a time relative to the reference.
1013+-- | @timesWith g@ returns a stream of time value tuples. The first component
1014+-- of the tuple is an absolute time reference (epoch) denoting the start of the
1015+-- stream and the second component is a time relative to the reference.
10131016--
10141017-- The argument @g@ specifies the granularity of the relative time in seconds.
10151018-- A lower granularity clock gives higher precision but is more expensive in
10161019-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
10171020--
1021+-- @
1022+-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.timesWith 0.01
1023+-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))
1024+-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))
1025+-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))
1026+-- @
1027+--
10181028-- Note: This API is not safe on 32-bit machines.
10191029--
10201030-- /Internal/
10211031--
1022−{-# INLINE times #-}
1023−times :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
1024−times g = fromStreamD $ D.times g
1032+{-# INLINE timesWith #-}
1033+timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
1034+timesWith g = fromStreamD $ D.times g
10251035
1026−-- TBD: rename to clock/absTimes?
1036+-- | @times@ returns a stream of time value tuples with clock of 10 ms
1037+-- granularity. The first component of the tuple is an absolute time reference
1038+-- (epoch) denoting the start of the stream and the second component is a time
1039+-- relative to the reference.
1040+--
1041+-- @
1042+-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.times
1043+-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))
1044+-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))
1045+-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))
1046+-- @
1047+--
1048+-- Note: This API is not safe on 32-bit machines.
10271049--
1028−-- | @absTimes g@ returns a stream of absolute timestamps using a clock of
1050+-- /Internal/
1051+--
1052+{-# INLINE times #-}
1053+times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
1054+times = timesWith 0.01
1055+
1056+-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
10291057-- granularity @g@ specified in seconds. A low granularity clock is more
10301058-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
10311059-- as 1 ms.
10321060--
1061+-- @
1062+-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimesWith 0.01
1063+-- @
1064+--
10331065-- Note: This API is not safe on 32-bit machines.
10341066--
10351067-- /Internal/
10361068--
1037−{-# INLINE absTimes #-}
1038−absTimes :: (IsStream t, MonadAsync m, Functor (t m))
1069+{-# INLINE absTimesWith #-}
1070+absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
10391071 => Double -> t m AbsTime
1040−absTimes = fmap (uncurry addToAbsTime64) . times
1072+absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
1073+
1074+-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
1075+-- granularity.
1076+--
1077+-- @
1078+-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimes
1079+-- @
1080+--
1081+-- Note: This API is not safe on 32-bit machines.
1082+--
1083+-- /Internal/
1084+--
1085+{-# INLINE absTimes #-}
1086+absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
1087+absTimes = fmap (uncurry addToAbsTime64) times
10411088
10421089{-# DEPRECATED currentTime "Please use absTimes instead" #-}
10431090{-# INLINE currentTime #-}
10441091currentTime :: (IsStream t, MonadAsync m, Functor (t m))
10451092 => Double -> t m AbsTime
1046−currentTime = absTimes
1093+currentTime = absTimesWith
10471094
1048−-- | @relTimes g@ returns a stream of relative time values starting from 0,
1095+-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
10491096-- using a clock of granularity @g@ specified in seconds. A low granularity
10501097-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
10511098-- ms is treated as 1 ms.
10521099--
1100+-- @
1101+-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimesWith 0.01
1102+-- > RelTime64 (NanoSecond64 0)
1103+-- > RelTime64 (NanoSecond64 91139000)
1104+-- > RelTime64 (NanoSecond64 204052000)
1105+-- @
1106+--
10531107-- Note: This API is not safe on 32-bit machines.
10541108--
10551109-- /Internal/
10561110--
1057−{-# INLINE relTimes #-}
1058−relTimes :: (IsStream t, MonadAsync m, Functor (t m))
1111+{-# INLINE relTimesWith #-}
1112+relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
10591113 => Double -> t m RelTime64
1060−relTimes = fmap snd . times
1114+relTimesWith = fmap snd . timesWith
1115+
1116+-- | @relTimes@ returns a stream of relative time values starting from 0,
1117+-- using a clock of granularity 10 ms.
1118+--
1119+-- @
1120+-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimes
1121+-- @
1122+--
1123+-- Note: This API is not safe on 32-bit machines.
1124+--
1125+-- /Internal/
1126+--
1127+{-# INLINE relTimes #-}
1128+relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
1129+relTimes = fmap snd times
10611130
10621131-- | @durations g@ returns a stream of relative time values measuring the time
10631132-- elapsed since the immediate predecessor element of the stream was generated.
@@ -2462,46 +2531,30 @@ reverse' s = fromStreamD $ D.reverse' $ toStreamD s
24622531
24632532-- intersperseM = intersperseBySpan 1
24642533
2465−-- | Insert the result of an effectful action between consecutive elements of
2466−-- the stream.
2534+-- | Insert an effect and its output before consuming an element of a stream
2535+-- except the first one.
24672536--
24682537-- @
24692538-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"
24702539-- > h.,e.,l.,l.,o"h,e,l,l,o"
24712540-- @
24722541--
2473−-- After yielding the first element, for every next element the effectful
2474−-- action is evaluated /after/ generating the element, however, its result is
2475−-- yielded /before/ the element.
2476−--
2477−-- @
2478−-- > S.toList $ S.intersperseM (putChar '.' >> return ',') $ S.trace putChar $ S.fromList "hello"
2479−-- he.l.l.o."h,e,l,l,o"
2480−-- @
2481−--
24822542-- @since 0.5.0
24832543{-# INLINE intersperseM #-}
24842544intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
24852545intersperseM m = fromStreamS . S.intersperseM m . toStreamS
24862546
2487−-- | Insert a side effect between consecutive elements of the stream.
2547+-- | Insert a side effect before consuming an element of a stream except the
2548+-- first one.
24882549--
24892550-- @
24902551-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
24912552-- > h.e.l.l.o
24922553-- @
24932554--
2494−-- After yielding the first element, for every next element the effect
2495−-- is evaluated after generating the element but before yielding it.
2496−--
2497−-- @
2498−-- >>> S.drain $ S.intersperseM_ (putChar '.') $ S.trace putChar $ S.fromList "hello"
2499−-- > he.l.l.o.
2500−-- @
2501−--
25022555-- /Internal/
25032556{-# INLINE intersperseM_ #-}
2504−intersperseM_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
2557+intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
25052558intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
25062559
25072560-- | Insert a pure value between successive elements of a stream.
@@ -2516,7 +2569,7 @@ intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
25162569intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
25172570intersperse a = fromStreamS . S.intersperse a . toStreamS
25182571
2519−-- | Insert an effectful action after each element of a stream.
2572+-- | Insert an effect and its output after consuming an element of a stream.
25202573--
25212574-- @
25222575-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
@@ -2528,7 +2581,7 @@ intersperse a = fromStreamS . S.intersperse a . toStreamS
25282581intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
25292582intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
25302583
2531−-- | Insert a side effect after each element of a stream.
2584+-- | Insert a side effect after consuming an element of a stream.
25322585--
25332586-- @
25342587-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
@@ -2540,7 +2593,7 @@ intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
25402593intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
25412594intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
25422595
2543−-- | Insert a side effect before each element of a stream.
2596+-- | Insert a side effect before consuming an element of a stream.
25442597--
25452598-- @
25462599-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
@@ -2559,7 +2612,33 @@ interspersePrefix_ m = mapM (\x -> void m >> return x)
25592612
25602613-- Note: delay must be serial.
25612614--
2562−-- | Introduce a delay of specified seconds after each element of a stream.
2615+-- | Introduce a delay of specified seconds before consuming an element of the
2616+-- stream except the first one.
2617+--
2618+-- @
2619+-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2620+-- > (AbsTime (TimeSpec {sec = 2502706, nsec = 751137000}),1)
2621+-- > (AbsTime (TimeSpec {sec = 2502707, nsec = 743535000}),2)
2622+-- > (AbsTime (TimeSpec {sec = 2502708, nsec = 749758000}),3)
2623+-- @
2624+--
2625+-- /Internal/
2626+--
2627+{-# INLINE delay #-}
2628+delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2629+delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
2630+
2631+-- Note: delay must be serial.
2632+--
2633+-- | Introduce a delay of specified seconds after consuming an element of a
2634+-- stream.
2635+--
2636+-- @
2637+-- >>> S.mapM_ print $ S.timestamped $ S.delayPost 1 $ S.enumerateFromTo 1 3
2638+-- > (AbsTime (TimeSpec {sec = 2502826, nsec = 119030000}),1)
2639+-- > (AbsTime (TimeSpec {sec = 2502827, nsec = 111393000}),2)
2640+-- > (AbsTime (TimeSpec {sec = 2502828, nsec = 112221000}),3)
2641+-- @
25632642--
25642643-- /Internal/
25652644--
@@ -2569,7 +2648,15 @@ delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
25692648
25702649-- Note: delay must be serial, that's why 'trace_' is used.
25712650--
2572−-- | Introduce a delay of specified seconds before each element of a stream.
2651+-- | Introduce a delay of specified seconds before consuming an element of a
2652+-- stream.
2653+--
2654+-- @
2655+-- >>> S.mapM_ print $ S.timestamped $ S.delayPre 1 $ S.enumerateFromTo 1 3
2656+-- > (AbsTime (TimeSpec {sec = 2502207, nsec = 533177000}),1)
2657+-- > (AbsTime (TimeSpec {sec = 2502208, nsec = 530859000}),2)
2658+-- > (AbsTime (TimeSpec {sec = 2502209, nsec = 531619000}),3)
2659+-- @
25732660--
25742661-- /Internal/
25752662--
@@ -2696,16 +2783,80 @@ indexed = fromStreamD . D.indexed . toStreamD
26962783indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
26972784indexedR n = fromStreamD . D.indexedR n . toStreamD
26982785
2786+-- Note: The timestamp stream must be the second stream in the zip so that the
2787+-- timestamp is generated after generating the stream element and not before.
2788+-- If we do not do that then the following example will generate the same
2789+-- timestamp for first two elements:
2790+--
2791+-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
2792+--
2793+-- | Pair each element in a stream with an absolute timestamp, using a clock of
2794+-- specified granularity. The timestamp is generated just before the element
2795+-- is consumed.
2796+--
2797+-- @
2798+-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2799+-- @
2800+--
2801+-- /Internal/
2802+--
2803+{-# INLINE timestampWith #-}
2804+timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
2805+ => Double -> t m a -> t m (AbsTime, a)
2806+timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
2807+
26992808-- TBD: check performance vs a custom implementation without using zipWith.
27002809--
2701−-- | Pair each element in a stream with the current wall clock timestamp.
2810+-- | Pair each element in a stream with an absolute timestamp, using a clock of
2811+-- 10 ms granularity. The timestamp is generated just before the element is
2812+-- consumed.
2813+--
2814+-- @
2815+-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2816+-- (AbsTime (TimeSpec {sec = 2460689, nsec = 641121000}),1)
2817+-- (AbsTime (TimeSpec {sec = 2460690, nsec = 639334000}),2)
2818+-- (AbsTime (TimeSpec {sec = 2460691, nsec = 644479000}),3)
2819+-- @
27022820--
27032821-- /Internal/
27042822--
27052823{-# INLINE timestamped #-}
2706−timestamped :: (IsStream t, MonadAsync m)
2707− => Double -> t m a -> t m (AbsTime, a)
2708−timestamped g = Z.zipWith (,) (absTimes g)
2824+timestamped :: (IsStream t, MonadAsync m, Functor (t m))
2825+ => t m a -> t m (AbsTime, a)
2826+timestamped = timestampWith 0.01
2827+
2828+-- | Pair each element in a stream with relative times starting from 0, using a
2829+-- clock with the specified granularity. The time is measured just before the
2830+-- element is consumed.
2831+--
2832+-- @
2833+-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2834+-- @
2835+--
2836+-- /Internal/
2837+--
2838+{-# INLINE timeIndexWith #-}
2839+timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
2840+ => Double -> t m a -> t m (RelTime64, a)
2841+timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
2842+
2843+-- | Pair each element in a stream with relative times starting from 0, using a
2844+-- 10 ms granularity clock. The time is measured just before the element is
2845+-- consumed.
2846+--
2847+-- @
2848+-- >>> S.mapM_ print $ S.timeIndexed $ S.delay 1 $ S.enumerateFromTo 1 3
2849+-- (RelTime64 (NanoSecond64 0),1)
2850+-- (RelTime64 (NanoSecond64 996239000),2)
2851+-- (RelTime64 (NanoSecond64 1996712000),3)
2852+-- @
2853+--
2854+-- /Internal/
2855+--
2856+{-# INLINE timeIndexed #-}
2857+timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
2858+ => t m a -> t m (RelTime64, a)
2859+timeIndexed = timeIndexWith 0.01
27092860
27102861------------------------------------------------------------------------------
27112862-- Comparison