@@ -28,6 +28,12 @@
2828-- Similarly, a partitioning applicative can partition the input among the
2929-- constituent folds. There are many other ways to combine folds.
3030--
31+-- All combinators in this module are of true streaming nature. For example,
32+-- the splitting combinators accept a fold which consumes the elements in the
33+-- split segments as soon as they are generated by the producer, segments are
34+-- not buffered in memory, resulting in constant memory usage. The folds used
35+-- in these combinators can be considered as continuations.
36+--
3137-- > import qualified Streamly.Fold as FL
3238--
3339--
@@ -192,6 +198,7 @@ module Streamly.Fold
192198 , break -- breakPre
193199 -- , breakPost
194200 -- , breakOn
201+ -- , breakAround
195202 -- , spanBy
196203 -- , spanRollingBy
197204
@@ -199,19 +206,16 @@ module Streamly.Fold
199206 -- breakOnSeq
200207
201208 -- ** Splitting
209+ -- | Streams can be split into segments in space or in time. We use the
210+ -- term @chunk@ to refer to a spatial length of the stream (spatial window)
211+ -- and the term @session@ to refer to a length in time (time window).
202212
203− -- In imperative terms grouped folding can be considered as a nested loop
213+ -- In imperative terms, grouped folding can be considered as a nested loop
204214 -- where we loop over the stream to group elements and then loop over
205215 -- individual groups to fold them to a single value that is yielded in the
206216 -- output stream.
207− --
208− -- Note that these grouping folds are true streaming folds that never
209− -- accumulate the group in memory before folding, i.e. the group elements
210− -- are consumed by the folds as they are yielded by the stream. Therefore,
211− -- the whole computation runs in constant space.
212− -- In contrast, we can simply use a scan on the stream to buffer the whole
213− -- groups in memory and then map a fold on it to fold the groups. This kind
214− -- of grouping and folding would not work well when the group size is big.
217+
218+ -- , groupScan
215219
216220 -- *** By Chunks
217221 , chunksOf
@@ -335,7 +339,9 @@ module Streamly.Fold
335339 , classifySlidingChunks
336340 , classifySlidingSessions
337341 -}
338−
342+ -- ** Sliding Window Buffers
343+ -- , slidingChunkBuffer
344+ -- , slidingSessionBuffer
339345 )
340346where
341347
@@ -362,9 +368,9 @@ import qualified Data.Map.Strict as Map
362368import Streamly (MonadAsync, parallel)
363369import Streamly.Fold.Types (Fold(..))
364370import Streamly.Mem.Array (Array)
371+-- import Streamly.Mem.Ring (Ring)
365372import Streamly.Streams.Serial (SerialT)
366373import Streamly.Streams.StreamK (IsStream())
367−-- import Streamly.Time.Clock (Clock(..), getTime)
368374import Streamly.Time.Units
369375 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
370376 toAbsTime)
@@ -1044,6 +1050,45 @@ splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
10441050-- N-ary APIs
10451051------------------------------------------------------------------------------
10461052
1053+------------------------------------------------------------------------------
1054+-- Generalized grouping
1055+------------------------------------------------------------------------------
1056+
1057+-- This combinator is the most general grouping combinator and can be used to
1058+-- implement all other grouping combinators.
1059+--
1060+-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1061+-- new elements, slide out old elements and incrementally compute the hash.
1062+-- Also, can we implement the windowed classification combinators using this?
1063+--
1064+-- In fact this is a parse. Instead of using a special return value in the fold
1065+-- we are using a mapping function.
1066+--
1067+-- Note that 'scanl'' (usually followed by a map to extract the desired value
1068+-- from the accumulator) can be used to realize many implementations e.g. a
1069+-- sliding window implementation. A scan followed by a mapMaybe is also a good
1070+-- pattern to express many problems where we want to emit a filtered output and
1071+-- not emit an output on every input.
1072+--
1073+-- Passing on of the initial accumulator value to the next fold is equivalent
1074+-- to returning the leftover concept.
1075+
1076+{-
1077+-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1078+-- @splitter@ is applied on the accumulator of the fold every time an item is
1079+-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1080+-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1081+-- in the output stream and the initial value of the accumulator for the next
1082+-- group's fold. This allows us to control whether to start fresh for the next
1083+-- fold or to continue from the previous fold's output.
1084+--
1085+{-# INLINE groupScan #-}
1086+groupScan
1087+ :: (IsStream t, Monad m)
1088+ => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1089+groupScan split fold m = undefined
1090+-}
1091+
10471092-- | Group the input stream into groups of @n@ elements each and then fold each
10481093-- group using the provided fold function.
10491094--
@@ -1688,21 +1733,21 @@ splitSuffixOnAny subseq f m = undefined
16881733-}
16891734
16901735------------------------------------------------------------------------------
1691−-- Grouped by order
1736+-- Reorder in sequence
16921737------------------------------------------------------------------------------
16931738
16941739{-
16951740-- Buffer until the next element in sequence arrives. The function argument
16961741-- determines the difference in sequence numbers. This could be useful in
16971742-- implementing sequenced streams, for example, TCP reassembly.
1698−{-# INLINE foldOrderedBy #-}
1699−foldOrderedBy
1743+{-# INLINE reassembleBy #-}
1744+reassembleBy
17001745 :: (IsStream t, Monad m)
1701− => (forall n. Monad n => Fold n a b)
1746+ => Fold m a b
17021747 -> (a -> a -> Int)
17031748 -> t m a
17041749 -> t m b
1705−foldOrderedBy = undefined
1750+reassembleBy = undefined
17061751-}
17071752
17081753------------------------------------------------------------------------------
@@ -2187,6 +2232,48 @@ classifySlidingSessions tick interval slide (Fold step initial extract) str
21872232 = undefined
21882233-}
21892234
2235+------------------------------------------------------------------------------
2236+-- Sliding Window Buffers
2237+------------------------------------------------------------------------------
2238+
2239+-- These buffered versions could be faster than concurrent incremental folds of
2240+-- all overlapping windows as in many cases we may not need all the values to
2241+-- compute the fold, we can just compute the result using the old value and new
2242+-- value. However, we may need the buffer once in a while, for example for
2243+-- string search we usually compute the hash incrementally but when the hash
2244+-- matches the hash of the pattern we need to compare the whole string.
2245+--
2246+-- XXX we should be able to implement sequence based splitting combinators
2247+-- using this combinator.
2248+
2249+{-
2250+-- | Buffer n elements of the input in a ring buffer. When t new elements are
2251+-- collected, slide the window to remove the same number of oldest elements,
2252+-- insert the new elements, and apply an incremental fold on the sliding
2253+-- window, supplying the outgoing elements, the new ring buffer as arguments.
2254+slidingChunkBuffer
2255+ :: (IsStream t, Monad m, Ord a, Storable a)
2256+ => Int -- window size
2257+ -> Int -- window slide
2258+ -> Fold m (Ring a, Array a) b
2259+ -> t m a
2260+ -> t m b
2261+slidingChunkBuffer = undefined
2262+
2263+-- Buffer n seconds worth of stream elements of the input in a radix tree.
2264+-- Every t seconds, remove the items that are older than n seconds, and apply
2265+-- an incremental fold on the sliding window, supplying the outgoing elements,
2266+-- and the new radix tree buffer as arguments.
2267+slidingSessionBuffer
2268+ :: (IsStream t, Monad m, Ord a, Storable a)
2269+ => Int -- window size
2270+ -> Int -- tick size
2271+ -> Fold m (RTree a, Array a) b
2272+ -> t m a
2273+ -> t m b
2274+slidingSessionBuffer = undefined
2275+-}
2276+
21902277------------------------------------------------------------------------------
21912278-- Keyed Session Windows
21922279------------------------------------------------------------------------------