@@ -70,6 +70,12 @@ module Streamly.Internal.Data.Stream.Transform
71 , indexedR
71 , indexedR
73 -- * Searching
79 -- * Searching
74 , findIndices -- XXX indicesBy
80 , findIndices -- XXX indicesBy
75 , elemIndices -- XXX indicesOf
81 , elemIndices -- XXX indicesOf
@@ -138,6 +144,7 @@ import Data.Maybe (isJust, fromJust)
139import Streamly.Internal.Data.Fold.Type (Fold)
145import Streamly.Internal.Data.Fold.Type (Fold)
140import Streamly.Internal.Data.Pipe (Pipe)
146import Streamly.Internal.Data.Pipe (Pipe)
147+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
142import qualified Streamly.Internal.Data.Fold as FL
149import qualified Streamly.Internal.Data.Fold as FL
143-- import qualified Streamly.Internal.Data.Fold.Window as Window
150-- import qualified Streamly.Internal.Data.Fold.Window as Window
@@ -145,6 +152,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
145import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
152import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
146import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
153import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
148import Streamly.Internal.Data.Stream.Bottom
156import Streamly.Internal.Data.Stream.Bottom
149import Streamly.Internal.Data.Stream.Type
157import Streamly.Internal.Data.Stream.Type
@@ -631,7 +639,7 @@ intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
631-- /Unimplemented/
639-- /Unimplemented/
632{-# INLINE intersperseBySpan #-}
640{-# INLINE intersperseBySpan #-}
633intersperseBySpan :: -- Monad m =>
641intersperseBySpan :: -- Monad m =>
634− Int -> m a -> t m a -> t m a
642+ Int -> m a -> Stream m a -> Stream m a
635intersperseBySpan _n _f _xs = undefined
643intersperseBySpan _n _f _xs = undefined
637-- | Insert an effect and its output after consuming an element of a stream.
645-- | Insert an effect and its output after consuming an element of a stream.
@@ -753,8 +761,8 @@ reassembleBy
753 :: -- Monad m =>
761 :: -- Monad m =>
754 Fold m a b
762 Fold m a b
755 -> (a -> a -> Int)
763 -> (a -> a -> Int)
756− -> t m a
764+ -> Stream m a
757− -> t m b
765+ -> Stream m b
758reassembleBy = undefined
766reassembleBy = undefined
760------------------------------------------------------------------------------
768------------------------------------------------------------------------------
@@ -796,6 +804,68 @@ indexedR :: Monad m => Int -> Stream m a -> Stream m (Int, a)
796-- indexedR n = scanMaybe (FL.indexingRev n)
804-- indexedR n = scanMaybe (FL.indexingRev n)
797indexedR n = fromStreamD . D.indexedR n . toStreamD
805indexedR n = fromStreamD . D.indexedR n . toStreamD
807+-------------------------------------------------------------------------------
809+-------------------------------------------------------------------------------
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:
816+-- Stream.fold Fold.toList $ Stream.timestamped $ Stream.delay $ Stream.enumerateFromTo 1 3
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
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)]
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)
832+-- TBD: check performance vs a custom implementation without using zipWith.
836+{-# INLINE timestamped #-}
837+timestamped :: (MonadIO m)
838+ => Stream m a -> Stream m(AbsTime, a)
839+timestamped = timestampWith 0.01
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.
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)]
848+-- /Pre-release/Monad
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)
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
859+-- >>> Stream.fold Fold.toList $ Stream.timeIndexed $ Stream.delay 1 $ Stream.enumerateFromTo 1 3
860+-- [(RelTime64 (NanoSecond64 ...),1),(RelTime64 (NanoSecond64 ...),2),(RelTime64 (NanoSecond64 ...),3)]
864+{-# INLINE timeIndexed #-}
865+timeIndexed :: (MonadIO m)
866+ => Stream m a -> Stream m(RelTime64, a)
867+timeIndexed = timeIndexWith 0.01
799------------------------------------------------------------------------------
869------------------------------------------------------------------------------
800-- Searching
870-- Searching
801------------------------------------------------------------------------------
871------------------------------------------------------------------------------