@@ -24,6 +24,7 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
2424 , smapM
2525
2626 -- * Mapping Side Effects (Observation)
27+ -- | See also the intersperse*_ combinators.
2728 , trace
2829 , trace_
2930 , tap
@@ -62,59 +63,72 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
6263 -- , lprescanlM'
6364
6465 -- * 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.
6669
70+ , with
71+ , deleteBy
6772 , filter
6873 , filterM
69− , deleteBy
7074 , uniq
71− -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
72− -- , uniqOn -- to remove duplicate sequences
73− -- , pruneBy -- dropAround + uniqBy - like words
75+ , uniqBy
76+ , nubBy
77+ , nubWindowBy
78+ , prune
79+ , repeated
7480
7581 -- * Trimming
7682 -- | Produce a subset of the stream trimmed at ends.
7783
7884 , take
79− -- , takeGE
80− -- , takeBetween
81− , takeByTime
82− -- , takeEnd
85+ , takeByTime -- takeInterval
86+ , takeEnd
87+ , takeEndInterval
8388 , takeWhile
8489 , takeWhileM
85− -- , takeWhileEnd
90+ , takeWhileEnd
91+ , takeWhileAround
8692 , drop
87− , dropByTime
88− -- , dropEnd
93+ , dropByTime -- dropInterval
94+ , dropEnd
95+ , dropEndInterval
8996 , dropWhile
9097 , dropWhileM
91− -- , dropWhileEnd
92− -- , dropAround
98+ , dropWhileEnd
99+ , dropWhileAround
93100
94101 -- * 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.
96105
97− , insertBy
106+ -- Element agnostic (Opposite of sampling)
98107 , intersperse
99108 , intersperseM -- XXX naming
109+ , intersperseBySpan
110+
100111 , intersperseSuffix
101112 , intersperseSuffixBySpan
102− -- , intersperseBySpan
103− -- , intersperseByIndices -- using an index function/stream
113+ , interjectSuffix
104114
105− -- * Inserting Side Effects
106− , intersperseM_ -- XXX naming
107− , intersperseSuffix_
108− , interspersePrefix_
115+ -- , interspersePrefix
116+ -- , interspersePrefixBySpan
109117
110− -- * Inserting Time
111− -- , intersperseByTime
112− -- , intersperseByEvent
113− , interjectSuffix
118+ -- * Inserting Side Effects/Time
119+ , intersperseM_ -- XXX naming
114120 , delay
121+ , intersperseSuffix_
115122 , delayPost
123+ , interspersePrefix_
116124 , delayPre
117125
126+ -- * Element Aware Insertion
127+ -- | Opposite of filtering
128+ , insertBy
129+ -- , intersperseByBefore
130+ -- , intersperseByAfter
131+
118132 -- * Reordering
119133 , reverse
120134 , reverse'
@@ -151,7 +165,7 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
151165 , lefts
152166 , rights
153167
154− -- * Threading
168+ -- * Concurrent Evaluation
155169 -- ** Concurrent Pipelines
156170 -- | Run streaming stages concurrently.
157171
@@ -162,9 +176,20 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
162176
163177 -- ** Concurrency Control
164178 , maxThreads
179+
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.
165185 , maxBuffer
186+ , sampleOld
187+ , sampleNew
188+ , sampleRate
166189
167190 -- ** Rate Limiting
191+ -- | Evaluate the stream at uniform intervals to maintain a specified
192+ -- evaluation rate.
168193 , Rate (..)
169194 , rate
170195 , avgRate
@@ -215,7 +240,7 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)
215240import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
216241import Streamly.Internal.Data.Stream.StreamK (IsStream)
217242import 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)
219244
220245import qualified Streamly.Internal.Data.Fold as FL
221246import qualified Streamly.Internal.Data.Stream.Parallel as Par
@@ -519,7 +544,9 @@ trace f = mapM (\x -> void (f x) >> return x)
519544-- "got here"
520545-- @
521546--
522−-- See also: 'trace', 'interspersePrefix_'
547+-- Same as 'interspersePrefix_' but always serial.
548+--
549+-- See also: 'trace'
523550--
524551-- /Internal/
525552{-# INLINE trace_ #-}
@@ -693,6 +720,24 @@ scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
693720-- Filtering
694721------------------------------------------------------------------------------
695722
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.
726+--
727+-- @
728+-- filterWithIndex = with indexed filter
729+-- filterWithAbsTime = with timestamped filter
730+-- filterWithRelTime = with timeIndexed filter
731+-- @
732+--
733+-- /Internal/
734+{-# INLINE with #-}
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
740+
696741-- | Include only those elements that pass a predicate.
697742--
698743-- @since 0.1.0
@@ -713,6 +758,31 @@ filter = K.filter
713758filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
714759filterM p m = fromStreamD $ D.filterM p $ toStreamD m
715760
761+-- | Drop repeated elements that are adjacent to each other using the supplied
762+-- comparison function.
763+--
764+-- @uniq = uniqBy (==)
765+--
766+-- To strip duplicate path separators:
767+--
768+-- @
769+-- f x y = x == '/' && x == y
770+-- Stream.toList $ Stream.uniqBy f $ Stream.fromList "//a//b"
771+-- "/a/b"
772+-- @
773+--
774+-- Space: @O(1)@
775+--
776+-- See also: 'nubBy'.
777+--
778+-- /Internal/
779+--
780+{-# INLINE uniqBy #-}
781+uniqBy :: (IsStream t, Monad m, Functor (t m)) =>
782+ (a -> a -> Bool) -> t m a -> t m a
783+uniqBy eq =
784+ catMaybes . rollingMap (\x y -> if x `eq` y then Nothing else Just y)
785+
716786-- | Drop repeated elements that are adjacent to each other.
717787--
718788-- @since 0.6.0
@@ -720,6 +790,67 @@ filterM p m = fromStreamD $ D.filterM p $ toStreamD m
720790uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
721791uniq = fromStreamD . D.uniq . toStreamD
722792
793+-- | Strip all leading and trailing occurrences of an element passing a
794+-- predicate and make all other consecutive occurrences uniq.
795+--
796+-- @
797+-- prune p = dropWhileAround p $ uniqBy (x y -> p x && p y)
798+-- @
799+--
800+-- >>> Stream.pruneBy isSpace (Stream.fromList " hello world! ")
801+-- "hello world!"
802+--
803+-- Space: @O(1)@
804+--
805+-- /Unimplemented/
806+{-# INLINE prune #-}
807+prune ::
808+ -- (IsStream t, Monad m, Eq a) =>
809+ (a -> Bool) -> t m a -> t m a
810+prune = error "Not implemented yet!"
811+
812+-- Possible implementation:
813+-- @repeated =
814+-- Stream.catMaybes . Stream.parseMany (Parser.groupBy (==) Fold.repeated)@
815+--
816+-- 'Fold.repeated' should return 'Just' when repeated, and 'Nothing' for a
817+-- single element.
818+--
819+-- | Emit only repeated elements, once.
820+--
821+-- /Unimplemented/
822+repeated :: -- (IsStream t, Monad m, Eq a) =>
823+ t m a -> t m a
824+repeated = undefined
825+
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.
829+--
830+-- | Drop repeated elements anywhere in the stream.
831+--
832+-- /Caution: not scalable for infinite streams/
833+--
834+-- /See also: nubWindowBy/
835+--
836+-- /Unimplemented/
837+--
838+{-# INLINE nubBy #-}
839+nubBy :: -- (IsStream t, Monad m) =>
840+ (a -> a -> Bool) -> t m a -> t m a
841+nubBy = undefined -- fromStreamD . D.nubBy . toStreamD
842+
843+-- | Drop repeated elements within the specified tumbling window in the stream.
844+--
845+-- @nubBy = nubWindowBy maxBound@
846+--
847+-- /Unimplemented/
848+--
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
853+
723854-- | Deletes the first occurrence of the element in the stream that satisfies
724855-- the given equality predicate.
725856--
@@ -733,6 +864,51 @@ uniq = fromStreamD . D.uniq . toStreamD
733864deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
734865deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
735866
867+------------------------------------------------------------------------------
868+-- Lossy Buffering
869+------------------------------------------------------------------------------
870+
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.
876+
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
880+-- FIFO manner.
881+--
882+-- /Unimplemented/
883+--
884+{-# INLINE sampleOld #-}
885+sampleOld :: -- (IsStream t, Monad m) =>
886+ Int -> t m a -> t m a
887+sampleOld = undefined
888+
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.
893+--
894+-- /Unimplemented/
895+--
896+{-# INLINE sampleNew #-}
897+sampleNew :: -- (IsStream t, Monad m) =>
898+ Int -> t m a -> t m a
899+sampleNew = undefined
900+
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
903+-- consumer pattern.
904+--
905+-- /Unimplemented/
906+--
907+{-# INLINE sampleRate #-}
908+sampleRate :: -- (IsStream t, Monad m) =>
909+ Double -> t m a -> t m a
910+sampleRate = undefined
911+
736912------------------------------------------------------------------------------
737913-- Trimming
738914------------------------------------------------------------------------------
@@ -744,6 +920,51 @@ deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
744920takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
745921takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
746922
923+-- See the lastN fold for impl hints. Use a Data.Array based ring buffer.
924+--
925+-- takeEnd n = Stream.concatM . fmap Array.toStream . fold Fold.lastN
926+--
927+-- | Take @n@ elements at the end of the stream.
928+--
929+-- O(n) space, where n is the number elements taken.
930+--
931+-- /Unimplemented/
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
936+
937+-- | Take time interval @i@ seconds at the end of the stream.
938+--
939+-- O(n) space, where n is the number elements taken.
940+--
941+-- /Unimplemented/
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
946+
947+-- | Take all consecutive elements at the end of the stream for which the
948+-- predicate is true.
949+--
950+-- O(n) space, where n is the number elements taken.
951+--
952+-- /Unimplemented/
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
957+
958+-- | Like 'takeWhile' and 'takeWhileEnd' combined.
959+--
960+-- O(n) space, where n is the number elements taken from the end.
961+--
962+-- /Unimplemented/
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
967+
747968-- | @takeByTime duration@ yields stream elements upto specified time
748969-- @duration@. The duration starts when the stream is evaluated for the first
749970-- 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
7971018dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
7981019dropByTime d = fromStreamD . D.dropByTime d . toStreamD
7991020
1021+-- | Drop @n@ elements at the end of the stream.
1022+--
1023+-- O(n) space, where n is the number elements dropped.
1024+--
1025+-- /Unimplemented/
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
1030+
1031+-- | Drop time interval @i@ seconds at the end of the stream.
1032+--
1033+-- O(n) space, where n is the number elements dropped.
1034+--
1035+-- /Unimplemented/
1036+{-# INLINE dropEndInterval #-}
1037+dropEndInterval :: -- (IsStream t, Monad m) =>
1038+ Int -> t m a -> t m a
1039+dropEndInterval = undefined
1040+
1041+-- | Drop all consecutive elements at the end of the stream for which the
1042+-- predicate is true.
1043+--
1044+-- O(n) space, where n is the number elements dropped.
1045+--
1046+-- /Unimplemented/
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
1051+
1052+-- | Like 'dropWhile' and 'dropWhileEnd' combined.
1053+--
1054+-- O(n) space, where n is the number elements dropped from the end.
1055+--
1056+-- /Unimplemented/
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
1061+
8001062------------------------------------------------------------------------------
8011063-- Inserting Elements
8021064------------------------------------------------------------------------------
@@ -844,7 +1106,6 @@ intersperse a = fromStreamS . S.intersperse a . toStreamS
8441106intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
8451107intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
8461108
847−{-
8481109-- | Intersperse a monadic action into the input stream after every @n@
8491110-- elements.
8501111--
@@ -853,11 +1114,11 @@ intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
8531114-- "he,ll,o"
8541115-- @
8551116--
856−-- @since 0.7.0
1117+-- /Unimplemented/
8571118{-# 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
8591121intersperseBySpan _n _f _xs = undefined
860−-}
8611122
8621123-- | Insert an effect and its output after consuming an element of a stream.
8631124--
@@ -883,6 +1144,8 @@ intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
8831144intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
8841145intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
8851146
1147+-- XXX Use an offset argument, like tapOffsetEvery
1148+--
8861149-- | Like 'intersperseSuffix' but intersperses an effectful action into the
8871150-- input stream after every @n@ elements and after the last element.
8881151--
@@ -906,7 +1169,7 @@ intersperseSuffixBySpan n eff =
9061169-- > .h.e.l.l.o"hello"
9071170-- @
9081171--
909−-- See also: 'trace_'
1172+-- Same as 'trace_' but may be concurrent.
9101173--
9111174-- /Concurrent/
9121175--