2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE FlexibleContexts #-}
5#if __GLASGOW_HASKELL__ >= 800
6{-# OPTIONS_GHC -Wno-orphans #-}
12-- Module : Streamly.Internal.Prelude
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : streamly@composewell.com
17-- Stability : experimental
21module Streamly.Internal.Prelude
83 -- -- ** To Summary (Full Folds)
91 -- -- ** To Summary (Maybe) (Full Folds)
100 -- -- ** To Elements (Partial Folds)
104 -- -- | Folds that extract selected elements of a stream or their properties.
114 -- -- ** To Boolean (Partial Folds)
129 -- ** Composable Left Folds
132 , toStream -- XXX rename to write?
133 , toStreamRev -- XXX rename to writeRev?
135 -- * Function application
161 -- ** Scan Using Fold
179 -- , timestampedR -- timer
186 -- ** Stateful Filters
202 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
203 -- , uniqOn -- to remove duplicate sequences
204 -- , pruneBy -- dropAround + uniqBy - like words
206 -- ** Mapping Filters
212 -- ** Scanning Filters
215 -- , seqIndices -- search a sequence in the stream
222 , intersperseSuffixBySpan
223 -- , intersperseBySpan
231 -- * Multi-Stream Operations
269 , concatUnfoldInterleave
270 , concatUnfoldRoundrobin
279 , concatMapIterateWith
282 , concatMapTreeYieldLeavesWith
288 -- , splitIn -- sessionN
292 , break -- breakBefore
311 -- -- *** Using Element Separators
319 , wordsBy -- stripAndCompactBy
321 -- -- *** Using Sequence Separators
324 -- , splitOnPrefixSeq
326 -- Keeping the delimiters
329 -- , splitByPrefixSeq
332 -- Splitting using multiple sequence separators
334 -- , splitOnAnySuffixSeq
335 -- , splitOnAnyPrefixSeq
354 -- * Windowed Classification
356 -- ** Tumbling Windows
357 -- , classifyChunksOf
361 -- ** Keep Alive Windows
362 -- , classifyKeepAliveChunks
363 , classifyKeepAliveSessions
366 -- ** Sliding Windows
367 , classifySlidingChunks
368 , classifySlidingSessions
370 -- ** Sliding Window Buffers
371 -- , slidingChunkBuffer
372 -- , slidingSessionBuffer
374 -- ** Containers of Streams
401 -- * Generalize Inner Monad
405 -- * Transform Inner Monad
436import Control.Concurrent (threadDelay)
437import Control.Exception (Exception)
438import Control.Monad (void)
439import Control.Monad.Catch (MonadCatch)
440import Control.Monad.IO.Class (MonadIO(..))
441import Control.Monad.Reader (ReaderT)
442import Control.Monad.State.Strict (StateT)
443import Control.Monad.Trans (MonadTrans(..))
444import Control.Monad.Trans.Control (MonadBaseControl)
445import Data.Functor.Identity (Identity (..))
446import Data.Heap (Entry(..))
447import Data.Maybe (isJust, fromJust, isNothing)
448import Foreign.Storable (Storable)
450 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
451 foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
452 notElem, maximum, minimum, head, last, tail, length, null,
453 reverse, iterate, init, and, or, lookup, foldr1, (!!),
454 scanl, scanl1, replicate, concatMap, span, splitAt, break,
457import qualified Data.Heap as H
458import qualified Data.Map.Strict as Map
459import qualified Prelude
460import qualified System.IO as IO
462import Streamly.Internal.Data.Stream.Enumeration (Enumerable(..), enumerate, enumerateTo)
463import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))
464import Streamly.Internal.Data.Unfold.Types (Unfold)
465import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)
466-- import Streamly.Memory.Ring (Ring)
467import Streamly.Internal.Data.SVar (MonadAsync, State, defState)
468import Streamly.Internal.Data.Stream.Async (mkAsync')
469import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)
470import Streamly.Internal.Data.Stream.Prelude
471 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)
472import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
473import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM))
474import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)
475import Streamly.Internal.Data.Stream.Zip (ZipSerialM)
476import Streamly.Internal.Data.Pipe.Types (Pipe (..))
477import Streamly.Internal.Data.Time.Units
478 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
479 toAbsTime, TimeUnit64)
480import Streamly.Internal.Mutable.Prim.Var (Prim, Var)
482import Streamly.Internal.Data.Strict
484import qualified Streamly.Internal.Memory.Array as A
485import qualified Streamly.Data.Fold as FL
486import qualified Streamly.Internal.Data.Fold.Types as FL
487import qualified Streamly.Internal.Data.Stream.Prelude as P
488import qualified Streamly.Internal.Data.Stream.StreamK as K
489import qualified Streamly.Internal.Data.Stream.StreamD as D
491#ifdef USE_STREAMK_ONLY
492import qualified Streamly.Internal.Data.Stream.StreamK as S
494import qualified Streamly.Internal.Data.Stream.StreamD as S
497import qualified Streamly.Internal.Data.Stream.Serial as Serial
498import qualified Streamly.Internal.Data.Stream.Parallel as Par
499import qualified Streamly.Internal.Data.Stream.Zip as Z
501------------------------------------------------------------------------------
503------------------------------------------------------------------------------
505-- | Decompose a stream into its head and tail. If the stream is empty, returns
506-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
507-- the head of the stream and @ma@ its tail.
509-- This is a brute force primitive. Avoid using it as long as possible, use it
510-- when no other combinator can do the job. This can be used to do pretty much
511-- anything in an imperative manner, as it just breaks down the stream into
512-- individual elements and we can loop over them as we deem fit. For example,
513-- this can be used to convert a streamly stream into other stream types.
517uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
518uncons m = K.uncons (K.adapt m)
520------------------------------------------------------------------------------
521-- Generation by Unfolding
522------------------------------------------------------------------------------
529-- Just (a, b) -> a \`cons` unfoldr step b
532-- Build a stream by unfolding a /pure/ step function @step@ starting from a
533-- seed @s@. The step function returns the next element in the stream and the
534-- next seed value. When it is done it returns 'Nothing' and the stream ends.
541-- else Just (b, b + 1)
542-- in toList $ unfoldr f 0
549{-# INLINE_EARLY unfoldr #-}
550unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
551unfoldr step seed = fromStreamS (S.unfoldr step seed)
552{-# RULES "unfoldr fallback to StreamK" [1]
553 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
555-- | Build a stream by unfolding a /monadic/ step function starting from a
556-- seed. The step function returns the next element in the stream and the next
557-- seed value. When it is done it returns 'Nothing' and the stream ends. For
563-- then return Nothing
564-- else print b >> return (Just (b, b + 1))
565-- in drain $ unfoldrM f 0
573-- When run concurrently, the next unfold step can run concurrently with the
574-- processing of the output of the previous step. Note that more than one step
575-- cannot run concurrently as the next step depends on the output of the
579-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
580-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
586{-# INLINE_EARLY unfoldrM #-}
587unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
590{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
591{-# INLINE_EARLY unfoldrMSerial #-}
592unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
593unfoldrMSerial = Serial.unfoldrM
595{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}
596{-# INLINE_EARLY unfoldrMWSerial #-}
597unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a
598unfoldrMWSerial = Serial.unfoldrM
600{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}
601{-# INLINE_EARLY unfoldrMZipSerial #-}
602unfoldrMZipSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a
603unfoldrMZipSerial = Serial.unfoldrM
605-- | Convert an 'Unfold' into a stream by supplying it an input seed.
607-- >>> unfold (UF.replicateM 10) (putStrLn "hello")
611unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
612unfold unf x = fromStreamD $ D.unfold unf x
614------------------------------------------------------------------------------
615-- Specialized Generation
616------------------------------------------------------------------------------
618-- Faster than yieldM because there is no bind.
622-- yield a = a \`cons` nil
625-- Create a singleton stream from a pure value.
627-- The following holds in monadic streams, but not in Zip streams:
631-- yield = yieldM . pure
634-- In Zip applicative streams 'yield' is not the same as 'pure' because in that
635-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are
636-- equally efficient, in other cases 'yield' may be slightly more efficient
637-- than the other equivalent definitions.
641yield :: IsStream t => a -> t m a
646-- yieldM m = m \`consM` nil
649-- Create a singleton stream from a monadic action.
652-- > toList $ yieldM getLine
659yieldM :: (Monad m, IsStream t) => m a -> t m a
664-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
667-- Generate an infinite stream, whose values are the output of a function @f@
668-- applied on the corresponding index. Index starts at 0.
671-- > S.toList $ S.take 5 $ S.fromIndices id
676{-# INLINE fromIndices #-}
677fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
678fromIndices = fromStreamS . S.fromIndices
683-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
686-- Generate an infinite stream, whose values are the output of a monadic
687-- function @f@ applied on the corresponding index. Index starts at 0.
692{-# INLINE_EARLY fromIndicesM #-}
693fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
694fromIndicesM = K.fromIndicesM
696{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
697{-# INLINE fromIndicesMSerial #-}
698fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
699fromIndicesMSerial = fromStreamS . S.fromIndicesM
703-- replicateM = take n . repeatM
706-- Generate a stream by performing a monadic action @n@ times. Same as:
709-- drain $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
710-- drain $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
716{-# INLINE_EARLY replicateM #-}
717replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
718replicateM = K.replicateM
720{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
721{-# INLINE replicateMSerial #-}
722replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
723replicateMSerial n = fromStreamS . S.replicateM n
727-- replicate = take n . repeat
730-- Generate a stream of length @n@ by repeating a value @n@ times.
733{-# INLINE_NORMAL replicate #-}
734replicate :: (IsStream t, Monad m) => Int -> a -> t m a
735replicate n = fromStreamS . S.replicate n
738-- Generate an infinite stream by repeating a pure value.
741{-# INLINE_NORMAL repeat #-}
742repeat :: (IsStream t, Monad m) => a -> t m a
743repeat = fromStreamS . S.repeat
747-- repeatM = fix . consM
748-- repeatM = cycle1 . yieldM
751-- Generate a stream by repeatedly executing a monadic action forever.
754-- drain $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
755-- drain $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
758-- /Concurrent, infinite (do not use with 'parallely')/
761{-# INLINE_EARLY repeatM #-}
762repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
765{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
766{-# INLINE repeatMSerial #-}
767repeatMSerial :: MonadAsync m => m a -> SerialT m a
768repeatMSerial = fromStreamS . S.repeatM
772-- iterate f x = x \`cons` iterate f x
775-- Generate an infinite stream with @x@ as the first element and each
776-- successive element derived by applying the function @f@ on the previous
780-- > S.toList $ S.take 5 $ S.iterate (+1) 1
785{-# INLINE_NORMAL iterate #-}
786iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a
787iterate step = fromStreamS . S.iterate step
791-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)
794-- Generate an infinite stream with the first element generated by the action
795-- @m@ and each successive element derived by applying the monadic function
796-- @f@ on the previous element.
798-- When run concurrently, the next iteration can run concurrently with the
799-- processing of the previous iteration. Note that more than one iteration
800-- cannot run concurrently as the next iteration depends on the output of the
801-- previous iteration.
804-- drain $ serially $ S.take 10 $ S.iterateM
805-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
807-- drain $ asyncly $ S.take 10 $ S.iterateM
808-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
813-- /Since: 0.7.0 (signature change)/
816{-# INLINE_EARLY iterateM #-}
817iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a
820{-# RULES "iterateM serial" iterateM = iterateMSerial #-}
821{-# INLINE iterateMSerial #-}
822iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a
823iterateMSerial step = fromStreamS . S.iterateM step
825------------------------------------------------------------------------------
827------------------------------------------------------------------------------
831-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
834-- Construct a stream from a list of monadic actions. This is more efficient
835-- than 'fromFoldableM' for serial streams.
838{-# INLINE_EARLY fromListM #-}
839fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
840fromListM = fromStreamD . D.fromListM
841{-# RULES "fromListM fallback to StreamK" [1]
842 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
846-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
849-- Construct a stream from a 'Foldable' containing monadic actions.
852-- drain $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
853-- drain $ asyncly $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
856-- /Concurrent (do not use with 'parallely' on infinite containers)/
859{-# INLINE fromFoldableM #-}
860fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
861fromFoldableM = Prelude.foldr consM K.nil
863-- | Same as 'fromFoldable'.
866{-# DEPRECATED each "Please use fromFoldable instead." #-}
868each :: (IsStream t, Foldable f) => f a -> t m a
871-- | Read lines from an IO Handle into a stream of Strings.
874{-# DEPRECATED fromHandle
875 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
876fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
879 go = K.mkStream $ \_ yld _ stp -> do
880 eof <- liftIO $ IO.hIsEOF h
884 str <- liftIO $ IO.hGetLine h
887-- | Construct a stream by reading a 'Prim' 'Var' repeatedly.
891{-# INLINE fromPrimVar #-}
892fromPrimVar :: (IsStream t, MonadIO m, Prim a) => Var IO a -> t m a
893fromPrimVar = fromStreamD . D.fromPrimVar
895------------------------------------------------------------------------------
896-- Elimination by Folding
897------------------------------------------------------------------------------
899-- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
900-- an output structure using the step function @build@. @build@ is invoked with
901-- the next input element and the remaining (lazy) tail of the output
902-- structure. It builds a lazy output expression using the two. When the "tail
903-- structure" in the output expression is evaluated it calls @build@ again thus
904-- lazily consuming the input @stream@ until either the output expression built
905-- by @build@ is free of the "tail" or the input is exhausted in which case
906-- @final@ is used as the terminating case for the output structure. For more
907-- details see the description in the previous section.
909-- Example, determine if any element is 'odd' in a stream:
911-- >>> S.foldrM (\x xs -> if odd x then return True else xs) (return False) $ S.fromList (2:4:5:undefined)
914-- /Since: 0.7.0 (signature changed)/
916-- /Since: 0.2.0 (signature changed)/
920foldrM :: Monad m => (a -> m b -> m b) -> m b -> SerialT m a -> m b
923-- | Right fold to a streaming monad.
925-- > foldrS S.cons S.nil === id
927-- 'foldrS' can be used to perform stateless stream to stream transformations
928-- like map and filter in general. It can be coupled with a scan to perform
929-- stateful transformations. However, note that the custom map and filter
930-- routines can be much more efficient than this due to better stream fusion.
932-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
935-- Find if any element in the stream is 'True':
937-- >>> S.toList $ S.foldrS (\x xs -> if odd x then return True else xs) (return False) $ (S.fromList (2:4:5:undefined) :: SerialT IO Int)
940-- Map (+2) on odd elements and filter out the even elements:
942-- >>> S.toList $ S.foldrS (\x xs -> if odd x then (x + 2) `S.cons` xs else xs) S.nil $ (S.fromList [1..5] :: SerialT IO Int)
945-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
946-- is much more efficient:
948-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
952foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
955-- | Right fold to a transformer monad. This is the most general right fold
956-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
957-- implementation can be more efficient:
960-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
962-- 'foldrT' can be used to translate streamly streams to other transformer
963-- monads e.g. to a different streaming type.
967foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
968 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
969foldrT f z s = S.foldrT f z (toStreamS s)
971-- | Right fold, lazy for lazy monads and pure streams, and strict for strict
974-- Please avoid using this routine in strict monads like IO unless you need a
975-- strict right fold. This is provided only for use in lazy monads (e.g.
976-- Identity) or pure streams. Note that with this signature it is not possible
977-- to implement a lazy foldr when the monad @m@ is strict. In that case it
978-- would be strict in its accumulator and therefore would necessarily consume
983foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
986-- XXX This seems to be of limited use as it cannot be used to construct
987-- recursive structures and for reduction foldl1' is better.
989-- | Lazy right fold for non-empty streams, using first element as the starting
990-- value. Returns 'Nothing' if the stream is empty.
994{-# DEPRECATED foldr1 "Use foldrM instead." #-}
995foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
996foldr1 f m = S.foldr1 f (toStreamS m)
998-- | Strict left fold with an extraction function. Like the standard strict
999-- left fold, but applies a user supplied extraction function (the third
1000-- argument) to the folded value at the end. This is designed to work with the
1001-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
1004{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
1006foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
1009-- | Left associative/strict push fold. @foldl' reduce initial stream@ invokes
1010-- @reduce@ with the accumulator and the next input in the input stream, using
1011-- @initial@ as the initial value of the current value of the accumulator. When
1012-- the input is exhausted the current value of the accumulator is returned.
1013-- Make sure to use a strict data structure for accumulator to not build
1014-- unnecessary lazy expressions unless that's what you want. See the previous
1015-- section for more details.
1018{-# INLINE foldl' #-}
1019foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
1022-- | Strict left fold, for non-empty streams, using first element as the
1023-- starting value. Returns 'Nothing' if the stream is empty.
1026{-# INLINE foldl1' #-}
1027foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1031 Nothing -> return Nothing
1033 res <- foldl' step h t
1036-- | Like 'foldx', but with a monadic step function.
1039{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
1040{-# INLINE foldxM #-}
1041foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
1044-- | Like 'foldl'' but with a monadic step function.
1047{-# INLINE foldlM' #-}
1048foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
1049foldlM' step begin m = S.foldlM' step begin $ toStreamS m
1051------------------------------------------------------------------------------
1053------------------------------------------------------------------------------
1055-- | Fold a stream using the supplied left fold.
1057-- >>> S.fold FL.sum (S.enumerateFromTo 1 100)
1062fold :: Monad m => Fold m a b -> SerialT m a -> m b
1065------------------------------------------------------------------------------
1067------------------------------------------------------------------------------
1070-- | Drain a stream to a 'Sink'.
1071{-# INLINE runSink #-}
1072runSink :: Monad m => Sink m a -> SerialT m a -> m ()
1073runSink = fold . toFold
1076------------------------------------------------------------------------------
1078------------------------------------------------------------------------------
1081-- > drain = mapM_ (\_ -> return ())
1083-- Run a stream, discarding the results. By default it interprets the stream
1084-- as 'SerialT', to run other types of streams use the type adapting
1085-- combinators for example @drain . 'asyncly'@.
1089drain :: Monad m => SerialT m a -> m ()
1092-- | Run a stream, discarding the results. By default it interprets the stream
1093-- as 'SerialT', to run other types of streams use the type adapting
1094-- combinators for example @runStream . 'asyncly'@.
1097{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
1098{-# INLINE runStream #-}
1099runStream :: Monad m => SerialT m a -> m ()
1103-- > drainN n = drain . take n
1105-- Run maximum up to @n@ iterations of a stream.
1108{-# INLINE drainN #-}
1109drainN :: Monad m => Int -> SerialT m a -> m ()
1110drainN n = drain . take n
1113-- > runN n = runStream . take n
1115-- Run maximum up to @n@ iterations of a stream.
1118{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
1120runN :: Monad m => Int -> SerialT m a -> m ()
1124-- > drainWhile p = drain . takeWhile p
1126-- Run a stream as long as the predicate holds true.
1129{-# INLINE drainWhile #-}
1130drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1131drainWhile p = drain . takeWhile p
1134-- > runWhile p = runStream . takeWhile p
1136-- Run a stream as long as the predicate holds true.
1139{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
1140{-# INLINE runWhile #-}
1141runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1142runWhile = drainWhile
1144-- | Determine whether the stream is empty.
1148null :: Monad m => SerialT m a -> m Bool
1149null = S.null . toStreamS
1151-- | Extract the first element of the stream, if any.
1157head :: Monad m => SerialT m a -> m (Maybe a)
1158head = S.head . toStreamS
1160-- | Extract the first element of the stream, if any, otherwise use the
1161-- supplied default value. It can help avoid one branch in high performance
1165{-# INLINE headElse #-}
1166headElse :: Monad m => a -> SerialT m a -> m a
1167headElse x = D.headElse x . toStreamD
1170-- > tail = fmap (fmap snd) . uncons
1172-- Extract all but the first element of the stream, if any.
1176tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1177tail m = K.tail (K.adapt m)
1179-- | Extract all but the last element of the stream, if any.
1183init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1184init m = K.init (K.adapt m)
1186-- | Extract the last element of the stream, if any.
1188-- > last xs = xs !! (length xs - 1)
1192last :: Monad m => SerialT m a -> m (Maybe a)
1193last m = S.last $ toStreamS m
1195-- | Determine whether an element is present in the stream.
1199elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1200elem e m = S.elem e (toStreamS m)
1202-- | Determine whether an element is not present in the stream.
1205{-# INLINE notElem #-}
1206notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1207notElem e m = S.notElem e (toStreamS m)
1209-- | Determine the length of the stream.
1212{-# INLINE length #-}
1213length :: Monad m => SerialT m a -> m Int
1214length = foldl' (\n _ -> n + 1) 0
1216-- | Determine whether all elements of a stream satisfy a predicate.
1220all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1221all p m = S.all p (toStreamS m)
1223-- | Determine whether any of the elements of a stream satisfy a predicate.
1227any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1228any p m = S.any p (toStreamS m)
1230-- | Determines if all elements of a boolean stream are True.
1234and :: Monad m => SerialT m Bool -> m Bool
1237-- | Determines whether at least one element of a boolean stream is True.
1241or :: Monad m => SerialT m Bool -> m Bool
1244-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
1245-- the stream is empty. Note that this is not numerically stable for floating
1250sum :: (Monad m, Num a) => SerialT m a -> m a
1253-- | Determine the product of all elements of a stream of numbers. Returns @1@
1254-- when the stream is empty.
1257{-# INLINE product #-}
1258product :: (Monad m, Num a) => SerialT m a -> m a
1259product = foldl' (*) 1
1263-- minimum = 'minimumBy' compare
1266-- Determine the minimum element in a stream.
1269{-# INLINE minimum #-}
1270minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1271minimum m = S.minimum (toStreamS m)
1273-- | Determine the minimum element in a stream using the supplied comparison
1277{-# INLINE minimumBy #-}
1278minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1279minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1283-- maximum = 'maximumBy' compare
1286-- Determine the maximum element in a stream.
1289{-# INLINE maximum #-}
1290maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1291maximum m = S.maximum (toStreamS m)
1293-- | Determine the maximum element in a stream using the supplied comparison
1297{-# INLINE maximumBy #-}
1298maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1299maximumBy cmp m = S.maximumBy cmp (toStreamS m)
1301-- | Lookup the element at the given index.
1305(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
1306m !! i = toStreamS m S.!! i
1308-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1309-- first pair where the key equals the given value @a@.
1311-- > lookup = snd <$> find ((==) . fst)
1314{-# INLINE lookup #-}
1315lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
1316lookup a m = S.lookup a (toStreamS m)
1318-- | Like 'findM' but with a non-monadic predicate.
1320-- > find p = findM (return . p)
1324find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
1325find p m = S.find p (toStreamS m)
1327-- | Returns the first element that satisfies the given predicate.
1331findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
1332findM p m = S.findM p (toStreamS m)
1334-- | Find all the indices where the element in the stream satisfies the given
1338{-# INLINE findIndices #-}
1339findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
1340findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
1342-- | Returns the first index that satisfies the given predicate.
1345{-# INLINE findIndex #-}
1346findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
1347findIndex p = head . findIndices p
1349-- | Find all the indices where the value of the element in the stream is equal
1350-- to the given value.
1353{-# INLINE elemIndices #-}
1354elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1355elemIndices a = findIndices (==a)
1357-- | Returns the first index where a given value is found in the stream.
1359-- > elemIndex a = findIndex (== a)
1362{-# INLINE elemIndex #-}
1363elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
1364elemIndex a = findIndex (== a)
1366------------------------------------------------------------------------------
1368------------------------------------------------------------------------------
1370-- | Returns 'True' if the first stream is the same as or a prefix of the
1371-- second. A stream is a prefix of itself.
1374-- > S.isPrefixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1379{-# INLINE isPrefixOf #-}
1380isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1381isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
1383-- | Returns 'True' if all the elements of the first stream occur, in order, in
1384-- the second stream. The elements do not have to occur consecutively. A stream
1385-- is a subsequence of itself.
1388-- > S.isSubsequenceOf (S.fromList "hlo") (S.fromList "hello" :: SerialT IO Char)
1393{-# INLINE isSubsequenceOf #-}
1394isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1395isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
1397-- | Drops the given prefix from a stream. Returns 'Nothing' if the stream does
1398-- not start with the given prefix. Returns @Just nil@ when the prefix is the
1399-- same as the stream.
1402{-# INLINE stripPrefix #-}
1404 :: (Eq a, IsStream t, Monad m)
1405 => t m a -> t m a -> m (Maybe (t m a))
1406stripPrefix m1 m2 = fmap fromStreamD <$>
1407 D.stripPrefix (toStreamD m1) (toStreamD m2)
1409------------------------------------------------------------------------------
1411------------------------------------------------------------------------------
1413-- XXX this can utilize parallel mapping if we implement it as drain . mapM
1415-- > mapM_ = drain . mapM
1417-- Apply a monadic action to each element of the stream and discard the output
1418-- of the action. This is not really a pure transformation operation but a
1419-- transformation followed by fold.
1423mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
1424mapM_ f m = S.mapM_ f $ toStreamS m
1426------------------------------------------------------------------------------
1428------------------------------------------------------------------------------
1432-- toList = S.foldr (:) []
1435-- Convert a stream into a list in the underlying monad. The list can be
1436-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
1437-- IO) the whole list is generated and buffered before it can be consumed.
1439-- /Warning!/ working on large lists accumulated as buffers in memory could be
1440-- very inefficient, consider using "Streamly.Array" instead.
1443{-# INLINE toList #-}
1444toList :: Monad m => SerialT m a -> m [a]
1449-- toListRev = S.foldl' (flip (:)) []
1452-- Convert a stream into a list in reverse order in the underlying monad.
1454-- /Warning!/ working on large lists accumulated as buffers in memory could be
1455-- very inefficient, consider using "Streamly.Array" instead.
1458{-# INLINE toListRev #-}
1459toListRev :: Monad m => SerialT m a -> m [a]
1460toListRev = D.toListRev . toStreamD
1464-- toHandle h = S.mapM_ $ hPutStrLn h
1467-- Write a stream of Strings to an IO Handle.
1470{-# DEPRECATED toHandle
1471 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1472toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
1476 let stop = return ()
1477 single a = liftIO (IO.hPutStrLn h a)
1478 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1479 in K.foldStream defState yieldk single stop m1
1481-- XXX rename these to write/writeRev to make the naming consistent with folds
1484-- | A fold that buffers its input to a pure stream.
1486-- /Warning!/ working on large streams accumulated as buffers in memory could
1487-- be very inefficient, consider using "Streamly.Array" instead.
1490{-# INLINE toStream #-}
1491toStream :: Monad m => Fold m a (SerialT Identity a)
1492toStream = Fold (\f x -> return $ f . (x `K.cons`))
1494 (return . ($ K.nil))
1496-- This is more efficient than 'toStream'. toStream is exactly the same as
1497-- reversing the stream after toStreamRev.
1499-- | Buffers the input stream to a pure stream in the reverse order of the
1502-- /Warning!/ working on large streams accumulated as buffers in memory could
1503-- be very inefficient, consider using "Streamly.Array" instead.
1507-- xn : ... : x2 : x1 : []
1508{-# INLINABLE toStreamRev #-}
1509toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1510toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
1512-- | Convert a stream to a pure stream.
1515-- toPure = foldr cons nil
1520{-# INLINE toPure #-}
1521toPure :: Monad m => SerialT m a -> m (SerialT Identity a)
1522toPure = foldr K.cons K.nil
1524-- | Convert a stream to a pure stream in reverse order.
1527-- toPureRev = foldl' (flip cons) nil
1532{-# INLINE toPureRev #-}
1533toPureRev :: Monad m => SerialT m a -> m (SerialT Identity a)
1534toPureRev = foldl' (flip K.cons) K.nil
1536------------------------------------------------------------------------------
1537-- Concurrent Application
1538------------------------------------------------------------------------------
1540-- XXX mkParallel is usually used to make serial streams concurrent by
1541-- evaluating the stream in another thread. Ideally, we should not be creating
1542-- an SVar if the stream itself is concurrent. For example, if we are zipping
1543-- two streams that are concurrently composed, we should be able to pull from
1544-- the SVar of that concurrent stream instead of creating another redundant
1545-- SVar on top of it.
1547{-# INLINE_NORMAL mkParallel #-}
1548mkParallel :: (IsStream t, MonadAsync m)
1549 => State K.Stream m a -> t m a -> m (t m a)
1550mkParallel st m = fmap fromStreamD $ D.mkParallel st (toStreamD m)
1551-- mkParallel = Par.mkParallel
1559-- | Parallel function application operator for streams; just like the regular
1560-- function application operator '$' except that it is concurrent. The
1561-- following code prints a value every second even though each stage adds a 1
1567-- S.mapM (\\x -> threadDelay 1000000 >> print x)
1568-- |$ S.repeatM (threadDelay 1000000 >> return 1)
1575(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
1576f |$ x = D.fromStreamD $
1577 D.applyParallel (D.toStreamD . f . D.fromStreamD) (D.toStreamD x)
1578-- (|$) = Par.applyParallel
1580-- | Parallel reverse function application operator for streams; just like the
1581-- regular reverse function application operator '&' except that it is
1586-- S.repeatM (threadDelay 1000000 >> return 1)
1587-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
1594(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
1597-- | Parallel function application operator; applies a @run@ or @fold@ function
1598-- to a stream such that the fold consumer and the stream producer run in
1599-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
1600-- underlying monad. The @.@ at the end of the operator is a mnemonic for
1601-- termination of the stream.
1604-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
1605-- |$. S.repeatM (threadDelay 1000000 >> return 1)
1612(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
1613(|$.) = D.foldParallel
1614-- (|$.) = Par.foldParallel
1616-- | Parallel reverse function application operator for applying a run or fold
1617-- functions to a stream. Just like '|$.' except that the operands are reversed.
1620-- S.repeatM (threadDelay 1000000 >> return 1)
1621-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
1628(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
1631------------------------------------------------------------------------------
1632-- General Transformation
1633------------------------------------------------------------------------------
1635-- | Use a 'Pipe' to transform a stream.
1636{-# INLINE transform #-}
1637transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
1638transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
1640------------------------------------------------------------------------------
1641-- Transformation by Folding (Scans)
1642------------------------------------------------------------------------------
1644-- XXX It may be useful to have a version of scan where we can keep the
1645-- accumulator independent of the value emitted. So that we do not necessarily
1646-- have to keep a value in the accumulator which we are not using. We can pass
1647-- an extraction function that will take the accumulator and the current value
1648-- of the element and emit the next value in the stream. That will also make it
1649-- possible to modify the accumulator after using it. In fact, the step function
1650-- can return new accumulator and the value to be emitted. The signature would
1651-- be more like mapAccumL. Or we can change the signature of scanx to
1654-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
1655-- user supplied extraction function (the third argument) at each step. This is
1656-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
1659-- /Since: 0.7.0 (Monad m constraint)/
1662{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
1664scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
1667-- XXX this needs to be concurrent
1668-- | Like 'scanl'' but with a monadic fold function.
1671{-# INLINE scanlM' #-}
1672scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1673scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
1675-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
1676-- however it adds an extra element.
1679-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
1684-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
1685-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
1688-- The output of 'scanl'' is the initial value of the accumulator followed by
1689-- all the intermediate steps and the final result of 'foldl''.
1691-- By streaming the accumulated state after each fold step, we can share the
1692-- state across multiple stages of stream composition. Each stage can modify or
1693-- extend the state, do some processing with it and emit it for the next stage,
1694-- thus modularizing the stream processing. This can be useful in
1695-- stateful or event-driven programming.
1697-- Consider the following monolithic example, computing the sum and the product
1698-- of the elements in a stream in one go using a @foldl'@:
1701-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
1705-- Using @scanl'@ we can make it modular by computing the sum in the first
1706-- stage and passing it down to the next stage for computing the product:
1709-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
1710-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
1711-- $ S.fromList \[1,2,3,4]
1715-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
1716-- lazy expressions inside the accumulator, it is recommended that a strict
1717-- data structure is used for accumulator.
1720{-# INLINE scanl' #-}
1721scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1722scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
1724-- | Like 'scanl'' but does not stream the initial value of the accumulator.
1726-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
1729{-# INLINE postscanl' #-}
1730postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1731postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
1733-- XXX this needs to be concurrent
1734-- | Like 'postscanl'' but with a monadic step function.
1737{-# INLINE postscanlM' #-}
1738postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
1739postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
1741-- XXX prescanl does not sound very useful, enable only if there is a
1742-- compelling use case.
1744-- | Like scanl' but does not stream the final value of the accumulator.
1747{-# INLINE prescanl' #-}
1748prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
1749prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
1751-- XXX this needs to be concurrent
1752-- | Like postscanl' but with a monadic step function.
1755{-# INLINE prescanlM' #-}
1756prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
1757prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
1759-- XXX this needs to be concurrent
1760-- | Like 'scanl1'' but with a monadic step function.
1763{-# INLINE scanl1M' #-}
1764scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
1765scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
1767-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
1768-- is used as the initial value of the accumulator. Does nothing if the stream
1772-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
1777{-# INLINE scanl1' #-}
1778scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
1779scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
1781------------------------------------------------------------------------------
1782-- Scanning with a Fold
1783------------------------------------------------------------------------------
1785-- | Scan a stream using the given monadic fold.
1789scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
1790scan (Fold step begin done) = P.scanlMx' step begin done
1792-- | Postscan a stream using the given monadic fold.
1795{-# INLINE postscan #-}
1796postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
1797postscan (Fold step begin done) = P.postscanlMx' step begin done
1799------------------------------------------------------------------------------
1800-- Stateful Transformations
1801------------------------------------------------------------------------------
1803-- | Apply a function on every two successive elements of a stream. If the
1804-- stream consists of a single element the output is an empty stream.
1808{-# INLINE rollingMap #-}
1809rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
1810rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
1812-- | Like 'rollingMap' but with an effectful map function.
1816{-# INLINE rollingMapM #-}
1817rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
1818rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
1820------------------------------------------------------------------------------
1821-- Transformation by Filtering
1822------------------------------------------------------------------------------
1824-- | Include only those elements that pass a predicate.
1827{-# INLINE filter #-}
1828#if __GLASGOW_HASKELL__ != 802
1829-- GHC 8.2.2 crashes with this code, when used with "stack"
1830filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1831filter p m = fromStreamS $ S.filter p $ toStreamS m
1833filter :: IsStream t => (a -> Bool) -> t m a -> t m a
1837-- | Same as 'filter' but with a monadic predicate.
1840{-# INLINE filterM #-}
1841filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1842filterM p m = fromStreamD $ D.filterM p $ toStreamD m
1844-- | Drop repeated elements that are adjacent to each other.
1848uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
1849uniq = fromStreamD . D.uniq . toStreamD
1851-- | Ensures that all the elements of the stream are identical and then returns
1852-- that unique element.
1856the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
1857the m = S.the (toStreamS m)
1859-- | Take first 'n' elements from the stream and discard the rest.
1863take :: (IsStream t, Monad m) => Int -> t m a -> t m a
1864take n m = fromStreamS $ S.take n $ toStreamS
1865 (maxYields (Just (fromIntegral n)) m)
1867-- | End the stream as soon as the predicate fails on an element.
1870{-# INLINE takeWhile #-}
1871takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1872takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
1874-- | Same as 'takeWhile' but with a monadic predicate.
1877{-# INLINE takeWhileM #-}
1878takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1879takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
1881-- | @takeByTime duration@ yields stream elements upto specified time
1882-- @duration@. The duration starts when the stream is evaluated for the first
1883-- time, before the first element is yielded. The time duration is checked
1884-- before generating each element, if the duration has expired the stream
1887-- The total time taken in executing the stream is guaranteed to be /at least/
1888-- @duration@, however, because the duration is checked before generating an
1889-- element, the upper bound is indeterminate and depends on the time taken in
1890-- generating and processing the last element.
1892-- No element is yielded if the duration is zero. At least one element is
1893-- yielded if the duration is non-zero.
1897{-# INLINE takeByTime #-}
1898takeByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1899takeByTime d = fromStreamD . D.takeByTime d . toStreamD
1901-- | Discard first 'n' elements from the stream and take the rest.
1905drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
1906drop n m = fromStreamS $ S.drop n $ toStreamS m
1908-- | Drop elements in the stream as long as the predicate succeeds and then
1909-- take the rest of the stream.
1912{-# INLINE dropWhile #-}
1913dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
1914dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
1916-- | Same as 'dropWhile' but with a monadic predicate.
1919{-# INLINE dropWhileM #-}
1920dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
1921dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
1923-- | @dropByTime duration@ drops stream elements until specified @duration@ has
1924-- passed. The duration begins when the stream is evaluated for the first
1925-- time. The time duration is checked /after/ generating a stream element, the
1926-- element is yielded if the duration has expired otherwise it is dropped.
1928-- The time elapsed before starting to generate the first element is /at most/
1929-- @duration@, however, because the duration expiry is checked after the
1930-- element is generated, the lower bound is indeterminate and depends on the
1931-- time taken in generating an element.
1933-- All elements are yielded if the duration is zero.
1937{-# INLINE dropByTime #-}
1938dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1939dropByTime d = fromStreamD . D.dropByTime d . toStreamD
1941------------------------------------------------------------------------------
1942-- Transformation by Mapping
1943------------------------------------------------------------------------------
1947-- mapM f = sequence . map f
1950-- Apply a monadic function to each element of the stream and replace it with
1951-- the output of the resulting action.
1954-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
1957-- drain $ S.replicateM 10 (return 1)
1958-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
1960-- drain $ S.replicateM 10 (return 1)
1961-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
1964-- /Concurrent (do not use with 'parallely' on infinite streams)/
1967{-# INLINE_EARLY mapM #-}
1968mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
1971{-# RULES "mapM serial" mapM = mapMSerial #-}
1972{-# INLINE mapMSerial #-}
1973mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
1974mapMSerial = Serial.mapM
1978-- sequence = mapM id
1981-- Replace the elements of a stream of monadic actions with the outputs of
1985-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
1988-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1989-- & (serially . S.sequence)
1991-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
1992-- & (asyncly . S.sequence)
1995-- /Concurrent (do not use with 'parallely' on infinite streams)/
1998{-# INLINE sequence #-}
1999sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
2000sequence m = fromStreamS $ S.sequence (toStreamS m)
2002------------------------------------------------------------------------------
2003-- Transformation by Map and Filter
2004------------------------------------------------------------------------------
2006-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
2007-- elements, and return a stream of values extracted from 'Just'.
2012-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
2016{-# INLINE mapMaybe #-}
2017mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
2018mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
2020-- | Like 'mapMaybe' but maps a monadic function.
2025-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
2028-- /Concurrent (do not use with 'parallely' on infinite streams)/
2031{-# INLINE_EARLY mapMaybeM #-}
2032mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
2033 => (a -> m (Maybe b)) -> t m a -> t m b
2034mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
2036{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
2037{-# INLINE mapMaybeMSerial #-}
2038mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
2039mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
2041------------------------------------------------------------------------------
2042-- Transformation by Reordering
2043------------------------------------------------------------------------------
2045-- XXX Use a compact region list to temporarily store the list, in both reverse
2046-- as well as in reverse'.
2048-- /Note:/ 'reverse'' is much faster than this, use that when performance
2051-- > reverse = S.foldlT (flip S.cons) S.nil
2053-- | Returns the elements of the stream in reverse order. The stream must be
2054-- finite. Note that this necessarily buffers the entire stream in memory.
2056-- /Since 0.7.0 (Monad m constraint)/
2059{-# INLINE reverse #-}
2060reverse :: (IsStream t, Monad m) => t m a -> t m a
2061reverse s = fromStreamS $ S.reverse $ toStreamS s
2063-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
2066{-# INLINE reverse' #-}
2067reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
2068reverse' s = fromStreamD $ D.reverse' $ toStreamD s
2070------------------------------------------------------------------------------
2071-- Transformation by Inserting
2072------------------------------------------------------------------------------
2074-- intersperseM = intersperseBySpan 1
2076-- | Generate a stream by inserting the result of a monadic action between
2077-- consecutive elements of the given stream. Note that the monadic action is
2078-- performed after the stream action before which its result is inserted.
2081-- > S.toList $ S.intersperseM (return ',') $ S.fromList "hello"
2086{-# INLINE intersperseM #-}
2087intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2088intersperseM m = fromStreamS . S.intersperseM m . toStreamS
2090-- | Generate a stream by inserting a given element between consecutive
2091-- elements of the given stream.
2094-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
2099{-# INLINE intersperse #-}
2100intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2101intersperse a = fromStreamS . S.intersperse a . toStreamS
2103-- | Insert a monadic action after each element in the stream.
2106{-# INLINE intersperseSuffix #-}
2107intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2108intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
2110-- | Perform a side effect after each element of a stream. The output of the
2111-- effectful action is discarded, therefore, the input stream remains
2115-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
2120{-# INLINE intersperseSuffix_ #-}
2121intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2122intersperseSuffix_ m = Serial.mapM (\x -> void m >> return x)
2124-- | Introduces a delay of specified seconds after each element of a stream.
2128{-# INLINE delayPost #-}
2129delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2130delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
2132-- | Like 'intersperseSuffix' but intersperses a monadic action into the input
2133-- stream after every @n@ elements and after the last element.
2136-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"
2142{-# INLINE intersperseSuffixBySpan #-}
2143intersperseSuffixBySpan :: (IsStream t, MonadAsync m)
2144 => Int -> m a -> t m a -> t m a
2145intersperseSuffixBySpan n eff =
2146 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
2149-- | Intersperse a monadic action into the input stream after every @n@
2153-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
2158{-# INLINE intersperseBySpan #-}
2159intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
2160intersperseBySpan _n _f _xs = undefined
2163-- | Intersperse a monadic action into the input stream after every @n@
2167-- > S.drain $ S.interjectSuffix 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"
2172{-# INLINE interjectSuffix #-}
2174 :: (IsStream t, MonadAsync m)
2175 => Double -> m a -> t m a -> t m a
2176interjectSuffix n f xs = xs `Par.parallelFst` repeatM timed
2177 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
2179-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
2180-- @stream@ that is less than @elem@ when compared using @cmp@.
2183-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
2187-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
2192{-# INLINE insertBy #-}
2194 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
2195insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
2197------------------------------------------------------------------------------
2199------------------------------------------------------------------------------
2201-- | Deletes the first occurrence of the element in the stream that satisfies
2202-- the given equality predicate.
2205-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
2210{-# INLINE deleteBy #-}
2211deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
2212deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
2214------------------------------------------------------------------------------
2216------------------------------------------------------------------------------
2219-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
2220-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
2222-- Pair each element in a stream with its index, starting from index 0.
2225-- > S.toList $ S.indexed $ S.fromList "hello"
2226-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
2230{-# INLINE indexed #-}
2231indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
2232indexed = fromStreamD . D.indexed . toStreamD
2235-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
2236-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
2238-- Pair each element in a stream with its index, starting from the
2239-- given index @n@ and counting down.
2242-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
2243-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
2247{-# INLINE indexedR #-}
2248indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2249indexedR n = fromStreamD . D.indexedR n . toStreamD
2251------------------------------------------------------------------------------
2253------------------------------------------------------------------------------
2255-- | Compare two streams for equality using an equality function.
2258{-# INLINABLE eqBy #-}
2259eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
2262-- | Compare two streams lexicographically using a comparison function.
2265{-# INLINABLE cmpBy #-}
2267 :: (IsStream t, Monad m)
2268 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
2271------------------------------------------------------------------------------
2273------------------------------------------------------------------------------
2275-- | Merge two streams using a comparison function. The head elements of both
2276-- the streams are compared and the smaller of the two elements is emitted, if
2277-- both elements are equal then the element from the first stream is used
2280-- If the streams are sorted in ascending order, the resulting stream would
2281-- also remain sorted in ascending order.
2284-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2289{-# INLINABLE mergeBy #-}
2291 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2292mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
2294-- | Like 'mergeBy' but with a monadic comparison function.
2296-- Merge two streams randomly:
2299-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
2300-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
2304-- Merge two streams in a proportion of 2:1:
2307-- proportionately m n = do
2308-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
2309-- return $ \\_ _ -> do
2310-- r <- readIORef ref
2311-- writeIORef ref $ tail r
2315-- f <- proportionately 2 1
2316-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
2320-- [1,1,2,1,1,2,1,1,2]
2324{-# INLINABLE mergeByM #-}
2326 :: (IsStream t, Monad m)
2327 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2328mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
2331-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
2332{-# INLINABLE mergeEndByAny #-}
2334 :: (IsStream t, Monad m)
2335 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2336mergeEndByAny f m1 m2 = fromStreamD $
2337 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
2339-- Like 'mergeByM' but stops merging as soon as the first stream stops.
2340{-# INLINABLE mergeEndByFirst #-}
2342 :: (IsStream t, Monad m)
2343 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2344mergeEndByFirst f m1 m2 = fromStreamS $
2345 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
2348-- Holding this back for now, we may want to use the name "merge" differently
2350-- | Same as @'mergeBy' 'compare'@.
2353-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2358{-# INLINABLE merge #-}
2360 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
2361merge = mergeBy compare
2364-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
2365-- merged are generated concurrently).
2368mergeAsyncBy :: (IsStream t, MonadAsync m)
2369 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2370mergeAsyncBy f m1 m2 = K.mkStream $ \st stp sng yld -> do
2371 ma <- mkAsync' st m1
2372 mb <- mkAsync' st m2
2373 K.foldStream st stp sng yld (K.mergeBy f ma mb)
2375-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
2376-- merged are generated concurrently).
2379mergeAsyncByM :: (IsStream t, MonadAsync m)
2380 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2381mergeAsyncByM f m1 m2 = K.mkStream $ \st stp sng yld -> do
2382 ma <- mkAsync' st m1
2383 mb <- mkAsync' st m2
2384 K.foldStream st stp sng yld (K.mergeByM f ma mb)
2386------------------------------------------------------------------------------
2388------------------------------------------------------------------------------
2390-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
2391-- The first argument specifies a merge or concat function that is used to
2392-- merge the streams generated by applying the second argument i.e. the @map@
2393-- function to each element of the input stream. The concat function could be
2394-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
2395-- and the second argument could be any stream generation function using a
2398-- /Compare 'foldMapWith'/
2401{-# INLINE concatMapWith #-}
2404 => (forall c. t m c -> t m c -> t m c)
2408concatMapWith = K.concatMapBy
2410-- | Map a stream producing function on each element of the stream and then
2411-- flatten the results into a single stream.
2414-- concatMap = 'concatMapWith' 'Serial.serial'
2415-- concatMap f = 'concatMapM' (return . f)
2419{-# INLINE concatMap #-}
2420concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
2421concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
2423-- | Append the outputs of two streams, yielding all the elements from the
2424-- first stream and then yielding all the elements from the second stream.
2426-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
2427-- few (say 100) streams because it can fuse via stream fusion. However, it
2428-- does not scale for a large number of streams (say 1000s) and becomes
2429-- qudartically slow. Therefore use this for custom appending of a few streams
2430-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
2431-- infinite containers of streams.
2434{-# INLINE append #-}
2435append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2436append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
2438-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
2439-- XXX Document the interleaving behavior of side effects in all the
2440-- interleaving combinators.
2441-- XXX Write time-domain equivalents of these. In the time domain we can
2442-- interleave two streams such that the value of second stream is always taken
2443-- from its last value even if no new value is being yielded, like
2444-- zipWithLatest. It would be something like interleaveWithLatest.
2446-- | Interleaves the outputs of two streams, yielding elements from each stream
2447-- alternately, starting from the first stream. If any of the streams finishes
2448-- early the other stream continues alone until it too finishes.
2450-- >>> :set -XOverloadedStrings
2451-- >>> interleave "ab" ",,,," :: SerialT Identity Char
2453-- >>> interleave "abcd" ",," :: SerialT Identity Char
2456-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
2458-- Do not use at scale in concatMapWith.
2461{-# INLINE interleave #-}
2462interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2463interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
2465-- | Interleaves the outputs of two streams, yielding elements from each stream
2466-- alternately, starting from the first stream. As soon as the first stream
2467-- finishes, the output stops, discarding the remaining part of the second
2468-- stream. In this case, the last element in the resulting stream would be from
2469-- the second stream. If the second stream finishes early then the first stream
2470-- still continues to yield elements until it finishes.
2472-- >>> :set -XOverloadedStrings
2473-- >>> interleaveSuffix "abc" ",,,," :: SerialT Identity Char
2475-- >>> interleaveSuffix "abc" "," :: SerialT Identity Char
2478-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
2480-- Do not use at scale in concatMapWith.
2483{-# INLINE interleaveSuffix #-}
2484interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2485interleaveSuffix m1 m2 =
2486 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
2488-- | Interleaves the outputs of two streams, yielding elements from each stream
2489-- alternately, starting from the first stream and ending at the first stream.
2490-- If the second stream is longer than the first, elements from the second
2491-- stream are infixed with elements from the first stream. If the first stream
2492-- is longer then it continues yielding elements even after the second stream
2495-- >>> :set -XOverloadedStrings
2496-- >>> interleaveInfix "abc" ",,,," :: SerialT Identity Char
2498-- >>> interleaveInfix "abc" "," :: SerialT Identity Char
2501-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
2503-- Do not use at scale in concatMapWith.
2506{-# INLINE interleaveInfix #-}
2507interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2508interleaveInfix m1 m2 =
2509 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
2511-- | Interleaves the outputs of two streams, yielding elements from each stream
2512-- alternately, starting from the first stream. The output stops as soon as any
2513-- of the two streams finishes, discarding the remaining part of the other
2514-- stream. The last element of the resulting stream would be from the longer
2517-- >>> :set -XOverloadedStrings
2518-- >>> interleaveMin "ab" ",,,," :: SerialT Identity Char
2520-- >>> interleaveMin "abcd" ",," :: SerialT Identity Char
2523-- 'interleaveMin' is dual to 'interleave'.
2525-- Do not use at scale in concatMapWith.
2528{-# INLINE interleaveMin #-}
2529interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2530interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
2532-- | Schedule the execution of two streams in a fair round-robin manner,
2533-- executing each stream once, alternately. Execution of a stream may not
2534-- necessarily result in an output, a stream may chose to @Skip@ producing an
2535-- element until later giving the other stream a chance to run. Therefore, this
2536-- combinator fairly interleaves the execution of two streams rather than
2537-- fairly interleaving the output of the two streams. This can be useful in
2538-- co-operative multitasking without using explicit threads. This can be used
2539-- as an alternative to `async`.
2541-- Do not use at scale in concatMapWith.
2544{-# INLINE roundrobin #-}
2545roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
2546roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
2548-- | Map a stream producing monadic function on each element of the stream
2549-- and then flatten the results into a single stream. Since the stream
2550-- generation function is monadic, unlike 'concatMap', it can produce an
2551-- effect at the beginning of each iteration of the inner loop.
2554{-# INLINE concatMapM #-}
2555concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
2556concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
2558-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
2559-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
2560-- therefore provide many times better performance.
2563{-# INLINE concatUnfold #-}
2564concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
2565concatUnfold u m = fromStreamD $ D.concatMapU u (toStreamD m)
2567-- | Like 'concatUnfold' but interleaves the streams in the same way as
2568-- 'interleave' behaves instead of appending them.
2571{-# INLINE concatUnfoldInterleave #-}
2572concatUnfoldInterleave ::(IsStream t, Monad m)
2573 => Unfold m a b -> t m a -> t m b
2574concatUnfoldInterleave u m =
2575 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
2577-- | Like 'concatUnfold' but executes the streams in the same way as
2581{-# INLINE concatUnfoldRoundrobin #-}
2582concatUnfoldRoundrobin ::(IsStream t, Monad m)
2583 => Unfold m a b -> t m a -> t m b
2584concatUnfoldRoundrobin u m =
2585 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
2587-- XXX we can swap the order of arguments to gintercalate so that the
2588-- definition of concatUnfold becomes simpler? The first stream should be
2589-- infixed inside the second one. However, if we change the order in
2590-- "interleave" as well similarly, then that will make it a bit unintuitive.
2592-- > concatUnfold unf str =
2593-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
2595-- | 'interleaveInfix' followed by unfold and concat.
2598{-# INLINE gintercalate #-}
2600 :: (IsStream t, Monad m)
2601 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
2602gintercalate unf1 str1 unf2 str2 =
2603 D.fromStreamD $ D.gintercalate
2604 unf1 (D.toStreamD str1)
2605 unf2 (D.toStreamD str2)
2607-- XXX The order of arguments in "intercalate" is consistent with the list
2608-- intercalate but inconsistent with gintercalate and other stream interleaving
2609-- combinators. We can change the order of the arguments in other combinators
2610-- but then 'interleave' combinator may become a bit unintuitive because we
2611-- will be starting with the second stream.
2613-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
2614-- > intercalate a unf str = concatUnfold unf $ intersperse a str
2616-- | 'intersperse' followed by unfold and concat.
2618-- > unwords = intercalate " " UF.fromList
2620-- >>> intercalate " " UF.fromList ["abc", "def", "ghi"]
2623{-# INLINE intercalate #-}
2624intercalate :: (IsStream t, Monad m)
2625 => b -> Unfold m b c -> t m b -> t m c
2626intercalate seed unf str = D.fromStreamD $
2627 D.concatMapU unf $ D.intersperse seed (toStreamD str)
2629-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
2631-- | Unfold the elements of a stream, intersperse the given element between the
2632-- unfolded streams and then concat them into a single stream.
2634-- > unwords = S.interpose ' '
2637{-# INLINE interpose #-}
2638interpose :: (IsStream t, Monad m)
2639 => c -> Unfold m b c -> t m b -> t m c
2640interpose x unf str =
2641 D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
2643-- | 'interleaveSuffix' followed by unfold and concat.
2646{-# INLINE gintercalateSuffix #-}
2648 :: (IsStream t, Monad m)
2649 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
2650gintercalateSuffix unf1 str1 unf2 str2 =
2651 D.fromStreamD $ D.gintercalateSuffix
2652 unf1 (D.toStreamD str1)
2653 unf2 (D.toStreamD str2)
2655-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
2656-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
2658-- | 'intersperseSuffix' followed by unfold and concat.
2660-- > unlines = intercalateSuffix "\n" UF.fromList
2662-- >>> intercalate "\n" UF.fromList ["abc", "def", "ghi"]
2663-- > "abc\ndef\nghi\n"
2665{-# INLINE intercalateSuffix #-}
2666intercalateSuffix :: (IsStream t, Monad m)
2667 => b -> Unfold m b c -> t m b -> t m c
2668intercalateSuffix seed unf str = fromStreamD $ D.concatMapU unf
2669 $ D.intersperseSuffix (return seed) (D.toStreamD str)
2671-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
2673-- | Unfold the elements of a stream, append the given element after each
2674-- unfolded stream and then concat them into a single stream.
2676-- > unlines = S.interposeSuffix '\n'
2679{-# INLINE interposeSuffix #-}
2680interposeSuffix :: (IsStream t, Monad m)
2681 => c -> Unfold m b c -> t m b -> t m c
2682interposeSuffix x unf str =
2683 D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
2685------------------------------------------------------------------------------
2687------------------------------------------------------------------------------
2689-- | Like 'iterateM' but using a stream generator function.
2693{-# INLINE concatMapIterateWith #-}
2696 => (forall c. t m c -> t m c -> t m c)
2700concatMapIterateWith combine f xs = concatMapWith combine go xs
2702 go x = yield x `combine` concatMapWith combine go (f x)
2704-- concatMapIterateLeftsWith
2706-- | Traverse a forest with recursive tree structures whose non-leaf nodes are
2707-- of type @a@ and leaf nodes are of type @b@, flattening all the trees into
2708-- streams and combining the streams into a single stream consisting of both
2709-- leaf and non-leaf nodes.
2711-- 'concatMapTreeWith' is a generalization of 'concatMap', using a recursive
2712-- feedback loop to append the non-leaf nodes back to the input stream enabling
2713-- recursive traversal. 'concatMap' flattens a single level nesting whereas
2714-- 'concatMapTreeWith' flattens a recursively nested structure.
2716-- Traversing a directory tree recursively is a canonical use case of
2717-- 'concatMapTreeWith'.
2720-- concatMapTreeWith combine f xs = concatMapIterateWith combine g xs
2722-- g (Left tree) = f tree
2723-- g (Right leaf) = nil
2728{-# INLINE concatMapTreeWith #-}
2731 => (forall c. t m c -> t m c -> t m c)
2732 -> (a -> t m (Either a b))
2733 -> t m (Either a b) -- Should be t m a?
2735concatMapTreeWith combine f xs = concatMapWith combine go xs
2737 go (Left tree) = yield (Left tree) `combine` concatMapWith combine go (f tree)
2738 go (Right leaf) = yield $ Right leaf
2741-- | Like concatMapTreeWith but produces only stream of leaf elements.
2742-- On an either stream, iterate lefts but yield only rights.
2744-- concatMapEitherYieldRightsWith combine f xs =
2745-- catRights $ concatMapTreeWith combine f xs
2747{-# INLINE concatMapEitherYieldRightsWith #-}
2748concatMapEitherYieldRightsWith :: (IsStream t, MonadAsync m)
2749 => _ -> (a -> t m (Either a b)) -> t m (Either a b) -> t m b
2750concatMapEitherYieldRightsWith combine f xs = undefined
2754{-# INLINE concatUnfoldTree #-}
2755concatUnfoldTree :: (IsStream t, MonadAsync m)
2756 => Unfold m a (Either a b) -> t m (Either a b) -> t m (Either a b)
2757concatUnfoldTree unf xs = undefined
2760------------------------------------------------------------------------------
2762------------------------------------------------------------------------------
2764-- Flatten a stream with a feedback loop back into the input. For example, if
2765-- the output stream may generate an exception the exceptions can be fed back
2766-- to the input to take any corrective action. For example, we may retry the
2767-- action again or do nothing or log the errors. For the retry case we need a
2770-- We can perhaps even implement the SVar using this. The stream we are mapping
2771-- on is the work queue. When evaluated it results in either a leaf element to
2772-- yield or a tail stream to queue back to the work queue.
2774-- | Feedback a component of the output back to the input stream.
2778{-# INLINE concatMapLoopWith #-}
2780 :: (IsStream t, MonadAsync m)
2781 => (forall x. t m x -> t m x -> t m x)
2782 -> (a -> t m (Either b c))
2783 -> (b -> t m a) -- feedback function to feed b back into input
2786concatMapLoopWith combine f fb xs =
2787 concatMapWith combine go $ concatMapWith combine f xs
2789 go (Left b) = concatMapLoopWith combine f fb $ fb b
2790 go (Right c) = yield c
2792-- Concat a stream of trees, generating only leaves.
2796{-# INLINE concatMapTreeYieldLeavesWith #-}
2797concatMapTreeYieldLeavesWith
2798 :: (IsStream t, MonadAsync m)
2799 => (forall x. t m x -> t m x -> t m x)
2800 -> (a -> t m (Either a b))
2803concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield
2805------------------------------------------------------------------------------
2806-- Grouping/Splitting
2807------------------------------------------------------------------------------
2809------------------------------------------------------------------------------
2810-- Grouping without looking at elements
2811------------------------------------------------------------------------------
2813------------------------------------------------------------------------------
2815------------------------------------------------------------------------------
2818-- | @splitAt n f1 f2@ composes folds @f1@ and @f2@ such that first @n@
2819-- elements of its input are consumed by fold @f1@ and the rest of the stream
2820-- is consumed by fold @f2@.
2822-- > let splitAt_ n xs = S.fold (FL.splitAt n FL.toList FL.toList) $ S.fromList xs
2824-- >>> splitAt_ 6 "Hello World!"
2825-- > ("Hello ","World!")
2827-- >>> splitAt_ (-1) [1,2,3]
2830-- >>> splitAt_ 0 [1,2,3]
2833-- >>> splitAt_ 1 [1,2,3]
2836-- >>> splitAt_ 3 [1,2,3]
2839-- >>> splitAt_ 4 [1,2,3]
2844-- This can be considered as a two-fold version of 'ltake' where we take both
2845-- the segments instead of discarding the leftover.
2847{-# INLINE splitAt #-}
2854splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2855 Fold step initial extract
2857 initial = Tuple3' <$> return n <*> initialL <*> initialR
2859 step (Tuple3' i xL xR) input =
2861 then stepL xL input >>= (\a -> return (Tuple3' (i - 1) a xR))
2862 else stepR xR input >>= (\b -> return (Tuple3' i xL b))
2864 extract (Tuple3' _ a b) = (,) <$> extractL a <*> extractR b
2866------------------------------------------------------------------------------
2868------------------------------------------------------------------------------
2870------------------------------------------------------------------------------
2871-- Generalized grouping
2872------------------------------------------------------------------------------
2874-- This combinator is the most general grouping combinator and can be used to
2875-- implement all other grouping combinators.
2877-- XXX check if this can implement the splitOn combinator i.e. we can slide in
2878-- new elements, slide out old elements and incrementally compute the hash.
2879-- Also, can we implement the windowed classification combinators using this?
2881-- In fact this is a parse. Instead of using a special return value in the fold
2882-- we are using a mapping function.
2884-- Note that 'scanl'' (usually followed by a map to extract the desired value
2885-- from the accumulator) can be used to realize many implementations e.g. a
2886-- sliding window implementation. A scan followed by a mapMaybe is also a good
2887-- pattern to express many problems where we want to emit a filtered output and
2888-- not emit an output on every input.
2890-- Passing on of the initial accumulator value to the next fold is equivalent
2891-- to returning the leftover concept.
2894-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
2895-- @splitter@ is applied on the accumulator of the fold every time an item is
2896-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
2897-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
2898-- in the output stream and the initial value of the accumulator for the next
2899-- group's fold. This allows us to control whether to start fresh for the next
2900-- fold or to continue from the previous fold's output.
2902{-# INLINE groupScan #-}
2904 :: (IsStream t, Monad m)
2905 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
2906groupScan split fold m = undefined
2909-- | Group the input stream into groups of @n@ elements each and then fold each
2910-- group using the provided fold function.
2912-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
2915-- This can be considered as an n-fold version of 'ltake' where we apply
2916-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
2919{-# INLINE chunksOf #-}
2921 :: (IsStream t, Monad m)
2922 => Int -> Fold m a b -> t m a -> t m b
2923chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
2925{-# INLINE chunksOf2 #-}
2927 :: (IsStream t, Monad m)
2928 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
2929chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
2931-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
2932-- @n@ elements each.
2934-- Same as the following but may be more efficient:
2936-- > arraysOf n = S.chunksOf n (A.writeN n)
2939{-# INLINE arraysOf #-}
2940arraysOf :: (IsStream t, MonadIO m, Storable a)
2941 => Int -> t m a -> t m (Array a)
2942arraysOf n = chunksOf n (writeNUnsafe n)
2944-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
2945-- XXX add this example after fixing the serial stream rate control
2946-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
2949-- | Group the input stream into windows of @n@ second each and then fold each
2950-- group using the provided fold function.
2953{-# INLINE intervalsOf #-}
2955 :: (IsStream t, MonadAsync m)
2956 => Double -> Fold m a b -> t m a -> t m b
2958 splitWithSuffix isNothing (FL.lcatMaybes f)
2959 (interjectSuffix n (return Nothing) (Serial.map Just xs))
2961------------------------------------------------------------------------------
2962-- Element Aware APIs
2963------------------------------------------------------------------------------
2965------------------------------------------------------------------------------
2967------------------------------------------------------------------------------
2969-- | Break the input stream into two groups, the first group takes the input as
2970-- long as the predicate applied to the first element of the stream and next
2971-- input element holds 'True', the second group takes the rest of the input.
2979spanBy cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
2980 Fold step initial extract
2983 initial = Tuple3' <$> initialL <*> initialR <*> return (Tuple' Nothing True)
2985 step (Tuple3' a b (Tuple' (Just frst) isFirstG)) input =
2986 if cmp frst input && isFirstG
2988 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just frst) isFirstG)))
2990 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2992 step (Tuple3' a b (Tuple' Nothing isFirstG)) input =
2995 >>= (\a' -> return (Tuple3' a' b (Tuple' (Just input) isFirstG)))
2997 >>= (\a' -> return (Tuple3' a a' (Tuple' Nothing False)))
2999 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
3001-- | @span p f1 f2@ composes folds @f1@ and @f2@ such that @f1@ consumes the
3002-- input as long as the predicate @p@ is 'True'. @f2@ consumes the rest of the
3005-- > let span_ p xs = S.fold (S.span p FL.toList FL.toList) $ S.fromList xs
3007-- >>> span_ (< 1) [1,2,3]
3010-- >>> span_ (< 2) [1,2,3]
3013-- >>> span_ (< 4) [1,2,3]
3018-- This can be considered as a two-fold version of 'ltakeWhile' where we take
3019-- both the segments instead of discarding the leftover.
3027span p (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
3028 Fold step initial extract
3032 initial = Tuple3' <$> initialL <*> initialR <*> return True
3034 step (Tuple3' a b isFirstG) input =
3035 if isFirstG && p input
3036 then stepL a input >>= (\a' -> return (Tuple3' a' b True))
3037 else stepR b input >>= (\a' -> return (Tuple3' a a' False))
3039 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
3042-- > break p = span (not . p)
3044-- Break as soon as the predicate becomes 'True'. @break p f1 f2@ composes
3045-- folds @f1@ and @f2@ such that @f1@ stops consuming input as soon as the
3046-- predicate @p@ becomes 'True'. The rest of the input is consumed @f2@.
3048-- This is the binary version of 'splitBy'.
3050-- > let break_ p xs = S.fold (S.break p FL.toList FL.toList) $ S.fromList xs
3052-- >>> break_ (< 1) [3,2,1]
3055-- >>> break_ (< 2) [3,2,1]
3058-- >>> break_ (< 4) [3,2,1]
3069break p = span (not . p)
3071-- | Like 'spanBy' but applies the predicate in a rolling fashion i.e.
3072-- predicate is applied to the previous and the next input elements.
3073{-# INLINE spanByRolling #-}
3080spanByRolling cmp (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
3081 Fold step initial extract
3084 initial = Tuple3' <$> initialL <*> initialR <*> return Nothing
3086 step (Tuple3' a b (Just frst)) input =
3088 then stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
3089 else stepR b input >>= (\b' -> return (Tuple3' a b' (Just input)))
3091 step (Tuple3' a b Nothing) input =
3092 stepL a input >>= (\a' -> return (Tuple3' a' b (Just input)))
3094 extract (Tuple3' a b _) = (,) <$> extractL a <*> extractR b
3096------------------------------------------------------------------------------
3098------------------------------------------------------------------------------
3100-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
3101-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
3102-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
3103-- group and so on. When the comparison fails a new group is started. Each
3104-- group is folded using the fold @f@ and the result of the fold is emitted in
3105-- the output stream.
3107-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
3108-- > [[1,3,7],[0,2,5]]
3111{-# INLINE groupsBy #-}
3113 :: (IsStream t, Monad m)
3118groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
3120-- | Unlike @groupsBy@ this function performs a rolling comparison of two
3121-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
3122-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
3123-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
3124-- 'True' then @c@ is also assigned to the same group and so on. When the
3125-- comparison fails a new group is started. Each group is folded using the fold
3128-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
3129-- > [[1,2,3],[7,8,9]]
3132{-# INLINE groupsByRolling #-}
3134 :: (IsStream t, Monad m)
3139groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
3142-- > groups = groupsBy (==)
3143-- > groups = groupsByRolling (==)
3145-- Groups contiguous spans of equal elements together in individual groups.
3147-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
3151groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
3152groups = groupsBy (==)
3154------------------------------------------------------------------------------
3155-- Binary splitting on a separator
3156------------------------------------------------------------------------------
3159-- | Find the first occurrence of the specified sequence in the input stream
3160-- and break the input stream into two parts, the first part consisting of the
3161-- stream before the sequence and the second part consisting of the sequence
3162-- and the rest of the stream.
3164-- > let breakOn_ pat xs = S.fold (S.breakOn pat FL.toList FL.toList) $ S.fromList xs
3166-- >>> breakOn_ "dear" "Hello dear world!"
3167-- > ("Hello ","dear world!")
3169{-# INLINE breakOn #-}
3170breakOn :: Monad m => Array a -> Fold m a b -> Fold m a c -> Fold m a (b,c)
3171breakOn pat f m = undefined
3174------------------------------------------------------------------------------
3175-- N-ary split on a predicate
3176------------------------------------------------------------------------------
3178-- TODO: Use a Splitter configuration similar to the "split" package to make it
3179-- possible to express all splitting combinations. In general, we can have
3180-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
3181-- separators. We can have a single split operation taking the splitter config
3184-- | Split on an infixed separator element, dropping the separator. Splits the
3185-- stream on separator elements determined by the supplied predicate, separator
3186-- is considered as infixed between two segments, if one side of the separator
3187-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
3188-- applied on the split segments. With '-' representing non-separator elements
3189-- and '.' as separator, 'splitOn' splits as follows:
3192-- "--.--" => "--" "--"
3197-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
3199-- Let's use the following definition for illustration:
3201-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
3203-- >>> splitOn' (== '.') ""
3206-- >>> splitOn' (== '.') "."
3209-- >>> splitOn' (== '.') ".a"
3212-- >>> splitOn' (== '.') "a."
3215-- >>> splitOn' (== '.') "a.b"
3218-- >>> splitOn' (== '.') "a..b"
3223-- This can be considered as an n-fold version of 'breakOn' where we apply
3224-- 'breakOn' successively on the input stream, dropping the first element
3225-- of the second segment after each break.
3227{-# INLINE splitOn #-}
3229 :: (IsStream t, Monad m)
3230 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3231splitOn predicate f m =
3232 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
3234-- | Like 'splitOn' but the separator is considered as suffixed to the segments
3235-- in the stream. A missing suffix at the end is allowed. A separator at the
3236-- beginning is parsed as empty segment. With '-' representing elements and
3237-- '.' as separator, 'splitOnSuffix' splits as follows:
3240-- "--.--." => "--" "--"
3241-- "--.--" => "--" "--"
3245-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
3247-- >>> splitOnSuffix' (== '.') ""
3250-- >>> splitOnSuffix' (== '.') "."
3253-- >>> splitOnSuffix' (== '.') "a"
3256-- >>> splitOnSuffix' (== '.') ".a"
3259-- >>> splitOnSuffix' (== '.') "a."
3262-- >>> splitOnSuffix' (== '.') "a.b"
3265-- >>> splitOnSuffix' (== '.') "a.b."
3268-- >>> splitOnSuffix' (== '.') "a..b.."
3271-- > lines = splitOnSuffix (== '\n')
3275-- This can be considered as an n-fold version of 'breakPost' where we apply
3276-- 'breakPost' successively on the input stream, dropping the first element
3277-- of the second segment after each break.
3279{-# INLINE splitOnSuffix #-}
3281 :: (IsStream t, Monad m)
3282 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3283splitOnSuffix predicate f m =
3284 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
3286-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
3287-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
3288-- @["a","b"]@. In other words, its like parsing words from whitespace
3291-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
3293-- >>> wordsBy' (== ',') ""
3296-- >>> wordsBy' (== ',') ","
3299-- >>> wordsBy' (== ',') ",a,,b,"
3302-- > words = wordsBy isSpace
3306-- It is equivalent to splitting in any of the infix/prefix/suffix styles
3307-- followed by removal of empty segments.
3308{-# INLINE wordsBy #-}
3310 :: (IsStream t, Monad m)
3311 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3312wordsBy predicate f m =
3313 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
3315-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
3318-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
3320-- >>> splitWithSuffix' (== '.') ""
3323-- >>> splitWithSuffix' (== '.') "."
3326-- >>> splitWithSuffix' (== '.') "a"
3329-- >>> splitWithSuffix' (== '.') ".a"
3332-- >>> splitWithSuffix' (== '.') "a."
3335-- >>> splitWithSuffix' (== '.') "a.b"
3338-- >>> splitWithSuffix' (== '.') "a.b."
3341-- >>> splitWithSuffix' (== '.') "a..b.."
3342-- > ["a.",".","b.","."]
3346-- This can be considered as an n-fold version of 'breakPost' where we apply
3347-- 'breakPost' successively on the input stream.
3349{-# INLINE splitWithSuffix #-}
3351 :: (IsStream t, Monad m)
3352 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3353splitWithSuffix predicate f m =
3354 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
3356------------------------------------------------------------------------------
3357-- Split on a delimiter sequence
3358------------------------------------------------------------------------------
3360-- Int list examples for splitOn:
3362-- >>> splitList [] [1,2,3,3,4]
3363-- > [[1],[2],[3],[3],[4]]
3365-- >>> splitList [5] [1,2,3,3,4]
3368-- >>> splitList [1] [1,2,3,3,4]
3371-- >>> splitList [4] [1,2,3,3,4]
3374-- >>> splitList [2] [1,2,3,3,4]
3377-- >>> splitList [3] [1,2,3,3,4]
3380-- >>> splitList [3,3] [1,2,3,3,4]
3383-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
3386-- | Like 'splitOn' but the separator is a sequence of elements instead of a
3389-- For illustration, let's define a function that operates on pure lists:
3392-- splitOnSeq' pat xs = S.toList $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
3395-- >>> splitOnSeq' "" "hello"
3396-- > ["h","e","l","l","o"]
3398-- >>> splitOnSeq' "hello" ""
3401-- >>> splitOnSeq' "hello" "hello"
3404-- >>> splitOnSeq' "x" "hello"
3407-- >>> splitOnSeq' "h" "hello"
3410-- >>> splitOnSeq' "o" "hello"
3413-- >>> splitOnSeq' "e" "hello"
3416-- >>> splitOnSeq' "l" "hello"
3419-- >>> splitOnSeq' "ll" "hello"
3422-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
3424-- > intercalate . splitOn == id
3426-- The following law holds when the separator is non-empty and contains none of
3427-- the elements present in the input lists:
3429-- > splitOn . intercalate == id
3433-- XXX We can use a polymorphic vector implemented by Array# to represent the
3434-- sequence, that way we can avoid the Storable constraint. If we still need
3435-- Storable Array for performance, we can use a separate splitOnArray API for
3436-- that. We can also have an API where the sequence itself is a lazy stream, so
3437-- that we can search files in files for example.
3438{-# INLINE splitOnSeq #-}
3440 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3441 => Array a -> Fold m a b -> t m a -> t m b
3442splitOnSeq patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
3445-- This can be implemented easily using Rabin Karp
3446-- | Split on any one of the given patterns.
3447{-# INLINE splitOnAny #-}
3449 :: (IsStream t, Monad m, Storable a, Integral a)
3450 => [Array a] -> Fold m a b -> t m a -> t m b
3451splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
3454-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
3455-- of a predicate for a single element.
3457-- > splitSuffixOn_ pat xs = S.toList $ S.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
3459-- >>> splitSuffixOn_ "." ""
3462-- >>> splitSuffixOn_ "." "."
3465-- >>> splitSuffixOn_ "." "a"
3468-- >>> splitSuffixOn_ "." ".a"
3471-- >>> splitSuffixOn_ "." "a."
3474-- >>> splitSuffixOn_ "." "a.b"
3477-- >>> splitSuffixOn_ "." "a.b."
3480-- >>> splitSuffixOn_ "." "a..b.."
3483-- > lines = splitSuffixOn "\n"
3486{-# INLINE splitOnSuffixSeq #-}
3488 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3489 => Array a -> Fold m a b -> t m a -> t m b
3490splitOnSuffixSeq patt f m =
3491 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
3494-- | Like 'splitOn' but drops any empty splits.
3496{-# INLINE wordsOn #-}
3498 :: (IsStream t, Monad m, Storable a, Eq a)
3499 => Array a -> Fold m a b -> t m a -> t m b
3500wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
3503-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
3505-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
3507-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3509-- >>> splitOn'_ "" "hello"
3510-- > ["h","","e","","l","","l","","o"]
3512-- >>> splitOn'_ "hello" ""
3515-- >>> splitOn'_ "hello" "hello"
3518-- >>> splitOn'_ "x" "hello"
3521-- >>> splitOn'_ "h" "hello"
3524-- >>> splitOn'_ "o" "hello"
3527-- >>> splitOn'_ "e" "hello"
3530-- >>> splitOn'_ "l" "hello"
3531-- > ["he","l","","l","o"]
3533-- >>> splitOn'_ "ll" "hello"
3537{-# INLINE splitBySeq #-}
3539 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
3540 => Array a -> Fold m a b -> t m a -> t m b
3541splitBySeq patt f m =
3542 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
3544-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
3546-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
3548-- >>> splitSuffixOn'_ "." ""
3551-- >>> splitSuffixOn'_ "." "."
3554-- >>> splitSuffixOn'_ "." "a"
3557-- >>> splitSuffixOn'_ "." ".a"
3560-- >>> splitSuffixOn'_ "." "a."
3563-- >>> splitSuffixOn'_ "." "a.b"
3566-- >>> splitSuffixOn'_ "." "a.b."
3569-- >>> splitSuffixOn'_ "." "a..b.."
3570-- > ["a.",".","b.","."]
3573{-# INLINE splitWithSuffixSeq #-}
3575 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3576 => Array a -> Fold m a b -> t m a -> t m b
3577splitWithSuffixSeq patt f m =
3578 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
3581-- This can be implemented easily using Rabin Karp
3582-- | Split post any one of the given patterns.
3583{-# INLINE splitSuffixOnAny #-}
3585 :: (IsStream t, Monad m, Storable a, Integral a)
3586 => [Array a] -> Fold m a b -> t m a -> t m b
3587splitSuffixOnAny subseq f m = undefined
3588 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
3591------------------------------------------------------------------------------
3593------------------------------------------------------------------------------
3595-- | Consider a chunked stream of container elements e.g. a stream of @Word8@
3596-- chunked as a stream of arrays of @Word8@. @splitInnerBy splitter joiner
3597-- stream@ splits the inner containers @f a@ using the @splitter@ function and
3598-- joins back the resulting fragments from splitting across multiple containers
3599-- using the @joiner@ function such that the transformed output stream is
3600-- consolidated as one container per segment of the split.
3602-- CAUTION! This is not a true streaming function as the container size after
3603-- the split and merge may not be bounded.
3606{-# INLINE splitInnerBy #-}
3608 :: (IsStream t, Monad m)
3609 => (f a -> m (f a, Maybe (f a))) -- splitter
3610 -> (f a -> f a -> m (f a)) -- joiner
3613splitInnerBy splitter joiner xs =
3614 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
3616-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
3620{-# INLINE splitInnerBySuffix #-}
3622 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
3623 => (f a -> m (f a, Maybe (f a))) -- splitter
3624 -> (f a -> f a -> m (f a)) -- joiner
3627splitInnerBySuffix splitter joiner xs =
3628 D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
3630------------------------------------------------------------------------------
3631-- Reorder in sequence
3632------------------------------------------------------------------------------
3635-- Buffer until the next element in sequence arrives. The function argument
3636-- determines the difference in sequence numbers. This could be useful in
3637-- implementing sequenced streams, for example, TCP reassembly.
3638{-# INLINE reassembleBy #-}
3640 :: (IsStream t, Monad m)
3645reassembleBy = undefined
3648------------------------------------------------------------------------------
3650------------------------------------------------------------------------------
3652-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
3653-- add a tap to log the contents flowing through the stream. The fold is used
3654-- only for effects, its result is discarded.
3659-- -----stream m a ---------------stream m a-----
3664-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
3669-- Compare with 'trace'.
3673tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
3674tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
3676-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
3677-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
3678-- means start at the first element in the stream. If the offset is outside
3679-- this range then @offset `mod` n@ is used as offset.
3682-- >>> S.drain $ S.tapOffsetEvery 0 2 (FL.mapM print FL.toList) $ S.enumerateFromTo 0 10
3688{-# INLINE tapOffsetEvery #-}
3689tapOffsetEvery :: (IsStream t, Monad m)
3690 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
3691tapOffsetEvery offset n f xs =
3692 D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
3694-- | Redirect a copy of the stream to a supplied fold and run it concurrently
3695-- in an independent thread. The fold may buffer some elements. The buffer size
3696-- is determined by the prevailing 'maxBuffer' setting.
3701-- -----stream m a ---------------stream m a-----
3706-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
3711-- Exceptions from the concurrently running fold are propagated to the current
3712-- computation. Note that, because of buffering in the fold, exceptions may be
3713-- delayed and may not correspond to the current element being processed in the
3714-- parent stream, but we guarantee that before the parent stream stops the tap
3715-- finishes and all exceptions from it are drained.
3718-- Compare with 'tap'.
3721{-# INLINE tapAsync #-}
3722tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
3723tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)
3725-- | Maintain the count of elements flowing in the stream and poll the count
3726-- asynchronously from another thread. The count stream is transformed using
3727-- the supplied transform and then folded using the supplied fold. The thread
3728-- is automatically cleaned up if the stream stops or aborts due to exception.
3730-- For example, to print the count of elements processed every second:
3733-- > S.drain $ S.pollCounts (rollingMap (-) . delayPost 1) (FL.drainBy print)
3734-- $ S.enumerateFrom 0
3737{-# INLINE pollCounts #-}
3739 (IsStream t, MonadAsync m)
3740 => (t m Int -> t m Int)
3744pollCounts transf f xs =
3746 $ D.pollCounts (D.toStreamD . transf . D.fromStreamD) f
3749-- | Calls the supplied function with the number of elements consumed
3750-- every @n@ seconds. The given function is run in a separate thread
3751-- until the end of the stream. In case there is an exception in the
3752-- stream the thread is killed during the next major GC.
3754-- Note: The action is not guaranteed to run if the main thread exits.
3757-- > delay n = threadDelay (round $ n * 1000000) >> return n
3758-- > S.drain $ S.tapRate 2 (\\n -> print $ show n ++ " elements processed") (delay 1 S.|: delay 0.5 S.|: delay 0.5 S.|: S.nil)
3759-- 2 elements processed
3760-- 1 elements processed
3763{-# INLINE tapRate #-}
3765 (IsStream t, MonadAsync m, MonadCatch m)
3770tapRate n f xs = D.fromStreamD $ D.tapRate n f $ (D.toStreamD xs)
3772-- | Apply a monadic function to each element flowing through the stream and
3773-- discard the results.
3776-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
3781-- Compare with 'tap'.
3785trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
3786trace f = mapM (\x -> void (f x) >> return x)
3788------------------------------------------------------------------------------
3789-- Windowed classification
3790------------------------------------------------------------------------------
3792-- We divide the stream into windows or chunks in space or time and each window
3793-- can be associated with a key, all events associated with a particular key in
3794-- the window can be folded to a single result. The stream can be split into
3795-- windows by size or by using a split predicate on the elements in the stream.
3796-- For example, when we receive a closing flag, we can close the window.
3798-- A "chunk" is a space window and a "session" is a time window. Are there any
3799-- other better short words to describe them. An alternative is to use
3800-- "swindow" and "twindow". Another word for "session" could be "spell".
3802-- TODO: To mark the position in space or time we can have Indexed or
3803-- TimeStamped types. That can make it easy to deal with the position indices
3806------------------------------------------------------------------------------
3807-- Keyed Sliding Windows
3808------------------------------------------------------------------------------
3811{-# INLINABLE classifySlidingChunks #-}
3812classifySlidingChunks
3813 :: (IsStream t, MonadAsync m, Ord k)
3814 => Int -- ^ window size
3815 -> Int -- ^ window slide
3816 -> Fold m a b -- ^ Fold to be applied to window events
3817 -> t m (k, a, Bool) -- ^ window key, data, close event
3819classifySlidingChunks wsize wslide (Fold step initial extract) str
3822-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
3823-- slide the send window when an ack is received and we slide the receive
3824-- window when a sequence is complete. Sliding is stateful in case of TCP,
3825-- sliding releases the send buffer or makes data available to the user from
3826-- the receive buffer.
3827{-# INLINABLE classifySlidingSessions #-}
3828classifySlidingSessions
3829 :: (IsStream t, MonadAsync m, Ord k)
3830 => Double -- ^ timer tick in seconds
3831 -> Double -- ^ time window size
3832 -> Double -- ^ window slide
3833 -> Fold m a b -- ^ Fold to be applied to window events
3834 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
3836classifySlidingSessions tick interval slide (Fold step initial extract) str
3840------------------------------------------------------------------------------
3841-- Sliding Window Buffers
3842------------------------------------------------------------------------------
3844-- These buffered versions could be faster than concurrent incremental folds of
3845-- all overlapping windows as in many cases we may not need all the values to
3846-- compute the fold, we can just compute the result using the old value and new
3847-- value. However, we may need the buffer once in a while, for example for
3848-- string search we usually compute the hash incrementally but when the hash
3849-- matches the hash of the pattern we need to compare the whole string.
3851-- XXX we should be able to implement sequence based splitting combinators
3852-- using this combinator.
3855-- | Buffer n elements of the input in a ring buffer. When t new elements are
3856-- collected, slide the window to remove the same number of oldest elements,
3857-- insert the new elements, and apply an incremental fold on the sliding
3858-- window, supplying the outgoing elements, the new ring buffer as arguments.
3860 :: (IsStream t, Monad m, Ord a, Storable a)
3861 => Int -- window size
3862 -> Int -- window slide
3863 -> Fold m (Ring a, Array a) b
3866slidingChunkBuffer = undefined
3868-- Buffer n seconds worth of stream elements of the input in a radix tree.
3869-- Every t seconds, remove the items that are older than n seconds, and apply
3870-- an incremental fold on the sliding window, supplying the outgoing elements,
3871-- and the new radix tree buffer as arguments.
3873 :: (IsStream t, Monad m, Ord a, Storable a)
3874 => Int -- window size
3876 -> Fold m (RTree a, Array a) b
3879slidingSessionBuffer = undefined
3882------------------------------------------------------------------------------
3883-- Keyed Session Windows
3884------------------------------------------------------------------------------
3887-- | Keyed variable size space windows. Close the window if we do not receive a
3888-- window event in the next "spaceout" elements.
3889{-# INLINABLE classifyChunksBy #-}
3891 :: (IsStream t, MonadAsync m, Ord k)
3892 => Int -- ^ window spaceout (spread)
3893 -> Bool -- ^ reset the spaceout when a chunk window element is received
3894 -> Fold m a b -- ^ Fold to be applied to chunk window elements
3895 -> t m (k, a, Bool) -- ^ chunk key, data, last element
3897classifyChunksBy spanout reset (Fold step initial extract) str = undefined
3899-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
3900-- received within the chunk size window. The chunk gets closed only if no
3901-- element is received within the chunk window.
3903{-# INLINABLE classifyKeepAliveChunks #-}
3904classifyKeepAliveChunks
3905 :: (IsStream t, MonadAsync m, Ord k)
3906 => Int -- ^ window spaceout (spread)
3907 -> Fold m a b -- ^ Fold to be applied to chunk window elements
3908 -> t m (k, a, Bool) -- ^ chunk key, data, last element
3910classifyKeepAliveChunks spanout = classifyChunksBy spanout True
3913-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
3914-- stream elements that belong to the same session. @timeout@ is the maximum
3915-- lifetime of a session in seconds. All elements belonging to a session are
3916-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
3917-- after every event received in the session. Session duration is measured
3918-- using the timestamp of the first element seen for that session. To detect
3919-- session timeouts, a monotonic event time clock is maintained using the
3920-- timestamps seen in the inputs and a timer with a tick duration specified by
3923-- @session key@ is a key that uniquely identifies the session for the given
3924-- element, @timestamp@ characterizes the time when the input element was
3925-- generated, this is an absolute time measured from some @Epoch@. @session
3926-- close@ is a boolean indicating whether this element marks the closing of the
3927-- session. When an input element with @session close@ set to @True@ is seen
3928-- the session is purged immediately.
3930-- All the input elements belonging to a session are collected using the fold
3931-- @f@. The session key and the fold result are emitted in the output stream
3932-- when the session is purged either via the session close event or via the
3933-- session lifetime timeout.
3936{-# INLINABLE classifySessionsBy #-}
3938 :: (IsStream t, MonadAsync m, Ord k)
3939 => Double -- ^ timer tick in seconds
3940 -> Double -- ^ session timeout
3941 -> Bool -- ^ reset the timeout when an event is received
3942 -> Fold m a b -- ^ Fold to be applied to session events
3943 -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
3945classifySessionsBy tick timeout reset (Fold step initial extract) str =
3946 concatMap (\(Tuple4' _ _ _ s) -> s) $ scanlM' sstep szero stream
3950 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
3951 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
3952 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty K.nil
3954 -- Got a new stream input element
3955 sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
3956 -- XXX we should use a heap in pinned memory to scale it to a large
3959 -- deleting a key from the heap is expensive, so we never delete a
3960 -- key, we just purge it from the Map and it gets purged from the
3961 -- heap on timeout. We just need an extra lookup in the Map when
3962 -- the key is purged from the heap, that should not be expensive.
3964 -- To detect session inactivity we keep a timestamp of the latest event
3965 -- in the Map along with the fold result. When we purge the session
3966 -- from the heap we match the timestamp in the heap with the timestamp
3967 -- in the Map, if the latest timestamp is newer and has not expired we
3968 -- reinsert the key in the heap.
3970 -- XXX if the key is an Int, we can also use an IntMap for slightly
3971 -- better performance.
3973 let accumulate v = do
3974 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
3976 return $ Tuple' ts new
3979 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
3980 Tuple' _ acc <- accumulate r
3982 return $ Tuple4' evTime hp mp' (yield (key, res))
3984 let r = Map.lookup key mp
3986 let mp' = Map.insert key acc mp
3990 let expiry = addToAbsTime ts timeoutMs
3991 in H.insert (Entry expiry key) hp
3993 -- Event time is maintained as monotonically increasing
3994 -- time. If we have lagged behind any of the timestamps
3995 -- seen then we increase it to match the latest time seen
3996 -- in the timestamps. We also increase it on timer ticks.
3997 return $ Tuple4' (max evTime ts) hp' mp' K.nil
3999 -- Got a timer tick event
4000 -- XXX can we yield the entries without accumulating them?
4001 sstep (Tuple4' evTime heap sessions _) Nothing = do
4002 (hp', mp', out) <- go heap sessions K.nil
4003 return $ Tuple4' curTime hp' mp' out
4007 curTime = addToAbsTime evTime tickMs
4009 let hres = H.uncons hp
4011 Just (Entry ts key, hp') -> do
4012 let duration = diffAbsTime curTime ts
4013 if duration >= timeoutMs
4015 let (r, mp') = Map.updateLookupWithKey
4016 (\_ _ -> Nothing) key mp
4018 Nothing -> go hp' mp' out
4019 Just (Tuple' latestTS acc) -> do
4020 let dur = diffAbsTime curTime latestTS
4021 if dur >= timeoutMs || not reset
4024 go hp' mp' ((key, sess) `K.cons` out)
4026 -- reset the session timeout
4027 let expiry = addToAbsTime latestTS timeoutMs
4028 hp'' = H.insert (Entry expiry key) hp'
4029 mp'' = Map.insert key (Tuple' latestTS acc) mp'
4031 else return (hp, mp, out)
4032 Nothing -> return (hp, mp, out)
4034 -- merge timer events in the stream
4035 stream = Serial.map Just str `Par.parallel` repeatM timer
4037 liftIO $ threadDelay (round $ tick * 1000000)
4040-- | Like 'classifySessionsOf' but the session is kept alive if an event is
4041-- received within the session window. The session times out and gets closed
4042-- only if no event is received within the specified session window size.
4045{-# INLINABLE classifyKeepAliveSessions #-}
4046classifyKeepAliveSessions
4047 :: (IsStream t, MonadAsync m, Ord k)
4048 => Double -- ^ session inactive timeout
4049 -> Fold m a b -- ^ Fold to be applied to session payload data
4050 -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
4052classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
4054------------------------------------------------------------------------------
4055-- Keyed tumbling windows
4056------------------------------------------------------------------------------
4058-- Tumbling windows is a special case of sliding windows where the window slide
4059-- is the same as the window size. Or it can be a special case of session
4060-- windows where the reset flag is set to False.
4062-- XXX instead of using the early termination flag in the stream, we can use an
4063-- early terminating fold instead.
4066-- | Split the stream into fixed size chunks of specified size. Within each
4067-- such chunk fold the elements in buckets identified by the keys. A particular
4068-- bucket fold can be terminated early if a closing flag is encountered in an
4069-- element for that key.
4072{-# INLINABLE classifyChunksOf #-}
4074 :: (IsStream t, MonadAsync m, Ord k)
4075 => Int -- ^ window size
4076 -> Fold m a b -- ^ Fold to be applied to window events
4077 -> t m (k, a, Bool) -- ^ window key, data, close event
4079classifyChunksOf wsize = classifyChunksBy wsize False
4082-- | Split the stream into fixed size time windows of specified interval in
4083-- seconds. Within each such window, fold the elements in buckets identified by
4084-- the keys. A particular bucket fold can be terminated early if a closing flag
4085-- is encountered in an element for that key. Once a fold is terminated the key
4086-- and value for that bucket are emitted in the output stream.
4088-- Session @timestamp@ in the input stream is an absolute time from some epoch,
4089-- characterizing the time when the input element was generated. To detect
4090-- session window end, a monotonic event time clock is maintained synced with
4091-- the timestamps with a clock resolution of 1 second.
4094{-# INLINABLE classifySessionsOf #-}
4096 :: (IsStream t, MonadAsync m, Ord k)
4097 => Double -- ^ time window size
4098 -> Fold m a b -- ^ Fold to be applied to window events
4099 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
4101classifySessionsOf interval = classifySessionsBy 1 interval False
4103------------------------------------------------------------------------------
4105------------------------------------------------------------------------------
4107-- | Run a side effect before the stream yields its first element.
4110{-# INLINE before #-}
4111before :: (IsStream t, Monad m) => m b -> t m a -> t m a
4112before action xs = D.fromStreamD $ D.before action $ D.toStreamD xs
4114-- | Run a side effect whenever the stream stops normally.
4116-- Prefer 'afterIO' over this as the @after@ action in this combinator is not
4117-- executed if the unfold is partially evaluated lazily and then garbage
4122after :: (IsStream t, Monad m) => m b -> t m a -> t m a
4123after action xs = D.fromStreamD $ D.after action $ D.toStreamD xs
4125-- | Run a side effect whenever the stream stops normally
4126-- or is garbage collected after a partial lazy evaluation.
4130{-# INLINE afterIO #-}
4131afterIO :: (IsStream t, MonadIO m, MonadBaseControl IO m) => m b -> t m a -> t m a
4132afterIO action xs = D.fromStreamD $ D.afterIO action $ D.toStreamD xs
4134-- | Run a side effect whenever the stream aborts due to an exception.
4137{-# INLINE onException #-}
4138onException :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
4139onException action xs = D.fromStreamD $ D.onException action $ D.toStreamD xs
4141-- | Run a side effect whenever the stream stops normally or aborts due to an
4144-- Prefer 'finallyIO' over this as the @after@ action in this combinator is not
4145-- executed if the unfold is partially evaluated lazily and then garbage
4149{-# INLINE finally #-}
4150finally :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
4151finally action xs = D.fromStreamD $ D.finally action $ D.toStreamD xs
4153-- | Run a side effect whenever the stream stops normally, aborts due to an
4154-- exception or if it is garbage collected after a partial lazy evaluation.
4158{-# INLINE finallyIO #-}
4159finallyIO :: (IsStream t, MonadAsync m, MonadCatch m) => m b -> t m a -> t m a
4160finallyIO action xs = D.fromStreamD $ D.finallyIO action $ D.toStreamD xs
4162-- | Run the first action before the stream starts and remember its output,
4163-- generate a stream using the output, run the second action using the
4164-- remembered value as an argument whenever the stream ends normally or due to
4167-- Prefer 'bracketIO' over this as the @after@ action in this combinator is not
4168-- executed if the unfold is partially evaluated lazily and then garbage
4172{-# INLINE bracket #-}
4173bracket :: (IsStream t, MonadCatch m)
4174 => m b -> (b -> m c) -> (b -> t m a) -> t m a
4175bracket bef aft bet = D.fromStreamD $
4176 D.bracket bef aft (\x -> toStreamD $ bet x)
4178-- | Run the first action before the stream starts and remember its output,
4179-- generate a stream using the output, run the second action using the
4180-- remembered value as an argument whenever the stream ends normally, due to
4181-- an exception or if it is garbage collected after a partial lazy evaluation.
4185{-# INLINE bracketIO #-}
4186bracketIO :: (IsStream t, MonadAsync m, MonadCatch m)
4187 => m b -> (b -> m c) -> (b -> t m a) -> t m a
4188bracketIO bef aft bet = D.fromStreamD $
4189 D.bracketIO bef aft (\x -> toStreamD $ bet x)
4191-- | When evaluating a stream if an exception occurs, stream evaluation aborts
4192-- and the specified exception handler is run with the exception as argument.
4195{-# INLINE handle #-}
4196handle :: (IsStream t, MonadCatch m, Exception e)
4197 => (e -> t m a) -> t m a -> t m a
4199 D.fromStreamD $ D.handle (\e -> D.toStreamD $ handler e) $ D.toStreamD xs
4201------------------------------------------------------------------------------
4202-- Generalize the underlying monad
4203------------------------------------------------------------------------------
4205-- | Transform the inner monad of a stream using a natural transformation.
4210hoist :: (Monad m, Monad n)
4211 => (forall x. m x -> n x) -> SerialT m a -> SerialT n a
4212hoist f xs = fromStreamS $ S.hoist f (toStreamS xs)
4214-- | Generalize the inner monad of the stream from 'Identity' to any monad.
4218{-# INLINE generally #-}
4219generally :: (IsStream t, Monad m) => t Identity a -> t m a
4220generally xs = fromStreamS $ S.hoist (return . runIdentity) (toStreamS xs)
4222------------------------------------------------------------------------------
4223-- Add and remove a monad transformer
4224------------------------------------------------------------------------------
4226-- | Lift the inner monad of a stream using a monad transformer.
4230{-# INLINE liftInner #-}
4231liftInner :: (Monad m, IsStream t, MonadTrans tr, Monad (tr m))
4232 => t m a -> t (tr m) a
4233liftInner xs = fromStreamD $ D.liftInner (toStreamD xs)
4235-- | Evaluate the inner monad of a stream as 'ReaderT'.
4239{-# INLINE runReaderT #-}
4240runReaderT :: (IsStream t, Monad m) => s -> t (ReaderT s m) a -> t m a
4241runReaderT s xs = fromStreamD $ D.runReaderT s (toStreamD xs)
4243-- | Evaluate the inner monad of a stream as 'StateT'.
4245-- This is supported only for 'SerialT' as concurrent state updation may not be
4250{-# INLINE evalStateT #-}
4251evalStateT :: Monad m => s -> SerialT (StateT s m) a -> SerialT m a
4252evalStateT s xs = fromStreamD $ D.evalStateT s (toStreamD xs)
4254-- | Run a stateful (StateT) stream transformation using a given state.
4256-- This is supported only for 'SerialT' as concurrent state updation may not be
4261{-# INLINE usingStateT #-}
4265 -> (SerialT (StateT s m) a -> SerialT (StateT s m) a)
4268usingStateT s f xs = evalStateT s $ f $ liftInner xs
4270-- | Evaluate the inner monad of a stream as 'StateT' and emit the resulting
4271-- state and value pair after each step.
4273-- This is supported only for 'SerialT' as concurrent state updation may not be
4278{-# INLINE runStateT #-}
4279runStateT :: Monad m => s -> SerialT (StateT s m) a -> SerialT m (s, a)
4280runStateT s xs = fromStreamD $ D.runStateT s (toStreamD xs)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list