@@ -24,6 +24,7 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
26 -- * Mapping Side Effects (Observation)
26 -- * Mapping Side Effects (Observation)
27+ -- | See also the intersperse*_ combinators.
@@ -62,59 +63,72 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
62 -- , lprescanlM'
63 -- , lprescanlM'
64 -- * Filtering
65 -- * Filtering
65− -- | Produce a subset of the stream.
66+ -- | Produce a subset of the stream using criteria based on the values of
67+ -- the elements. We can use a concatMap and scan for filtering but these
68+ -- combinators are more efficient and convenient.
71− -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
75+ , uniqBy
72− -- , uniqOn -- to remove duplicate sequences
76+ , nubBy
73− -- , pruneBy -- dropAround + uniqBy - like words
77+ , nubWindowBy
75 -- * Trimming
81 -- * Trimming
76 -- | Produce a subset of the stream trimmed at ends.
82 -- | Produce a subset of the stream trimmed at ends.
81− , takeByTime
85+ , takeByTime -- takeInterval
82− -- , takeEnd
86+ , takeEnd
83 , takeWhile
88 , takeWhile
84 , takeWhileM
89 , takeWhileM
85− -- , takeWhileEnd
90+ , takeWhileEnd
87− , dropByTime
93+ , dropByTime -- dropInterval
88− -- , dropEnd
94+ , dropEnd
89 , dropWhile
96 , dropWhile
90 , dropWhileM
97 , dropWhileM
91− -- , dropWhileEnd
98+ , dropWhileEnd
92− -- , dropAround
99+ , dropWhileAround
94 -- * Inserting Elements
101 -- * Inserting Elements
95− -- | Produce a superset of the stream.
102+ -- | Produce a superset of the stream. This is the opposite of
103+ -- filtering/sampling. We can always use concatMap and scan for inserting
104+ -- but these combinators are more efficient and convenient.
97− , insertBy
106+ -- Element agnostic (Opposite of sampling)
98 , intersperse
107 , intersperse
99 , intersperseM -- XXX naming
108 , intersperseM -- XXX naming
100 , intersperseSuffix
111 , intersperseSuffix
101 , intersperseSuffixBySpan
112 , intersperseSuffixBySpan
102− -- , intersperseBySpan
113+ , interjectSuffix
103− -- , intersperseByIndices -- using an index function/stream
105− -- * Inserting Side Effects
115+ -- , interspersePrefix
106− , intersperseM_ -- XXX naming
116+ -- , interspersePrefixBySpan
107− , intersperseSuffix_
108− , interspersePrefix_
110− -- * Inserting Time
118+ -- * Inserting Side Effects/Time
111− -- , intersperseByTime
119+ , intersperseM_ -- XXX naming
112− -- , intersperseByEvent
121+ , intersperseSuffix_
115 , delayPost
122 , delayPost
123+ , interspersePrefix_
116 , delayPre
124 , delayPre
126+ -- * Element Aware Insertion
127+ -- | Opposite of filtering
129+ -- , intersperseByBefore
130+ -- , intersperseByAfter
118 -- * Reordering
132 -- * Reordering
119 , reverse
133 , reverse
120 , reverse'
134 , reverse'
@@ -151,7 +165,7 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
154− -- * Threading
168+ -- * Concurrent Evaluation
155 -- ** Concurrent Pipelines
169 -- ** Concurrent Pipelines
156 -- | Run streaming stages concurrently.
170 -- | Run streaming stages concurrently.
@@ -162,9 +176,20 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
163 -- ** Concurrency Control
177 -- ** Concurrency Control
164 , maxThreads
178 , maxThreads
180+ -- ** Buffering and Sampling
181+ -- | Evaluate strictly using a buffer of results. When the buffer becomes
182+ -- full we can block, drop the new elements, drop the oldest element and
183+ -- insert the new at the end or keep dropping elements uniformly to match
184+ -- the rate of the consumer.
165 , maxBuffer
185 , maxBuffer
167 -- ** Rate Limiting
190 -- ** Rate Limiting
191+ -- | Evaluate the stream at uniform intervals to maintain a specified
168 , Rate (..)
193 , Rate (..)
170 , avgRate
195 , avgRate
@@ -215,7 +240,7 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)
215import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
240import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
216import Streamly.Internal.Data.Stream.StreamK (IsStream)
241import Streamly.Internal.Data.Stream.StreamK (IsStream)
217import Streamly.Internal.Data.SVar (MonadAsync, Rate(..))
242import Streamly.Internal.Data.SVar (MonadAsync, Rate(..))
218−import Streamly.Internal.Data.Time.Units ( TimeUnit64, AbsTime, RelTime64)
243+import Streamly.Internal.Data.Time.Units (TimeUnit64, AbsTime, RelTime64)
220import qualified Streamly.Internal.Data.Fold as FL
245import qualified Streamly.Internal.Data.Fold as FL
221import qualified Streamly.Internal.Data.Stream.Parallel as Par
246import qualified Streamly.Internal.Data.Stream.Parallel as Par
@@ -519,7 +544,9 @@ trace f = mapM (\x -> void (f x) >> return x)
519-- "got here"
544-- "got here"
547+-- Same as 'interspersePrefix_' but always serial.
522−-- See also: 'trace', 'interspersePrefix_'
549+-- See also: 'trace'
524-- /Internal/
551-- /Internal/
525{-# INLINE trace_ #-}
552{-# INLINE trace_ #-}
@@ -693,6 +720,24 @@ scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
693-- Filtering
720-- Filtering
694------------------------------------------------------------------------------
721------------------------------------------------------------------------------
723+-- | Modify a @t m a -> t m a@ stream transformation that accepts a predicate
724+-- @(a -> b)@ to accept @((s, a) -> b)@ instead, provided a transformation @t m
725+-- a -> t m (s, a)@. Convenient to filter with index or time.
728+-- filterWithIndex = with indexed filter
729+-- filterWithAbsTime = with timestamped filter
730+-- filterWithRelTime = with timeIndexed filter
735+with :: Functor (t m) =>
736+ (t m a -> t m (s, a))
737+ -> (((s, a) -> c) -> t m (s, a) -> t m (s, a))
738+ -> (((s, a) -> c) -> t m a -> t m a)
739+with f comb g = fmap snd . comb g . f
696-- | Include only those elements that pass a predicate.
741-- | Include only those elements that pass a predicate.
698-- @since 0.1.0
743-- @since 0.1.0
@@ -713,6 +758,31 @@ filter = K.filter
713filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
758filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
714filterM p m = fromStreamD $ D.filterM p $ toStreamD m
759filterM p m = fromStreamD $ D.filterM p $ toStreamD m
761+-- | Drop repeated elements that are adjacent to each other using the supplied
762+-- comparison function.
764+-- @uniq = uniqBy (==)
766+-- To strip duplicate path separators:
769+-- f x y = x == '/' && x == y
770+-- Stream.toList $ Stream.uniqBy f $ Stream.fromList "//a//b"
776+-- See also: 'nubBy'.
780+{-# INLINE uniqBy #-}
781+uniqBy :: (IsStream t, Monad m, Functor (t m)) =>
782+ (a -> a -> Bool) -> t m a -> t m a
784+ catMaybes . rollingMap (\x y -> if x `eq` y then Nothing else Just y)
716-- | Drop repeated elements that are adjacent to each other.
786-- | Drop repeated elements that are adjacent to each other.
718-- @since 0.6.0
788-- @since 0.6.0
@@ -720,6 +790,67 @@ filterM p m = fromStreamD $ D.filterM p $ toStreamD m
720uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
790uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
721uniq = fromStreamD . D.uniq . toStreamD
791uniq = fromStreamD . D.uniq . toStreamD
793+-- | Strip all leading and trailing occurrences of an element passing a
794+-- predicate and make all other consecutive occurrences uniq.
797+-- prune p = dropWhileAround p $ uniqBy (x y -> p x && p y)
800+-- >>> Stream.pruneBy isSpace (Stream.fromList " hello world! ")
808+ -- (IsStream t, Monad m, Eq a) =>
809+ (a -> Bool) -> t m a -> t m a
810+prune = error "Not implemented yet!"
812+-- Possible implementation:
814+-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
816+-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
819+-- | Emit only repeated elements, once.
822+repeated :: -- (IsStream t, Monad m, Eq a) =>
826+-- We can have more efficient implementations for nubOrd and nubInt by using
827+-- Set and IntSet to find/remove duplication. For Hashable we can use a
828+-- hashmap. Use rewrite rules to specialize to more efficient impls.
830+-- | Drop repeated elements anywhere in the stream.
832+-- /Caution: not scalable for infinite streams/
834+-- /See also: nubWindowBy/
839+nubBy :: -- (IsStream t, Monad m) =>
840+ (a -> a -> Bool) -> t m a -> t m a
841+nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
843+-- | Drop repeated elements within the specified tumbling window in the stream.
845+-- @nubBy = nubWindowBy maxBound@
849+{-# INLINE nubWindowBy #-}
850+nubWindowBy :: -- (IsStream t, Monad m) =>
851+ Int -> (a -> a -> Bool) -> t m a -> t m a
852+nubWindowBy = undefined -- fromStreamD . D.nubWithinBy . toStreamD
723-- | Deletes the first occurrence of the element in the stream that satisfies
854-- | Deletes the first occurrence of the element in the stream that satisfies
724-- the given equality predicate.
855-- the given equality predicate.
@@ -733,6 +864,51 @@ uniq = fromStreamD . D.uniq . toStreamD
733deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
864deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
734deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
865deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
867+------------------------------------------------------------------------------
869+------------------------------------------------------------------------------
871+-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
872+-- may want to have the evaluation rate independent of the sampling rate. To
873+-- support that we can decouple evaluation and sampling in independent stages.
874+-- The sampling stage would strictly evaluate and sample, the evaluation stage
875+-- would control the evaluation rate.
877+-- | Evaluate the input stream continuously and keep only the oldest @n@
878+-- elements in the buffer, discard the new ones when the buffer is full. When
879+-- the output stream is evaluated it consumes the values from the buffer in a
884+{-# INLINE sampleOld #-}
885+sampleOld :: -- (IsStream t, Monad m) =>
886+ Int -> t m a -> t m a
887+sampleOld = undefined
889+-- | Evaluate the input stream continuously and keep only the latest @n@
890+-- elements in a ring buffer, keep discarding the older ones to make space for
891+-- the new ones. When the output stream is evaluated it consumes the values
892+-- from the buffer in a FIFO manner.
896+{-# INLINE sampleNew #-}
897+sampleNew :: -- (IsStream t, Monad m) =>
898+ Int -> t m a -> t m a
899+sampleNew = undefined
901+-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
902+-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
907+{-# INLINE sampleRate #-}
908+sampleRate :: -- (IsStream t, Monad m) =>
909+ Double -> t m a -> t m a
910+sampleRate = undefined
736------------------------------------------------------------------------------
912------------------------------------------------------------------------------
737-- Trimming
913-- Trimming
738------------------------------------------------------------------------------
914------------------------------------------------------------------------------
@@ -744,6 +920,51 @@ deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
744takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
920takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
745takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
921takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
923+-- See the lastN fold for impl hints. Use a Data.Array based ring buffer.
925+-- takeEnd n = Stream.concatM . fmap Array.toStream . fold Fold.lastN
927+-- | Take @n@ elements at the end of the stream.
929+-- O(n) space, where n is the number elements taken.
932+{-# INLINE takeEnd #-}
933+takeEnd :: -- (IsStream t, Monad m) =>
934+ Int -> t m a -> t m a
935+takeEnd = undefined -- fromStreamD $ D.takeEnd n $ toStreamD m
937+-- | Take time interval @i@ seconds at the end of the stream.
939+-- O(n) space, where n is the number elements taken.
942+{-# INLINE takeEndInterval #-}
943+takeEndInterval :: -- (IsStream t, Monad m) =>
944+ Double -> t m a -> t m a
945+takeEndInterval = undefined -- fromStreamD $ D.takeEnd n $ toStreamD m
947+-- | Take all consecutive elements at the end of the stream for which the
948+-- predicate is true.
950+-- O(n) space, where n is the number elements taken.
953+{-# INLINE takeWhileEnd #-}
954+takeWhileEnd :: -- (IsStream t, Monad m) =>
955+ (a -> Bool) -> t m a -> t m a
956+takeWhileEnd = undefined -- fromStreamD $ D.takeWhileEnd n $ toStreamD m
958+-- | Like 'takeWhile' and 'takeWhileEnd' combined.
960+-- O(n) space, where n is the number elements taken from the end.
963+{-# INLINE takeWhileAround #-}
964+takeWhileAround :: -- (IsStream t, Monad m) =>
965+ (a -> Bool) -> t m a -> t m a
966+takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
747-- | @takeByTime duration@ yields stream elements upto specified time
968-- | @takeByTime duration@ yields stream elements upto specified time
748-- @duration@. The duration starts when the stream is evaluated for the first
969-- @duration@. The duration starts when the stream is evaluated for the first
749-- time, before the first element is yielded. The time duration is checked
970-- time, before the first element is yielded. The time duration is checked
@@ -797,6 +1018,47 @@ dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
797dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1018dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
798dropByTime d = fromStreamD . D.dropByTime d . toStreamD
1019dropByTime d = fromStreamD . D.dropByTime d . toStreamD
1021+-- | Drop @n@ elements at the end of the stream.
1023+-- O(n) space, where n is the number elements dropped.
1026+{-# INLINE dropEnd #-}
1027+dropEnd :: -- (IsStream t, Monad m) =>
1028+ Int -> t m a -> t m a
1029+dropEnd = undefined -- fromStreamD $ D.dropEnd n $ toStreamD m
1031+-- | Drop time interval @i@ seconds at the end of the stream.
1033+-- O(n) space, where n is the number elements dropped.
1036+{-# INLINE dropEndInterval #-}
1037+dropEndInterval :: -- (IsStream t, Monad m) =>
1038+ Int -> t m a -> t m a
1039+dropEndInterval = undefined
1041+-- | Drop all consecutive elements at the end of the stream for which the
1042+-- predicate is true.
1044+-- O(n) space, where n is the number elements dropped.
1047+{-# INLINE dropWhileEnd #-}
1048+dropWhileEnd :: -- (IsStream t, Monad m) =>
1049+ (a -> Bool) -> t m a -> t m a
1050+dropWhileEnd = undefined -- fromStreamD $ D.dropWhileEnd n $ toStreamD m
1052+-- | Like 'dropWhile' and 'dropWhileEnd' combined.
1054+-- O(n) space, where n is the number elements dropped from the end.
1057+{-# INLINE dropWhileAround #-}
1058+dropWhileAround :: -- (IsStream t, Monad m) =>
1059+ (a -> Bool) -> t m a -> t m a
1060+dropWhileAround = undefined -- fromStreamD $ D.dropWhileAround n $ toStreamD m
800------------------------------------------------------------------------------
1062------------------------------------------------------------------------------
801-- Inserting Elements
1063-- Inserting Elements
802------------------------------------------------------------------------------
1064------------------------------------------------------------------------------
@@ -844,7 +1106,6 @@ intersperse a = fromStreamS . S.intersperse a . toStreamS
844intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1106intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
845intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
1107intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
848-- | Intersperse a monadic action into the input stream after every @n@
1109-- | Intersperse a monadic action into the input stream after every @n@
849-- elements.
1110-- elements.
@@ -853,11 +1114,11 @@ intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
853-- "he,ll,o"
1114-- "he,ll,o"
856−-- @since 0.7.0
1117+-- /Unimplemented/
857{-# INLINE intersperseBySpan #-}
1118{-# INLINE intersperseBySpan #-}
858−intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
1119+intersperseBySpan :: -- IsStream t =>
1120+ Int -> m a -> t m a -> t m a
859intersperseBySpan _n _f _xs = undefined
1121intersperseBySpan _n _f _xs = undefined
862-- | Insert an effect and its output after consuming an element of a stream.
1123-- | Insert an effect and its output after consuming an element of a stream.
@@ -883,6 +1144,8 @@ intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
883intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
1144intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
884intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
1145intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
1147+-- XXX Use an offset argument, like tapOffsetEvery
886-- | Like 'intersperseSuffix' but intersperses an effectful action into the
1149-- | Like 'intersperseSuffix' but intersperses an effectful action into the
887-- input stream after every @n@ elements and after the last element.
1150-- input stream after every @n@ elements and after the last element.
@@ -906,7 +1169,7 @@ intersperseSuffixBySpan n eff =
906-- > .h.e.l.l.o"hello"
1169-- > .h.e.l.l.o"hello"
909−-- See also: 'trace_'
1172+-- Same as 'trace_' but may be concurrent.
911-- /Concurrent/
1174-- /Concurrent/