1{-# OPTIONS_GHC -Wno-orphans #-}
6-- Module : Streamly.Internal.Data.Stream.IsStream
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
14-- This is an Internal module consisting of released, unreleased and
15-- unimplemented APIs. For stable and released APIs please see
16-- "Streamly.Prelude" module. This module provides documentation only for the
17-- unreleased and unimplemented APIs. For documentation on released APIs please
18-- see "Streamly.Prelude" module.
20module Streamly.Internal.Data.Stream.IsStream
29 -- ** Speculative Streams
33 -- ** Asynchronous Streams
48 -- * Stream Type Adapters
87 -- ** Time Enumeration
108 -- ** Cyclic Elements
111 -- ** From Containers
138 -- ** Composable Left Folds
144 -- ** Concurrent Folds
151 -- -- ** To Summary (Full Folds)
159 -- -- ** To Summary (Maybe) (Full Folds)
167 -- -- ** To Containers (Full Folds)
173 -- ** Composable Left Folds
175 , toStream -- XXX rename to write?
176 , toStreamRev -- XXX rename to writeRev?
180 -- -- ** To Elements (Partial Folds)
184 -- -- | Folds that extract selected elements of a stream or their properties.
194 -- -- ** To Boolean (Partial Folds)
203 -- ** Multi-Stream folds
208 -- finding subsequences
214 -- trimming sequences
254 -- ** Scan Using Fold
258 -- XXX Once we have pipes the contravariant transformations can be
259 -- represented by attaching pipes before a transformation.
271 -- ** Concurrent Transformation
283 -- ** Deleting Elements
286 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
287 -- , uniqOn -- to remove duplicate sequences
288 -- , pruneBy -- dropAround + uniqBy - like words
290 -- ** Inserting Elements
299 , intersperseSuffixBySpan
300 -- , intersperseBySpan
301 -- , intersperseByIndices -- using an index function/stream
303 -- time domain intersperse
304 -- , intersperseByTime
305 -- , intersperseByEvent
316 -- , timestampedR -- timer
354 -- -- *** Searching Elements
358 -- -- *** Searching Sequences
359 -- , seqIndices -- search a sequence in the stream
361 -- -- *** Searching Multiple Sequences
362 -- , seqIndicesAny -- search any of the given sequence in the stream
364 -- -- -- ** Searching Streams
365 -- -- | Finding a stream within another stream.
368 -- | Streams can be sliced into segments in space or in time. We use the
369 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
370 -- and the term @session@ to refer to a length in time (time window).
372 -- -- *** Using Element Separators
380 , wordsBy -- stripAndCompactBy
382 -- -- *** Splitting By Sequences
385 -- , splitOnPrefixSeq
387 -- Keeping the delimiters
390 -- , splitByPrefixSeq
393 -- Splitting using multiple sequence separators
395 -- , splitOnAnySuffixSeq
396 -- , splitOnAnyPrefixSeq
398 -- -- *** Splitting By Streams
399 -- -- | Splitting a stream using another stream as separator.
406 -- In imperative terms, grouped folding can be considered as a nested loop
407 -- where we loop over the stream to group elements and then loop over
408 -- individual groups to fold them to a single value that is yielded in the
421 -- * Windowed Classification
423 -- | Split the stream into windows or chunks in space or time. Each window
424 -- can be associated with a key, all events associated with a particular
425 -- key in the window can be folded to a single result. The stream is split
426 -- into windows of specified size, the window can be terminated early if
427 -- the closing flag is specified in the input stream.
429 -- The term "chunk" is used for a space window and the term "session" is
430 -- used for a time window.
432 -- ** Tumbling Windows
433 -- | A new window starts after the previous window is finished.
435 -- , classifyChunksOf
439 -- ** Keep Alive Windows
440 -- | The window size is extended if an event arrives within the specified
441 -- window size. This can represent sessions with idle or inactive timeout.
443 -- , classifyKeepAliveChunks
444 , classifyKeepAliveSessions
447 -- ** Sliding Windows
448 -- | A new window starts after the specified slide from the previous
449 -- window. Therefore windows can overlap.
450 , classifySlidingChunks
451 , classifySlidingSessions
453 -- ** Sliding Window Buffers
454 -- , slidingChunkBuffer
455 -- , slidingSessionBuffer
457 -- * Combining Streams
497 -- ** Flattening a Container of Streams
499 , concatMapFoldableWith
500 , concatForFoldableWith
502 -- ** Flattening a Stream of Streams
511 -- ** Flattening Using Unfolds
513 , concatUnfoldInterleave
514 , concatUnfoldRoundrobin
516 -- ** Flattening a Tree of Streams
519 -- ** Flattening a Graph of Streams
522 -- ** Inserting Streams in Streams
543 -- * Generalize Inner Monad
547 -- * Transform Inner Monad
563 , iterateMapLeftsWith
565 -- * Concurrency Control
595import Control.Concurrent (threadDelay)
596import Control.Exception (Exception, assert)
597import Control.Monad (void)
598import Control.Monad.Catch (MonadCatch, MonadThrow)
599import Control.Monad.IO.Class (MonadIO(..))
600import Control.Monad.Trans.Reader (ReaderT)
601import Control.Monad.Trans.State.Strict (StateT)
602import Control.Monad.Trans.Class (MonadTrans(..))
603import Control.Monad.Trans.Control (MonadBaseControl)
604import Data.Either (isLeft, isRight)
605import Data.Functor.Identity (Identity (..))
606import Data.Kind (Type)
607import Data.Heap (Entry(..))
608import Data.Maybe (isJust, fromJust, isNothing)
609import Data.Void (Void)
610import Foreign.Storable (Storable)
611import Streamly.Internal.BaseCompat (fromLeft, fromRight)
612import Streamly.Internal.Data.Stream.Enumeration
613 (Enumerable(..), enumerate, enumerateTo)
614import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))
615import Streamly.Internal.Data.Parser (Parser (..))
616import Streamly.Internal.Data.Unfold.Types (Unfold)
617import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array, writeNUnsafe)
618import Streamly.Internal.Data.SVar (MonadAsync, defState, Rate (..))
619import Streamly.Internal.Data.Stream.Ahead (AheadT, Ahead, ahead, aheadly)
620import Streamly.Internal.Data.Stream.Async
621 ( AsyncT, Async, WAsyncT, WAsync, mkAsync, async, asyncly, wAsync
623import Streamly.Internal.Data.Stream.Combinators
624 ( inspectMode, maxBuffer, maxThreads, maxYields, rate, avgRate, minRate
625 , maxRate, constRate)
626import Streamly.Internal.Data.Stream.Parallel
627 ( ParallelT, Parallel, parallel, parallely)
628import Streamly.Internal.Data.Stream.Prelude
629 (fromStreamS, toStreamS, concatFoldableWith, concatMapFoldableWith
630 , concatForFoldableWith)
631import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
632import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM), adapt)
633import Streamly.Internal.Data.Stream.Serial
634 ( SerialT, WSerialT, Serial, WSerial, serial, wSerial, serially
636import Streamly.Internal.Data.Stream.Zip
637 ( ZipSerialM, ZipSerial, ZipAsyncM, ZipAsync, zipSerially, zipAsyncly)
638import Streamly.Internal.Data.Pipe.Types (Pipe (..))
639import Streamly.Internal.Data.Time.Units
640 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
641 , toAbsTime, TimeUnit64, RelTime64, addToAbsTime64)
642import Streamly.Internal.Data.IORef.Prim (Prim, IORef)
643import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
645import qualified Streamly.Internal.Data.Array.Storable.Foreign as A
646import qualified Streamly.Data.Fold as FL
647import qualified Streamly.Internal.Data.Fold.Types as FL
648import qualified Streamly.Internal.Data.Stream.Prelude as P
649import qualified Streamly.Internal.Data.Stream.StreamK as K
650import qualified Streamly.Internal.Data.Stream.StreamD as D
651import qualified Streamly.Internal.Data.Stream.Serial as Serial
652import qualified Streamly.Internal.Data.Stream.Parallel as Par
653import qualified Streamly.Internal.Data.Stream.Zip as Z
654import qualified Streamly.Internal.Data.Parser.ParserK.Types as PRK
655import qualified Streamly.Internal.Data.Parser.ParserD as PRD
656import qualified Data.Heap as H
657import qualified Data.Map.Strict as Map
658import qualified Prelude
659import qualified System.IO as IO
660#ifdef USE_STREAMK_ONLY
661import qualified Streamly.Internal.Data.Stream.StreamK as S
663import qualified Streamly.Internal.Data.Stream.StreamD as S
667 ( filter, drop, dropWhile, take, takeWhile, zipWith, foldr
668 , foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem
669 , notElem, maximum, minimum, head, last, tail, length, null
670 , reverse, iterate, init, and, or, lookup, foldr1, (!!)
671 , scanl, scanl1, replicate, concatMap, span, splitAt, break
672 , repeat, concat, mconcat)
674------------------------------------------------------------------------------
676------------------------------------------------------------------------------
678-- | Decompose a stream into its head and tail. If the stream is empty, returns
679-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
680-- the head of the stream and @ma@ its tail.
682-- This is a brute force primitive. Avoid using it as long as possible, use it
683-- when no other combinator can do the job. This can be used to do pretty much
684-- anything in an imperative manner, as it just breaks down the stream into
685-- individual elements and we can loop over them as we deem fit. For example,
686-- this can be used to convert a streamly stream into other stream types.
690uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
691uncons m = K.uncons (K.adapt m)
693------------------------------------------------------------------------------
694-- Generation by Unfolding
695------------------------------------------------------------------------------
702-- Just (a, b) -> a \`cons` unfoldr step b
705-- Build a stream by unfolding a /pure/ step function @step@ starting from a
706-- seed @s@. The step function returns the next element in the stream and the
707-- next seed value. When it is done it returns 'Nothing' and the stream ends.
714-- else Just (b, b + 1)
715-- in toList $ unfoldr f 0
722{-# INLINE_EARLY unfoldr #-}
723unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
724unfoldr step seed = fromStreamS (S.unfoldr step seed)
725{-# RULES "unfoldr fallback to StreamK" [1]
726 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
728-- | Build a stream by unfolding a /monadic/ step function starting from a
729-- seed. The step function returns the next element in the stream and the next
730-- seed value. When it is done it returns 'Nothing' and the stream ends. For
736-- then return Nothing
737-- else print b >> return (Just (b, b + 1))
738-- in drain $ unfoldrM f 0
746-- When run concurrently, the next unfold step can run concurrently with the
747-- processing of the output of the previous step. Note that more than one step
748-- cannot run concurrently as the next step depends on the output of the
752-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
753-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
759{-# INLINE_EARLY unfoldrM #-}
760unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
763{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
764{-# INLINE_EARLY unfoldrMSerial #-}
765unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
766unfoldrMSerial = Serial.unfoldrM
768{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}
769{-# INLINE_EARLY unfoldrMWSerial #-}
770unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a
771unfoldrMWSerial = Serial.unfoldrM
773{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}
774{-# INLINE_EARLY unfoldrMZipSerial #-}
775unfoldrMZipSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a
776unfoldrMZipSerial = Serial.unfoldrM
778-- | Convert an 'Unfold' into a stream by supplying it an input seed.
780-- >>> unfold (UF.replicateM 10) (putStrLn "hello")
784unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
785unfold unf x = fromStreamD $ D.unfold unf x
787-- | Convert an 'Unfold' with a closed input end into a stream.
790{-# INLINE unfold0 #-}
791unfold0 :: (IsStream t, Monad m) => Unfold m Void b -> t m b
792unfold0 unf = unfold unf (error "unfold0: unexpected void evaluation")
794------------------------------------------------------------------------------
795-- Specialized Generation
796------------------------------------------------------------------------------
798-- Faster than yieldM because there is no bind.
802-- yield a = a \`cons` nil
805-- Create a singleton stream from a pure value.
807-- The following holds in monadic streams, but not in Zip streams:
811-- yield = yieldM . pure
814-- In Zip applicative streams 'yield' is not the same as 'pure' because in that
815-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are
816-- equally efficient, in other cases 'yield' may be slightly more efficient
817-- than the other equivalent definitions.
821yield :: IsStream t => a -> t m a
826-- yieldM m = m \`consM` nil
829-- Create a singleton stream from a monadic action.
832-- > toList $ yieldM getLine
839yieldM :: (Monad m, IsStream t) => m a -> t m a
844-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
847-- Generate an infinite stream, whose values are the output of a function @f@
848-- applied on the corresponding index. Index starts at 0.
851-- > S.toList $ S.take 5 $ S.fromIndices id
856{-# INLINE fromIndices #-}
857fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
858fromIndices = fromStreamS . S.fromIndices
863-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
866-- Generate an infinite stream, whose values are the output of a monadic
867-- function @f@ applied on the corresponding index. Index starts at 0.
872{-# INLINE_EARLY fromIndicesM #-}
873fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
874fromIndicesM = K.fromIndicesM
876{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
877{-# INLINE fromIndicesMSerial #-}
878fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
879fromIndicesMSerial = fromStreamS . S.fromIndicesM
883-- replicateM = take n . repeatM
886-- Generate a stream by performing a monadic action @n@ times. Same as:
889-- drain $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
890-- drain $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
896{-# INLINE_EARLY replicateM #-}
897replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
898replicateM = K.replicateM
900{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
901{-# INLINE replicateMSerial #-}
902replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
903replicateMSerial n = fromStreamS . S.replicateM n
907-- replicate = take n . repeat
910-- Generate a stream of length @n@ by repeating a value @n@ times.
913{-# INLINE_NORMAL replicate #-}
914replicate :: (IsStream t, Monad m) => Int -> a -> t m a
915replicate n = fromStreamS . S.replicate n
918-- Generate an infinite stream by repeating a pure value.
921{-# INLINE_NORMAL repeat #-}
922repeat :: (IsStream t, Monad m) => a -> t m a
923repeat = fromStreamS . S.repeat
927-- repeatM = fix . consM
928-- repeatM = cycle1 . yieldM
931-- Generate a stream by repeatedly executing a monadic action forever.
934-- drain $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
935-- drain $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
938-- /Concurrent, infinite (do not use with 'parallely')/
941{-# INLINE_EARLY repeatM #-}
942repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
945{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
946{-# INLINE repeatMSerial #-}
947repeatMSerial :: MonadAsync m => m a -> SerialT m a
948repeatMSerial = fromStreamS . S.repeatM
952-- iterate f x = x \`cons` iterate f x
955-- Generate an infinite stream with @x@ as the first element and each
956-- successive element derived by applying the function @f@ on the previous
960-- > S.toList $ S.take 5 $ S.iterate (+1) 1
965{-# INLINE_NORMAL iterate #-}
966iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a
967iterate step = fromStreamS . S.iterate step
971-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)
974-- Generate an infinite stream with the first element generated by the action
975-- @m@ and each successive element derived by applying the monadic function
976-- @f@ on the previous element.
978-- When run concurrently, the next iteration can run concurrently with the
979-- processing of the previous iteration. Note that more than one iteration
980-- cannot run concurrently as the next iteration depends on the output of the
981-- previous iteration.
984-- drain $ serially $ S.take 10 $ S.iterateM
985-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
987-- drain $ asyncly $ S.take 10 $ S.iterateM
988-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
995-- /Since: 0.7.0 (signature change)/
996{-# INLINE_EARLY iterateM #-}
997iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a
1000{-# RULES "iterateM serial" iterateM = iterateMSerial #-}
1001{-# INLINE iterateMSerial #-}
1002iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a
1003iterateMSerial step = fromStreamS . S.iterateM step
1005------------------------------------------------------------------------------
1007------------------------------------------------------------------------------
1011-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
1014-- Construct a stream from a list of monadic actions. This is more efficient
1015-- than 'fromFoldableM' for serial streams.
1018{-# INLINE_EARLY fromListM #-}
1019fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
1020fromListM = fromStreamD . D.fromListM
1021{-# RULES "fromListM fallback to StreamK" [1]
1022 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
1026-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
1029-- Construct a stream from a 'Foldable' containing monadic actions.
1032-- drain $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
1033-- drain $ asyncly $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
1036-- /Concurrent (do not use with 'parallely' on infinite containers)/
1039{-# INLINE fromFoldableM #-}
1040fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
1041fromFoldableM = Prelude.foldr consM K.nil
1043-- | Same as 'fromFoldable'.
1046{-# DEPRECATED each "Please use fromFoldable instead." #-}
1048each :: (IsStream t, Foldable f) => f a -> t m a
1049each = K.fromFoldable
1051-- | Read lines from an IO Handle into a stream of Strings.
1054{-# DEPRECATED fromHandle
1055 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1056fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
1059 go = K.mkStream $ \_ yld _ stp -> do
1060 eof <- liftIO $ IO.hIsEOF h
1064 str <- liftIO $ IO.hGetLine h
1067-- | Construct a stream by reading a 'Prim' 'IORef' repeatedly.
1071{-# INLINE fromPrimIORef #-}
1072fromPrimIORef :: (IsStream t, MonadIO m, Prim a) => IORef a -> t m a
1073fromPrimIORef = fromStreamD . D.fromPrimIORef
1075-- | Takes a callback setter function and provides it with a callback. The
1076-- callback when invoked adds a value at the tail of the stream. Returns a
1077-- stream of values generated by the callback.
1081{-# INLINE fromCallback #-}
1082fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
1083fromCallback setCallback = concatM $ do
1084 (callback, stream) <- D.newCallbackStream
1085 setCallback callback
1088------------------------------------------------------------------------------
1090------------------------------------------------------------------------------
1092-- | @timesWith g@ returns a stream of time value tuples. The first component
1093-- of the tuple is an absolute time reference (epoch) denoting the start of the
1094-- stream and the second component is a time relative to the reference.
1096-- The argument @g@ specifies the granularity of the relative time in seconds.
1097-- A lower granularity clock gives higher precision but is more expensive in
1098-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
1101-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.timesWith 0.01
1102-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))
1103-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))
1104-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))
1107-- Note: This API is not safe on 32-bit machines.
1111{-# INLINE timesWith #-}
1112timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
1113timesWith g = fromStreamD $ D.times g
1115-- | @times@ returns a stream of time value tuples with clock of 10 ms
1116-- granularity. The first component of the tuple is an absolute time reference
1117-- (epoch) denoting the start of the stream and the second component is a time
1118-- relative to the reference.
1121-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.times
1122-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))
1123-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))
1124-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))
1127-- Note: This API is not safe on 32-bit machines.
1132times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
1133times = timesWith 0.01
1135-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
1136-- granularity @g@ specified in seconds. A low granularity clock is more
1137-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
1141-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimesWith 0.01
1144-- Note: This API is not safe on 32-bit machines.
1148{-# INLINE absTimesWith #-}
1149absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
1150 => Double -> t m AbsTime
1151absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
1153-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
1157-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimes
1160-- Note: This API is not safe on 32-bit machines.
1164{-# INLINE absTimes #-}
1165absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
1166absTimes = fmap (uncurry addToAbsTime64) times
1168{-# DEPRECATED currentTime "Please use absTimes instead" #-}
1169{-# INLINE currentTime #-}
1170currentTime :: (IsStream t, MonadAsync m, Functor (t m))
1171 => Double -> t m AbsTime
1172currentTime = absTimesWith
1174-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
1175-- using a clock of granularity @g@ specified in seconds. A low granularity
1176-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
1177-- ms is treated as 1 ms.
1180-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimesWith 0.01
1181-- > RelTime64 (NanoSecond64 0)
1182-- > RelTime64 (NanoSecond64 91139000)
1183-- > RelTime64 (NanoSecond64 204052000)
1186-- Note: This API is not safe on 32-bit machines.
1190{-# INLINE relTimesWith #-}
1191relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
1192 => Double -> t m RelTime64
1193relTimesWith = fmap snd . timesWith
1195-- | @relTimes@ returns a stream of relative time values starting from 0,
1196-- using a clock of granularity 10 ms.
1199-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimes
1202-- Note: This API is not safe on 32-bit machines.
1206{-# INLINE relTimes #-}
1207relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
1208relTimes = fmap snd times
1210-- | @durations g@ returns a stream of relative time values measuring the time
1211-- elapsed since the immediate predecessor element of the stream was generated.
1212-- The first element of the stream is always 0. @durations@ uses a clock of
1213-- granularity @g@ specified in seconds. A low granularity clock is more
1214-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
1215-- Durations lower than 1 ms will be 0.
1217-- Note: This API is not safe on 32-bit machines.
1221{-# INLINE durations #-}
1222durations :: -- (IsStream t, MonadAsync m) =>
1223 Double -> t m RelTime64
1224durations = undefined
1226-- | Generate ticks at the specified rate. The rate is adaptive, the tick
1227-- generation speed can be increased or decreased at different times to achieve
1228-- the specified rate. The specific behavior for different styles of 'Rate'
1229-- specifications is documented under 'Rate'. The effective maximum rate
1230-- achieved by a stream is governed by the processor speed.
1235ticks :: -- (IsStream t, MonadAsync m) =>
1239-- | Generate a singleton event at or after the specified absolute time. Note
1240-- that this is different from a threadDelay, a threadDelay starts from the
1241-- time when the action is evaluated, whereas if we use AbsTime based timeout
1242-- it will immediately expire if the action is evaluated too late.
1246{-# INLINE timeout #-}
1247timeout :: -- (IsStream t, MonadAsync m) =>
1251------------------------------------------------------------------------------
1252-- Elimination by Folding
1253------------------------------------------------------------------------------
1255-- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
1256-- an output structure using the step function @build@. @build@ is invoked with
1257-- the next input element and the remaining (lazy) tail of the output
1258-- structure. It builds a lazy output expression using the two. When the "tail
1259-- structure" in the output expression is evaluated it calls @build@ again thus
1260-- lazily consuming the input @stream@ until either the output expression built
1261-- by @build@ is free of the "tail" or the input is exhausted in which case
1262-- @final@ is used as the terminating case for the output structure. For more
1263-- details see the description in the previous section.
1265-- Example, determine if any element is 'odd' in a stream:
1267-- >>> S.foldrM (\x xs -> if odd x then return True else xs) (return False) $ S.fromList (2:4:5:undefined)
1270-- /Since: 0.7.0 (signature changed)/
1272-- /Since: 0.2.0 (signature changed)/
1275{-# INLINE foldrM #-}
1276foldrM :: Monad m => (a -> m b -> m b) -> m b -> SerialT m a -> m b
1279-- | Right fold to a streaming monad.
1281-- > foldrS S.cons S.nil === id
1283-- 'foldrS' can be used to perform stateless stream to stream transformations
1284-- like map and filter in general. It can be coupled with a scan to perform
1285-- stateful transformations. However, note that the custom map and filter
1286-- routines can be much more efficient than this due to better stream fusion.
1288-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
1291-- Find if any element in the stream is 'True':
1293-- >>> 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)
1296-- Map (+2) on odd elements and filter out the even elements:
1298-- >>> 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)
1301-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
1302-- is much more efficient:
1304-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1307{-# INLINE foldrS #-}
1308foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
1311-- | Right fold to a transformer monad. This is the most general right fold
1312-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
1313-- implementation can be more efficient:
1316-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1318-- 'foldrT' can be used to translate streamly streams to other transformer
1319-- monads e.g. to a different streaming type.
1322{-# INLINE foldrT #-}
1323foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
1324 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
1325foldrT f z s = S.foldrT f z (toStreamS s)
1327-- | Right fold, lazy for lazy monads and pure streams, and strict for strict
1330-- Please avoid using this routine in strict monads like IO unless you need a
1331-- strict right fold. This is provided only for use in lazy monads (e.g.
1332-- Identity) or pure streams. Note that with this signature it is not possible
1333-- to implement a lazy foldr when the monad @m@ is strict. In that case it
1334-- would be strict in its accumulator and therefore would necessarily consume
1339foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
1342-- XXX This seems to be of limited use as it cannot be used to construct
1343-- recursive structures and for reduction foldl1' is better.
1345-- | Lazy right fold for non-empty streams, using first element as the starting
1346-- value. Returns 'Nothing' if the stream is empty.
1349{-# INLINE foldr1 #-}
1350{-# DEPRECATED foldr1 "Use foldrM instead." #-}
1351foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1352foldr1 f m = S.foldr1 f (toStreamS m)
1354-- | Strict left fold with an extraction function. Like the standard strict
1355-- left fold, but applies a user supplied extraction function (the third
1356-- argument) to the folded value at the end. This is designed to work with the
1357-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
1360{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
1362foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
1365-- | Left associative/strict push fold. @foldl' reduce initial stream@ invokes
1366-- @reduce@ with the accumulator and the next input in the input stream, using
1367-- @initial@ as the initial value of the current value of the accumulator. When
1368-- the input is exhausted the current value of the accumulator is returned.
1369-- Make sure to use a strict data structure for accumulator to not build
1370-- unnecessary lazy expressions unless that's what you want. See the previous
1371-- section for more details.
1374{-# INLINE foldl' #-}
1375foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
1378-- | Strict left fold, for non-empty streams, using first element as the
1379-- starting value. Returns 'Nothing' if the stream is empty.
1382{-# INLINE foldl1' #-}
1383foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1387 Nothing -> return Nothing
1389 res <- foldl' step h t
1392-- | Like 'foldx', but with a monadic step function.
1395{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
1396{-# INLINE foldxM #-}
1397foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
1400-- | Like 'foldl'' but with a monadic step function.
1404-- /Since: 0.8.0 (signature change)/
1405{-# INLINE foldlM' #-}
1406foldlM' :: Monad m => (b -> a -> m b) -> m b -> SerialT m a -> m b
1407foldlM' step begin m = S.foldlM' step begin $ toStreamS m
1409------------------------------------------------------------------------------
1411------------------------------------------------------------------------------
1413-- | Fold a stream using the supplied left fold.
1415-- >>> S.fold FL.sum (S.enumerateFromTo 1 100)
1420fold :: Monad m => Fold m a b -> SerialT m a -> m b
1423------------------------------------------------------------------------------
1425------------------------------------------------------------------------------
1428-- | Drain a stream to a 'Sink'.
1429{-# INLINE runSink #-}
1430runSink :: Monad m => Sink m a -> SerialT m a -> m ()
1431runSink = fold . toFold
1434------------------------------------------------------------------------------
1436------------------------------------------------------------------------------
1438-- | Parse a stream using the supplied 'Parser'.
1442{-# INLINE_NORMAL parseD #-}
1443parseD :: MonadThrow m => PRD.Parser m a b -> SerialT m a -> m b
1444parseD (PRD.Parser step initial extract) = P.parselMx' step initial extract
1446{-# INLINE parseK #-}
1447parseK :: MonadThrow m => PRK.Parser m a b -> SerialT m a -> m b
1450-- | Parse a stream using the supplied 'Parser'.
1454{-# INLINE [3] parse #-}
1455parse :: MonadThrow m => Parser m a b -> SerialT m a -> m b
1456parse = parseD . PRK.fromParserK
1458------------------------------------------------------------------------------
1460------------------------------------------------------------------------------
1463-- > drain = mapM_ (\_ -> return ())
1465-- Run a stream, discarding the results. By default it interprets the stream
1466-- as 'SerialT', to run other types of streams use the type adapting
1467-- combinators for example @drain . 'asyncly'@.
1471drain :: Monad m => SerialT m a -> m ()
1474-- | Run a stream, discarding the results. By default it interprets the stream
1475-- as 'SerialT', to run other types of streams use the type adapting
1476-- combinators for example @runStream . 'asyncly'@.
1479{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
1480{-# INLINE runStream #-}
1481runStream :: Monad m => SerialT m a -> m ()
1485-- > drainN n = drain . take n
1487-- Run maximum up to @n@ iterations of a stream.
1490{-# INLINE drainN #-}
1491drainN :: Monad m => Int -> SerialT m a -> m ()
1492drainN n = drain . take n
1495-- > runN n = runStream . take n
1497-- Run maximum up to @n@ iterations of a stream.
1500{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
1502runN :: Monad m => Int -> SerialT m a -> m ()
1506-- > drainWhile p = drain . takeWhile p
1508-- Run a stream as long as the predicate holds true.
1511{-# INLINE drainWhile #-}
1512drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1513drainWhile p = drain . takeWhile p
1516-- > runWhile p = runStream . takeWhile p
1518-- Run a stream as long as the predicate holds true.
1521{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
1522{-# INLINE runWhile #-}
1523runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1524runWhile = drainWhile
1526-- | Determine whether the stream is empty.
1530null :: Monad m => SerialT m a -> m Bool
1531null = S.null . toStreamS
1533-- | Extract the first element of the stream, if any.
1539head :: Monad m => SerialT m a -> m (Maybe a)
1540head = S.head . toStreamS
1542-- | Extract the first element of the stream, if any, otherwise use the
1543-- supplied default value. It can help avoid one branch in high performance
1547{-# INLINE headElse #-}
1548headElse :: Monad m => a -> SerialT m a -> m a
1549headElse x = D.headElse x . toStreamD
1552-- > tail = fmap (fmap snd) . uncons
1554-- Extract all but the first element of the stream, if any.
1558tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1559tail m = K.tail (K.adapt m)
1561-- | Extract all but the last element of the stream, if any.
1565init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1566init m = K.init (K.adapt m)
1568-- | Extract the last element of the stream, if any.
1570-- > last xs = xs !! (length xs - 1)
1574last :: Monad m => SerialT m a -> m (Maybe a)
1575last m = S.last $ toStreamS m
1577-- | Determine whether an element is present in the stream.
1581elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1582elem e m = S.elem e (toStreamS m)
1584-- | Determine whether an element is not present in the stream.
1587{-# INLINE notElem #-}
1588notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1589notElem e m = S.notElem e (toStreamS m)
1591-- | Determine the length of the stream.
1594{-# INLINE length #-}
1595length :: Monad m => SerialT m a -> m Int
1596length = foldl' (\n _ -> n + 1) 0
1598-- | Determine whether all elements of a stream satisfy a predicate.
1602all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1603all p m = S.all p (toStreamS m)
1605-- | Determine whether any of the elements of a stream satisfy a predicate.
1609any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1610any p m = S.any p (toStreamS m)
1612-- | Determines if all elements of a boolean stream are True.
1616and :: Monad m => SerialT m Bool -> m Bool
1619-- | Determines whether at least one element of a boolean stream is True.
1623or :: Monad m => SerialT m Bool -> m Bool
1626-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
1627-- the stream is empty. Note that this is not numerically stable for floating
1632sum :: (Monad m, Num a) => SerialT m a -> m a
1635-- | Determine the product of all elements of a stream of numbers. Returns @1@
1636-- when the stream is empty.
1639{-# INLINE product #-}
1640product :: (Monad m, Num a) => SerialT m a -> m a
1641product = foldl' (*) 1
1643-- | Fold a stream of monoid elements by appending them.
1646{-# INLINE mconcat #-}
1647mconcat :: (Monad m, Monoid a) => SerialT m a -> m a
1648mconcat = foldr mappend mempty
1652-- minimum = 'minimumBy' compare
1655-- Determine the minimum element in a stream.
1658{-# INLINE minimum #-}
1659minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1660minimum m = S.minimum (toStreamS m)
1662-- | Determine the minimum element in a stream using the supplied comparison
1666{-# INLINE minimumBy #-}
1667minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1668minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1672-- maximum = 'maximumBy' compare
1675-- Determine the maximum element in a stream.
1678{-# INLINE maximum #-}
1679maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1682-- | Determine the maximum element in a stream using the supplied comparison
1686{-# INLINE maximumBy #-}
1687maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1688maximumBy cmp m = S.maximumBy cmp (toStreamS m)
1690-- | Lookup the element at the given index.
1694(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
1695m !! i = toStreamS m S.!! i
1697-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1698-- first pair where the key equals the given value @a@.
1700-- > lookup = snd <$> find ((==) . fst)
1703{-# INLINE lookup #-}
1704lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
1705lookup a m = S.lookup a (toStreamS m)
1707-- | Like 'findM' but with a non-monadic predicate.
1709-- > find p = findM (return . p)
1713find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
1714find p m = S.find p (toStreamS m)
1716-- | Returns the first element that satisfies the given predicate.
1720findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
1721findM p m = S.findM p (toStreamS m)
1723-- | Find all the indices where the element in the stream satisfies the given
1727{-# INLINE findIndices #-}
1728findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
1729findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
1731-- | Returns the first index that satisfies the given predicate.
1734{-# INLINE findIndex #-}
1735findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
1736findIndex p = head . findIndices p
1738-- | Find all the indices where the value of the element in the stream is equal
1739-- to the given value.
1742{-# INLINE elemIndices #-}
1743elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1744elemIndices a = findIndices (==a)
1746-- | Returns the first index where a given value is found in the stream.
1748-- > elemIndex a = findIndex (== a)
1751{-# INLINE elemIndex #-}
1752elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
1753elemIndex a = findIndex (== a)
1755------------------------------------------------------------------------------
1757------------------------------------------------------------------------------
1759-- | Returns 'True' if the first stream is the same as or a prefix of the
1760-- second. A stream is a prefix of itself.
1763-- > S.isPrefixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1768{-# INLINE isPrefixOf #-}
1769isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1770isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
1772-- Note: isPrefixOf uses the prefix stream only once. In contrast, isSuffixOf
1773-- may use the suffix stream many times. To run in optimal memory we do not
1774-- want to buffer the suffix stream in memory therefore we need an ability to
1775-- clone (or consume it multiple times) the suffix stream without any side
1776-- effects so that multiple potential suffix matches can proceed in parallel
1777-- without buffering the suffix stream. For example, we may create the suffix
1778-- stream from a file handle, however, if we evaluate the stream multiple
1779-- times, once for each match, we will need a different file handle each time
1780-- which may exhaust the file descriptors. Instead, we want to share the same
1781-- underlying file descriptor, use pread on it to generate the stream and clone
1782-- the stream for each match. Therefore the suffix stream should be built in
1783-- such a way that it can be consumed multiple times without any problems.
1785-- XXX Can be implemented with better space/time complexity.
1786-- Space: @O(n)@ worst case where @n@ is the length of the suffix.
1788-- | Returns 'True' if the first stream is a suffix of the second. A stream is
1789-- considered a suffix of itself.
1792-- > S.isSuffixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1796-- Space: @O(n)@, buffers entire input stream and the suffix.
1800-- /Suboptimal/ - Help wanted.
1802{-# INLINE isSuffixOf #-}
1803isSuffixOf :: (Monad m, Eq a) => SerialT m a -> SerialT m a -> m Bool
1804isSuffixOf suffix stream = reverse suffix `isPrefixOf` reverse stream
1806-- | Returns 'True' if the first stream is an infix of the second. A stream is
1807-- considered an infix of itself.
1810-- > S.isInfixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1814-- Space: @O(n)@ worst case where @n@ is the length of the infix.
1818-- /Requires 'Storable' constraint/ - Help wanted.
1820{-# INLINE isInfixOf #-}
1821isInfixOf :: (MonadIO m, Eq a, Enum a, Storable a)
1822 => SerialT m a -> SerialT m a -> m Bool
1823isInfixOf infx stream = do
1824 arr <- fold A.write infx
1825 -- XXX can use breakOnSeq instead (when available)
1826 r <- null $ drop 1 $ splitOnSeq arr FL.drain stream
1829-- | Returns 'True' if all the elements of the first stream occur, in order, in
1830-- the second stream. The elements do not have to occur consecutively. A stream
1831-- is a subsequence of itself.
1834-- > S.isSubsequenceOf (S.fromList "hlo") (S.fromList "hello" :: SerialT IO Char)
1839{-# INLINE isSubsequenceOf #-}
1840isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1841isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
1843-- | @stripPrefix prefix stream@ strips @prefix@ from @stream@ if it is a
1844-- prefix of stream. Returns 'Nothing' if the stream does not start with the
1845-- given prefix, stripped stream otherwise. Returns @Just nil@ when the prefix
1846-- is the same as the stream.
1851{-# INLINE stripPrefix #-}
1853 :: (Eq a, IsStream t, Monad m)
1854 => t m a -> t m a -> m (Maybe (t m a))
1855stripPrefix m1 m2 = fmap fromStreamD <$>
1856 D.stripPrefix (toStreamD m1) (toStreamD m2)
1858-- Note: If we want to return a Maybe value to know whether the
1859-- suffix/infix was present or not along with the stripped stream then
1860-- we need to buffer the whole input stream.
1862-- | Drops the given suffix from a stream. Returns 'Nothing' if the stream does
1863-- not end with the given suffix. Returns @Just nil@ when the suffix is the
1864-- same as the stream.
1866-- It may be more efficient to convert the stream to an Array and use
1867-- stripSuffix on that especially if the elements have a Storable or Prim
1870-- Space: @O(n)@, buffers the entire input stream as well as the suffix
1873{-# INLINE stripSuffix #-}
1876 => SerialT m a -> SerialT m a -> m (Maybe (SerialT m a))
1877stripSuffix m1 m2 = fmap reverse <$> stripPrefix (reverse m1) (reverse m2)
1879-- | Drop prefix from the input stream if present.
1883-- /Unimplemented/ - Help wanted.
1884{-# INLINE dropPrefix #-}
1886 -- (Eq a, IsStream t, Monad m) =>
1887 t m a -> t m a -> t m a
1888dropPrefix = error "Not implemented yet!"
1890-- | Drop all matching infix from the input stream if present. Infix stream
1891-- may be consumed multiple times.
1893-- Space: @O(n)@ where n is the length of the infix.
1895-- /Unimplemented/ - Help wanted.
1896{-# INLINE dropInfix #-}
1898 -- (Eq a, IsStream t, Monad m) =>
1899 t m a -> t m a -> t m a
1900dropInfix = error "Not implemented yet!"
1902-- | Drop suffix from the input stream if present. Suffix stream may be
1903-- consumed multiple times.
1905-- Space: @O(n)@ where n is the length of the suffix.
1907-- /Unimplemented/ - Help wanted.
1908{-# INLINE dropSuffix #-}
1910 -- (Eq a, IsStream t, Monad m) =>
1911 t m a -> t m a -> t m a
1912dropSuffix = error "Not implemented yet!"
1914------------------------------------------------------------------------------
1916------------------------------------------------------------------------------
1918-- XXX this can utilize parallel mapping if we implement it as drain . mapM
1920-- > mapM_ = drain . mapM
1922-- Apply a monadic action to each element of the stream and discard the output
1923-- of the action. This is not really a pure transformation operation but a
1924-- transformation followed by fold.
1928mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
1929mapM_ f m = S.mapM_ f $ toStreamS m
1931------------------------------------------------------------------------------
1933------------------------------------------------------------------------------
1937-- toList = S.foldr (:) []
1940-- Convert a stream into a list in the underlying monad. The list can be
1941-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
1942-- IO) the whole list is generated and buffered before it can be consumed.
1944-- /Warning!/ working on large lists accumulated as buffers in memory could be
1945-- very inefficient, consider using "Streamly.Array" instead.
1948{-# INLINE toList #-}
1949toList :: Monad m => SerialT m a -> m [a]
1954-- toListRev = S.foldl' (flip (:)) []
1957-- Convert a stream into a list in reverse order in the underlying monad.
1959-- /Warning!/ working on large lists accumulated as buffers in memory could be
1960-- very inefficient, consider using "Streamly.Array" instead.
1963{-# INLINE toListRev #-}
1964toListRev :: Monad m => SerialT m a -> m [a]
1965toListRev = D.toListRev . toStreamD
1969-- toHandle h = S.mapM_ $ hPutStrLn h
1972-- Write a stream of Strings to an IO Handle.
1975{-# DEPRECATED toHandle
1976 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1977toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
1981 let stop = return ()
1982 single a = liftIO (IO.hPutStrLn h a)
1983 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1984 in K.foldStream defState yieldk single stop m1
1986-- XXX rename these to write/writeRev to make the naming consistent with folds
1989-- | A fold that buffers its input to a pure stream.
1991-- /Warning!/ working on large streams accumulated as buffers in memory could
1992-- be very inefficient, consider using "Streamly.Data.Array" instead.
1995{-# INLINE toStream #-}
1996toStream :: Monad m => Fold m a (SerialT Identity a)
1997toStream = Fold (\f x -> return $ f . (x `K.cons`))
1999 (return . ($ K.nil))
2001-- This is more efficient than 'toStream'. toStream is exactly the same as
2002-- reversing the stream after toStreamRev.
2004-- | Buffers the input stream to a pure stream in the reverse order of the
2007-- /Warning!/ working on large streams accumulated as buffers in memory could
2008-- be very inefficient, consider using "Streamly.Data.Array" instead.
2012-- xn : ... : x2 : x1 : []
2013{-# INLINABLE toStreamRev #-}
2014toStreamRev :: Monad m => Fold m a (SerialT Identity a)
2015toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
2017-- | Convert a stream to a pure stream.
2020-- toPure = foldr cons nil
2025{-# INLINE toPure #-}
2026toPure :: Monad m => SerialT m a -> m (SerialT Identity a)
2027toPure = foldr K.cons K.nil
2029-- | Convert a stream to a pure stream in reverse order.
2032-- toPureRev = foldl' (flip cons) nil
2037{-# INLINE toPureRev #-}
2038toPureRev :: Monad m => SerialT m a -> m (SerialT Identity a)
2039toPureRev = foldl' (flip K.cons) K.nil
2041------------------------------------------------------------------------------
2042-- Concurrent Application
2043------------------------------------------------------------------------------
2051-- | Parallel transform application operator; applies a stream transformation
2052-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
2053-- is evaluated asynchronously in an independent thread yielding elements to a
2054-- buffer and the transformation function runs in another thread consuming the
2055-- input from the buffer. '|$' is just like regular function application
2056-- operator '$' except that it is concurrent.
2058-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
2059-- look at it as a transformation that converts a transform function to a
2060-- buffered concurrent transform function.
2062-- The following code prints a value every second even though each stage adds a
2068-- S.mapM (\\x -> threadDelay 1000000 >> print x)
2069-- |$ S.repeatM (threadDelay 1000000 >> return 1)
2074-- /Since: 0.3.0 ("Streamly")/
2078(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
2079-- (|$) f = f . Async.mkAsync
2080(|$) f = f . D.mkParallel
2086{-# INLINE applyAsync #-}
2087applyAsync :: (IsStream t, MonadAsync m)
2088 => (t m a -> t m b) -> (t m a -> t m b)
2091-- | Parallel reverse function application operator for streams; just like the
2092-- regular reverse function application operator '&' except that it is
2097-- S.repeatM (threadDelay 1000000 >> return 1)
2098-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
2103-- /Since: 0.3.0 ("Streamly")/
2107(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
2110-- | Parallel fold application operator; applies a fold function @t m a -> m b@
2111-- to a stream @t m a@ concurrently; The the input stream is evaluated
2112-- asynchronously in an independent thread yielding elements to a buffer and
2113-- the folding action runs in another thread consuming the input from the
2116-- If you read the signature as @(t m a -> m b) -> (t m a -> m b)@ you can look
2117-- at it as a transformation that converts a fold function to a buffered
2118-- concurrent fold function.
2120-- The @.@ at the end of the operator is a mnemonic for termination of the
2124-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
2125-- |$. S.repeatM (threadDelay 1000000 >> return 1)
2130-- /Since: 0.3.0 ("Streamly")/
2134(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
2135-- (|$.) f = f . Async.mkAsync
2136(|$.) f = f . D.mkParallel
2142{-# INLINE foldAsync #-}
2143foldAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
2146-- | Parallel reverse function application operator for applying a run or fold
2147-- functions to a stream. Just like '|$.' except that the operands are reversed.
2150-- S.repeatM (threadDelay 1000000 >> return 1)
2151-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
2156-- /Since: 0.3.0 ("Streamly")/
2160(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
2163------------------------------------------------------------------------------
2164-- General Transformation
2165------------------------------------------------------------------------------
2167-- | Use a 'Pipe' to transform a stream.
2171{-# INLINE transform #-}
2172transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
2173transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
2175------------------------------------------------------------------------------
2176-- Transformation by Folding (Scans)
2177------------------------------------------------------------------------------
2179-- XXX It may be useful to have a version of scan where we can keep the
2180-- accumulator independent of the value emitted. So that we do not necessarily
2181-- have to keep a value in the accumulator which we are not using. We can pass
2182-- an extraction function that will take the accumulator and the current value
2183-- of the element and emit the next value in the stream. That will also make it
2184-- possible to modify the accumulator after using it. In fact, the step function
2185-- can return new accumulator and the value to be emitted. The signature would
2186-- be more like mapAccumL. Or we can change the signature of scanx to
2189-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
2190-- user supplied extraction function (the third argument) at each step. This is
2191-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
2196-- /Since: 0.7.0 (Monad m constraint)/
2197{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
2199scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
2202-- XXX this needs to be concurrent
2203-- XXX because of the use of D.cons for appending, scanlM' has quadratic
2204-- complexity when iterated over a stream. We should use StreamK style scanlM'
2205-- for linear performance on iteration.
2207-- | Like 'scanl'' but with a monadic step function and a monadic seed.
2211-- /Since: 0.8.0 (signature change)/
2212{-# INLINE scanlM' #-}
2213scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
2214scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
2216-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
2217-- that it provides an additional @done@ function to be applied on the
2218-- accumulator when the stream stops. The result of @done@ is also emitted in
2221-- This function can be used to allocate a resource in the beginning of the
2222-- scan and release it when the stream ends or to flush the internal state of
2223-- the scan at the end.
2227{-# INLINE scanlMAfter' #-}
2228scanlMAfter' :: (IsStream t, Monad m)
2229 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
2230scanlMAfter' step initial done stream =
2231 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
2233-- XXX because of the use of D.cons for appending, scanl' has quadratic
2234-- complexity when iterated over a stream. We should use StreamK style scanlM'
2235-- for linear performance on iteration.
2237-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
2238-- however it adds an extra element.
2241-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
2246-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
2247-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
2250-- The output of 'scanl'' is the initial value of the accumulator followed by
2251-- all the intermediate steps and the final result of 'foldl''.
2253-- By streaming the accumulated state after each fold step, we can share the
2254-- state across multiple stages of stream composition. Each stage can modify or
2255-- extend the state, do some processing with it and emit it for the next stage,
2256-- thus modularizing the stream processing. This can be useful in
2257-- stateful or event-driven programming.
2259-- Consider the following monolithic example, computing the sum and the product
2260-- of the elements in a stream in one go using a @foldl'@:
2263-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
2267-- Using @scanl'@ we can make it modular by computing the sum in the first
2268-- stage and passing it down to the next stage for computing the product:
2271-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
2272-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
2273-- $ S.fromList \[1,2,3,4]
2277-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
2278-- lazy expressions inside the accumulator, it is recommended that a strict
2279-- data structure is used for accumulator.
2281-- See also: 'usingStateT'
2284{-# INLINE scanl' #-}
2285scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
2286scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
2288-- | Like 'scanl'' but does not stream the initial value of the accumulator.
2290-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
2293{-# INLINE postscanl' #-}
2294postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
2295postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
2297-- XXX this needs to be concurrent
2298-- | Like 'postscanl'' but with a monadic step function and a monadic seed.
2302-- /Since: 0.8.0 (signature change)/
2303{-# INLINE postscanlM' #-}
2304postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
2305postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
2307-- XXX prescanl does not sound very useful, enable only if there is a
2308-- compelling use case.
2310-- | Like scanl' but does not stream the final value of the accumulator.
2313{-# INLINE prescanl' #-}
2314prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
2315prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
2317-- XXX this needs to be concurrent
2318-- | Like prescanl' but with a monadic step function and a monadic seed.
2321{-# INLINE prescanlM' #-}
2322prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
2323prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
2325-- XXX this needs to be concurrent
2326-- | Like 'scanl1'' but with a monadic step function.
2329{-# INLINE scanl1M' #-}
2330scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
2331scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
2333-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
2334-- is used as the initial value of the accumulator. Does nothing if the stream
2338-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
2343{-# INLINE scanl1' #-}
2344scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
2345scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
2347------------------------------------------------------------------------------
2348-- Scanning with a Fold
2349------------------------------------------------------------------------------
2351-- | Scan a stream using the given monadic fold.
2355scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
2356scan (Fold step begin done) = P.scanlMx' step begin done
2358-- | Postscan a stream using the given monadic fold.
2361{-# INLINE postscan #-}
2362postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
2363postscan (Fold step begin done) = P.postscanlMx' step begin done
2365------------------------------------------------------------------------------
2366-- Stateful Transformations
2367------------------------------------------------------------------------------
2369-- XXX this is not a one-to-one map so calling it map may not be right.
2370-- We can perhaps call it zipWithTail or rollWith.
2372-- | Apply a function on every two successive elements of a stream. If the
2373-- stream consists of a single element the output is an empty stream.
2375-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
2379{-# INLINE rollingMap #-}
2380rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
2381rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
2383-- | Like 'rollingMap' but with an effectful map function.
2387{-# INLINE rollingMapM #-}
2388rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
2389rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
2391------------------------------------------------------------------------------
2392-- Transformation by Filtering
2393------------------------------------------------------------------------------
2395-- | Include only those elements that pass a predicate.
2398{-# INLINE filter #-}
2399#if __GLASGOW_HASKELL__ != 802
2400-- GHC 8.2.2 crashes with this code, when used with "stack"
2401filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2402filter p m = fromStreamS $ S.filter p $ toStreamS m
2404filter :: IsStream t => (a -> Bool) -> t m a -> t m a
2408-- | Same as 'filter' but with a monadic predicate.
2411{-# INLINE filterM #-}
2412filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2413filterM p m = fromStreamD $ D.filterM p $ toStreamD m
2415-- | Drop repeated elements that are adjacent to each other.
2419uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
2420uniq = fromStreamD . D.uniq . toStreamD
2422-- | Ensures that all the elements of the stream are identical and then returns
2423-- that unique element.
2427the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
2428the m = S.the (toStreamS m)
2430-- | Take first 'n' elements from the stream and discard the rest.
2434take :: (IsStream t, Monad m) => Int -> t m a -> t m a
2435take n m = fromStreamS $ S.take n $ toStreamS
2436 (maxYields (Just (fromIntegral n)) m)
2438-- | End the stream as soon as the predicate fails on an element.
2441{-# INLINE takeWhile #-}
2442takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2443takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
2445-- | Same as 'takeWhile' but with a monadic predicate.
2448{-# INLINE takeWhileM #-}
2449takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2450takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
2452-- | @takeByTime duration@ yields stream elements upto specified time
2453-- @duration@. The duration starts when the stream is evaluated for the first
2454-- time, before the first element is yielded. The time duration is checked
2455-- before generating each element, if the duration has expired the stream
2458-- The total time taken in executing the stream is guaranteed to be /at least/
2459-- @duration@, however, because the duration is checked before generating an
2460-- element, the upper bound is indeterminate and depends on the time taken in
2461-- generating and processing the last element.
2463-- No element is yielded if the duration is zero. At least one element is
2464-- yielded if the duration is non-zero.
2468{-# INLINE takeByTime #-}
2469takeByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
2470takeByTime d = fromStreamD . D.takeByTime d . toStreamD
2472-- | Discard first 'n' elements from the stream and take the rest.
2476drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
2477drop n m = fromStreamS $ S.drop n $ toStreamS m
2479-- | Drop elements in the stream as long as the predicate succeeds and then
2480-- take the rest of the stream.
2483{-# INLINE dropWhile #-}
2484dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2485dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
2487-- | Same as 'dropWhile' but with a monadic predicate.
2490{-# INLINE dropWhileM #-}
2491dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2492dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
2494-- | @dropByTime duration@ drops stream elements until specified @duration@ has
2495-- passed. The duration begins when the stream is evaluated for the first
2496-- time. The time duration is checked /after/ generating a stream element, the
2497-- element is yielded if the duration has expired otherwise it is dropped.
2499-- The time elapsed before starting to generate the first element is /at most/
2500-- @duration@, however, because the duration expiry is checked after the
2501-- element is generated, the lower bound is indeterminate and depends on the
2502-- time taken in generating an element.
2504-- All elements are yielded if the duration is zero.
2508{-# INLINE dropByTime #-}
2509dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
2510dropByTime d = fromStreamD . D.dropByTime d . toStreamD
2512------------------------------------------------------------------------------
2513-- Transformation by Mapping
2514------------------------------------------------------------------------------
2518-- mapM f = sequence . map f
2521-- Apply a monadic function to each element of the stream and replace it with
2522-- the output of the resulting action.
2525-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
2528-- drain $ S.replicateM 10 (return 1)
2529-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
2531-- drain $ S.replicateM 10 (return 1)
2532-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
2535-- /Concurrent (do not use with 'parallely' on infinite streams)/
2538{-# INLINE_EARLY mapM #-}
2539mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
2542{-# RULES "mapM serial" mapM = mapMSerial #-}
2543{-# INLINE mapMSerial #-}
2544mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
2545mapMSerial = Serial.mapM
2549-- sequence = mapM id
2552-- Replace the elements of a stream of monadic actions with the outputs of
2556-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
2559-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
2560-- & (serially . S.sequence)
2562-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
2563-- & (asyncly . S.sequence)
2566-- /Concurrent (do not use with 'parallely' on infinite streams)/
2569{-# INLINE sequence #-}
2570sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
2571sequence m = fromStreamS $ S.sequence (toStreamS m)
2575-- The stateful step function can be simplified to @(s -> a -> m b)@ to provide
2576-- a read-only environment. However, that would just be 'mapM'.
2578-- The initial action could be @m (s, Maybe b)@, and we can also add a final
2579-- action @s -> m (Maybe b)@. This can be used to get pre/post scan like
2580-- functionality and also to flush the state in the end like scanlMAfter'.
2581-- We can also use it along with a fusible version of bracket to get
2582-- scanlMAfter' like functionality. See issue #677.
2584-- This can be further generalized to a type similar to Fold/Parser, giving it
2585-- filtering and parsing capability as well (this is in fact equivalent to
2588-- smapM :: (s -> a -> m (Step s b)) -> m s -> t m a -> t m b
2591-- | A stateful 'mapM', equivalent to a left scan, more like mapAccumL.
2592-- Hopefully, this is a better alternative to @scan@. Separation of state from
2593-- the output makes it easier to think in terms of a shared state, and also
2594-- makes it easier to keep the state fully strict and the output lazy.
2596-- See also: 'scanlM''
2601smapM :: (IsStream t, Monad m) =>
2602 (s -> a -> m (s, b))
2606smapM step initial stream =
2607 -- XXX implement this directly instead of using scanlM'
2608 -- Once we have postscanlM' with monadic initial we can use this code
2609 -- let r = postscanlM'
2610 -- (\(s, _) a -> step s a)
2611 -- (fmap (,undefined) initial)
2614 (\(s, _) a -> step s a)
2615 (fmap (,undefined) initial)
2619------------------------------------------------------------------------------
2621------------------------------------------------------------------------------
2623-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
2624-- elements, and return a stream of values extracted from 'Just'.
2629-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
2633{-# INLINE mapMaybe #-}
2634mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
2635mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
2637-- | Like 'mapMaybe' but maps a monadic function.
2642-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
2645-- /Concurrent (do not use with 'parallely' on infinite streams)/
2648{-# INLINE_EARLY mapMaybeM #-}
2649mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
2650 => (a -> m (Maybe b)) -> t m a -> t m b
2651mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
2653{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
2654{-# INLINE mapMaybeMSerial #-}
2655mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
2656mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
2658-- | In a stream of 'Maybe's, discard 'Nothing's and unwrap 'Just's.
2662catMaybes :: (IsStream t, Monad m, Functor (t m)) => t m (Maybe a) -> t m a
2663catMaybes = fmap fromJust . filter isJust
2665------------------------------------------------------------------------------
2666-- Transformation by Reordering
2667------------------------------------------------------------------------------
2669-- XXX Use a compact region list to temporarily store the list, in both reverse
2670-- as well as in reverse'.
2672-- /Note:/ 'reverse'' is much faster than this, use that when performance
2675-- > reverse = S.foldlT (flip S.cons) S.nil
2677-- | Returns the elements of the stream in reverse order. The stream must be
2678-- finite. Note that this necessarily buffers the entire stream in memory.
2680-- /Since 0.7.0 (Monad m constraint)/
2683{-# INLINE reverse #-}
2684reverse :: (IsStream t, Monad m) => t m a -> t m a
2685reverse s = fromStreamS $ S.reverse $ toStreamS s
2687-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
2690{-# INLINE reverse' #-}
2691reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
2692reverse' s = fromStreamD $ D.reverse' $ toStreamD s
2694------------------------------------------------------------------------------
2695-- Transformation by Inserting
2696------------------------------------------------------------------------------
2698-- intersperseM = intersperseBySpan 1
2700-- | Insert an effect and its output before consuming an element of a stream
2701-- except the first one.
2704-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"
2705-- > h.,e.,l.,l.,o"h,e,l,l,o"
2709{-# INLINE intersperseM #-}
2710intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2711intersperseM m = fromStreamS . S.intersperseM m . toStreamS
2713-- | Insert a side effect before consuming an element of a stream except the
2717-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
2722{-# INLINE intersperseM_ #-}
2723intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2724intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
2726-- | Insert a pure value between successive elements of a stream.
2729-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
2734{-# INLINE intersperse #-}
2735intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2736intersperse a = fromStreamS . S.intersperse a . toStreamS
2738-- | Insert an effect and its output after consuming an element of a stream.
2741-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
2742-- > h.,e.,l.,l.,o.,"h,e,l,l,o,"
2746{-# INLINE intersperseSuffix #-}
2747intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2748intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
2750-- | Insert a side effect after consuming an element of a stream.
2753-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
2758{-# INLINE intersperseSuffix_ #-}
2759intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2760intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
2762-- | Insert a side effect before consuming an element of a stream.
2765-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
2766-- > .h.e.l.l.o"hello"
2769-- See also: 'trace_'
2775{-# INLINE interspersePrefix_ #-}
2776interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
2777interspersePrefix_ m = mapM (\x -> void m >> return x)
2779-- Note: delay must be serial.
2781-- | Introduce a delay of specified seconds before consuming an element of the
2782-- stream except the first one.
2785-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2786-- > (AbsTime (TimeSpec {sec = 2502706, nsec = 751137000}),1)
2787-- > (AbsTime (TimeSpec {sec = 2502707, nsec = 743535000}),2)
2788-- > (AbsTime (TimeSpec {sec = 2502708, nsec = 749758000}),3)
2794delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2795delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
2797-- Note: delay must be serial.
2799-- | Introduce a delay of specified seconds after consuming an element of a
2803-- >>> S.mapM_ print $ S.timestamped $ S.delayPost 1 $ S.enumerateFromTo 1 3
2804-- > (AbsTime (TimeSpec {sec = 2502826, nsec = 119030000}),1)
2805-- > (AbsTime (TimeSpec {sec = 2502827, nsec = 111393000}),2)
2806-- > (AbsTime (TimeSpec {sec = 2502828, nsec = 112221000}),3)
2811{-# INLINE delayPost #-}
2812delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2813delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
2815-- Note: delay must be serial, that's why 'trace_' is used.
2817-- | Introduce a delay of specified seconds before consuming an element of a
2821-- >>> S.mapM_ print $ S.timestamped $ S.delayPre 1 $ S.enumerateFromTo 1 3
2822-- > (AbsTime (TimeSpec {sec = 2502207, nsec = 533177000}),1)
2823-- > (AbsTime (TimeSpec {sec = 2502208, nsec = 530859000}),2)
2824-- > (AbsTime (TimeSpec {sec = 2502209, nsec = 531619000}),3)
2829{-# INLINE delayPre #-}
2830delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2831delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
2833-- | Like 'intersperseSuffix' but intersperses an effectful action into the
2834-- input stream after every @n@ elements and after the last element.
2837-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"
2843{-# INLINE intersperseSuffixBySpan #-}
2844intersperseSuffixBySpan :: (IsStream t, MonadAsync m)
2845 => Int -> m a -> t m a -> t m a
2846intersperseSuffixBySpan n eff =
2847 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
2850-- | Intersperse a monadic action into the input stream after every @n@
2854-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
2859{-# INLINE intersperseBySpan #-}
2860intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
2861intersperseBySpan _n _f _xs = undefined
2864-- | Intersperse a monadic action into the input stream after every @n@
2868-- > S.drain $ S.interjectSuffix 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"
2873{-# INLINE interjectSuffix #-}
2875 :: (IsStream t, MonadAsync m)
2876 => Double -> m a -> t m a -> t m a
2877interjectSuffix n f xs = xs `Par.parallelFst` repeatM timed
2878 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
2880-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
2881-- @stream@ that is less than @elem@ when compared using @cmp@.
2884-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
2888-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
2893{-# INLINE insertBy #-}
2895 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
2896insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
2898------------------------------------------------------------------------------
2900------------------------------------------------------------------------------
2902-- | Deletes the first occurrence of the element in the stream that satisfies
2903-- the given equality predicate.
2906-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
2911{-# INLINE deleteBy #-}
2912deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
2913deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
2915------------------------------------------------------------------------------
2917------------------------------------------------------------------------------
2920-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
2921-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
2923-- Pair each element in a stream with its index, starting from index 0.
2926-- > S.toList $ S.indexed $ S.fromList "hello"
2927-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
2931{-# INLINE indexed #-}
2932indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
2933indexed = fromStreamD . D.indexed . toStreamD
2936-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
2937-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
2939-- Pair each element in a stream with its index, starting from the
2940-- given index @n@ and counting down.
2943-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
2944-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
2948{-# INLINE indexedR #-}
2949indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2950indexedR n = fromStreamD . D.indexedR n . toStreamD
2952-- Note: The timestamp stream must be the second stream in the zip so that the
2953-- timestamp is generated after generating the stream element and not before.
2954-- If we do not do that then the following example will generate the same
2955-- timestamp for first two elements:
2957-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
2959-- | Pair each element in a stream with an absolute timestamp, using a clock of
2960-- specified granularity. The timestamp is generated just before the element
2964-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2969{-# INLINE timestampWith #-}
2970timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
2971 => Double -> t m a -> t m (AbsTime, a)
2972timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
2974-- TBD: check performance vs a custom implementation without using zipWith.
2976-- | Pair each element in a stream with an absolute timestamp, using a clock of
2977-- 10 ms granularity. The timestamp is generated just before the element is
2981-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2982-- (AbsTime (TimeSpec {sec = 2460689, nsec = 641121000}),1)
2983-- (AbsTime (TimeSpec {sec = 2460690, nsec = 639334000}),2)
2984-- (AbsTime (TimeSpec {sec = 2460691, nsec = 644479000}),3)
2989{-# INLINE timestamped #-}
2990timestamped :: (IsStream t, MonadAsync m, Functor (t m))
2991 => t m a -> t m (AbsTime, a)
2992timestamped = timestampWith 0.01
2994-- | Pair each element in a stream with relative times starting from 0, using a
2995-- clock with the specified granularity. The time is measured just before the
2996-- element is consumed.
2999-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
3004{-# INLINE timeIndexWith #-}
3005timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
3006 => Double -> t m a -> t m (RelTime64, a)
3007timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
3009-- | Pair each element in a stream with relative times starting from 0, using a
3010-- 10 ms granularity clock. The time is measured just before the element is
3014-- >>> S.mapM_ print $ S.timeIndexed $ S.delay 1 $ S.enumerateFromTo 1 3
3015-- (RelTime64 (NanoSecond64 0),1)
3016-- (RelTime64 (NanoSecond64 996239000),2)
3017-- (RelTime64 (NanoSecond64 1996712000),3)
3022{-# INLINE timeIndexed #-}
3023timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
3024 => t m a -> t m (RelTime64, a)
3025timeIndexed = timeIndexWith 0.01
3027------------------------------------------------------------------------------
3029------------------------------------------------------------------------------
3031-- | Compare two streams for equality using an equality function.
3034{-# INLINABLE eqBy #-}
3035eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
3038-- | Compare two streams lexicographically using a comparison function.
3041{-# INLINABLE cmpBy #-}
3043 :: (IsStream t, Monad m)
3044 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
3047------------------------------------------------------------------------------
3049------------------------------------------------------------------------------
3051-- | Merge two streams using a comparison function. The head elements of both
3052-- the streams are compared and the smaller of the two elements is emitted, if
3053-- both elements are equal then the element from the first stream is used
3056-- If the streams are sorted in ascending order, the resulting stream would
3057-- also remain sorted in ascending order.
3060-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
3065{-# INLINABLE mergeBy #-}
3067 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
3068mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
3070-- | Like 'mergeBy' but with a monadic comparison function.
3072-- Merge two streams randomly:
3075-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
3076-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
3080-- Merge two streams in a proportion of 2:1:
3083-- proportionately m n = do
3084-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
3085-- return $ \\_ _ -> do
3086-- r <- readIORef ref
3087-- writeIORef ref $ tail r
3091-- f <- proportionately 2 1
3092-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
3096-- [1,1,2,1,1,2,1,1,2]
3100{-# INLINABLE mergeByM #-}
3102 :: (IsStream t, Monad m)
3103 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
3104mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
3107-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
3108{-# INLINABLE mergeEndByAny #-}
3110 :: (IsStream t, Monad m)
3111 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
3112mergeEndByAny f m1 m2 = fromStreamD $
3113 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
3115-- Like 'mergeByM' but stops merging as soon as the first stream stops.
3116{-# INLINABLE mergeEndByFirst #-}
3118 :: (IsStream t, Monad m)
3119 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
3120mergeEndByFirst f m1 m2 = fromStreamS $
3121 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
3124-- Holding this back for now, we may want to use the name "merge" differently
3126-- | Same as @'mergeBy' 'compare'@.
3129-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
3134{-# INLINABLE merge #-}
3136 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
3137merge = mergeBy compare
3140-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
3141-- merged are generated concurrently).
3144{-# INLINE mergeAsyncBy #-}
3145mergeAsyncBy :: (IsStream t, MonadAsync m)
3146 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
3147mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
3149-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
3150-- merged are generated concurrently).
3153{-# INLINE mergeAsyncByM #-}
3154mergeAsyncByM :: (IsStream t, MonadAsync m)
3155 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
3156mergeAsyncByM f m1 m2 = fromStreamD $
3157 D.mergeByM f (D.mkParallelD $ toStreamD m1) (D.mkParallelD $ toStreamD m2)
3159------------------------------------------------------------------------------
3161------------------------------------------------------------------------------
3163-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
3164-- The first argument specifies a merge or concat function that is used to
3165-- merge the streams generated by applying the second argument i.e. the @map@
3166-- function to each element of the input stream. The concat function could be
3167-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
3168-- and the second argument could be any stream generation function using a
3171-- /Compare 'foldMapWith'/
3175-- /Since: 0.8.0 (signature change)/
3176{-# INLINE concatMapWith #-}
3179 => (t m b -> t m b -> t m b)
3183concatMapWith = K.concatMapBy
3185-- | Like 'concatMapWith' but carries a state which can be used to share
3186-- information across multiple steps of concat.
3189-- concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
3194{-# INLINE concatSmapMWith #-}
3196 :: (IsStream t, Monad m)
3197 => (t m b -> t m b -> t m b)
3198 -> (s -> a -> m (s, t m b))
3202concatSmapMWith combine f initial = concatMapWith combine id . smapM f initial
3204-- | Map a stream producing function on each element of the stream and then
3205-- flatten the results into a single stream.
3208-- concatMap f = 'concat . map f'
3209-- concatMap = 'concatMapWith' 'Serial.serial'
3210-- concatMap f = 'concatMapM' (return . f)
3211-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
3215{-# INLINE concatMap #-}
3216concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
3217concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
3219-- | Flatten a stream of streams to a single stream.
3222-- concat = concatMap id
3226{-# INLINE concat #-}
3227concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
3228concat = concatMap id
3230-- | Append the outputs of two streams, yielding all the elements from the
3231-- first stream and then yielding all the elements from the second stream.
3233-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
3234-- few (say 100) streams because it can fuse via stream fusion. However, it
3235-- does not scale for a large number of streams (say 1000s) and becomes
3236-- qudartically slow. Therefore use this for custom appending of a few streams
3237-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
3238-- infinite containers of streams.
3241{-# INLINE append #-}
3242append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3243append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
3245-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
3246-- XXX Document the interleaving behavior of side effects in all the
3247-- interleaving combinators.
3248-- XXX Write time-domain equivalents of these. In the time domain we can
3249-- interleave two streams such that the value of second stream is always taken
3250-- from its last value even if no new value is being yielded, like
3251-- zipWithLatest. It would be something like interleaveWithLatest.
3253-- | Interleaves the outputs of two streams, yielding elements from each stream
3254-- alternately, starting from the first stream. If any of the streams finishes
3255-- early the other stream continues alone until it too finishes.
3257-- >>> :set -XOverloadedStrings
3258-- >>> interleave "ab" ",,,," :: SerialT Identity Char
3260-- >>> interleave "abcd" ",," :: SerialT Identity Char
3263-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
3265-- Do not use at scale in concatMapWith.
3268{-# INLINE interleave #-}
3269interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3270interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
3272-- | Interleaves the outputs of two streams, yielding elements from each stream
3273-- alternately, starting from the first stream. As soon as the first stream
3274-- finishes, the output stops, discarding the remaining part of the second
3275-- stream. In this case, the last element in the resulting stream would be from
3276-- the second stream. If the second stream finishes early then the first stream
3277-- still continues to yield elements until it finishes.
3279-- >>> :set -XOverloadedStrings
3280-- >>> interleaveSuffix "abc" ",,,," :: SerialT Identity Char
3282-- >>> interleaveSuffix "abc" "," :: SerialT Identity Char
3285-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
3287-- Do not use at scale in concatMapWith.
3290{-# INLINE interleaveSuffix #-}
3291interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3292interleaveSuffix m1 m2 =
3293 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
3295-- | Interleaves the outputs of two streams, yielding elements from each stream
3296-- alternately, starting from the first stream and ending at the first stream.
3297-- If the second stream is longer than the first, elements from the second
3298-- stream are infixed with elements from the first stream. If the first stream
3299-- is longer then it continues yielding elements even after the second stream
3302-- >>> :set -XOverloadedStrings
3303-- >>> interleaveInfix "abc" ",,,," :: SerialT Identity Char
3305-- >>> interleaveInfix "abc" "," :: SerialT Identity Char
3308-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
3310-- Do not use at scale in concatMapWith.
3313{-# INLINE interleaveInfix #-}
3314interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3315interleaveInfix m1 m2 =
3316 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
3318-- | Interleaves the outputs of two streams, yielding elements from each stream
3319-- alternately, starting from the first stream. The output stops as soon as any
3320-- of the two streams finishes, discarding the remaining part of the other
3321-- stream. The last element of the resulting stream would be from the longer
3324-- >>> :set -XOverloadedStrings
3325-- >>> interleaveMin "ab" ",,,," :: SerialT Identity Char
3327-- >>> interleaveMin "abcd" ",," :: SerialT Identity Char
3330-- 'interleaveMin' is dual to 'interleave'.
3332-- Do not use at scale in concatMapWith.
3335{-# INLINE interleaveMin #-}
3336interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3337interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
3339-- | Schedule the execution of two streams in a fair round-robin manner,
3340-- executing each stream once, alternately. Execution of a stream may not
3341-- necessarily result in an output, a stream may chose to @Skip@ producing an
3342-- element until later giving the other stream a chance to run. Therefore, this
3343-- combinator fairly interleaves the execution of two streams rather than
3344-- fairly interleaving the output of the two streams. This can be useful in
3345-- co-operative multitasking without using explicit threads. This can be used
3346-- as an alternative to `async`.
3348-- Do not use at scale in concatMapWith.
3351{-# INLINE roundrobin #-}
3352roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3353roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
3355-- | Map a stream producing monadic function on each element of the stream
3356-- and then flatten the results into a single stream. Since the stream
3357-- generation function is monadic, unlike 'concatMap', it can produce an
3358-- effect at the beginning of each iteration of the inner loop.
3361{-# INLINE concatMapM #-}
3362concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
3363concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
3365-- | Given a stream value in the underlying monad, lift and join the underlying
3366-- monad with the stream monad.
3369-- concatM = concat . yieldM
3370-- concatM = concat . lift -- requires @(MonadTrans t)@
3371-- concatM = join . lift -- requires @(MonadTrans t@, @Monad (t m))@
3374-- See also: 'concat', 'sequence'
3378{-# INLINE concatM #-}
3379concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
3380concatM generator = concatMapM (\() -> generator) (yield ())
3382-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
3383-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
3384-- therefore provide many times better performance.
3387{-# INLINE concatUnfold #-}
3388concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
3389concatUnfold u m = fromStreamD $ D.concatMapU u (toStreamD m)
3391-- | Like 'concatUnfold' but interleaves the streams in the same way as
3392-- 'interleave' behaves instead of appending them.
3395{-# INLINE concatUnfoldInterleave #-}
3396concatUnfoldInterleave ::(IsStream t, Monad m)
3397 => Unfold m a b -> t m a -> t m b
3398concatUnfoldInterleave u m =
3399 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
3401-- | Like 'concatUnfold' but executes the streams in the same way as
3405{-# INLINE concatUnfoldRoundrobin #-}
3406concatUnfoldRoundrobin ::(IsStream t, Monad m)
3407 => Unfold m a b -> t m a -> t m b
3408concatUnfoldRoundrobin u m =
3409 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
3411-- XXX we can swap the order of arguments to gintercalate so that the
3412-- definition of concatUnfold becomes simpler? The first stream should be
3413-- infixed inside the second one. However, if we change the order in
3414-- "interleave" as well similarly, then that will make it a bit unintuitive.
3416-- > concatUnfold unf str =
3417-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
3419-- | 'interleaveInfix' followed by unfold and concat.
3422{-# INLINE gintercalate #-}
3424 :: (IsStream t, Monad m)
3425 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
3426gintercalate unf1 str1 unf2 str2 =
3427 D.fromStreamD $ D.gintercalate
3428 unf1 (D.toStreamD str1)
3429 unf2 (D.toStreamD str2)
3431-- XXX The order of arguments in "intercalate" is consistent with the list
3432-- intercalate but inconsistent with gintercalate and other stream interleaving
3433-- combinators. We can change the order of the arguments in other combinators
3434-- but then 'interleave' combinator may become a bit unintuitive because we
3435-- will be starting with the second stream.
3437-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
3438-- > intercalate a unf str = concatUnfold unf $ intersperse a str
3440-- | 'intersperse' followed by unfold and concat.
3442-- > unwords = intercalate " " UF.fromList
3444-- >>> intercalate " " UF.fromList ["abc", "def", "ghi"]
3448{-# INLINE intercalate #-}
3449intercalate :: (IsStream t, Monad m)
3450 => b -> Unfold m b c -> t m b -> t m c
3451intercalate seed unf str = D.fromStreamD $
3452 D.concatMapU unf $ D.intersperse seed (toStreamD str)
3454-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
3456-- | Unfold the elements of a stream, intersperse the given element between the
3457-- unfolded streams and then concat them into a single stream.
3459-- > unwords = S.interpose ' '
3462{-# INLINE interpose #-}
3463interpose :: (IsStream t, Monad m)
3464 => c -> Unfold m b c -> t m b -> t m c
3465interpose x unf str =
3466 D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
3468-- | 'interleaveSuffix' followed by unfold and concat.
3471{-# INLINE gintercalateSuffix #-}
3473 :: (IsStream t, Monad m)
3474 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
3475gintercalateSuffix unf1 str1 unf2 str2 =
3476 D.fromStreamD $ D.gintercalateSuffix
3477 unf1 (D.toStreamD str1)
3478 unf2 (D.toStreamD str2)
3480-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
3481-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
3483-- | 'intersperseSuffix' followed by unfold and concat.
3485-- > unlines = intercalateSuffix "\n" UF.fromList
3487-- >>> intercalate "\n" UF.fromList ["abc", "def", "ghi"]
3488-- > "abc\ndef\nghi\n"
3491{-# INLINE intercalateSuffix #-}
3492intercalateSuffix :: (IsStream t, Monad m)
3493 => b -> Unfold m b c -> t m b -> t m c
3494intercalateSuffix seed unf str = fromStreamD $ D.concatMapU unf
3495 $ D.intersperseSuffix (return seed) (D.toStreamD str)
3497-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
3499-- | Unfold the elements of a stream, append the given element after each
3500-- unfolded stream and then concat them into a single stream.
3502-- > unlines = S.interposeSuffix '\n'
3505{-# INLINE interposeSuffix #-}
3506interposeSuffix :: (IsStream t, Monad m)
3507 => c -> Unfold m b c -> t m b -> t m c
3508interposeSuffix x unf str =
3509 D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
3511------------------------------------------------------------------------------
3513------------------------------------------------------------------------------
3515-- | Like 'iterateM' but iterates after mapping a stream generator on the
3518-- Yield an input element in the output stream, map a stream generator on it
3519-- and then do the same on the resulting stream. This can be used for a depth
3520-- first traversal of a tree like structure.
3522-- Note that 'iterateM' is a special case of 'iterateMapWith':
3525-- iterateM f = iterateMapWith serial (yieldM . f) . yieldM
3528-- It can be used to traverse a tree structure. For example, to list a
3532-- Stream.iterateMapWith Stream.serial
3533-- (either Dir.toEither (const nil))
3534-- (yield (Left "tmp"))
3539{-# INLINE iterateMapWith #-}
3542 => (t m a -> t m a -> t m a)
3546iterateMapWith combine f = concatMapWith combine go
3548 go x = yield x `combine` concatMapWith combine go (f x)
3551{-# INLINE iterateUnfold #-}
3552iterateUnfold :: (IsStream t, MonadAsync m)
3553 => Unfold m a a -> t m a -> t m a
3554iterateUnfold unf xs = undefined
3557------------------------------------------------------------------------------
3559------------------------------------------------------------------------------
3561-- To traverse graphs we need a state to be carried around in the traversal.
3562-- For example, we can use a hashmap to store the visited status of nodes.
3564-- | Like 'iterateMap' but carries a state in the stream generation function.
3565-- This can be used to traverse graph like structures, we can remember the
3566-- visited nodes in the state to avoid cycles.
3568-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
3569-- traverse graphs. However, this function provides a more localized state
3570-- instead of using a global state.
3576{-# INLINE iterateSmapMWith #-}
3578 :: (IsStream t, Monad m)
3579 => (t m a -> t m a -> t m a)
3580 -> (b -> a -> m (b, t m a))
3584iterateSmapMWith combine f initial stream =
3585 concatMap (\b -> concatMapWith combine (go b) stream) (yieldM initial)
3589 go b a = yield a `combine` feedback b a
3593 (\(b1, s) -> concatMapWith combine (go b1) s)
3596------------------------------------------------------------------------------
3598------------------------------------------------------------------------------
3600-- | Discard 'Right's and unwrap 'Left's in an 'Either' stream.
3604lefts :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m a
3605lefts = fmap (fromLeft undefined) . filter isLeft
3607-- | Discard 'Left's and unwrap 'Right's in an 'Either' stream.
3611rights :: (IsStream t, Monad m, Functor (t m)) => t m (Either a b) -> t m b
3612rights = fmap (fromRight undefined) . filter isRight
3614-- | In an 'Either' stream iterate on 'Left's. This is a special case of
3618-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
3621-- To traverse a directory tree:
3624-- iterateMapLeftsWith serial Dir.toEither (yield (Left "tmp"))
3629{-# INLINE iterateMapLeftsWith #-}
3632 => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
3633 -> (a -> t m (Either a b))
3636iterateMapLeftsWith combine f = iterateMapWith combine (either f (const K.nil))
3638------------------------------------------------------------------------------
3640------------------------------------------------------------------------------
3642-- Splitting operations that take a predicate and a Fold can be
3643-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
3644-- can be expressed using parseMany when used with an appropriate Parser.
3646-- XXX We need takeGE/takeBetween to implement "some" using "many".
3648-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
3651-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
3654-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
3657-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
3658-- > ["hello\n","world"]
3662{-# INLINE parseMany #-}
3664 :: (IsStream t, MonadThrow m)
3669 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
3671{-# INLINE parseManyD #-}
3673 :: (IsStream t, MonadThrow m)
3678 D.fromStreamD $ D.parseMany p (D.toStreamD m)
3680-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
3681-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
3682-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
3683-- output of @test@ is discarded and the output of @collect@ is emitted in the
3684-- output stream. The parser fails if @collect@ fails.
3688{-# INLINE parseManyTill #-}
3690 -- (IsStream t, MonadThrow m) =>
3695parseManyTill = undefined
3697-- | Iterate a parser generating function on a stream. The initial value @b@ is
3698-- used to generate the first parser, the parser is applied on the stream and
3699-- the result is used to generate the next parser and so on.
3701-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
3702-- > [3,10,21,36,55,55]
3704-- This is the streaming equivalent of monad like sequenced application of
3705-- parsers where next parser is dependent on the previous parser.
3709{-# INLINE parseIterate #-}
3711 :: (IsStream t, MonadThrow m)
3712 => (b -> Parser m a b)
3716parseIterate f i m = D.fromStreamD $
3717 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
3719------------------------------------------------------------------------------
3720-- Grouping/Splitting
3721------------------------------------------------------------------------------
3723------------------------------------------------------------------------------
3724-- Grouping without looking at elements
3725------------------------------------------------------------------------------
3727------------------------------------------------------------------------------
3729------------------------------------------------------------------------------
3732------------------------------------------------------------------------------
3734------------------------------------------------------------------------------
3736------------------------------------------------------------------------------
3737-- Generalized grouping
3738------------------------------------------------------------------------------
3740-- This combinator is the most general grouping combinator and can be used to
3741-- implement all other grouping combinators.
3743-- XXX check if this can implement the splitOn combinator i.e. we can slide in
3744-- new elements, slide out old elements and incrementally compute the hash.
3745-- Also, can we implement the windowed classification combinators using this?
3747-- In fact this is a parse. Instead of using a special return value in the fold
3748-- we are using a mapping function.
3750-- Note that 'scanl'' (usually followed by a map to extract the desired value
3751-- from the accumulator) can be used to realize many implementations e.g. a
3752-- sliding window implementation. A scan followed by a mapMaybe is also a good
3753-- pattern to express many problems where we want to emit a filtered output and
3754-- not emit an output on every input.
3756-- Passing on of the initial accumulator value to the next fold is equivalent
3757-- to returning the leftover concept.
3760-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
3761-- @splitter@ is applied on the accumulator of the fold every time an item is
3762-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
3763-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
3764-- in the output stream and the initial value of the accumulator for the next
3765-- group's fold. This allows us to control whether to start fresh for the next
3766-- fold or to continue from the previous fold's output.
3768{-# INLINE groupScan #-}
3770 :: (IsStream t, Monad m)
3771 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
3772groupScan split fold m = undefined
3775-- | Group the input stream into groups of @n@ elements each and then fold each
3776-- group using the provided fold function.
3778-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
3781-- This can be considered as an n-fold version of 'ltake' where we apply
3782-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
3785{-# INLINE chunksOf #-}
3787 :: (IsStream t, Monad m)
3788 => Int -> Fold m a b -> t m a -> t m b
3789chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
3794{-# INLINE chunksOf2 #-}
3796 :: (IsStream t, Monad m)
3797 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
3798chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
3800-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
3801-- @n@ elements each.
3803-- Same as the following but may be more efficient:
3805-- > arraysOf n = S.chunksOf n (A.writeN n)
3808{-# INLINE arraysOf #-}
3809arraysOf :: (IsStream t, MonadIO m, Storable a)
3810 => Int -> t m a -> t m (Array a)
3811arraysOf n = chunksOf n (writeNUnsafe n)
3813-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
3814-- XXX add this example after fixing the serial stream rate control
3815-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
3818-- | Group the input stream into windows of @n@ second each and then fold each
3819-- group using the provided fold function.
3822{-# INLINE intervalsOf #-}
3824 :: (IsStream t, MonadAsync m)
3825 => Double -> Fold m a b -> t m a -> t m b
3827 splitWithSuffix isNothing (FL.lcatMaybes f)
3828 (interjectSuffix n (return Nothing) (Serial.map Just xs))
3830------------------------------------------------------------------------------
3832------------------------------------------------------------------------------
3834-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
3835-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
3836-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
3837-- group and so on. When the comparison fails a new group is started. Each
3838-- group is folded using the fold @f@ and the result of the fold is emitted in
3839-- the output stream.
3841-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
3842-- > [[1,3,7],[0,2,5]]
3845{-# INLINE groupsBy #-}
3847 :: (IsStream t, Monad m)
3852groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
3854-- | Unlike @groupsBy@ this function performs a rolling comparison of two
3855-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
3856-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
3857-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
3858-- 'True' then @c@ is also assigned to the same group and so on. When the
3859-- comparison fails a new group is started. Each group is folded using the fold
3862-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
3863-- > [[1,2,3],[7,8,9]]
3866{-# INLINE groupsByRolling #-}
3868 :: (IsStream t, Monad m)
3873groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
3876-- > groups = groupsBy (==)
3877-- > groups = groupsByRolling (==)
3879-- Groups contiguous spans of equal elements together in individual groups.
3881-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
3885groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
3886groups = groupsBy (==)
3888------------------------------------------------------------------------------
3889-- N-ary split on a predicate
3890------------------------------------------------------------------------------
3892-- TODO: Use a Splitter configuration similar to the "split" package to make it
3893-- possible to express all splitting combinations. In general, we can have
3894-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
3895-- separators. We can have a single split operation taking the splitter config
3898-- | Split on an infixed separator element, dropping the separator. Splits the
3899-- stream on separator elements determined by the supplied predicate, separator
3900-- is considered as infixed between two segments, if one side of the separator
3901-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
3902-- applied on the split segments. With '-' representing non-separator elements
3903-- and '.' as separator, 'splitOn' splits as follows:
3906-- "--.--" => "--" "--"
3911-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
3913-- Let's use the following definition for illustration:
3915-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
3917-- >>> splitOn' (== '.') ""
3920-- >>> splitOn' (== '.') "."
3923-- >>> splitOn' (== '.') ".a"
3926-- >>> splitOn' (== '.') "a."
3929-- >>> splitOn' (== '.') "a.b"
3932-- >>> splitOn' (== '.') "a..b"
3937-- This can be considered as an n-fold version of 'breakOn' where we apply
3938-- 'breakOn' successively on the input stream, dropping the first element
3939-- of the second segment after each break.
3941{-# INLINE splitOn #-}
3943 :: (IsStream t, Monad m)
3944 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3945splitOn predicate f m =
3946 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
3948-- | Like 'splitOn' but the separator is considered as suffixed to the segments
3949-- in the stream. A missing suffix at the end is allowed. A separator at the
3950-- beginning is parsed as empty segment. With '-' representing elements and
3951-- '.' as separator, 'splitOnSuffix' splits as follows:
3954-- "--.--." => "--" "--"
3955-- "--.--" => "--" "--"
3959-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
3961-- >>> splitOnSuffix' (== '.') ""
3964-- >>> splitOnSuffix' (== '.') "."
3967-- >>> splitOnSuffix' (== '.') "a"
3970-- >>> splitOnSuffix' (== '.') ".a"
3973-- >>> splitOnSuffix' (== '.') "a."
3976-- >>> splitOnSuffix' (== '.') "a.b"
3979-- >>> splitOnSuffix' (== '.') "a.b."
3982-- >>> splitOnSuffix' (== '.') "a..b.."
3985-- > lines = splitOnSuffix (== '\n')
3989-- This can be considered as an n-fold version of 'breakPost' where we apply
3990-- 'breakPost' successively on the input stream, dropping the first element
3991-- of the second segment after each break.
3993{-# INLINE splitOnSuffix #-}
3995 :: (IsStream t, Monad m)
3996 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3997splitOnSuffix predicate f m =
3998 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
4000-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
4001-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
4002-- @["a","b"]@. In other words, its like parsing words from whitespace
4005-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
4007-- >>> wordsBy' (== ',') ""
4010-- >>> wordsBy' (== ',') ","
4013-- >>> wordsBy' (== ',') ",a,,b,"
4016-- > words = wordsBy isSpace
4020-- It is equivalent to splitting in any of the infix/prefix/suffix styles
4021-- followed by removal of empty segments.
4022{-# INLINE wordsBy #-}
4024 :: (IsStream t, Monad m)
4025 => (a -> Bool) -> Fold m a b -> t m a -> t m b
4026wordsBy predicate f m =
4027 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
4029-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
4032-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
4034-- >>> splitWithSuffix' (== '.') ""
4037-- >>> splitWithSuffix' (== '.') "."
4040-- >>> splitWithSuffix' (== '.') "a"
4043-- >>> splitWithSuffix' (== '.') ".a"
4046-- >>> splitWithSuffix' (== '.') "a."
4049-- >>> splitWithSuffix' (== '.') "a.b"
4052-- >>> splitWithSuffix' (== '.') "a.b."
4055-- >>> splitWithSuffix' (== '.') "a..b.."
4056-- > ["a.",".","b.","."]
4060-- This can be considered as an n-fold version of 'breakPost' where we apply
4061-- 'breakPost' successively on the input stream.
4063{-# INLINE splitWithSuffix #-}
4065 :: (IsStream t, Monad m)
4066 => (a -> Bool) -> Fold m a b -> t m a -> t m b
4067splitWithSuffix predicate f m =
4068 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
4070------------------------------------------------------------------------------
4071-- Split on a delimiter sequence
4072------------------------------------------------------------------------------
4074-- Int list examples for splitOn:
4076-- >>> splitList [] [1,2,3,3,4]
4077-- > [[1],[2],[3],[3],[4]]
4079-- >>> splitList [5] [1,2,3,3,4]
4082-- >>> splitList [1] [1,2,3,3,4]
4085-- >>> splitList [4] [1,2,3,3,4]
4088-- >>> splitList [2] [1,2,3,3,4]
4091-- >>> splitList [3] [1,2,3,3,4]
4094-- >>> splitList [3,3] [1,2,3,3,4]
4097-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
4100-- | Like 'splitOn' but the separator is a sequence of elements instead of a
4103-- For illustration, let's define a function that operates on pure lists:
4106-- splitOnSeq' pat xs = S.toList $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
4109-- >>> splitOnSeq' "" "hello"
4110-- > ["h","e","l","l","o"]
4112-- >>> splitOnSeq' "hello" ""
4115-- >>> splitOnSeq' "hello" "hello"
4118-- >>> splitOnSeq' "x" "hello"
4121-- >>> splitOnSeq' "h" "hello"
4124-- >>> splitOnSeq' "o" "hello"
4127-- >>> splitOnSeq' "e" "hello"
4130-- >>> splitOnSeq' "l" "hello"
4133-- >>> splitOnSeq' "ll" "hello"
4136-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
4138-- > intercalate . splitOn == id
4140-- The following law holds when the separator is non-empty and contains none of
4141-- the elements present in the input lists:
4143-- > splitOn . intercalate == id
4147-- XXX We can use a polymorphic vector implemented by Array# to represent the
4148-- sequence, that way we can avoid the Storable constraint. If we still need
4149-- Storable Array for performance, we can use a separate splitOnArray API for
4150-- that. We can also have an API where the sequence itself is a lazy stream, so
4151-- that we can search files in files for example.
4152{-# INLINE splitOnSeq #-}
4154 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
4155 => Array a -> Fold m a b -> t m a -> t m b
4156splitOnSeq patt f m = D.fromStreamD $ D.splitOnSeq patt f (D.toStreamD m)
4159-- This can be implemented easily using Rabin Karp
4160-- | Split on any one of the given patterns.
4161{-# INLINE splitOnAny #-}
4163 :: (IsStream t, Monad m, Storable a, Integral a)
4164 => [Array a] -> Fold m a b -> t m a -> t m b
4165splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
4168-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
4169-- of a predicate for a single element.
4171-- > splitOnSuffixSeq_ pat xs = S.toList $ S.splitOnSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
4173-- >>> splitOnSuffixSeq_ "." ""
4176-- >>> splitOnSuffixSeq_ "." "."
4179-- >>> splitOnSuffixSeq_ "." "a"
4182-- >>> splitOnSuffixSeq_ "." ".a"
4185-- >>> splitOnSuffixSeq_ "." "a."
4188-- >>> splitOnSuffixSeq_ "." "a.b"
4191-- >>> splitOnSuffixSeq_ "." "a.b."
4194-- >>> splitOnSuffixSeq_ "." "a..b.."
4197-- > lines = splitOnSuffixSeq "\n"
4199-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
4202-- > intercalateSuffix . splitOnSuffixSeq == id
4204-- The following law holds when the separator is non-empty and contains none of
4205-- the elements present in the input lists:
4207-- > splitSuffixOn . intercalateSuffix == id
4210{-# INLINE splitOnSuffixSeq #-}
4212 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
4213 => Array a -> Fold m a b -> t m a -> t m b
4214splitOnSuffixSeq patt f m =
4215 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
4218-- | Like 'splitOn' but drops any empty splits.
4220{-# INLINE wordsOn #-}
4222 :: (IsStream t, Monad m, Storable a, Eq a)
4223 => Array a -> Fold m a b -> t m a -> t m b
4224wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
4227-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
4229-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
4231-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
4233-- >>> splitOn'_ "" "hello"
4234-- > ["h","","e","","l","","l","","o"]
4236-- >>> splitOn'_ "hello" ""
4239-- >>> splitOn'_ "hello" "hello"
4242-- >>> splitOn'_ "x" "hello"
4245-- >>> splitOn'_ "h" "hello"
4248-- >>> splitOn'_ "o" "hello"
4251-- >>> splitOn'_ "e" "hello"
4254-- >>> splitOn'_ "l" "hello"
4255-- > ["he","l","","l","o"]
4257-- >>> splitOn'_ "ll" "hello"
4261{-# INLINE splitBySeq #-}
4263 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
4264 => Array a -> Fold m a b -> t m a -> t m b
4265splitBySeq patt f m =
4266 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
4268-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
4270-- > splitWithSuffixSeq'_ pat xs = S.toList $ S.splitWithSuffixSeq (A.fromList pat) (FL.toList) (S.fromList xs)
4272-- >>> splitWithSuffixSeq' "." ""
4275-- >>> splitWithSuffixSeq' "." "."
4278-- >>> splitWithSuffixSeq' "." "a"
4281-- >>> splitWithSuffixSeq' "." ".a"
4284-- >>> splitWithSuffixSeq' "." "a."
4287-- >>> splitWithSuffixSeq' "." "a.b"
4290-- >>> splitWithSuffixSeq' "." "a.b."
4293-- >>> splitWithSuffixSeq' "." "a..b.."
4294-- > ["a.",".","b.","."]
4297{-# INLINE splitWithSuffixSeq #-}
4299 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
4300 => Array a -> Fold m a b -> t m a -> t m b
4301splitWithSuffixSeq patt f m =
4302 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
4305-- This can be implemented easily using Rabin Karp
4306-- | Split post any one of the given patterns.
4307{-# INLINE splitOnSuffixSeqAny #-}
4309 :: (IsStream t, Monad m, Storable a, Integral a)
4310 => [Array a] -> Fold m a b -> t m a -> t m b
4311splitOnSuffixSeqAny subseq f m = undefined
4312 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
4315------------------------------------------------------------------------------
4317------------------------------------------------------------------------------
4319-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
4320-- an input stream @t m (f a)@ using the @splitter@ function. Container
4321-- elements @f a@ are collected until a split occurs, then all the elements
4322-- before the split are joined using the @joiner@ function.
4324-- For example, if we have a stream of @Array Word8@, we may want to split the
4325-- stream into arrays representing lines separated by '\n' byte such that the
4326-- resulting stream after a split would be one array for each line.
4328-- CAUTION! This is not a true streaming function as the container size after
4329-- the split and merge may not be bounded.
4332{-# INLINE splitInnerBy #-}
4334 :: (IsStream t, Monad m)
4335 => (f a -> m (f a, Maybe (f a))) -- splitter
4336 -> (f a -> f a -> m (f a)) -- joiner
4339splitInnerBy splitter joiner xs =
4340 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
4342-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
4346{-# INLINE splitInnerBySuffix #-}
4348 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
4349 => (f a -> m (f a, Maybe (f a))) -- splitter
4350 -> (f a -> f a -> m (f a)) -- joiner
4353splitInnerBySuffix splitter joiner xs =
4354 D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
4356------------------------------------------------------------------------------
4357-- Reorder in sequence
4358------------------------------------------------------------------------------
4361-- Buffer until the next element in sequence arrives. The function argument
4362-- determines the difference in sequence numbers. This could be useful in
4363-- implementing sequenced streams, for example, TCP reassembly.
4364{-# INLINE reassembleBy #-}
4366 :: (IsStream t, Monad m)
4371reassembleBy = undefined
4374------------------------------------------------------------------------------
4376------------------------------------------------------------------------------
4378-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
4379-- add a tap to log the contents flowing through the stream. The fold is used
4380-- only for effects, its result is discarded.
4385-- -----stream m a ---------------stream m a-----
4390-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
4395-- Compare with 'trace'.
4399tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
4400tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
4402-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
4403-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
4404-- means start at the first element in the stream. If the offset is outside
4405-- this range then @offset `mod` n@ is used as offset.
4408-- >>> S.drain $ S.tapOffsetEvery 0 2 (FL.mapM print FL.toList) $ S.enumerateFromTo 0 10
4414{-# INLINE tapOffsetEvery #-}
4415tapOffsetEvery :: (IsStream t, Monad m)
4416 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
4417tapOffsetEvery offset n f xs =
4418 D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
4420-- | Redirect a copy of the stream to a supplied fold and run it concurrently
4421-- in an independent thread. The fold may buffer some elements. The buffer size
4422-- is determined by the prevailing 'maxBuffer' setting.
4427-- -----stream m a ---------------stream m a-----
4432-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
4437-- Exceptions from the concurrently running fold are propagated to the current
4438-- computation. Note that, because of buffering in the fold, exceptions may be
4439-- delayed and may not correspond to the current element being processed in the
4440-- parent stream, but we guarantee that before the parent stream stops the tap
4441-- finishes and all exceptions from it are drained.
4444-- Compare with 'tap'.
4447{-# INLINE tapAsync #-}
4448tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
4449tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)
4451-- | @pollCounts predicate transform fold stream@ counts those elements in the
4452-- stream that pass the @predicate@. The resulting count stream is sent to
4453-- another thread which transforms it using @transform@ and then folds it using
4454-- @fold@. The thread is automatically cleaned up if the stream stops or
4455-- aborts due to exception.
4457-- For example, to print the count of elements processed every second:
4460-- > S.drain $ S.pollCounts (const True) (S.rollingMap (-) . S.delayPost 1) (FL.drainBy print)
4461-- $ S.enumerateFrom 0
4464-- Note: This may not work correctly on 32-bit machines.
4468{-# INLINE pollCounts #-}
4470 (IsStream t, MonadAsync m)
4472 -> (t m Int -> t m Int)
4476pollCounts predicate transf f xs =
4478 $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f
4481-- | Calls the supplied function with the number of elements consumed
4482-- every @n@ seconds. The given function is run in a separate thread
4483-- until the end of the stream. In case there is an exception in the
4484-- stream the thread is killed during the next major GC.
4486-- Note: The action is not guaranteed to run if the main thread exits.
4489-- > delay n = threadDelay (round $ n * 1000000) >> return n
4490-- > S.drain $ S.tapRate 2 (\\n -> print $ show n ++ " elements processed") (delay 1 S.|: delay 0.5 S.|: delay 0.5 S.|: S.nil)
4491-- 2 elements processed
4492-- 1 elements processed
4495-- Note: This may not work correctly on 32-bit machines.
4498{-# INLINE tapRate #-}
4500 (IsStream t, MonadAsync m, MonadCatch m)
4505tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs
4507-- | Apply a monadic function to each element flowing through the stream and
4508-- discard the results.
4511-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
4516-- Compare with 'tap'.
4520trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
4521trace f = mapM (\x -> void (f x) >> return x)
4523-- | Perform a side effect before yielding each element of the stream and
4524-- discard the results.
4527-- > S.drain $ S.trace_ (print "got here") (S.enumerateFromTo 1 2)
4532-- See also: 'trace', 'interspersePrefix_'
4535{-# INLINE trace_ #-}
4536trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
4537trace_ eff = Serial.mapM (\x -> eff >> return x)
4539------------------------------------------------------------------------------
4540-- Windowed classification
4541------------------------------------------------------------------------------
4543-- We divide the stream into windows or chunks in space or time and each window
4544-- can be associated with a key, all events associated with a particular key in
4545-- the window can be folded to a single result. The stream can be split into
4546-- windows by size or by using a split predicate on the elements in the stream.
4547-- For example, when we receive a closing flag, we can close the window.
4549-- A "chunk" is a space window and a "session" is a time window. Are there any
4550-- other better short words to describe them. An alternative is to use
4551-- "swindow" and "twindow". Another word for "session" could be "spell".
4553-- TODO: To mark the position in space or time we can have Indexed or
4554-- TimeStamped types. That can make it easy to deal with the position indices
4557------------------------------------------------------------------------------
4558-- Keyed Sliding Windows
4559------------------------------------------------------------------------------
4562{-# INLINABLE classifySlidingChunks #-}
4563classifySlidingChunks
4564 :: (IsStream t, MonadAsync m, Ord k)
4565 => Int -- ^ window size
4566 -> Int -- ^ window slide
4567 -> Fold m a b -- ^ Fold to be applied to window events
4568 -> t m (k, a, Bool) -- ^ window key, data, close event
4570classifySlidingChunks wsize wslide (Fold step initial extract) str
4573-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
4574-- slide the send window when an ack is received and we slide the receive
4575-- window when a sequence is complete. Sliding is stateful in case of TCP,
4576-- sliding releases the send buffer or makes data available to the user from
4577-- the receive buffer.
4578{-# INLINABLE classifySlidingSessions #-}
4579classifySlidingSessions
4580 :: (IsStream t, MonadAsync m, Ord k)
4581 => Double -- ^ timer tick in seconds
4582 -> Double -- ^ time window size
4583 -> Double -- ^ window slide
4584 -> Fold m a b -- ^ Fold to be applied to window events
4585 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
4587classifySlidingSessions tick interval slide (Fold step initial extract) str
4591------------------------------------------------------------------------------
4592-- Sliding Window Buffers
4593------------------------------------------------------------------------------
4595-- These buffered versions could be faster than concurrent incremental folds of
4596-- all overlapping windows as in many cases we may not need all the values to
4597-- compute the fold, we can just compute the result using the old value and new
4598-- value. However, we may need the buffer once in a while, for example for
4599-- string search we usually compute the hash incrementally but when the hash
4600-- matches the hash of the pattern we need to compare the whole string.
4602-- XXX we should be able to implement sequence based splitting combinators
4603-- using this combinator.
4606-- | Buffer n elements of the input in a ring buffer. When t new elements are
4607-- collected, slide the window to remove the same number of oldest elements,
4608-- insert the new elements, and apply an incremental fold on the sliding
4609-- window, supplying the outgoing elements, the new ring buffer as arguments.
4611 :: (IsStream t, Monad m, Ord a, Storable a)
4612 => Int -- window size
4613 -> Int -- window slide
4614 -> Fold m (Ring a, Array a) b
4617slidingChunkBuffer = undefined
4619-- Buffer n seconds worth of stream elements of the input in a radix tree.
4620-- Every t seconds, remove the items that are older than n seconds, and apply
4621-- an incremental fold on the sliding window, supplying the outgoing elements,
4622-- and the new radix tree buffer as arguments.
4624 :: (IsStream t, Monad m, Ord a, Storable a)
4625 => Int -- window size
4627 -> Fold m (RTree a, Array a) b
4630slidingSessionBuffer = undefined
4633------------------------------------------------------------------------------
4634-- Keyed Session Windows
4635------------------------------------------------------------------------------
4638-- | Keyed variable size space windows. Close the window if we do not receive a
4639-- window event in the next "spaceout" elements.
4640{-# INLINABLE classifyChunksBy #-}
4642 :: (IsStream t, MonadAsync m, Ord k)
4643 => Int -- ^ window spaceout (spread)
4644 -> Bool -- ^ reset the spaceout when a chunk window element is received
4645 -> Fold m a b -- ^ Fold to be applied to chunk window elements
4646 -> t m (k, a, Bool) -- ^ chunk key, data, last element
4648classifyChunksBy spanout reset (Fold step initial extract) str = undefined
4650-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
4651-- received within the chunk size window. The chunk gets closed only if no
4652-- element is received within the chunk window.
4654{-# INLINABLE classifyKeepAliveChunks #-}
4655classifyKeepAliveChunks
4656 :: (IsStream t, MonadAsync m, Ord k)
4657 => Int -- ^ window spaceout (spread)
4658 -> Fold m a b -- ^ Fold to be applied to chunk window elements
4659 -> t m (k, a, Bool) -- ^ chunk key, data, last element
4661classifyKeepAliveChunks spanout = classifyChunksBy spanout True
4664data SessionState t m k a b = SessionState
4665 { sessionCurTime :: !AbsTime -- ^ time since last event
4666 , sessionEventTime :: !AbsTime -- ^ time as per last event
4667 , sessionCount :: !Int -- ^ total number sessions in progress
4668 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
4669 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
4670 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
4675-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
4676-- events in an input event stream into sessions based on a session key. Each
4677-- element in the stream is an event consisting of a triple @(session key,
4678-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
4679-- the session. All the events belonging to a session are folded using the
4680-- fold @f@ until the fold returns a 'Left' result or a timeout has occurred.
4681-- The session key and the result of the fold are emitted in the output stream
4682-- when the session is purged.
4684-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
4685-- seconds, measured from the @timestamp@ of the first event in that session.
4686-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
4687-- every event received in the session.
4689-- @timestamp@ in an event characterizes the time when the input event was
4690-- generated, this is an absolute time measured from some @Epoch@. The notion
4691-- of current time is maintained by a monotonic event time clock using the
4692-- timestamps seen in the input stream. The latest timestamp seen till now is
4693-- used as the base for the current time. When no new events are seen, a timer
4694-- is started with a tick duration specified by @tick@. This timer is used to
4695-- detect session timeouts in the absence of new events.
4697-- The predicate @pred@ is invoked with the current session count, if it
4698-- returns 'True' a session is ejected from the session cache before inserting
4699-- a new session. This could be useful to alert or eject sessions when the
4700-- number of sessions becomes too high.
4705-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
4706{-# INLINABLE classifySessionsBy #-}
4708 :: (IsStream t, MonadAsync m, Ord k)
4709 => Double -- ^ timer tick in seconds
4710 -> Double -- ^ session timeout in seconds
4711 -> Bool -- ^ reset the timeout when an event is received
4712 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
4713 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4714 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4715 -> t m (k, b) -- ^ session key, fold result
4716classifySessionsBy tick tmout reset ejectPred
4717 (Fold step initial extract) str =
4718 concatMap sessionOutputStream $
4719 scanlMAfter' sstep (return szero) flush stream
4723 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
4724 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
4725 szero = SessionState
4726 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
4727 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
4729 , sessionTimerHeap = H.empty
4730 , sessionKeyValueMap = Map.empty
4731 , sessionOutputStream = K.nil
4734 -- We can eject sessions based on the current session count to limit
4735 -- memory consumption. There are two possible strategies:
4737 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
4738 -- threshold even before timeout, effectively reduce the timeout.
4739 -- 2) Drop creation of new sessions but keep accepting new events for the
4742 -- We use the first strategy as of now.
4744 -- Got a new stream input element
4745 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
4746 -- XXX we should use a heap in pinned memory to scale it to a large
4749 -- To detect session inactivity we keep a timestamp of the latest event
4750 -- in the Map along with the fold result. When we purge the session
4751 -- from the heap we match the timestamp in the heap with the timestamp
4752 -- in the Map, if the latest timestamp is newer and has not expired we
4753 -- reinsert the key in the heap.
4755 -- XXX if the key is an Int, we can also use an IntMap for slightly
4756 -- better performance.
4758 let curTime = max sessionEventTime timestamp
4762 Just (Tuple' _ acc) -> return acc
4763 new <- step old value
4764 return $ Tuple' timestamp new
4765 mOld = Map.lookup key sessionKeyValueMap
4767 acc@(Tuple' _ fres) <- accumulate mOld
4771 -- deleting a key from the heap is expensive, so we never
4772 -- delete a key from heap, we just purge it from the Map and it
4773 -- gets purged from the heap on timeout. We just need an extra
4774 -- lookup in the Map when the key is purged from the heap, that
4775 -- should not be expensive.
4777 let (mp, cnt) = case mOld of
4778 Nothing -> (sessionKeyValueMap, sessionCount)
4779 Just _ -> (Map.delete key sessionKeyValueMap
4782 { sessionCurTime = curTime
4783 , sessionEventTime = curTime
4784 , sessionCount = cnt
4785 , sessionKeyValueMap = mp
4786 , sessionOutputStream = yield (key, x)
4789 (hp1, mp1, out1, cnt1) <- do
4790 let vars = (sessionTimerHeap, sessionKeyValueMap,
4791 K.nil, sessionCount)
4793 -- inserting new entry
4795 -- Eject a session from heap and map is needed
4796 eject <- ejectPred sessionCount
4797 (hp, mp, out, cnt) <-
4802 -- Insert the new session in heap
4803 let expiry = addToAbsTime timestamp timeoutMs
4804 hp' = H.insert (Entry expiry key) hp
4805 in return (hp', mp, out, cnt + 1)
4806 -- updating old entry
4807 Just _ -> return vars
4809 let mp2 = Map.insert key acc mp1
4810 return $ SessionState
4811 { sessionCurTime = curTime
4812 , sessionEventTime = curTime
4813 , sessionCount = cnt1
4814 , sessionTimerHeap = hp1
4815 , sessionKeyValueMap = mp2
4816 , sessionOutputStream = out1
4819 -- Got a timer tick event
4820 sstep sessionState@SessionState{..} Nothing =
4821 let curTime = addToAbsTime sessionCurTime tickMs
4822 in ejectExpired sessionState curTime
4824 flush session@SessionState{..} = do
4825 (hp', mp', out, count) <-
4828 , sessionKeyValueMap
4833 { sessionCount = count
4834 , sessionTimerHeap = hp'
4835 , sessionKeyValueMap = mp'
4836 , sessionOutputStream = out
4844 -- delete from map and output the fold accumulator
4845 ejectEntry hp mp out cnt acc key = do
4847 let out1 = (key, fromEither sess) `K.cons` out
4848 let mp1 = Map.delete key mp
4849 return (hp, mp1, out1, cnt - 1)
4851 ejectAll (hp, mp, out, !cnt) = do
4852 let hres = H.uncons hp
4854 Just (Entry _ key, hp1) -> do
4855 r <- case Map.lookup key mp of
4856 Nothing -> return (hp1, mp, out, cnt)
4857 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
4860 assert (Map.null mp) (return ())
4861 return (hp, mp, out, cnt)
4863 ejectOne (hp, mp, out, !cnt) = do
4864 let hres = H.uncons hp
4866 Just (Entry expiry key, hp1) ->
4867 case Map.lookup key mp of
4868 Nothing -> ejectOne (hp1, mp, out, cnt)
4869 Just (Tuple' latestTS acc) -> do
4870 let expiry1 = addToAbsTime latestTS timeoutMs
4871 if not reset || expiry1 <= expiry
4872 then ejectEntry hp1 mp out cnt acc key
4874 -- reset the session timeout and continue
4875 let hp2 = H.insert (Entry expiry1 key) hp1
4876 in ejectOne (hp2, mp, out, cnt)
4878 assert (Map.null mp) (return ())
4879 return (hp, mp, out, cnt)
4881 ejectExpired session@SessionState{..} curTime = do
4882 (hp', mp', out, count) <-
4883 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
4885 { sessionCurTime = curTime
4886 , sessionCount = count
4887 , sessionTimerHeap = hp'
4888 , sessionKeyValueMap = mp'
4889 , sessionOutputStream = out
4894 ejectLoop hp mp out !cnt = do
4895 let hres = H.uncons hp
4897 Just (Entry expiry key, hp1) -> do
4899 if curTime >= expiry
4900 then return (True, False)
4906 case Map.lookup key mp of
4907 Nothing -> ejectLoop hp1 mp out cnt
4908 Just (Tuple' latestTS acc) -> do
4909 let expiry1 = addToAbsTime latestTS timeoutMs
4910 if expiry1 <= curTime || not reset || force
4912 (hp2,mp1,out1,cnt1) <-
4913 ejectEntry hp1 mp out cnt acc key
4914 ejectLoop hp2 mp1 out1 cnt1
4916 -- reset the session timeout and continue
4917 let hp2 = H.insert (Entry expiry1 key) hp1
4918 in ejectLoop hp2 mp out cnt
4919 else return (hp, mp, out, cnt)
4921 assert (Map.null mp) (return ())
4922 return (hp, mp, out, cnt)
4924 -- merge timer events in the stream
4925 stream = Serial.map Just str `Par.parallelFst` repeatM timer
4927 liftIO $ threadDelay (round $ tick * 1000000)
4930-- | Like 'classifySessionsOf' but the session is kept alive if an event is
4931-- received within the session window. The session times out and gets closed
4932-- only if no event is received within the specified session window size.
4934-- If the ejection predicate returns 'True', the session that was idle for
4935-- the longest time is ejected before inserting a new session.
4938-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
4943{-# INLINABLE classifyKeepAliveSessions #-}
4944classifyKeepAliveSessions
4945 :: (IsStream t, MonadAsync m, Ord k)
4946 => Double -- ^ session inactive timeout
4947 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
4948 -> Fold m a (Either b b) -- ^ Fold to be applied to session payload data
4949 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4951classifyKeepAliveSessions tmout =
4952 classifySessionsBy 1 tmout True
4954------------------------------------------------------------------------------
4955-- Keyed tumbling windows
4956------------------------------------------------------------------------------
4958-- Tumbling windows is a special case of sliding windows where the window slide
4959-- is the same as the window size. Or it can be a special case of session
4960-- windows where the reset flag is set to False.
4962-- XXX instead of using the early termination flag in the stream, we can use an
4963-- early terminating fold instead.
4966-- | Split the stream into fixed size chunks of specified size. Within each
4967-- such chunk fold the elements in buckets identified by the keys. A particular
4968-- bucket fold can be terminated early if a closing flag is encountered in an
4969-- element for that key.
4972{-# INLINABLE classifyChunksOf #-}
4974 :: (IsStream t, MonadAsync m, Ord k)
4975 => Int -- ^ window size
4976 -> Fold m a b -- ^ Fold to be applied to window events
4977 -> t m (k, a, Bool) -- ^ window key, data, close event
4979classifyChunksOf wsize = classifyChunksBy wsize False
4982-- | Split the stream into fixed size time windows of specified interval in
4983-- seconds. Within each such window, fold the elements in sessions identified
4984-- by the session keys. The fold result is emitted in the output stream if the
4985-- fold returns a 'Left' result or if the time window ends.
4987-- Session @timestamp@ in the input stream is an absolute time from some epoch,
4988-- characterizing the time when the input element was generated. To detect
4989-- session window end, a monotonic event time clock is maintained synced with
4990-- the timestamps with a clock resolution of 1 second.
4992-- If the ejection predicate returns 'True', the session with the longest
4993-- lifetime is ejected before inserting a new session.
4996-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
5001-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
5002-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
5005-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
5010{-# INLINABLE classifySessionsOf #-}
5012 :: (IsStream t, MonadAsync m, Ord k)
5013 => Double -- ^ time window size
5014 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
5015 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
5016 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
5018classifySessionsOf interval =
5019 classifySessionsBy 1 interval False
5021------------------------------------------------------------------------------
5023------------------------------------------------------------------------------
5025-- | Run the action @m b@ before the stream yields its first element.
5027-- > before action xs = 'nilM' action <> xs
5030{-# INLINE before #-}
5031before :: (IsStream t, Monad m) => m b -> t m a -> t m a
5032before action xs = D.fromStreamD $ D.before action $ D.toStreamD xs
5034-- | Like 'after', with following differences:
5036-- * action @m b@ won't run if the stream is garbage collected
5037-- after partial evaluation.
5038-- * Monad @m@ does not require any other constraints.
5039-- * has slightly better performance than 'after'.
5041-- Same as the following, but with stream fusion:
5043-- > after_ action xs = xs <> 'nilM' action
5047{-# INLINE after_ #-}
5048after_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
5049after_ action xs = D.fromStreamD $ D.after_ action $ D.toStreamD xs
5051-- | Run the action @m b@ whenever the stream @t m a@ stops normally, or if it
5052-- is garbage collected after a partial lazy evaluation.
5054-- The semantics of the action @m b@ are similar to the semantics of cleanup
5055-- action in 'bracket'.
5057-- /See also 'after_'/
5062after :: (IsStream t, MonadIO m, MonadBaseControl IO m)
5063 => m b -> t m a -> t m a
5064after action xs = D.fromStreamD $ D.after action $ D.toStreamD xs
5066-- | Run the action @m b@ if the stream aborts due to an exception. The
5067-- exception is not caught, simply rethrown.
5069-- /Inhibits stream fusion/
5072{-# INLINE onException #-}
5073onException :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
5074onException action xs = D.fromStreamD $ D.onException action $ D.toStreamD xs
5076-- | Like 'finally' with following differences:
5078-- * action @m b@ won't run if the stream is garbage collected
5079-- after partial evaluation.
5080-- * does not require a 'MonadAsync' constraint.
5081-- * has slightly better performance than 'finally'.
5083-- /Inhibits stream fusion/
5087{-# INLINE finally_ #-}
5088finally_ :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
5089finally_ action xs = D.fromStreamD $ D.finally_ action $ D.toStreamD xs
5091-- | Run the action @m b@ whenever the stream @t m a@ stops normally, aborts
5092-- due to an exception or if it is garbage collected after a partial lazy
5095-- The semantics of running the action @m b@ are similar to the cleanup action
5096-- semantics described in 'bracket'.
5099-- finally release = bracket (return ()) (const release)
5102-- /See also 'finally_'/
5104-- /Inhibits stream fusion/
5108{-# INLINE finally #-}
5109finally :: (IsStream t, MonadAsync m, MonadCatch m) => m b -> t m a -> t m a
5110finally action xs = D.fromStreamD $ D.finally action $ D.toStreamD xs
5112-- | Like 'bracket' but with following differences:
5114-- * alloc action @m b@ runs with async exceptions enabled
5115-- * cleanup action @b -> m c@ won't run if the stream is garbage collected
5116-- after partial evaluation.
5117-- * does not require a 'MonadAsync' constraint.
5118-- * has slightly better performance than 'bracket'.
5120-- /Inhibits stream fusion/
5124{-# INLINE bracket_ #-}
5125bracket_ :: (IsStream t, MonadCatch m)
5126 => m b -> (b -> m c) -> (b -> t m a) -> t m a
5127bracket_ bef aft bet = D.fromStreamD $
5128 D.bracket_ bef aft (toStreamD . bet)
5130-- | Run the alloc action @m b@ with async exceptions disabled but keeping
5131-- blocking operations interruptible (see 'Control.Exception.mask'). Use the
5132-- output @b@ as input to @b -> t m a@ to generate an output stream.
5134-- @b@ is usually a resource under the state of monad @m@, e.g. a file
5135-- handle, that requires a cleanup after use. The cleanup action @b -> m c@,
5136-- runs whenever the stream ends normally, due to a sync or async exception or
5137-- if it gets garbage collected after a partial lazy evaluation.
5139-- 'bracket' only guarantees that the cleanup action runs, and it runs with
5140-- async exceptions enabled. The action must ensure that it can successfully
5141-- cleanup the resource in the face of sync or async exceptions.
5143-- When the stream ends normally or on a sync exception, cleanup action runs
5144-- immediately in the current thread context, whereas in other cases it runs in
5145-- the GC context, therefore, cleanup may be delayed until the GC gets to run.
5147-- /See also: 'bracket_'/
5149-- /Inhibits stream fusion/
5153{-# INLINE bracket #-}
5154bracket :: (IsStream t, MonadAsync m, MonadCatch m)
5155 => m b -> (b -> m c) -> (b -> t m a) -> t m a
5156bracket bef aft bet = D.fromStreamD $
5157 D.bracket bef aft (toStreamD . bet)
5159-- | Like 'handle' but the exception handler is also provided with the stream
5160-- that generated the exception as input. The exception handler can thus
5161-- re-evaluate the stream to retry the action that failed. The exception
5162-- handler can again call 'ghandle' on it to retry the action multiple times.
5164-- This is highly experimental. In a stream of actions we can map the stream
5165-- with a retry combinator to retry each action on failure.
5167-- /Inhibits stream fusion/
5171{-# INLINE ghandle #-}
5172ghandle :: (IsStream t, MonadCatch m, Exception e)
5173 => (e -> t m a -> t m a) -> t m a -> t m a
5176 . D.ghandle (\e xs -> D.toStreamD $ handler e (D.fromStreamD xs))
5179-- | When evaluating a stream if an exception occurs, stream evaluation aborts
5180-- and the specified exception handler is run with the exception as argument.
5182-- /Inhibits stream fusion/
5185{-# INLINE handle #-}
5186handle :: (IsStream t, MonadCatch m, Exception e)
5187 => (e -> t m a) -> t m a -> t m a
5189 D.fromStreamD $ D.handle (D.toStreamD . handler) $ D.toStreamD xs
5191-- Keep concating either streams as long as rights are generated, stop as soon
5192-- as a left is generated and concat the left stream.
5194-- See also: 'handle'
5200 :: -- (IsStream t, MonadAsync m) =>
5201 (forall x. t m x -> t m x -> t m x)
5202 -> (a -> t m (Either (t m b) b))
5205concatMapEitherWith = undefined
5208------------------------------------------------------------------------------
5209-- Generalize the underlying monad
5210------------------------------------------------------------------------------
5212-- | Transform the inner monad of a stream using a natural transformation.
5217hoist :: (Monad m, Monad n)
5218 => (forall x. m x -> n x) -> SerialT m a -> SerialT n a
5219hoist f xs = fromStreamS $ S.hoist f (toStreamS xs)
5221-- | Generalize the inner monad of the stream from 'Identity' to any monad.
5225{-# INLINE generally #-}
5226generally :: (IsStream t, Monad m) => t Identity a -> t m a
5227generally xs = fromStreamS $ S.hoist (return . runIdentity) (toStreamS xs)
5229------------------------------------------------------------------------------
5230-- Add and remove a monad transformer
5231------------------------------------------------------------------------------
5233-- | Lift the inner monad @m@ of a stream @t m a@ to @tr m@ using the monad
5238{-# INLINE liftInner #-}
5239liftInner :: (Monad m, IsStream t, MonadTrans tr, Monad (tr m))
5240 => t m a -> t (tr m) a
5241liftInner xs = fromStreamD $ D.liftInner (toStreamD xs)
5243------------------------------------------------------------------------------
5244-- Sharing read only state in a stream
5245------------------------------------------------------------------------------
5247-- | Evaluate the inner monad of a stream as 'ReaderT'.
5251{-# INLINE runReaderT #-}
5252runReaderT :: (IsStream t, Monad m) => m s -> t (ReaderT s m) a -> t m a
5253runReaderT s xs = fromStreamD $ D.runReaderT s (toStreamD xs)
5255-- | Run a stream transformation using a given environment.
5257-- See also: 'Serial.map'
5261{-# INLINE usingReaderT #-}
5263 :: (Monad m, IsStream t)
5265 -> (t (ReaderT r m) a -> t (ReaderT r m) a)
5268usingReaderT r f xs = runReaderT r $ f $ liftInner xs
5270------------------------------------------------------------------------------
5271-- Sharing read write state in a stream
5272------------------------------------------------------------------------------
5274-- | Evaluate the inner monad of a stream as 'StateT'.
5276-- This is supported only for 'SerialT' as concurrent state updation may not be
5281{-# INLINE evalStateT #-}
5282evalStateT :: Monad m => m s -> SerialT (StateT s m) a -> SerialT m a
5283evalStateT s xs = fromStreamD $ D.evalStateT s (toStreamD xs)
5285-- | Run a stateful (StateT) stream transformation using a given state.
5287-- This is supported only for 'SerialT' as concurrent state updation may not be
5290-- See also: 'scanl''
5294{-# INLINE usingStateT #-}
5298 -> (SerialT (StateT s m) a -> SerialT (StateT s m) a)
5301usingStateT s f xs = evalStateT s $ f $ liftInner xs
5303-- | Evaluate the inner monad of a stream as 'StateT' and emit the resulting
5304-- state and value pair after each step.
5306-- This is supported only for 'SerialT' as concurrent state updation may not be
5311{-# INLINE runStateT #-}
5312runStateT :: Monad m => m s -> SerialT (StateT s m) a -> SerialT m (s, a)
5313runStateT 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