@@ -70,6 +70,12 @@ module Streamly.Internal.Data.Stream.Transform
7070 , indexed
7171 , indexedR
7272
73+ -- * Time Indexing
74+ , timestamped
75+ , timestampWith
76+ , timeIndexed
77+ , timeIndexWith
78+
7379 -- * Searching
7480 , findIndices -- XXX indicesBy
7581 , elemIndices -- XXX indicesOf
@@ -138,6 +144,7 @@ import Data.Maybe (isJust, fromJust)
138144
139145import Streamly.Internal.Data.Fold.Type (Fold)
140146import Streamly.Internal.Data.Pipe (Pipe)
147+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
141148
142149import qualified Streamly.Internal.Data.Fold as FL
143150-- import qualified Streamly.Internal.Data.Fold.Window as Window
@@ -145,6 +152,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
145152import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
146153import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
147154
155+
148156import Streamly.Internal.Data.Stream.Bottom
149157import Streamly.Internal.Data.Stream.Type
150158
@@ -631,7 +639,7 @@ intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
631639-- /Unimplemented/
632640{-# INLINE intersperseBySpan #-}
633641intersperseBySpan :: -- Monad m =>
634− Int -> m a -> t m a -> t m a
642+ Int -> m a -> Stream m a -> Stream m a
635643intersperseBySpan _n _f _xs = undefined
636644
637645-- | Insert an effect and its output after consuming an element of a stream.
@@ -753,8 +761,8 @@ reassembleBy
753761 :: -- Monad m =>
754762 Fold m a b
755763 -> (a -> a -> Int)
756− -> t m a
757− -> t m b
764+ -> Stream m a
765+ -> Stream m b
758766reassembleBy = undefined
759767
760768------------------------------------------------------------------------------
@@ -796,6 +804,68 @@ indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
796804-- indexedR n = scanMaybe (FL.indexingRev n)
797805indexedR n = fromStreamD . D.indexedR n . toStreamD
798806
807+-------------------------------------------------------------------------------
808+-- Time Indexing
809+-------------------------------------------------------------------------------
810+
811+-- Note: The timestamp stream must be the second stream in the zip so that the
812+-- timestamp is generated after generating the stream element and not before.
813+-- If we do not do that then the following example will generate the same
814+-- timestamp for first two elements:
815+--
816+-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
817+--
818+-- | Pair each element in a stream with an absolute timestamp, using a clock of
819+-- specified granularity. The timestamp is generated just before the element
820+-- is consumed.
821+--
822+-- >>> Stream.fold Fold.toList $ Stream.timestampWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
823+-- [(AbsTime (TimeSpec {sec = ..., nsec = ...}),1),(AbsTime (TimeSpec {sec = ..., nsec = ...}),2),(AbsTime (TimeSpec {sec = ..., nsec = ...}),3)]
824+--
825+-- /Pre-release/
826+--
827+{-# INLINE timestampWith #-}
828+timestampWith :: (MonadIO m)
829+ => Double -> Stream m a -> Stream m(AbsTime, a)
830+timestampWith g stream = zipWith (flip (,)) stream (absTimesWith g)
831+
832+-- TBD: check performance vs a custom implementation without using zipWith.
833+--
834+-- /Pre-release/
835+--
836+{-# INLINE timestamped #-}
837+timestamped :: (MonadIO m)
838+ => Stream m a -> Stream m(AbsTime, a)
839+timestamped = timestampWith 0.01
840+
841+-- | Pair each element in a stream with relative times starting from 0, using a
842+-- clock with the specified granularity. The time is measured just before the
843+-- element is consumed.
844+--
845+-- >>> Stream.fold Fold.toList $ Stream.timeIndexWith 0.01 $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
846+-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
847+--
848+-- /Pre-release/Monad
849+--
850+{-# INLINE timeIndexWith #-}
851+timeIndexWith :: (MonadIO m)
852+ => Double -> Stream m a -> Stream m(RelTime64, a)
853+timeIndexWith g stream = zipWith (flip (,)) stream (relTimesWith g)
854+
855+-- | Pair each element in a stream with relative times starting from 0, using a
856+-- 10 ms granularity clock. The time is measured just before the element is
857+-- consumed.
858+--
859+-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
860+-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
861+--
862+-- /Pre-release/
863+--
864+{-# INLINE timeIndexed #-}
865+timeIndexed :: (MonadIO m)
866+ => Stream m a -> Stream m(RelTime64, a)
867+timeIndexed = timeIndexWith 0.01
868+
799869------------------------------------------------------------------------------
800870-- Searching
801871------------------------------------------------------------------------------