@@ -256,6 +256,7 @@ module Streamly.Internal.Prelude
256 , intersperseM_
256 , intersperseM_
257 , intersperse
257 , intersperse
258 , intersperseSuffix
258 , intersperseSuffix
259+ , intersperseSuffix_
259 , interspersePrefix_
260 , interspersePrefix_
260 , intersperseSuffixBySpan
261 , intersperseSuffixBySpan
261 -- , intersperseBySpan
262 -- , intersperseBySpan
@@ -265,6 +266,7 @@ module Streamly.Internal.Prelude
265 -- , intersperseByTime
266 -- , intersperseByTime
266 -- , intersperseByEvent
267 -- , intersperseByEvent
267 , interjectSuffix
268 , interjectSuffix
268 , delayPost
270 , delayPost
269 , delayPre
271 , delayPre
@@ -272,6 +274,7 @@ module Streamly.Internal.Prelude
272 , indexed
274 , indexed
273 , indexedR
275 , indexedR
274 , timestamped
276 , timestamped
275 -- , timestampedR -- timer
278 -- , timestampedR -- timer
277 -- ** Reordering
280 -- ** Reordering
@@ -1007,57 +1010,123 @@ fromCallback setCallback = concatM $ do
1007-- Time related
1010-- Time related
1008------------------------------------------------------------------------------
1011------------------------------------------------------------------------------
1010−-- | @times g@ returns a stream of time value tuples. The first component of
1013+-- | @timesWith g@ returns a stream of time value tuples. The first component
1011−-- the tuple is an absolute time reference denoting the start of the stream
1014+-- of the tuple is an absolute time reference (epoch) denoting the start of the
1012−-- and the second component is a time relative to the reference.
1015+-- stream and the second component is a time relative to the reference.
1014-- The argument @g@ specifies the granularity of the relative time in seconds.
1017-- The argument @g@ specifies the granularity of the relative time in seconds.
1015-- A lower granularity clock gives higher precision but is more expensive in
1018-- A lower granularity clock gives higher precision but is more expensive in
1016-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
1019-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
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))
1018-- Note: This API is not safe on 32-bit machines.
1028-- Note: This API is not safe on 32-bit machines.
1020-- /Internal/
1030-- /Internal/
1022−{-# INLINE times #-}
1032+{-# INLINE timesWith #-}
1023−times :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
1033+timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
1024−times g = fromStreamD $ D.times g
1034+timesWith g = fromStreamD $ D.times g
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.
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))
1048+-- Note: This API is not safe on 32-bit machines.
1052+{-# INLINE times #-}
1053+times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
1054+times = timesWith 0.01
1028−-- | @absTimes g@ returns a stream of absolute timestamps using a clock of
1056+-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
1029-- granularity @g@ specified in seconds. A low granularity clock is more
1057-- granularity @g@ specified in seconds. A low granularity clock is more
1030-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
1058-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
1031-- as 1 ms.
1059-- as 1 ms.
1062+-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimesWith 0.01
1033-- Note: This API is not safe on 32-bit machines.
1065-- Note: This API is not safe on 32-bit machines.
1035-- /Internal/
1067-- /Internal/
1037−{-# INLINE absTimes #-}
1069+{-# INLINE absTimesWith #-}
1038−absTimes :: (IsStream t, MonadAsync m, Functor (t m))
1070+absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
1039 => Double -> t m AbsTime
1071 => Double -> t m AbsTime
1040−absTimes = fmap (uncurry addToAbsTime64) . times
1072+absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
1074+-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
1078+-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimes
1081+-- Note: This API is not safe on 32-bit machines.
1085+{-# INLINE absTimes #-}
1086+absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
1087+absTimes = fmap (uncurry addToAbsTime64) times
1042{-# DEPRECATED currentTime "Please use absTimes instead" #-}
1089{-# DEPRECATED currentTime "Please use absTimes instead" #-}
1043{-# INLINE currentTime #-}
1090{-# INLINE currentTime #-}
1044currentTime :: (IsStream t, MonadAsync m, Functor (t m))
1091currentTime :: (IsStream t, MonadAsync m, Functor (t m))
1045 => Double -> t m AbsTime
1092 => Double -> t m AbsTime
1046−currentTime = absTimes
1093+currentTime = absTimesWith
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,
1049-- using a clock of granularity @g@ specified in seconds. A low granularity
1096-- using a clock of granularity @g@ specified in seconds. A low granularity
1050-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
1097-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
1051-- ms is treated as 1 ms.
1098-- ms is treated as 1 ms.
1101+-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimesWith 0.01
1102+-- > RelTime64 (NanoSecond64 0)
1103+-- > RelTime64 (NanoSecond64 91139000)
1104+-- > RelTime64 (NanoSecond64 204052000)
1053-- Note: This API is not safe on 32-bit machines.
1107-- Note: This API is not safe on 32-bit machines.
1055-- /Internal/
1109-- /Internal/
1057−{-# INLINE relTimes #-}
1111+{-# INLINE relTimesWith #-}
1058−relTimes :: (IsStream t, MonadAsync m, Functor (t m))
1112+relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
1059 => Double -> t m RelTime64
1113 => Double -> t m RelTime64
1060−relTimes = fmap snd . times
1114+relTimesWith = fmap snd . timesWith
1116+-- | @relTimes@ returns a stream of relative time values starting from 0,
1117+-- using a clock of granularity 10 ms.
1120+-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimes
1123+-- Note: This API is not safe on 32-bit machines.
1127+{-# INLINE relTimes #-}
1128+relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
1129+relTimes = fmap snd times
1062-- | @durations g@ returns a stream of relative time values measuring the time
1131-- | @durations g@ returns a stream of relative time values measuring the time
1063-- elapsed since the immediate predecessor element of the stream was generated.
1132-- elapsed since the immediate predecessor element of the stream was generated.
@@ -2462,46 +2531,30 @@ reverse' s = fromStreamD $ D.reverse' $ toStreamD s
2463-- intersperseM = intersperseBySpan 1
2532-- intersperseM = intersperseBySpan 1
2465−-- | Insert the result of an effectful action between consecutive elements of
2534+-- | Insert an effect and its output before consuming an element of a stream
2466−-- the stream.
2535+-- except the first one.
2469-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"
2538-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"
2470-- > h.,e.,l.,l.,o"h,e,l,l,o"
2539-- > h.,e.,l.,l.,o"h,e,l,l,o"
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.
2478−-- > S.toList $ S.intersperseM (putChar '.' >> return ',') $ S.trace putChar $ S.fromList "hello"
2479−-- he.l.l.o."h,e,l,l,o"
2482-- @since 0.5.0
2542-- @since 0.5.0
2483{-# INLINE intersperseM #-}
2543{-# INLINE intersperseM #-}
2484intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2544intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2485intersperseM m = fromStreamS . S.intersperseM m . toStreamS
2545intersperseM m = fromStreamS . S.intersperseM m . toStreamS
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
2490-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
2551-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
2491-- > h.e.l.l.o
2552-- > h.e.l.l.o
2494−-- After yielding the first element, for every next element the effect
2495−-- is evaluated after generating the element but before yielding it.
2498−-- >>> S.drain $ S.intersperseM_ (putChar '.') $ S.trace putChar $ S.fromList "hello"
2502-- /Internal/
2555-- /Internal/
2503{-# INLINE intersperseM_ #-}
2556{-# 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
2505intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
2558intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
2507-- | Insert a pure value between successive elements of a stream.
2560-- | Insert a pure value between successive elements of a stream.
@@ -2516,7 +2569,7 @@ intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
2516intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2569intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2517intersperse a = fromStreamS . S.intersperse a . toStreamS
2570intersperse a = fromStreamS . S.intersperse a . toStreamS
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.
2522-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
2575-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
@@ -2528,7 +2581,7 @@ intersperse a = fromStreamS . S.intersperse a . toStreamS
2528intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2581intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2529intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
2582intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
2531−-- | Insert a side effect after each element of a stream.
2584+-- | Insert a side effect after consuming an element of a stream.
2534-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
2587-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
@@ -2540,7 +2593,7 @@ intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
2540intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2593intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2541intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
2594intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
2543−-- | Insert a side effect before each element of a stream.
2596+-- | Insert a side effect before consuming an element of a stream.
2546-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
2599-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
@@ -2559,7 +2612,33 @@ interspersePrefix_ m = mapM (\x -> void m >> return x)
2560-- Note: delay must be serial.
2613-- Note: delay must be serial.
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.
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)
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
2631+-- Note: delay must be serial.
2633+-- | Introduce a delay of specified seconds after consuming an element of a
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)
2564-- /Internal/
2643-- /Internal/
@@ -2569,7 +2648,15 @@ delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
2570-- Note: delay must be serial, that's why 'trace_' is used.
2649-- Note: delay must be serial, that's why 'trace_' is used.
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
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)
2574-- /Internal/
2661-- /Internal/
@@ -2696,16 +2783,80 @@ indexed = fromStreamD . D.indexed . toStreamD
2696indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2783indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2697indexedR n = fromStreamD . D.indexedR n . toStreamD
2784indexedR n = fromStreamD . D.indexedR n . toStreamD
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:
2791+-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
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
2798+-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
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)
2699-- TBD: check performance vs a custom implementation without using zipWith.
2808-- TBD: check performance vs a custom implementation without using zipWith.
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
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)
2703-- /Internal/
2821-- /Internal/
2705{-# INLINE timestamped #-}
2823{-# INLINE timestamped #-}
2706−timestamped :: (IsStream t, MonadAsync m)
2824+timestamped :: (IsStream t, MonadAsync m, Functor (t m))
2825+ => t m a -> t m (AbsTime, a)
2826+timestamped = timestampWith 0.01
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.
2833+-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2838+{-# INLINE timeIndexWith #-}
2839+timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
2707− => Double -> t m a -> t m (AbsTime, a)
2840+ => Double -> t m a -> t m (RelTime64, a)
2708−timestamped g = Z.zipWith (,) (absTimes g)
2841+timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
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
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)
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
2710------------------------------------------------------------------------------
2861------------------------------------------------------------------------------
2711-- Comparison
2862-- Comparison