@@ -242,14 +242,12 @@ import Prelude
242242-- import qualified Data.Map.Strict as Map
243243-- import qualified Prelude
244244
245−-- import Streamly.Prelude (MonadAsync, parallel)
246245-- import Streamly.Data.Fold.Types (Fold(..))
247246import Streamly.Internal.Data.Pipe.Type
248247 (Pipe(..), PipeState(..), Step(..), zipWith, tee, map, compose)
249248-- import Streamly.Internal.Data.Array.Unboxed.Type (Array)
250249-- import Streamly.Internal.Data.Ring.Unboxed (Ring)
251250-- import Streamly.Internal.Data.Stream (Stream)
252−-- import Streamly.Internal.Data.Stream.StreamK (IsStream())
253251-- import Streamly.Internal.Data.Time.Units
254252-- (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
255253-- toAbsTime)
@@ -276,1693 +274,3 @@ mapM f = Pipe consume undefined ()
276274 consume _ a = do
277275 r <- f a
278276 return $ Yield r (Consume ())
279−{-
280−------------------------------------------------------------------------------
281−-- Filtering
282−------------------------------------------------------------------------------
283−
284−-- | Include only those elements that pass a predicate.
285−--
286−-- >>> FL.foldl (lfilter (> 5) FL.sum) [1..10]
287−-- 40
288−--
289−-- @since 0.7.0
290−{-# INLINABLE lfilter #-}
291−lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
292−lfilter f (Fold step begin done) = Fold step' begin done
293− where
294− step' x a = if f a then step x a else return x
295−
296−-- | Like 'lfilter' but with a monadic predicate.
297−--
298−-- @since 0.7.0
299−{-# INLINABLE lfilterM #-}
300−lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
301−lfilterM f (Fold step begin done) = Fold step' begin done
302− where
303− step' x a = do
304− use <- f a
305− if use then step x a else return x
306−
307−-- | Take first 'n' elements from the stream and discard the rest.
308−--
309−-- @since 0.7.0
310−{-# INLINABLE ltake #-}
311−ltake :: Monad m => Int -> Fold m a b -> Fold m a b
312−ltake n (Fold step initial done) = Fold step' initial' done'
313− where
314− initial' = fmap (Tuple' 0) initial
315− step' (Tuple' i r) a = do
316− if i < n
317− then do
318− res <- step r a
319− return $ Tuple' (i + 1) res
320− else return $ Tuple' i r
321− done' (Tuple' _ r) = done r
322−
323−-- | Takes elements from the input as long as the predicate succeeds.
324−--
325−-- @since 0.7.0
326−{-# INLINABLE ltakeWhile #-}
327−ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
328−ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
329− where
330− initial' = fmap Left' initial
331− step' (Left' r) a = do
332− if predicate a
333− then fmap Left' $ step r a
334− else return (Right' r)
335− step' r _ = return r
336− done' (Left' r) = done r
337− done' (Right' r) = done r
338−
339−------------------------------------------------------------------------------
340−-- Grouping/Splitting
341−------------------------------------------------------------------------------
342−
343−------------------------------------------------------------------------------
344−-- Grouping without looking at elements
345−------------------------------------------------------------------------------
346−--
347−------------------------------------------------------------------------------
348−-- Binary APIs
349−------------------------------------------------------------------------------
350−--
351−
352−-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
353−-- elements of its input are sent to fold @f1@ and the rest of the stream is
354−-- sent to fold @f2@.
355−--
356−-- > let splitAt_ n xs = FL.foldl' (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
357−--
358−-- >>> splitAt_ 6 "Hello World!"
359−-- > ("Hello ","World!")
360−--
361−-- >>> splitAt_ (-1) [1,2,3]
362−-- > ([],[1,2,3])
363−--
364−-- >>> splitAt_ 0 [1,2,3]
365−-- > ([],[1,2,3])
366−--
367−-- >>> splitAt_ 1 [1,2,3]
368−-- > ([1],[2,3])
369−--
370−-- >>> splitAt_ 3 [1,2,3]
371−-- > ([1,2,3],[])
372−--
373−-- >>> splitAt_ 4 [1,2,3]
374−-- > ([1,2,3],[])
375−--
376−-- This can be considered as a two-fold version of 'ltake' where we take both
377−-- the segments instead of discarding the leftover.
378−--
379−-- @since 0.7.0
380−{-# INLINE splitAt #-}
381−splitAt
382− :: Monad m
383− => Int
384− -> Fold m a b
385− -> Fold m a c
386− -> Fold m a (b, c)
387−splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
388− Fold step init extract
389− where
390− init = Tuple3' <$> return n <*> initialL <*> initialR
391−
392− step (Tuple3' i xL xR) input =
393− if i > 0
394− then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
395− else stepR xR input >>= (\b -> return (Tuple3' i xL b))
396−
397− extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
398−
399−------------------------------------------------------------------------------
400−-- N-ary APIs
401−------------------------------------------------------------------------------
402−
403−------------------------------------------------------------------------------
404−-- Generalized grouping
405−------------------------------------------------------------------------------
406−
407−-- This combinator is the most general grouping combinator and can be used to
408−-- implement all other grouping combinators.
409−--
410−-- XXX check if this can implement the splitOn combinator i.e. we can slide in
411−-- new elements, slide out old elements and incrementally compute the hash.
412−-- Also, can we implement the windowed classification combinators using this?
413−--
414−-- In fact this is a parse. Instead of using a special return value in the fold
415−-- we are using a mapping function.
416−--
417−-- Note that 'scanl'' (usually followed by a map to extract the desired value
418−-- from the accumulator) can be used to realize many implementations e.g. a
419−-- sliding window implementation. A scan followed by a mapMaybe is also a good
420−-- pattern to express many problems where we want to emit a filtered output and
421−-- not emit an output on every input.
422−--
423−-- Passing on of the initial accumulator value to the next fold is equivalent
424−-- to returning the leftover concept.
425−
426−{-
427−-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
428−-- @splitter@ is applied on the accumulator of the fold every time an item is
429−-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
430−-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
431−-- in the output stream and the initial value of the accumulator for the next
432−-- group's fold. This allows us to control whether to start fresh for the next
433−-- fold or to continue from the previous fold's output.
434−--
435−{-# INLINE groupScan #-}
436−groupScan
437− :: (IsStream t, Monad m)
438− => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
439−groupScan split fold m = undefined
440−-}
441−
442−-- | Group the input stream into groups of @n@ elements each and then fold each
443−-- group using the provided fold function.
444−--
445−-- >> S.toList $ FL.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
446−-- > [3,7,11,15,19]
447−--
448−-- This can be considered as an n-fold version of 'ltake' where we apply
449−-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
450−--
451−-- @since 0.7.0
452−{-# INLINE chunksOf #-}
453−chunksOf
454− :: (IsStream t, Monad m)
455− => Int -> Fold m a b -> t m a -> t m b
456−chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
457−
458−-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
459−-- 'Just' values.
460−{-# INLINE lcatMaybes #-}
461−lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
462−lcatMaybes = lfilter isJust . lmap fromJust
463−
464−-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
465−-- XXX add this example after fixing the serial stream rate control
466−-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
467−-- > [3,7,11,15,19]
468−--
469−-- | Group the input stream into windows of @n@ second each and then fold each
470−-- group using the provided fold function.
471−--
472−-- @since 0.7.0
473−{-# INLINE sessionsOf #-}
474−sessionsOf
475− :: (IsStream t, MonadAsync m)
476− => Double -> Fold m a b -> t m a -> t m b
477−sessionsOf n f xs =
478− splitSuffixBy' isNothing (lcatMaybes f)
479− (S.intersperseByTime n (return Nothing) (S.map Just xs))
480−
481−------------------------------------------------------------------------------
482−-- Element Aware APIs
483−------------------------------------------------------------------------------
484−--
485−------------------------------------------------------------------------------
486−-- Binary APIs
487−------------------------------------------------------------------------------
488−
489−-- | Break the input stream into two groups, the first group takes the input as
490−-- long as the predicate applied to the first element of the stream and next
491−-- input element holds 'True', the second group takes the rest of the input.
492−--
493−_spanBy
494− :: Monad m
495− => (a -> a -> Bool)
496− -> Fold m a b
497− -> Fold m a c
498− -> Fold m a (b, c)
499−_spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
500− Fold step init extract
501−
502− where
503− init = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
504−
505− step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
506− if cmp frst input && isFirstG
507− then stepL a input
508− >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
509− else stepR b input
510− >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
511−
512− step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
513− if isFirstG
514− then stepL a input
515− >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
516− else stepR b input
517− >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
518−
519− extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
520−
521−-- | Span as long as the predicate is 'True'. @span p f1 f2@ composes folds
522−-- @f1@ and @f2@ such that the composed fold continues sending the input to
523−-- @f1@ as long as the predicate @p@ is 'True'. The rest of the input is sent
524−-- to @f2@.
525−--
526−-- > let span_ p xs = FL.foldl' (FL.span p FL.toList FL.toList) $ S.fromList xs
527−--
528−-- >>> span_ (< 1) [1,2,3]
529−-- > ([],[1,2,3])
530−--
531−-- >>> span_ (< 2) [1,2,3]
532−-- > ([1],[2,3])
533−--
534−-- >>> span_ (< 4) [1,2,3]
535−-- > ([1,2,3],[])
536−--
537−-- This can be considered as a two-fold version of 'ltakeWhile' where we take
538−-- both the segments instead of discarding the leftover.
539−--
540−-- @since 0.7.0
541−{-# INLINE span #-}
542−span
543− :: Monad m
544− => (a -> Bool)
545− -> Fold m a b
546− -> Fold m a c
547− -> Fold m a (b, c)
548−span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
549− Fold step init extract
550−
551− where
552− init = Tuple3' <$> initialL <*> initialR <*> return True
553−
554− step (Tuple3' a b isFirstG) input =
555− if isFirstG && p input
556− then stepL a input >>= (\a' -> return (Tuple3' a' b True))
557− else stepR b input >>= (\a' -> return (Tuple3' a a' False))
558−
559− extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
560−
561−-- |
562−-- > break p = span (not . p)
563−--
564−-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
565−-- folds @f1@ and @f2@ such that @f1@ stops receiving input as soon as the
566−-- predicate @p@ becomes 'True'. The rest of the input is sent to @f2@.
567−--
568−-- This is the binary version of 'splitBy'.
569−--
570−-- > let break_ p xs = FL.foldl' (FL.break p FL.toList FL.toList) $ S.fromList xs
571−--
572−-- >>> break_ (< 1) [3,2,1]
573−-- > ([3,2,1],[])
574−--
575−-- >>> break_ (< 2) [3,2,1]
576−-- > ([3,2],[1])
577−--
578−-- >>> break_ (< 4) [3,2,1]
579−-- > ([],[3,2,1])
580−--
581−-- @since 0.7.0
582−{-# INLINE break #-}
583−break
584− :: Monad m
585− => (a -> Bool)
586− -> Fold m a b
587− -> Fold m a c
588− -> Fold m a (b, c)
589−break p = span (not . p)
590−
591−-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
592−-- predicate is applied to the previous and the next input elements.
593−{-# INLINE _spanRollingBy #-}
594−_spanRollingBy
595− :: Monad m
596− => (a -> a -> Bool)
597− -> Fold m a b
598− -> Fold m a c
599− -> Fold m a (b, c)
600−_spanRollingBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
601− Fold step init extract
602−
603− where
604− init = Tuple3' <$> initialL <*> initialR <*> return Nothing
605−
606− step (Tuple3' a b (Just frst)) input =
607− if cmp input frst
608− then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
609− else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
610−
611− step (Tuple3' a b Nothing) input =
612− stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
613−
614− extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
615−
616−------------------------------------------------------------------------------
617−-- N-ary APIs
618−------------------------------------------------------------------------------
619−--
620−-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
621−-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
622−-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
623−-- group and so on. When the comparison fails a new group is started. Each
624−-- group is folded using the fold @f@.
625−--
626−-- >>> S.toList $ FL.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
627−-- > [[1,3,7],[0,2,5]]
628−--
629−-- @since 0.7.0
630−{-# INLINE groupsBy #-}
631−groupsBy
632− :: (IsStream t, Monad m)
633− => (a -> a -> Bool)
634− -> Fold m a b
635− -> t m a
636− -> t m b
637−groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
638−
639−-- | Unlike @groupsBy@ this function performs a rolling comparison of two
640−-- successive elements in the input stream. @groupsRollingBy cmp f $ S.fromList
641−-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
642−-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
643−-- 'True' then @c@ is also assigned to the same group and so on. When the
644−-- comparison fails a new group is started. Each group is folded using the fold
645−-- @f@.
646−--
647−-- >>> S.toList $ FL.groupsRollingBy (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
648−-- > [[1,2,3],[7,8,9]]
649−--
650−-- @since 0.7.0
651−{-# INLINE groupsRollingBy #-}
652−groupsRollingBy
653− :: (IsStream t, Monad m)
654− => (a -> a -> Bool)
655− -> Fold m a b
656− -> t m a
657− -> t m b
658−groupsRollingBy cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
659−
660−-- |
661−-- > groups = groupsBy (==)
662−-- > groups = groupsRollingBy (==)
663−--
664−-- Groups a contiguous span of equal elements together in one group.
665−--
666−-- >>> S.toList $ FL.groups FL.toList $ S.fromList [1,1,2,2]
667−-- > [[1,1],[2,2]]
668−--
669−-- @since 0.7.0
670−groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
671−groups = groupsBy (==)
672−
673−------------------------------------------------------------------------------
674−-- Binary splitting on a separator
675−------------------------------------------------------------------------------
676−
677−{-
678−-- | Find the first occurrence of the specified sequence in the input stream
679−-- and break the input stream into two parts, the first part consisting of the
680−-- stream before the sequence and the second part consisting of the sequence
681−-- and the rest of the stream.
682−--
683−-- > let breakOn_ pat xs = FL.foldl' (FL.breakOn pat FL.toList FL.toList) $ S.fromList xs
684−--
685−-- >>> breakOn_ "dear" "Hello dear world!"
686−-- > ("Hello ","dear world!")
687−--
688−{-# INLINE breakOn #-}
689−breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
690−breakOn pat f m = undefined
691−-}
692−
693−------------------------------------------------------------------------------
694−-- N-ary split on a predicate
695−------------------------------------------------------------------------------
696−
697−-- TODO: Use a Splitter configuration similar to the "split" package to make it
698−-- possible to express all splitting combinations. In general, we can have
699−-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
700−-- separators. We can have a single split operation taking the splitter config
701−-- as argument.
702−
703−-- | Split a stream on separator elements determined by a predicate, dropping
704−-- the separator. Separators are not considered part of stream segments on
705−-- either side of it instead they are treated as infixed between two stream
706−-- segments. For example, with @.@ as separator, @"a.b.c"@ would be parsed as
707−-- @["a","b","c"]@. When @.@ is in leading or trailing position it is still
708−-- considered as infixed, treating the first or the last segment as empty. For
709−-- example, @".a."@ would be parsed as @["","a",""]@. This operation is
710−-- opposite of 'intercalate'.
711−--
712−-- Let's use the following definition for illustration:
713−--
714−-- > splitBy_ p xs = S.toList $ FL.splitBy p (FL.toList) (S.fromList xs)
715−--
716−-- >>> splitBy_ (== '.') ""
717−-- [""]
718−--
719−-- >>> splitBy_ (== '.') "."
720−-- ["",""]
721−--
722−-- >>> splitBy_ (== '.') ".a"
723−-- > ["","a"]
724−--
725−-- >>> splitBy_ (== '.') "a."
726−-- > ["a",""]
727−--
728−-- >>> splitBy_ (== '.') "a.b"
729−-- > ["a","b"]
730−--
731−-- >>> splitBy_ (== '.') "a..b"
732−-- > ["a","","b"]
733−--
734−-- This can be considered as an n-fold version of 'break' where we apply
735−-- 'break' successively on the input stream, dropping the first element
736−-- of the second segment after each break.
737−--
738−-- @since 0.7.0
739−{-# INLINE splitBy #-}
740−splitBy
741− :: (IsStream t, Monad m)
742− => (a -> Bool) -> Fold m a b -> t m a -> t m b
743−splitBy predicate f m =
744− D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
745−
746−-- | Like 'splitBy' but the separator is treated as part of the previous
747−-- stream segment (suffix). Therefore, when the separator is in trailing
748−-- position, no empty segment is considered to follow it. For example, @"a.b."@
749−-- would be parsed as @["a","b"]@ instead of @["a","b",""]@ as in the case of
750−-- 'splitBy'.
751−--
752−-- > splitSuffixBy_ p xs = S.toList $ FL.splitSuffixBy p (FL.toList) (S.fromList xs)
753−--
754−-- >>> splitSuffixBy_ (== '.') ""
755−-- []
756−--
757−-- >>> splitSuffixBy_ (== '.') "."
758−-- [""]
759−--
760−-- >>> splitSuffixBy_ (== '.') "a"
761−-- ["a"]
762−--
763−-- >>> splitSuffixBy_ (== '.') ".a"
764−-- > ["","a"]
765−--
766−-- >>> splitSuffixBy_ (== '.') "a."
767−-- > ["a"]
768−--
769−-- >>> splitSuffixBy_ (== '.') "a.b"
770−-- > ["a","b"]
771−--
772−-- >>> splitSuffixBy_ (== '.') "a.b."
773−-- > ["a","b"]
774−--
775−-- >>> splitSuffixBy_ (== '.') "a..b.."
776−-- > ["a","","b",""]
777−--
778−-- > lines = splitSuffixBy (== '\n')
779−--
780−-- This can be considered as an n-fold version of 'breakPost' where we apply
781−-- 'breakPost' successively on the input stream, dropping the first element
782−-- of the second segment after each break.
783−--
784−-- @since 0.7.0
785−{-# INLINE splitSuffixBy #-}
786−splitSuffixBy
787− :: (IsStream t, Monad m)
788− => (a -> Bool) -> Fold m a b -> t m a -> t m b
789−splitSuffixBy predicate f m =
790− D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
791−
792−-- | Like 'splitBy' but ignores repeated separators or separators in leading
793−-- or trailing position. Therefore, @"..a..b.."@ would be parsed as
794−-- @["a","b"]@. In other words, it treats the input like words separated by
795−-- whitespace elements determined by the predicate.
796−--
797−-- > wordsBy' p xs = S.toList $ FL.wordsBy p (FL.toList) (S.fromList xs)
798−--
799−-- >>> wordsBy' (== ',') ""
800−-- > []
801−--
802−-- >>> wordsBy' (== ',') ","
803−-- > []
804−--
805−-- >>> wordsBy' (== ',') ",a,,b,"
806−-- > ["a","b"]
807−--
808−-- > words = wordsBy isSpace
809−--
810−-- @since 0.7.0
811−{-# INLINE wordsBy #-}
812−wordsBy
813− :: (IsStream t, Monad m)
814− => (a -> Bool) -> Fold m a b -> t m a -> t m b
815−wordsBy predicate f m =
816− D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
817−
818−-- XXX we should express this using the Splitter config.
819−--
820−-- We can get splitSuffixBy' by appending the suffix to the output segments
821−-- produced by splitSuffixBy. However, it may add an additional suffix if the last
822−-- fragment did not have a suffix in the first place.
823−
824−-- | Like 'splitSuffixBy' but keeps the suffix in the splits.
825−--
826−-- > splitSuffixBy'_ p xs = S.toList $ FL.splitSuffixBy' p (FL.toList) (S.fromList xs)
827−--
828−-- >>> splitSuffixBy'_ (== '.') ""
829−-- []
830−--
831−-- >>> splitSuffixBy'_ (== '.') "."
832−-- ["."]
833−--
834−-- >>> splitSuffixBy'_ (== '.') "a"
835−-- ["a"]
836−--
837−-- >>> splitSuffixBy'_ (== '.') ".a"
838−-- > [".","a"]
839−--
840−-- >>> splitSuffixBy'_ (== '.') "a."
841−-- > ["a."]
842−--
843−-- >>> splitSuffixBy'_ (== '.') "a.b"
844−-- > ["a.","b"]
845−--
846−-- >>> splitSuffixBy'_ (== '.') "a.b."
847−-- > ["a.","b."]
848−--
849−-- >>> splitSuffixBy'_ (== '.') "a..b.."
850−-- > ["a.",".","b.","."]
851−--
852−-- This can be considered as an n-fold version of 'breakPost' where we apply
853−-- 'breakPost' successively on the input stream.
854−--
855−-- @since 0.7.0
856−{-# INLINE splitSuffixBy' #-}
857−splitSuffixBy'
858− :: (IsStream t, Monad m)
859− => (a -> Bool) -> Fold m a b -> t m a -> t m b
860−splitSuffixBy' predicate f m =
861− D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
862−
863−------------------------------------------------------------------------------
864−-- Split on a delimiter sequence
865−------------------------------------------------------------------------------
866−
867−-- Int list examples for splitOn:
868−--
869−-- >>> splitList [] [1,2,3,3,4]
870−-- > [[1],[2],[3],[3],[4]]
871−--
872−-- >>> splitList [5] [1,2,3,3,4]
873−-- > [[1,2,3,3,4]]
874−--
875−-- >>> splitList [1] [1,2,3,3,4]
876−-- > [[],[2,3,3,4]]
877−--
878−-- >>> splitList [4] [1,2,3,3,4]
879−-- > [[1,2,3,3],[]]
880−--
881−-- >>> splitList [2] [1,2,3,3,4]
882−-- > [[1],[3,3,4]]
883−--
884−-- >>> splitList [3] [1,2,3,3,4]
885−-- > [[1,2],[],[4]]
886−--
887−-- >>> splitList [3,3] [1,2,3,3,4]
888−-- > [[1,2],[4]]
889−--
890−-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
891−-- > [[],[]]
892−
893−-- | Split the stream on both sides of a separator sequence, dropping the
894−-- separator.
895−--
896−-- For illustration, let's define a function that operates on pure lists:
897−--
898−-- @
899−-- splitOn_ pat xs = S.toList $ FL.splitOn (A.fromList pat) (FL.toList) (S.fromList xs)
900−-- @
901−--
902−-- >>> splitOn_ "" "hello"
903−-- > ["h","e","l","l","o"]
904−--
905−-- >>> splitOn_ "hello" ""
906−-- > [""]
907−--
908−-- >>> splitOn_ "hello" "hello"
909−-- > ["",""]
910−--
911−-- >>> splitOn_ "x" "hello"
912−-- > ["hello"]
913−--
914−-- >>> splitOn_ "h" "hello"
915−-- > ["","ello"]
916−--
917−-- >>> splitOn_ "o" "hello"
918−-- > ["hell",""]
919−--
920−-- >>> splitOn_ "e" "hello"
921−-- > ["h","llo"]
922−--
923−-- >>> splitOn_ "l" "hello"
924−-- > ["he","","o"]
925−--
926−-- >>> splitOn_ "ll" "hello"
927−-- > ["he","o"]
928−--
929−-- 'splitOn' is an inverse of 'intercalate'. The following law always holds:
930−--
931−-- > intercalate . splitOn == id
932−--
933−-- The following law holds when the separator is non-empty and contains none of
934−-- the elements present in the input lists:
935−--
936−-- > splitOn . intercalate == id
937−--
938−-- The following law always holds:
939−--
940−-- > concat . splitOn . intercalate == concat
941−--
942−-- @since 0.7.0
943−{-# INLINE splitOn #-}
944−splitOn
945− :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
946− => Array a -> Fold m a b -> t m a -> t m b
947−splitOn patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
948−
949−{-
950−-- This can be implemented easily using Rabin Karp
951−-- | Split on any one of the given patterns.
952−{-# INLINE splitOnAny #-}
953−splitOnAny
954− :: (IsStream t, Monad m, Storable a, Integral a)
955− => [Array a] -> Fold m a b -> t m a -> t m b
956−splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
957−-}
958−
959−-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
960−-- of a predicate for a single element.
961−--
962−-- > splitSuffixOn_ pat xs = S.toList $ FL.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
963−--
964−-- >>> splitSuffixOn_ "." ""
965−-- [""]
966−--
967−-- >>> splitSuffixOn_ "." "."
968−-- [""]
969−--
970−-- >>> splitSuffixOn_ "." "a"
971−-- ["a"]
972−--
973−-- >>> splitSuffixOn_ "." ".a"
974−-- > ["","a"]
975−--
976−-- >>> splitSuffixOn_ "." "a."
977−-- > ["a"]
978−--
979−-- >>> splitSuffixOn_ "." "a.b"
980−-- > ["a","b"]
981−--
982−-- >>> splitSuffixOn_ "." "a.b."
983−-- > ["a","b"]
984−--
985−-- >>> splitSuffixOn_ "." "a..b.."
986−-- > ["a","","b",""]
987−--
988−-- > lines = splitSuffixOn "\n"
989−--
990−-- @since 0.7.0
991−{-# INLINE splitSuffixOn #-}
992−splitSuffixOn
993− :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
994− => Array a -> Fold m a b -> t m a -> t m b
995−splitSuffixOn patt f m =
996− D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
997−
998−{-
999−-- | Like 'splitOn' but drops any empty splits.
1000−--
1001−{-# INLINE wordsOn #-}
1002−wordsOn
1003− :: (IsStream t, Monad m, Storable a, Eq a)
1004− => Array a -> Fold m a b -> t m a -> t m b
1005−wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
1006−-}
1007−
1008−-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
1009−--
1010−-- | Like 'splitOn' but splits the separator as well, as an infix token.
1011−--
1012−-- > splitOn'_ pat xs = S.toList $ FL.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1013−--
1014−-- >>> splitOn'_ "" "hello"
1015−-- > ["h","","e","","l","","l","","o"]
1016−--
1017−-- >>> splitOn'_ "hello" ""
1018−-- > [""]
1019−--
1020−-- >>> splitOn'_ "hello" "hello"
1021−-- > ["","hello",""]
1022−--
1023−-- >>> splitOn'_ "x" "hello"
1024−-- > ["hello"]
1025−--
1026−-- >>> splitOn'_ "h" "hello"
1027−-- > ["","h","ello"]
1028−--
1029−-- >>> splitOn'_ "o" "hello"
1030−-- > ["hell","o",""]
1031−--
1032−-- >>> splitOn'_ "e" "hello"
1033−-- > ["h","e","llo"]
1034−--
1035−-- >>> splitOn'_ "l" "hello"
1036−-- > ["he","l","","l","o"]
1037−--
1038−-- >>> splitOn'_ "ll" "hello"
1039−-- > ["he","ll","o"]
1040−--
1041−-- @since 0.7.0
1042−{-# INLINE splitOn' #-}
1043−splitOn'
1044− :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
1045− => Array a -> Fold m a b -> t m a -> t m b
1046−splitOn' patt f m = S.intersperseM (foldl' f (A.read patt)) $ splitOn patt f m
1047−
1048−-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
1049−--
1050−-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
1051−--
1052−-- >>> splitSuffixOn'_ "." ""
1053−-- [""]
1054−--
1055−-- >>> splitSuffixOn'_ "." "."
1056−-- ["."]
1057−--
1058−-- >>> splitSuffixOn'_ "." "a"
1059−-- ["a"]
1060−--
1061−-- >>> splitSuffixOn'_ "." ".a"
1062−-- > [".","a"]
1063−--
1064−-- >>> splitSuffixOn'_ "." "a."
1065−-- > ["a."]
1066−--
1067−-- >>> splitSuffixOn'_ "." "a.b"
1068−-- > ["a.","b"]
1069−--
1070−-- >>> splitSuffixOn'_ "." "a.b."
1071−-- > ["a.","b."]
1072−--
1073−-- >>> splitSuffixOn'_ "." "a..b.."
1074−-- > ["a.",".","b.","."]
1075−--
1076−-- @since 0.7.0
1077−{-# INLINE splitSuffixOn' #-}
1078−splitSuffixOn'
1079− :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
1080− => Array a -> Fold m a b -> t m a -> t m b
1081−splitSuffixOn' patt f m =
1082− D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
1083−
1084−{-
1085−-- This can be implemented easily using Rabin Karp
1086−-- | Split post any one of the given patterns.
1087−{-# INLINE splitSuffixOnAny #-}
1088−splitSuffixOnAny
1089− :: (IsStream t, Monad m, Storable a, Integral a)
1090− => [Array a] -> Fold m a b -> t m a -> t m b
1091−splitSuffixOnAny subseq f m = undefined
1092− -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
1093−-}
1094−
1095−------------------------------------------------------------------------------
1096−-- Reorder in sequence
1097−------------------------------------------------------------------------------
1098−
1099−{-
1100−-- Buffer until the next element in sequence arrives. The function argument
1101−-- determines the difference in sequence numbers. This could be useful in
1102−-- implementing sequenced streams, for example, TCP reassembly.
1103−{-# INLINE reassembleBy #-}
1104−reassembleBy
1105− :: (IsStream t, Monad m)
1106− => Fold m a b
1107− -> (a -> a -> Int)
1108− -> t m a
1109− -> t m b
1110−reassembleBy = undefined
1111−-}
1112−
1113−------------------------------------------------------------------------------
1114−-- Distributing
1115−------------------------------------------------------------------------------
1116−--
1117−-- | Distribute one copy of the stream to each fold and zip the results.
1118−--
1119−-- @
1120−-- |-------Fold m a b--------|
1121−-- ---stream m a---| |---m (b,c)
1122−-- |-------Fold m a c--------|
1123−-- @
1124−-- >>> FL.foldl' (FL.tee FL.sum FL.length) (S.enumerateFromTo 1.0 100.0)
1125−-- (5050.0,100)
1126−--
1127−-- @since 0.7.0
1128−{-# INLINE tee #-}
1129−tee :: Monad m => Fold m a b -> Fold m a c -> Fold m a (b,c)
1130−tee f1 f2 = (,) <$> f1 <*> f2
1131−
1132−{-# INLINE foldNil #-}
1133−foldNil :: Monad m => Fold m a [b]
1134−foldNil = Fold step begin done where
1135− begin = return []
1136− step _ _ = return []
1137− done = return
1138−
1139−{-# INLINE foldCons #-}
1140−foldCons :: Monad m => Fold m a b -> Fold m a [b] -> Fold m a [b]
1141−foldCons (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1142− Fold step begin done
1143−
1144− where
1145−
1146− begin = Tuple' <$> beginL <*> beginR
1147− step (Tuple' xL xR) a = Tuple' <$> stepL xL a <*> stepR xR a
1148− done (Tuple' xL xR) = (:) <$> (doneL xL) <*> (doneR xR)
1149−
1150−-- XXX use "List" instead of "[]"?, use Array for output to scale it to a large
1151−-- number of consumers?
1152−--
1153−-- | Distribute one copy of the stream to each fold and collect the results in
1154−-- a container.
1155−--
1156−-- @
1157−--
1158−-- |-------Fold m a b--------|
1159−-- ---stream m a---| |---m [b]
1160−-- |-------Fold m a b--------|
1161−-- | |
1162−-- ...
1163−-- @
1164−--
1165−-- >>> FL.foldl' (FL.distribute [FL.sum, FL.length]) (S.enumerateFromTo 1 5)
1166−-- [15,5]
1167−--
1168−-- This is the consumer side dual of the producer side 'sequence' operation.
1169−--
1170−-- @since 0.7.0
1171−{-# INLINE distribute #-}
1172−distribute :: Monad m => [Fold m a b] -> Fold m a [b]
1173−distribute [] = foldNil
1174−distribute (x:xs) = foldCons x (distribute xs)
1175−
1176−------------------------------------------------------------------------------
1177−-- Partitioning
1178−------------------------------------------------------------------------------
1179−--
1180−-- | Partition the input over two folds using an 'Either' partitioning
1181−-- predicate.
1182−--
1183−-- @
1184−--
1185−-- |-------Fold b x--------|
1186−-- -----stream m a --> (Either b c)----| |----(x,y)
1187−-- |-------Fold c y--------|
1188−-- @
1189−--
1190−-- Send input to either fold randomly:
1191−--
1192−-- >>> import System.Random (randomIO)
1193−-- >>> randomly a = randomIO >>= \x -> return $ if x then Left a else Right a
1194−-- >>> FL.foldl' (FL.partitionByM randomly FL.length FL.length) (S.enumerateFromTo 1 100)
1195−-- (59,41)
1196−--
1197−-- Send input to the two folds in a proportion of 2:1:
1198−--
1199−-- @
1200−-- import Data.IORef (newIORef, readIORef, writeIORef)
1201−-- proportionately m n = do
1202−-- ref <- newIORef $ cycle $ concat [replicate m Left, replicate n Right]
1203−-- return $ \\a -> do
1204−-- r <- readIORef ref
1205−-- writeIORef ref $ tail r
1206−-- return $ head r a
1207−--
1208−-- main = do
1209−-- f <- proportionately 2 1
1210−-- r <- FL.foldl' (FL.partitionByM f FL.length FL.length) (S.enumerateFromTo (1 :: Int) 100)
1211−-- print r
1212−-- @
1213−-- @
1214−-- (67,33)
1215−-- @
1216−--
1217−-- This is the consumer side dual of the producer side 'mergeBy' operation.
1218−--
1219−-- @since 0.7.0
1220−{-# INLINE partitionByM #-}
1221−partitionByM :: Monad m
1222− => (a -> m (Either b c)) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1223−partitionByM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1224−
1225− Fold step begin done
1226−
1227− where
1228−
1229− begin = Tuple' <$> beginL <*> beginR
1230− step (Tuple' xL xR) a = do
1231− r <- f a
1232− case r of
1233− Left b -> Tuple' <$> stepL xL b <*> return xR
1234− Right c -> Tuple' <$> return xL <*> stepR xR c
1235− done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1236−
1237−-- Note: we could use (a -> Bool) instead of (a -> Either b c), but the latter
1238−-- makes the signature clearer as to which case belongs to which fold.
1239−-- XXX need to check the performance in both cases.
1240−
1241−-- | Same as 'partitionByM' but with a pure partition function.
1242−--
1243−-- Count even and odd numbers in a stream:
1244−--
1245−-- @
1246−-- >>> let f = FL.partitionBy (\\n -> if even n then Left n else Right n)
1247−-- (fmap (("Even " ++) . show) FL.length)
1248−-- (fmap (("Odd " ++) . show) FL.length)
1249−-- in FL.foldl' f (S.enumerateFromTo 1 100)
1250−-- ("Even 50","Odd 50")
1251−-- @
1252−--
1253−-- @since 0.7.0
1254−{-# INLINE partitionBy #-}
1255−partitionBy :: Monad m
1256− => (a -> Either b c) -> Fold m b x -> Fold m c y -> Fold m a (x, y)
1257−partitionBy f = partitionByM (return . f)
1258−
1259−-- | Compose two folds such that the combined fold accepts a stream of 'Either'
1260−-- and routes the 'Left' values to the first fold and 'Right' values to the
1261−-- second fold.
1262−--
1263−-- > partition = partitionBy id
1264−--
1265−-- @since 0.7.0
1266−{-# INLINE partition #-}
1267−partition :: Monad m
1268− => Fold m b x -> Fold m c y -> Fold m (Either b c) (x, y)
1269−partition = partitionBy id
1270−
1271−{-
1272−-- | Send one item to each fold in a round-robin fashion. This is the consumer
1273−-- side dual of producer side 'mergeN' operation.
1274−--
1275−-- partitionN :: Monad m => [Fold m a b] -> Fold m a [b]
1276−-- partitionN fs = Fold step begin done
1277−-}
1278−
1279−-- Demultiplex an input element into a number of typed variants. We want to
1280−-- statically restrict the target values within a set of predefined types, an
1281−-- enumeration of a GADT. We also want to make sure that the Map contains only
1282−-- those types and the full set of those types. Instead of Map it should
1283−-- probably be a lookup-table using an array and not in GC memory.
1284−--
1285−-- This is the consumer side dual of the producer side 'mux' operation (XXX to
1286−-- be implemented).
1287−
1288−-- | Split the input stream based on a key field and fold each split using a
1289−-- specific fold collecting the results in a map from the keys to the results.
1290−-- Useful for cases like protocol handlers to handle different type of packets
1291−-- using different handlers.
1292−--
1293−-- @
1294−--
1295−-- |-------Fold m a b
1296−-- -----stream m a-----Map-----|
1297−-- |-------Fold m a b
1298−-- |
1299−-- ...
1300−-- @
1301−--
1302−-- @since 0.7.0
1303−{-# INLINE demuxWith #-}
1304−demuxWith :: (Monad m, Ord k)
1305− => (a -> k) -> Map k (Fold m a b) -> Fold m a (Map k b)
1306−demuxWith f kv = Fold step initial extract
1307−
1308− where
1309−
1310− initial = return kv
1311− step mp a =
1312− -- XXX should we raise an exception in Nothing case?
1313− -- Ideally we should enforce that it is a total map over k so that look
1314− -- up never fails
1315− -- XXX we could use a monadic update function for a single lookup and
1316− -- update in the map.
1317− let k = f a
1318− in case Map.lookup k mp of
1319− Nothing -> return mp
1320− Just (Fold step' acc extract') -> do
1321− !r <- acc >>= \x -> step' x a
1322− return $ Map.insert k (Fold step' (return r) extract') mp
1323− extract = Prelude.mapM (\(Fold _ acc e) -> acc >>= e)
1324−
1325−-- | Fold a stream of key value pairs using a map of specific folds for each
1326−-- key into a map from keys to the results of fold outputs of the corresponding
1327−-- values.
1328−--
1329−-- @
1330−-- > let table = Data.Map.fromList [(\"SUM", FL.sum), (\"PRODUCT", FL.product)]
1331−-- input = S.fromList [(\"SUM",1),(\"PRODUCT",2),(\"SUM",3),(\"PRODUCT",4)]
1332−-- in FL.foldl' (FL.demux table) input
1333−-- One 1
1334−-- Two 2
1335−-- @
1336−--
1337−-- @since 0.7.0
1338−{-# INLINE demux #-}
1339−demux :: (Monad m, Ord k)
1340− => Map k (Fold m a b) -> Fold m (k, a) (Map k b)
1341−demux fs = demuxWith fst (Map.map (lmap snd) fs)
1342−
1343−-- | Split the input stream based on a key field and fold each split using a
1344−-- specific fold without collecting the results. Useful for cases like protocol
1345−-- handlers to handle different type of packets.
1346−--
1347−-- @
1348−--
1349−-- |-------Fold m a ()
1350−-- -----stream m a-----Map-----|
1351−-- |-------Fold m a ()
1352−-- |
1353−-- ...
1354−-- @
1355−--
1356−--
1357−-- @since 0.7.0
1358−
1359−-- demuxWith_ can be slightly faster than demuxWith because we do not need to
1360−-- update the Map in this case. This may be significant only if the map is
1361−-- large.
1362−{-# INLINE demuxWith_ #-}
1363−demuxWith_ :: (Monad m, Ord k)
1364− => (a -> k) -> Map k (Fold m a b) -> Fold m a ()
1365−demuxWith_ f kv = Fold step initial extract
1366−
1367− where
1368−
1369− initial = do
1370− Prelude.mapM (\(Fold s i e) ->
1371− i >>= \r -> return (Fold s (return r) e)) kv
1372− step mp a =
1373− -- XXX should we raise an exception in Nothing case?
1374− -- Ideally we should enforce that it is a total map over k so that look
1375− -- up never fails
1376− case Map.lookup (f a) mp of
1377− Nothing -> return mp
1378− Just (Fold step' acc _) -> do
1379− _ <- acc >>= \x -> step' x a
1380− return mp
1381− extract mp = Prelude.mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
1382−
1383−-- | Given a stream of key value pairs and a map from keys to folds, fold the
1384−-- values for each key using the corresponding folds, discarding the outputs.
1385−--
1386−-- @
1387−-- > let prn = FL.drainBy print
1388−-- > let table = Data.Map.fromList [(\"ONE", prn), (\"TWO", prn)]
1389−-- input = S.fromList [(\"ONE",1),(\"TWO",2)]
1390−-- in FL.foldl' (FL.demux_ table) input
1391−-- One 1
1392−-- Two 2
1393−-- @
1394−--
1395−-- @since 0.7.0
1396−{-# INLINE demux_ #-}
1397−demux_ :: (Monad m, Ord k) => Map k (Fold m a ()) -> Fold m (k, a) ()
1398−demux_ fs = demuxWith_ fst (Map.map (lmap snd) fs)
1399−
1400−-- XXX instead of a Map we could yield the results as a pure stream as they
1401−-- complete. We could then concatMap the fold results to implement the
1402−-- windowing combinators.
1403−--
1404−-- | Split the input stream based on a key field and fold each split using the
1405−-- given fold. Useful for map/reduce, bucketizing the input in different bins
1406−-- or for generating histograms.
1407−--
1408−-- @
1409−-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1410−-- in FL.foldl' (FL.classify FL.toListRev) input
1411−-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1412−-- @
1413−--
1414−-- @since 0.7.0
1415−{-# INLINE classifyWith #-}
1416−classifyWith :: (Monad m, Ord k) => (a -> k) -> Fold m a b -> Fold m a (Map k b)
1417−classifyWith f (Fold step initial extract) = Fold step' initial' extract'
1418−
1419− where
1420−
1421− initial' = return Map.empty
1422− step' kv a =
1423− let k = f a
1424− in case Map.lookup k kv of
1425− Nothing -> do
1426− x <- initial
1427− r <- step x a
1428− return $ Map.insert k r kv
1429− Just x -> do
1430− r <- step x a
1431− return $ Map.insert k r kv
1432− extract' = Prelude.mapM extract
1433−
1434−-- | Given an input stream of key value pairs and a fold for values, fold all
1435−-- the values belonging to each key. Useful for map/reduce, bucketizing the
1436−-- input in different bins or for generating histograms.
1437−--
1438−-- @
1439−-- > let input = S.fromList [(\"ONE",1),(\"ONE",1.1),(\"TWO",2), (\"TWO",2.2)]
1440−-- in FL.foldl' (FL.classify FL.toListRev) input
1441−-- fromList [(\"ONE",[1.1,1.0]),(\"TWO",[2.2,2.0])]
1442−-- @
1443−--
1444−-- Same as:
1445−--
1446−-- > classify fld = classifyWith fst (lmap snd fld)
1447−--
1448−-- @since 0.7.0
1449−{-# INLINE classify #-}
1450−classify :: (Monad m, Ord k) => Fold m a b -> Fold m (k, a) (Map k b)
1451−classify fld = classifyWith fst (lmap snd fld)
1452−
1453−------------------------------------------------------------------------------
1454−-- Unzipping
1455−------------------------------------------------------------------------------
1456−--
1457−-- | Like 'unzipWith' but with a monadic splitter function.
1458−--
1459−-- @since 0.7.0
1460−{-# INLINE unzipWithM #-}
1461−unzipWithM :: Monad m
1462− => (a -> m (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1463−unzipWithM f (Fold stepL beginL doneL) (Fold stepR beginR doneR) =
1464− Fold step begin done
1465−
1466− where
1467−
1468− step (Tuple' xL xR) a = do
1469− (b,c) <- f a
1470− Tuple' <$> stepL xL b <*> stepR xR c
1471− begin = Tuple' <$> beginL <*> beginR
1472− done (Tuple' xL xR) = (,) <$> doneL xL <*> doneR xR
1473−
1474−-- | Split elements in the input stream into two parts using a pure splitter
1475−-- function, direct each part to a different fold and zip the results.
1476−--
1477−-- @since 0.7.0
1478−{-# INLINE unzipWith #-}
1479−unzipWith :: Monad m
1480− => (a -> (b,c)) -> Fold m b x -> Fold m c y -> Fold m a (x,y)
1481−unzipWith f = unzipWithM (return . f)
1482−
1483−-- | Send the elements of tuples in a stream of tuples through two different
1484−-- folds.
1485−--
1486−-- @
1487−--
1488−-- |-------Fold a x--------|
1489−-- -----Stream m x----(a,b)--| |----m (x,y)
1490−-- |-------Fold b y--------|
1491−--
1492−-- @
1493−--
1494−-- This is the consumer side dual of the producer side 'zip' operation.
1495−--
1496−-- @since 0.7.0
1497−{-# INLINE unzip #-}
1498−unzip :: Monad m => Fold m a x -> Fold m b y -> Fold m (a,b) (x,y)
1499−unzip = unzipWith id
1500−
1501−------------------------------------------------------------------------------
1502−-- Nesting
1503−------------------------------------------------------------------------------
1504−--
1505−-- | Modify the fold such that when the fold is done, instead of returning the
1506−-- accumulator, it returns a fold. The returned fold starts from where we left
1507−-- i.e. it uses the last accumulator value as the initial value of the
1508−-- accumulator. Thus we can resume the fold later and feed it more input.
1509−--
1510−-- >> do
1511−-- > more <- FL.foldl (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
1512−-- > evenMore <- FL.foldl (FL.duplicate more) (S.enumerateFromTo 11 20)
1513−-- > FL.foldl evenMore (S.enumerateFromTo 21 30)
1514−-- > 465
1515−--
1516−-- @since 0.7.0
1517−{-# INLINABLE duplicate #-}
1518−duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
1519−duplicate (Fold step begin done) =
1520− Fold step begin (\x -> pure (Fold step (pure x) done))
1521−
1522−{-
1523−-- All the stream flattening transformations can also be applied to a fold
1524−-- input stream.
1525−
1526−-- | This can be used to apply all the stream generation operations on folds.
1527−lconcatMap ::(IsStream t, Monad m) => (a -> t m b)
1528− -> Fold m b c
1529− -> Fold m a c
1530−lconcatMap s f1 f2 = undefined
1531−-}
1532−
1533−-- All the grouping transformation that we apply to a stream can also be
1534−-- applied to a fold input stream.
1535−
1536−{-
1537−-- | Group the input stream into groups of elements between @low@ and @high@.
1538−-- Collection starts in chunks of @low@ and then keeps doubling until we reach
1539−-- @high@. Each chunk is folded using the provided fold function.
1540−--
1541−-- This could be useful, for example, when we are folding a stream of unknown
1542−-- size to a stream of arrays and we want to minimize the number of
1543−-- allocations.
1544−--
1545−-- @
1546−--
1547−-- XXX we should be able to implement it with parsers/terminating folds.
1548−--
1549−{-# INLINE lchunksInRange #-}
1550−lchunksInRange :: Monad m
1551− => Int -> Int -> Fold m a b -> Fold m b c -> Fold m a c
1552−lchunksInRange low high (Fold step1 initial1 extract1)
1553− (Fold step2 initial2 extract2) = undefined
1554−-}
1555−
1556−-- | Group the input stream into groups of @n@ elements each and then fold each
1557−-- group using the provided fold function.
1558−--
1559−-- @
1560−--
1561−-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1562−--
1563−-- @
1564−--
1565−{-# INLINE lchunksOf #-}
1566−lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
1567−lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1568− Fold step' initial' extract'
1569−
1570− where
1571−
1572− initial' = (Tuple3' 0) <$> initial1 <*> initial2
1573− step' (Tuple3' i r1 r2) a = do
1574− if i < n
1575− then do
1576− res <- step1 r1 a
1577− return $ Tuple3' (i + 1) res r2
1578− else do
1579− res <- extract1 r1
1580− acc2 <- step2 r2 res
1581−
1582− i1 <- initial1
1583− acc1 <- step1 i1 a
1584− return $ Tuple3' 1 acc1 acc2
1585− extract' (Tuple3' _ _ r) = extract2 r
1586−
1587−-- | Group the input stream into windows of n second each and then fold each
1588−-- group using the provided fold function.
1589−--
1590−-- For example, we can copy and distribute a stream to multiple folds where
1591−-- each fold can group the input differently e.g. by one second, one minute and
1592−-- one hour windows respectively and fold each resulting stream of folds.
1593−--
1594−-- @
1595−--
1596−-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1597−--
1598−-- @
1599−{-# INLINE lsessionsOf #-}
1600−lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
1601−lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1602− Fold step' initial' extract'
1603−
1604− where
1605−
1606− -- XXX MVar may be expensive we need a cheaper synch mechanism here
1607− initial' = do
1608− i1 <- initial1
1609− i2 <- initial2
1610− mv1 <- liftIO $ newMVar i1
1611− mv2 <- liftIO $ newMVar (Right i2)
1612− t <- control $ \run ->
1613− mask $ \restore -> do
1614− tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
1615− (handleChildException mv2)
1616− run (return tid)
1617− return $ Tuple3' t mv1 mv2
1618− step' acc@(Tuple3' _ mv1 _) a = do
1619− r1 <- liftIO $ takeMVar mv1
1620− res <- step1 r1 a
1621− liftIO $ putMVar mv1 res
1622− return acc
1623− extract' (Tuple3' tid _ mv2) = do
1624− r2 <- liftIO $ takeMVar mv2
1625− liftIO $ killThread tid
1626− case r2 of
1627− Left e -> throwM e
1628− Right x -> extract2 x
1629−
1630− timerThread mv1 mv2 = do
1631− liftIO $ threadDelay (round $ n * 1000000)
1632−
1633− r1 <- liftIO $ takeMVar mv1
1634− i1 <- initial1
1635− liftIO $ putMVar mv1 i1
1636−
1637− res1 <- extract1 r1
1638− r2 <- liftIO $ takeMVar mv2
1639− res <- case r2 of
1640− Left _ -> return r2
1641− Right x -> fmap Right $ step2 x res1
1642− liftIO $ putMVar mv2 res
1643− timerThread mv1 mv2
1644−
1645− handleChildException ::
1646− MVar (Either SomeException a) -> SomeException -> IO ()
1647− handleChildException mv2 e = do
1648− r2 <- takeMVar mv2
1649− let r = case r2 of
1650− Left _ -> r2
1651− Right _ -> Left e
1652− putMVar mv2 r
1653−
1654−------------------------------------------------------------------------------
1655−-- Windowed classification
1656−------------------------------------------------------------------------------
1657−
1658−-- We divide the stream into windows or chunks in space or time and each window
1659−-- can be associated with a key, all events associated with a particular key in
1660−-- the window can be folded to a single result. The stream can be split into
1661−-- windows by size or by using a split predicate on the elements in the stream.
1662−-- For example, when we receive a closing flag, we can close the window.
1663−--
1664−-- A "chunk" is a space window and a "session" is a time window. Are there any
1665−-- other better short words to describe them. An alternative is to use
1666−-- "swindow" and "twindow". Another word for "session" could be "spell".
1667−--
1668−-- TODO: To mark the position in space or time we can have Indexed or
1669−-- TimeStamped types. That can make it easy to deal with the position indices
1670−-- or timestamps.
1671−
1672−------------------------------------------------------------------------------
1673−-- Keyed Sliding Windows
1674−------------------------------------------------------------------------------
1675−
1676−{-
1677−{-# INLINABLE classifySlidingChunks #-}
1678−classifySlidingChunks
1679− :: (IsStream t, MonadAsync m, Ord k)
1680− => Int -- ^ window size
1681− -> Int -- ^ window slide
1682− -> Fold m a b -- ^ Fold to be applied to window events
1683− -> t m (k, a, Bool) -- ^ window key, data, close event
1684− -> t m (k, b)
1685−classifySlidingChunks wsize wslide (Fold step initial extract) str
1686− = undefined
1687−
1688−-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1689−-- slide the send window when an ack is received and we slide the receive
1690−-- window when a sequence is complete. Sliding is stateful in case of TCP,
1691−-- sliding releases the send buffer or makes data available to the user from
1692−-- the receive buffer.
1693−{-# INLINABLE classifySlidingSessions #-}
1694−classifySlidingSessions
1695− :: (IsStream t, MonadAsync m, Ord k)
1696− => Double -- ^ timer tick in seconds
1697− -> Double -- ^ time window size
1698− -> Double -- ^ window slide
1699− -> Fold m a b -- ^ Fold to be applied to window events
1700− -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1701− -> t m (k, b)
1702−classifySlidingSessions tick interval slide (Fold step initial extract) str
1703− = undefined
1704−-}
1705−
1706−------------------------------------------------------------------------------
1707−-- Sliding Window Buffers
1708−------------------------------------------------------------------------------
1709−
1710−-- These buffered versions could be faster than concurrent incremental folds of
1711−-- all overlapping windows as in many cases we may not need all the values to
1712−-- compute the fold, we can just compute the result using the old value and new
1713−-- value. However, we may need the buffer once in a while, for example for
1714−-- string search we usually compute the hash incrementally but when the hash
1715−-- matches the hash of the pattern we need to compare the whole string.
1716−--
1717−-- XXX we should be able to implement sequence based splitting combinators
1718−-- using this combinator.
1719−
1720−{-
1721−-- | Buffer n elements of the input in a ring buffer. When t new elements are
1722−-- collected, slide the window to remove the same number of oldest elements,
1723−-- insert the new elements, and apply an incremental fold on the sliding
1724−-- window, supplying the outgoing elements, the new ring buffer as arguments.
1725−slidingChunkBuffer
1726− :: (IsStream t, Monad m, Ord a, Storable a)
1727− => Int -- window size
1728− -> Int -- window slide
1729− -> Fold m (Ring a, Array a) b
1730− -> t m a
1731− -> t m b
1732−slidingChunkBuffer = undefined
1733−
1734−-- Buffer n seconds worth of stream elements of the input in a radix tree.
1735−-- Every t seconds, remove the items that are older than n seconds, and apply
1736−-- an incremental fold on the sliding window, supplying the outgoing elements,
1737−-- and the new radix tree buffer as arguments.
1738−slidingSessionBuffer
1739− :: (IsStream t, Monad m, Ord a, Storable a)
1740− => Int -- window size
1741− -> Int -- tick size
1742− -> Fold m (RTree a, Array a) b
1743− -> t m a
1744− -> t m b
1745−slidingSessionBuffer = undefined
1746−-}
1747−
1748−------------------------------------------------------------------------------
1749−-- Keyed Session Windows
1750−------------------------------------------------------------------------------
1751−
1752−{-
1753−-- | Keyed variable size space windows. Close the window if we do not receive a
1754−-- window event in the next "spaceout" elements.
1755−{-# INLINABLE classifyChunksBy #-}
1756−classifyChunksBy
1757− :: (IsStream t, MonadAsync m, Ord k)
1758− => Int -- ^ window spaceout (spread)
1759− -> Bool -- ^ reset the spaceout when a chunk window element is received
1760− -> Fold m a b -- ^ Fold to be applied to chunk window elements
1761− -> t m (k, a, Bool) -- ^ chunk key, data, last element
1762− -> t m (k, b)
1763−classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1764−
1765−-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1766−-- received within the chunk size window. The chunk gets closed only if no
1767−-- element is received within the chunk window.
1768−--
1769−{-# INLINABLE classifyKeepAliveChunks #-}
1770−classifyKeepAliveChunks
1771− :: (IsStream t, MonadAsync m, Ord k)
1772− => Int -- ^ window spaceout (spread)
1773− -> Fold m a b -- ^ Fold to be applied to chunk window elements
1774− -> t m (k, a, Bool) -- ^ chunk key, data, last element
1775− -> t m (k, b)
1776−classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1777−-}
1778−
1779−-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
1780−-- stream elements that belong to the same session. @timeout@ is the maximum
1781−-- lifetime of a session in seconds. All elements belonging to a session are
1782−-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
1783−-- after every event received in the session. Session duration is measured
1784−-- using the timestamp of the first element seen for that session. To detect
1785−-- session timeouts, a monotonic event time clock is maintained using the
1786−-- timestamps seen in the inputs and a timer with a tick duration specified by
1787−-- @tick@.
1788−--
1789−-- @session key@ is a key that uniquely identifies the session for the given
1790−-- element, @timestamp@ characterizes the time when the input element was
1791−-- generated, this is an absolute time measured from some @Epoch@. @session
1792−-- close@ is a boolean indicating whether this element marks the closing of the
1793−-- session. When an input element with @session close@ set to @True@ is seen
1794−-- the session is purged immediately.
1795−--
1796−-- All the input elements belonging to a session are collected using the fold
1797−-- @f@. The session key and the fold result are emitted in the output stream
1798−-- when the session is purged either via the session close event or via the
1799−-- session lifetime timeout.
1800−--
1801−-- @since 0.7.0
1802−{-# INLINABLE classifySessionsBy #-}
1803−classifySessionsBy
1804− :: (IsStream t, MonadAsync m, Ord k)
1805− => Double -- ^ timer tick in seconds
1806− -> Double -- ^ session timeout
1807− -> Bool -- ^ reset the timeout when an event is received
1808− -> Fold m a b -- ^ Fold to be applied to session events
1809− -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
1810− -> t m (k, b)
1811−classifySessionsBy tick timeout reset (Fold step initial extract) str =
1812− S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
1813−
1814− where
1815−
1816− timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
1817− tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1818− szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
1819−
1820− -- Got a new stream input element
1821− sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
1822− -- XXX we should use a heap in pinned memory to scale it to a large
1823− -- size
1824− --
1825− -- deleting a key from the heap is expensive, so we never delete a
1826− -- key, we just purge it from the Map and it gets purged from the
1827− -- heap on timeout. We just need an extra lookup in the Map when
1828− -- the key is purged from the heap, that should not be expensive.
1829− --
1830− -- To detect session inactivity we keep a timestamp of the latest event
1831− -- in the Map along with the fold result. When we purge the session
1832− -- from the heap we match the timestamp in the heap with the timestamp
1833− -- in the Map, if the latest timestamp is newer and has not expired we
1834− -- reinsert the key in the heap.
1835− --
1836− -- XXX if the key is an Int, we can also use an IntMap for slightly
1837− -- better performance.
1838− --
1839− let accumulate v = do
1840− Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
1841− new <- step old a
1842− return $ Tuple' ts new
1843− in if closing
1844− then do
1845− let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
1846− Tuple' _ acc <- accumulate r
1847− res <- extract acc
1848− return $ Tuple4' evTime hp mp' (S.fromPure (key, res))
1849− else do
1850− let r = Map.lookup key mp
1851− acc <- accumulate r
1852− let mp' = Map.insert key acc mp
1853− let hp' =
1854− case r of
1855− Nothing ->
1856− let expiry = addToAbsTime ts timeoutMs
1857− in H.insert (Entry expiry key) hp
1858− Just _ -> hp
1859− -- Event time is maintained as monotonically increasing
1860− -- time. If we have lagged behind any of the timestamps
1861− -- seen then we increase it to match the latest time seen
1862− -- in the timestamps. We also increase it on timer ticks.
1863− return $ Tuple4' (max evTime ts) hp' mp' S.nil
1864−
1865− -- Got a timer tick event
1866− -- XXX can we yield the entries without accumulating them?
1867− sstep (Tuple4' evTime heap sessions _) Nothing = do
1868− (hp', mp', out) <- go heap sessions S.nil
1869− return $ Tuple4' curTime hp' mp' out
1870−
1871− where
1872−
1873− curTime = addToAbsTime evTime tickMs
1874− go hp mp out = do
1875− let hres = H.uncons hp
1876− case hres of
1877− Just (Entry ts key, hp') -> do
1878− let duration = diffAbsTime curTime ts
1879− if duration >= timeoutMs
1880− then do
1881− let (r, mp') = Map.updateLookupWithKey
1882− (\_ _ -> Nothing) key mp
1883− case r of
1884− Nothing -> go hp' mp' out
1885− Just (Tuple' latestTS acc) -> do
1886− let dur = diffAbsTime curTime latestTS
1887− if dur >= timeoutMs || not reset
1888− then do
1889− sess <- extract acc
1890− go hp' mp' ((key, sess) `S.cons` out)
1891− else
1892− -- reset the session timeout
1893− let expiry = addToAbsTime latestTS timeoutMs
1894− hp'' = H.insert (Entry expiry key) hp'
1895− mp'' = Map.insert key (Tuple' latestTS acc) mp'
1896− in go hp'' mp'' out
1897− else return (hp, mp, out)
1898− Nothing -> return (hp, mp, out)
1899−
1900− -- merge timer events in the stream
1901− stream = S.map Just str `parallel` S.repeatM timer
1902− timer = do
1903− liftIO $ threadDelay (round $ tick * 1000000)
1904− return Nothing
1905−
1906−-- | Like 'classifySessionsOf' but the session is kept alive if an event is
1907−-- received within the session window. The session times out and gets closed
1908−-- only if no event is received within the specified session window size.
1909−--
1910−-- @since 0.7.0
1911−{-# INLINABLE classifyKeepAliveSessions #-}
1912−classifyKeepAliveSessions
1913− :: (IsStream t, MonadAsync m, Ord k)
1914− => Double -- ^ session inactive timeout
1915− -> Fold m a b -- ^ Fold to be applied to session payload data
1916− -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
1917− -> t m (k, b)
1918−classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
1919−
1920−------------------------------------------------------------------------------
1921−-- Keyed tumbling windows
1922−------------------------------------------------------------------------------
1923−
1924−-- Tumbling windows is a special case of sliding windows where the window slide
1925−-- is the same as the window size. Or it can be a special case of session
1926−-- windows where the reset flag is set to False.
1927−
1928−-- XXX instead of using the early termination flag in the stream, we can use an
1929−-- early terminating fold instead.
1930−
1931−{-
1932−-- | Split the stream into fixed size chunks of specified size. Within each
1933−-- such chunk fold the elements in buckets identified by the keys. A particular
1934−-- bucket fold can be terminated early if a closing flag is encountered in an
1935−-- element for that key.
1936−--
1937−-- @since 0.7.0
1938−{-# INLINABLE classifyChunksOf #-}
1939−classifyChunksOf
1940− :: (IsStream t, MonadAsync m, Ord k)
1941− => Int -- ^ window size
1942− -> Fold m a b -- ^ Fold to be applied to window events
1943− -> t m (k, a, Bool) -- ^ window key, data, close event
1944− -> t m (k, b)
1945−classifyChunksOf wsize = classifyChunksBy wsize False
1946−-}
1947−
1948−-- | Split the stream into fixed size time windows of specified interval in
1949−-- seconds. Within each such window, fold the elements in buckets identified by
1950−-- the keys. A particular bucket fold can be terminated early if a closing flag
1951−-- is encountered in an element for that key. Once a fold is terminated the key
1952−-- and value for that bucket are emitted in the output stream.
1953−--
1954−-- Session @timestamp@ in the input stream is an absolute time from some epoch,
1955−-- characterizing the time when the input element was generated. To detect
1956−-- session window end, a monotonic event time clock is maintained synced with
1957−-- the timestamps with a clock resolution of 1 second.
1958−--
1959−-- @since 0.7.0
1960−{-# INLINABLE classifySessionsOf #-}
1961−classifySessionsOf
1962− :: (IsStream t, MonadAsync m, Ord k)
1963− => Double -- ^ time window size
1964− -> Fold m a b -- ^ Fold to be applied to window events
1965− -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1966− -> t m (k, b)
1967−classifySessionsOf interval = classifySessionsBy 1 interval False
1968−-}