1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE RankNTypes #-}
4{-# LANGUAGE RecordWildCards #-}
5{-# LANGUAGE KindSignatures #-}
6{-# LANGUAGE FlexibleContexts #-}
8#if __GLASGOW_HASKELL__ >= 800
9{-# OPTIONS_GHC -Wno-orphans #-}
15-- Module : Streamly.Internal.Prelude
16-- Copyright : (c) 2017 Harendra Kumar
19-- Maintainer : streamly@composewell.com
20-- Stability : experimental
23-- This is an Internal module consisting of released, unreleased and
24-- unimplemented APIs. For stable and released APIs please see
25-- "Streamly.Prelude" module. This module provides documentation only for the
26-- unreleased and unimplemented APIs. For documentation on released APIs please
27-- see "Streamly.Prelude" module.
29module Streamly.Internal.Prelude
54 -- ** Time Enumeration
99 -- ** Composable Left Folds
105 -- ** Concurrent Folds
112 -- -- ** To Summary (Full Folds)
120 -- -- ** To Summary (Maybe) (Full Folds)
128 -- -- ** To Containers (Full Folds)
134 -- ** Composable Left Folds
136 , toStream -- XXX rename to write?
137 , toStreamRev -- XXX rename to writeRev?
141 -- -- ** To Elements (Partial Folds)
145 -- -- | Folds that extract selected elements of a stream or their properties.
155 -- -- ** To Boolean (Partial Folds)
164 -- ** Multi-Stream folds
169 -- finding subsequences
175 -- trimming sequences
213 -- ** Scan Using Fold
217 -- XXX Once we have pipes the contravariant transformations can be
218 -- represented by attaching pipes before a transformation.
230 -- ** Concurrent Transformation
242 -- ** Mapping Filters
246 -- ** Deleting Elements
249 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
250 -- , uniqOn -- to remove duplicate sequences
251 -- , pruneBy -- dropAround + uniqBy - like words
253 -- ** Inserting Elements
262 , intersperseSuffixBySpan
263 -- , intersperseBySpan
264 -- , intersperseByIndices -- using an index function/stream
266 -- time domain intersperse
267 -- , intersperseByTime
268 -- , intersperseByEvent
279 -- , timestampedR -- timer
316 -- -- *** Searching Elements
320 -- -- *** Searching Sequences
321 -- , seqIndices -- search a sequence in the stream
323 -- -- *** Searching Multiple Sequences
324 -- , seqIndicesAny -- search any of the given sequence in the stream
326 -- -- -- ** Searching Streams
327 -- -- | Finding a stream within another stream.
330 -- | Streams can be sliced into segments in space or in time. We use the
331 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
332 -- and the term @session@ to refer to a length in time (time window).
334 -- -- *** Using Element Separators
342 , wordsBy -- stripAndCompactBy
344 -- -- *** Splitting By Sequences
347 -- , splitOnPrefixSeq
349 -- Keeping the delimiters
352 -- , splitByPrefixSeq
355 -- Splitting using multiple sequence separators
357 -- , splitOnAnySuffixSeq
358 -- , splitOnAnyPrefixSeq
360 -- -- *** Splitting By Streams
361 -- -- | Splitting a stream using another stream as separator.
368 -- In imperative terms, grouped folding can be considered as a nested loop
369 -- where we loop over the stream to group elements and then loop over
370 -- individual groups to fold them to a single value that is yielded in the
383 -- * Windowed Classification
385 -- | Split the stream into windows or chunks in space or time. Each window
386 -- can be associated with a key, all events associated with a particular
387 -- key in the window can be folded to a single result. The stream is split
388 -- into windows of specified size, the window can be terminated early if
389 -- the closing flag is specified in the input stream.
391 -- The term "chunk" is used for a space window and the term "session" is
392 -- used for a time window.
394 -- ** Tumbling Windows
395 -- | A new window starts after the previous window is finished.
397 -- , classifyChunksOf
401 -- ** Keep Alive Windows
402 -- | The window size is extended if an event arrives within the specified
403 -- window size. This can represent sessions with idle or inactive timeout.
405 -- , classifyKeepAliveChunks
406 , classifyKeepAliveSessions
409 -- ** Sliding Windows
410 -- | A new window starts after the specified slide from the previous
411 -- window. Therefore windows can overlap.
412 , classifySlidingChunks
413 , classifySlidingSessions
415 -- ** Sliding Window Buffers
416 -- , slidingChunkBuffer
417 -- , slidingSessionBuffer
419 -- * Combining Streams
454 -- ** Folding Containers of Streams
459 -- Flattening Nested Streams
460 -- ** Folding Streams of Streams
465 -- XXX add stateful concatMapWith?
469 -- ** Flattening Using Unfolds
471 , concatUnfoldInterleave
472 , concatUnfoldRoundrobin
475 , concatMapIterateWith
478 , concatMapTreeYieldLeavesWith
481 -- ** Inserting Streams in Streams
501 -- * Generalize Inner Monad
505 -- * Transform Inner Monad
531import Control.Concurrent (threadDelay)
532import Control.Exception (Exception, assert)
533import Control.Monad (void)
534import Control.Monad.Catch (MonadCatch, MonadThrow, throwM)
535import Control.Monad.IO.Class (MonadIO(..))
536import Control.Monad.Reader (ReaderT)
537import Control.Monad.State.Strict (StateT)
538import Control.Monad.Trans (MonadTrans(..))
539import Control.Monad.Trans.Control (MonadBaseControl)
540import Data.Functor.Identity (Identity (..))
541#if __GLASGOW_HASKELL__ >= 800
542import Data.Kind (Type)
544import Data.Heap (Entry(..))
545import Data.Maybe (isJust, fromJust, isNothing)
546import Foreign.Storable (Storable)
548 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
549 foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
550 notElem, maximum, minimum, head, last, tail, length, null,
551 reverse, iterate, init, and, or, lookup, foldr1, (!!),
552 scanl, scanl1, replicate, concatMap, span, splitAt, break,
553 repeat, concat, mconcat)
555import qualified Data.Heap as H
556import qualified Data.Map.Strict as Map
557import qualified Prelude
558import qualified System.IO as IO
560import Streamly.Internal.Data.Stream.Enumeration (Enumerable(..), enumerate, enumerateTo)
561import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))
562import Streamly.Internal.Data.Parser (Parser (..))
563import Streamly.Internal.Data.Unfold.Types (Unfold)
564import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)
565-- import Streamly.Memory.Ring (Ring)
566import Streamly.Internal.Data.SVar (MonadAsync, defState, Rate)
567import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)
568import Streamly.Internal.Data.Stream.Prelude
569 (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)
570import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)
571import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM))
572import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)
573import Streamly.Internal.Data.Stream.Zip (ZipSerialM)
574import Streamly.Internal.Data.Pipe.Types (Pipe (..))
575import Streamly.Internal.Data.Time.Units
576 (AbsTime, MilliSecond64(..), addToAbsTime, toRelTime,
577 toAbsTime, TimeUnit64, RelTime64, addToAbsTime64)
578import Streamly.Internal.Mutable.Prim.Var (Prim, Var)
580import Streamly.Internal.Data.Strict
582import qualified Streamly.Internal.Memory.Array as A
583import qualified Streamly.Data.Fold as FL
584import qualified Streamly.Internal.Data.Fold.Types as FL
585import qualified Streamly.Internal.Data.Stream.Prelude as P
586import qualified Streamly.Internal.Data.Stream.StreamK as K
587import qualified Streamly.Internal.Data.Stream.StreamD as D
589#ifdef USE_STREAMK_ONLY
590import qualified Streamly.Internal.Data.Stream.StreamK as S
592import qualified Streamly.Internal.Data.Stream.StreamD as S
595-- import qualified Streamly.Internal.Data.Stream.Async as Async
596import qualified Streamly.Internal.Data.Stream.Serial as Serial
597import qualified Streamly.Internal.Data.Stream.Parallel as Par
598import qualified Streamly.Internal.Data.Stream.Zip as Z
599import qualified Streamly.Internal.Data.Parser.ParserK.Types as PRK
600import qualified Streamly.Internal.Data.Parser.ParserD as PRD
601import qualified Streamly.Internal.Data.Zipper as ZR
603------------------------------------------------------------------------------
605------------------------------------------------------------------------------
607-- | Decompose a stream into its head and tail. If the stream is empty, returns
608-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is
609-- the head of the stream and @ma@ its tail.
611-- This is a brute force primitive. Avoid using it as long as possible, use it
612-- when no other combinator can do the job. This can be used to do pretty much
613-- anything in an imperative manner, as it just breaks down the stream into
614-- individual elements and we can loop over them as we deem fit. For example,
615-- this can be used to convert a streamly stream into other stream types.
619uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))
620uncons m = K.uncons (K.adapt m)
622------------------------------------------------------------------------------
623-- Generation by Unfolding
624------------------------------------------------------------------------------
631-- Just (a, b) -> a \`cons` unfoldr step b
634-- Build a stream by unfolding a /pure/ step function @step@ starting from a
635-- seed @s@. The step function returns the next element in the stream and the
636-- next seed value. When it is done it returns 'Nothing' and the stream ends.
643-- else Just (b, b + 1)
644-- in toList $ unfoldr f 0
651{-# INLINE_EARLY unfoldr #-}
652unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a
653unfoldr step seed = fromStreamS (S.unfoldr step seed)
654{-# RULES "unfoldr fallback to StreamK" [1]
655 forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}
657-- | Build a stream by unfolding a /monadic/ step function starting from a
658-- seed. The step function returns the next element in the stream and the next
659-- seed value. When it is done it returns 'Nothing' and the stream ends. For
665-- then return Nothing
666-- else print b >> return (Just (b, b + 1))
667-- in drain $ unfoldrM f 0
675-- When run concurrently, the next unfold step can run concurrently with the
676-- processing of the output of the previous step. Note that more than one step
677-- cannot run concurrently as the next step depends on the output of the
681-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)
682-- & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
688{-# INLINE_EARLY unfoldrM #-}
689unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
692{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}
693{-# INLINE_EARLY unfoldrMSerial #-}
694unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
695unfoldrMSerial = Serial.unfoldrM
697{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}
698{-# INLINE_EARLY unfoldrMWSerial #-}
699unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a
700unfoldrMWSerial = Serial.unfoldrM
702{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}
703{-# INLINE_EARLY unfoldrMZipSerial #-}
704unfoldrMZipSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a
705unfoldrMZipSerial = Serial.unfoldrM
707-- | Convert an 'Unfold' into a stream by supplying it an input seed.
709-- >>> unfold (UF.replicateM 10) (putStrLn "hello")
713unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b
714unfold unf x = fromStreamD $ D.unfold unf x
716------------------------------------------------------------------------------
717-- Specialized Generation
718------------------------------------------------------------------------------
720-- Faster than yieldM because there is no bind.
724-- yield a = a \`cons` nil
727-- Create a singleton stream from a pure value.
729-- The following holds in monadic streams, but not in Zip streams:
733-- yield = yieldM . pure
736-- In Zip applicative streams 'yield' is not the same as 'pure' because in that
737-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are
738-- equally efficient, in other cases 'yield' may be slightly more efficient
739-- than the other equivalent definitions.
743yield :: IsStream t => a -> t m a
748-- yieldM m = m \`consM` nil
751-- Create a singleton stream from a monadic action.
754-- > toList $ yieldM getLine
761yieldM :: (Monad m, IsStream t) => m a -> t m a
766-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0
769-- Generate an infinite stream, whose values are the output of a function @f@
770-- applied on the corresponding index. Index starts at 0.
773-- > S.toList $ S.take 5 $ S.fromIndices id
778{-# INLINE fromIndices #-}
779fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a
780fromIndices = fromStreamS . S.fromIndices
785-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0
788-- Generate an infinite stream, whose values are the output of a monadic
789-- function @f@ applied on the corresponding index. Index starts at 0.
794{-# INLINE_EARLY fromIndicesM #-}
795fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a
796fromIndicesM = K.fromIndicesM
798{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}
799{-# INLINE fromIndicesMSerial #-}
800fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a
801fromIndicesMSerial = fromStreamS . S.fromIndicesM
805-- replicateM = take n . repeatM
808-- Generate a stream by performing a monadic action @n@ times. Same as:
811-- drain $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
812-- drain $ asyncly $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)
818{-# INLINE_EARLY replicateM #-}
819replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a
820replicateM = K.replicateM
822{-# RULES "replicateM serial" replicateM = replicateMSerial #-}
823{-# INLINE replicateMSerial #-}
824replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a
825replicateMSerial n = fromStreamS . S.replicateM n
829-- replicate = take n . repeat
832-- Generate a stream of length @n@ by repeating a value @n@ times.
835{-# INLINE_NORMAL replicate #-}
836replicate :: (IsStream t, Monad m) => Int -> a -> t m a
837replicate n = fromStreamS . S.replicate n
840-- Generate an infinite stream by repeating a pure value.
843{-# INLINE_NORMAL repeat #-}
844repeat :: (IsStream t, Monad m) => a -> t m a
845repeat = fromStreamS . S.repeat
849-- repeatM = fix . consM
850-- repeatM = cycle1 . yieldM
853-- Generate a stream by repeatedly executing a monadic action forever.
856-- drain $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
857-- drain $ asyncly $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)
860-- /Concurrent, infinite (do not use with 'parallely')/
863{-# INLINE_EARLY repeatM #-}
864repeatM :: (IsStream t, MonadAsync m) => m a -> t m a
867{-# RULES "repeatM serial" repeatM = repeatMSerial #-}
868{-# INLINE repeatMSerial #-}
869repeatMSerial :: MonadAsync m => m a -> SerialT m a
870repeatMSerial = fromStreamS . S.repeatM
874-- iterate f x = x \`cons` iterate f x
877-- Generate an infinite stream with @x@ as the first element and each
878-- successive element derived by applying the function @f@ on the previous
882-- > S.toList $ S.take 5 $ S.iterate (+1) 1
887{-# INLINE_NORMAL iterate #-}
888iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a
889iterate step = fromStreamS . S.iterate step
893-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)
896-- Generate an infinite stream with the first element generated by the action
897-- @m@ and each successive element derived by applying the monadic function
898-- @f@ on the previous element.
900-- When run concurrently, the next iteration can run concurrently with the
901-- processing of the previous iteration. Note that more than one iteration
902-- cannot run concurrently as the next iteration depends on the output of the
903-- previous iteration.
906-- drain $ serially $ S.take 10 $ S.iterateM
907-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
909-- drain $ asyncly $ S.take 10 $ S.iterateM
910-- (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)
915-- /Since: 0.7.0 (signature change)/
918{-# INLINE_EARLY iterateM #-}
919iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a
922{-# RULES "iterateM serial" iterateM = iterateMSerial #-}
923{-# INLINE iterateMSerial #-}
924iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a
925iterateMSerial step = fromStreamS . S.iterateM step
927------------------------------------------------------------------------------
929------------------------------------------------------------------------------
933-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'
936-- Construct a stream from a list of monadic actions. This is more efficient
937-- than 'fromFoldableM' for serial streams.
940{-# INLINE_EARLY fromListM #-}
941fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a
942fromListM = fromStreamD . D.fromListM
943{-# RULES "fromListM fallback to StreamK" [1]
944 forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}
948-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'
951-- Construct a stream from a 'Foldable' containing monadic actions.
954-- drain $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
955-- drain $ asyncly $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)
958-- /Concurrent (do not use with 'parallely' on infinite containers)/
961{-# INLINE fromFoldableM #-}
962fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a
963fromFoldableM = Prelude.foldr consM K.nil
965-- | Same as 'fromFoldable'.
968{-# DEPRECATED each "Please use fromFoldable instead." #-}
970each :: (IsStream t, Foldable f) => f a -> t m a
973-- | Read lines from an IO Handle into a stream of Strings.
976{-# DEPRECATED fromHandle
977 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
978fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
981 go = K.mkStream $ \_ yld _ stp -> do
982 eof <- liftIO $ IO.hIsEOF h
986 str <- liftIO $ IO.hGetLine h
989-- | Construct a stream by reading a 'Prim' 'Var' repeatedly.
993{-# INLINE fromPrimVar #-}
994fromPrimVar :: (IsStream t, MonadIO m, Prim a) => Var IO a -> t m a
995fromPrimVar = fromStreamD . D.fromPrimVar
997-- | Takes a callback setter function and provides it with a callback. The
998-- callback when invoked adds a value at the tail of the stream. Returns a
999-- stream of values generated by the callback.
1003{-# INLINE fromCallback #-}
1004fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
1005fromCallback setCallback = concatM $ do
1006 (callback, stream) <- D.newCallbackStream
1007 setCallback callback
1010------------------------------------------------------------------------------
1012------------------------------------------------------------------------------
1014-- | @timesWith g@ returns a stream of time value tuples. The first component
1015-- of the tuple is an absolute time reference (epoch) denoting the start of the
1016-- stream and the second component is a time relative to the reference.
1018-- The argument @g@ specifies the granularity of the relative time in seconds.
1019-- A lower granularity clock gives higher precision but is more expensive in
1020-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
1023-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.timesWith 0.01
1024-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))
1025-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))
1026-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))
1029-- Note: This API is not safe on 32-bit machines.
1033{-# INLINE timesWith #-}
1034timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)
1035timesWith g = fromStreamD $ D.times g
1037-- | @times@ returns a stream of time value tuples with clock of 10 ms
1038-- granularity. The first component of the tuple is an absolute time reference
1039-- (epoch) denoting the start of the stream and the second component is a time
1040-- relative to the reference.
1043-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.times
1044-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))
1045-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))
1046-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))
1049-- Note: This API is not safe on 32-bit machines.
1054times :: (IsStream t, MonadAsync m) => t m (AbsTime, RelTime64)
1055times = timesWith 0.01
1057-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
1058-- granularity @g@ specified in seconds. A low granularity clock is more
1059-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
1063-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimesWith 0.01
1066-- Note: This API is not safe on 32-bit machines.
1070{-# INLINE absTimesWith #-}
1071absTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
1072 => Double -> t m AbsTime
1073absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
1075-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
1079-- >>> S.mapM_ print $ S.delayPre 1 $ S.absTimes
1082-- Note: This API is not safe on 32-bit machines.
1086{-# INLINE absTimes #-}
1087absTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m AbsTime
1088absTimes = fmap (uncurry addToAbsTime64) times
1090{-# DEPRECATED currentTime "Please use absTimes instead" #-}
1091{-# INLINE currentTime #-}
1092currentTime :: (IsStream t, MonadAsync m, Functor (t m))
1093 => Double -> t m AbsTime
1094currentTime = absTimesWith
1096-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
1097-- using a clock of granularity @g@ specified in seconds. A low granularity
1098-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
1099-- ms is treated as 1 ms.
1102-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimesWith 0.01
1103-- > RelTime64 (NanoSecond64 0)
1104-- > RelTime64 (NanoSecond64 91139000)
1105-- > RelTime64 (NanoSecond64 204052000)
1108-- Note: This API is not safe on 32-bit machines.
1112{-# INLINE relTimesWith #-}
1113relTimesWith :: (IsStream t, MonadAsync m, Functor (t m))
1114 => Double -> t m RelTime64
1115relTimesWith = fmap snd . timesWith
1117-- | @relTimes@ returns a stream of relative time values starting from 0,
1118-- using a clock of granularity 10 ms.
1121-- >>> S.mapM_ print $ S.delayPre 1 $ S.relTimes
1124-- Note: This API is not safe on 32-bit machines.
1128{-# INLINE relTimes #-}
1129relTimes :: (IsStream t, MonadAsync m, Functor (t m)) => t m RelTime64
1130relTimes = fmap snd times
1132-- | @durations g@ returns a stream of relative time values measuring the time
1133-- elapsed since the immediate predecessor element of the stream was generated.
1134-- The first element of the stream is always 0. @durations@ uses a clock of
1135-- granularity @g@ specified in seconds. A low granularity clock is more
1136-- expensive in terms of CPU usage. The minimum granularity is 1 millisecond.
1137-- Durations lower than 1 ms will be 0.
1139-- Note: This API is not safe on 32-bit machines.
1143{-# INLINE durations #-}
1144durations :: -- (IsStream t, MonadAsync m) =>
1145 Double -> t m RelTime64
1146durations = undefined
1148-- | Generate ticks at the specified rate. The rate is adaptive, the tick
1149-- generation speed can be increased or decreased at different times to achieve
1150-- the specified rate. The specific behavior for different styles of 'Rate'
1151-- specifications is documented under 'Rate'. The effective maximum rate
1152-- achieved by a stream is governed by the processor speed.
1157ticks :: -- (IsStream t, MonadAsync m) =>
1161-- | Generate a singleton event at or after the specified absolute time. Note
1162-- that this is different from a threadDelay, a threadDelay starts from the
1163-- time when the action is evaluated, whereas if we use AbsTime based timeout
1164-- it will immediately expire if the action is evaluated too late.
1168{-# INLINE timeout #-}
1169timeout :: -- (IsStream t, MonadAsync m) =>
1173------------------------------------------------------------------------------
1174-- Elimination by Folding
1175------------------------------------------------------------------------------
1177-- | Right associative/lazy pull fold. @foldrM build final stream@ constructs
1178-- an output structure using the step function @build@. @build@ is invoked with
1179-- the next input element and the remaining (lazy) tail of the output
1180-- structure. It builds a lazy output expression using the two. When the "tail
1181-- structure" in the output expression is evaluated it calls @build@ again thus
1182-- lazily consuming the input @stream@ until either the output expression built
1183-- by @build@ is free of the "tail" or the input is exhausted in which case
1184-- @final@ is used as the terminating case for the output structure. For more
1185-- details see the description in the previous section.
1187-- Example, determine if any element is 'odd' in a stream:
1189-- >>> S.foldrM (\x xs -> if odd x then return True else xs) (return False) $ S.fromList (2:4:5:undefined)
1192-- /Since: 0.7.0 (signature changed)/
1194-- /Since: 0.2.0 (signature changed)/
1197{-# INLINE foldrM #-}
1198foldrM :: Monad m => (a -> m b -> m b) -> m b -> SerialT m a -> m b
1201-- | Right fold to a streaming monad.
1203-- > foldrS S.cons S.nil === id
1205-- 'foldrS' can be used to perform stateless stream to stream transformations
1206-- like map and filter in general. It can be coupled with a scan to perform
1207-- stateful transformations. However, note that the custom map and filter
1208-- routines can be much more efficient than this due to better stream fusion.
1210-- >>> S.toList $ S.foldrS S.cons S.nil $ S.fromList [1..5]
1213-- Find if any element in the stream is 'True':
1215-- >>> 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)
1218-- Map (+2) on odd elements and filter out the even elements:
1220-- >>> 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)
1223-- 'foldrM' can also be represented in terms of 'foldrS', however, the former
1224-- is much more efficient:
1226-- > foldrM f z s = runIdentityT $ foldrS (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1229{-# INLINE foldrS #-}
1230foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
1233-- | Right fold to a transformer monad. This is the most general right fold
1234-- function. 'foldrS' is a special case of 'foldrT', however 'foldrS'
1235-- implementation can be more efficient:
1238-- > foldrM f z s = runIdentityT $ foldrT (\x xs -> lift $ f x (runIdentityT xs)) (lift z) s
1240-- 'foldrT' can be used to translate streamly streams to other transformer
1241-- monads e.g. to a different streaming type.
1244{-# INLINE foldrT #-}
1245foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s)
1246 => (a -> s m b -> s m b) -> s m b -> t m a -> s m b
1247foldrT f z s = S.foldrT f z (toStreamS s)
1249-- | Right fold, lazy for lazy monads and pure streams, and strict for strict
1252-- Please avoid using this routine in strict monads like IO unless you need a
1253-- strict right fold. This is provided only for use in lazy monads (e.g.
1254-- Identity) or pure streams. Note that with this signature it is not possible
1255-- to implement a lazy foldr when the monad @m@ is strict. In that case it
1256-- would be strict in its accumulator and therefore would necessarily consume
1261foldr :: Monad m => (a -> b -> b) -> b -> SerialT m a -> m b
1264-- XXX This seems to be of limited use as it cannot be used to construct
1265-- recursive structures and for reduction foldl1' is better.
1267-- | Lazy right fold for non-empty streams, using first element as the starting
1268-- value. Returns 'Nothing' if the stream is empty.
1271{-# INLINE foldr1 #-}
1272{-# DEPRECATED foldr1 "Use foldrM instead." #-}
1273foldr1 :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1274foldr1 f m = S.foldr1 f (toStreamS m)
1276-- | Strict left fold with an extraction function. Like the standard strict
1277-- left fold, but applies a user supplied extraction function (the third
1278-- argument) to the folded value at the end. This is designed to work with the
1279-- @foldl@ library. The suffix @x@ is a mnemonic for extraction.
1282{-# DEPRECATED foldx "Please use foldl' followed by fmap instead." #-}
1284foldx :: Monad m => (x -> a -> x) -> x -> (x -> b) -> SerialT m a -> m b
1287-- | Left associative/strict push fold. @foldl' reduce initial stream@ invokes
1288-- @reduce@ with the accumulator and the next input in the input stream, using
1289-- @initial@ as the initial value of the current value of the accumulator. When
1290-- the input is exhausted the current value of the accumulator is returned.
1291-- Make sure to use a strict data structure for accumulator to not build
1292-- unnecessary lazy expressions unless that's what you want. See the previous
1293-- section for more details.
1296{-# INLINE foldl' #-}
1297foldl' :: Monad m => (b -> a -> b) -> b -> SerialT m a -> m b
1300-- | Strict left fold, for non-empty streams, using first element as the
1301-- starting value. Returns 'Nothing' if the stream is empty.
1304{-# INLINE foldl1' #-}
1305foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
1309 Nothing -> return Nothing
1311 res <- foldl' step h t
1314-- | Like 'foldx', but with a monadic step function.
1317{-# DEPRECATED foldxM "Please use foldlM' followed by fmap instead." #-}
1318{-# INLINE foldxM #-}
1319foldxM :: Monad m => (x -> a -> m x) -> m x -> (x -> m b) -> SerialT m a -> m b
1322-- | Like 'foldl'' but with a monadic step function.
1325{-# INLINE foldlM' #-}
1326foldlM' :: Monad m => (b -> a -> m b) -> b -> SerialT m a -> m b
1327foldlM' step begin m = S.foldlM' step begin $ toStreamS m
1329------------------------------------------------------------------------------
1331------------------------------------------------------------------------------
1333-- | Fold a stream using the supplied left fold.
1335-- >>> S.fold FL.sum (S.enumerateFromTo 1 100)
1340fold :: Monad m => Fold m a b -> SerialT m a -> m b
1343------------------------------------------------------------------------------
1345------------------------------------------------------------------------------
1348-- | Drain a stream to a 'Sink'.
1349{-# INLINE runSink #-}
1350runSink :: Monad m => Sink m a -> SerialT m a -> m ()
1351runSink = fold . toFold
1354------------------------------------------------------------------------------
1356------------------------------------------------------------------------------
1358-- | Parse a stream using the supplied 'Parser'.
1362{-# INLINE_NORMAL parseD #-}
1363parseD :: MonadThrow m => PRD.Parser m a b -> SerialT m a -> m b
1364parseD (PRD.Parser step initial extract) = P.parselMx' step initial extract
1366{-# INLINE parseK #-}
1367parseK :: MonadThrow m => PRK.Parser m a b -> SerialT m a -> m b
1368parseK parser xs = do
1369 r <- PRK.runParser parser (ZR.fromStream $ K.toStream xs)
1370 (\(_, b) -> return b)
1372 Left err -> throwM (PRD.ParseError err)
1375-- | Parse a stream using the supplied 'Parser'.
1379{-# INLINE [3] parse #-}
1380parse :: MonadThrow m => Parser m a b -> SerialT m a -> m b
1381parse = parseD . PRD.fromParserK
1382{-# RULES "parse fallback to CPS" [2]
1383 forall a. parseD (PRD.fromParserK a) = parseK a #-}
1385------------------------------------------------------------------------------
1387------------------------------------------------------------------------------
1390-- > drain = mapM_ (\_ -> return ())
1392-- Run a stream, discarding the results. By default it interprets the stream
1393-- as 'SerialT', to run other types of streams use the type adapting
1394-- combinators for example @drain . 'asyncly'@.
1398drain :: Monad m => SerialT m a -> m ()
1401-- | Run a stream, discarding the results. By default it interprets the stream
1402-- as 'SerialT', to run other types of streams use the type adapting
1403-- combinators for example @runStream . 'asyncly'@.
1406{-# DEPRECATED runStream "Please use \"drain\" instead" #-}
1407{-# INLINE runStream #-}
1408runStream :: Monad m => SerialT m a -> m ()
1412-- > drainN n = drain . take n
1414-- Run maximum up to @n@ iterations of a stream.
1417{-# INLINE drainN #-}
1418drainN :: Monad m => Int -> SerialT m a -> m ()
1419drainN n = drain . take n
1422-- > runN n = runStream . take n
1424-- Run maximum up to @n@ iterations of a stream.
1427{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
1429runN :: Monad m => Int -> SerialT m a -> m ()
1433-- > drainWhile p = drain . takeWhile p
1435-- Run a stream as long as the predicate holds true.
1438{-# INLINE drainWhile #-}
1439drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1440drainWhile p = drain . takeWhile p
1443-- > runWhile p = runStream . takeWhile p
1445-- Run a stream as long as the predicate holds true.
1448{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
1449{-# INLINE runWhile #-}
1450runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
1451runWhile = drainWhile
1453-- | Determine whether the stream is empty.
1457null :: Monad m => SerialT m a -> m Bool
1458null = S.null . toStreamS
1460-- | Extract the first element of the stream, if any.
1466head :: Monad m => SerialT m a -> m (Maybe a)
1467head = S.head . toStreamS
1469-- | Extract the first element of the stream, if any, otherwise use the
1470-- supplied default value. It can help avoid one branch in high performance
1474{-# INLINE headElse #-}
1475headElse :: Monad m => a -> SerialT m a -> m a
1476headElse x = D.headElse x . toStreamD
1479-- > tail = fmap (fmap snd) . uncons
1481-- Extract all but the first element of the stream, if any.
1485tail :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1486tail m = K.tail (K.adapt m)
1488-- | Extract all but the last element of the stream, if any.
1492init :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (t m a))
1493init m = K.init (K.adapt m)
1495-- | Extract the last element of the stream, if any.
1497-- > last xs = xs !! (length xs - 1)
1501last :: Monad m => SerialT m a -> m (Maybe a)
1502last m = S.last $ toStreamS m
1504-- | Determine whether an element is present in the stream.
1508elem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1509elem e m = S.elem e (toStreamS m)
1511-- | Determine whether an element is not present in the stream.
1514{-# INLINE notElem #-}
1515notElem :: (Monad m, Eq a) => a -> SerialT m a -> m Bool
1516notElem e m = S.notElem e (toStreamS m)
1518-- | Determine the length of the stream.
1521{-# INLINE length #-}
1522length :: Monad m => SerialT m a -> m Int
1523length = foldl' (\n _ -> n + 1) 0
1525-- | Determine whether all elements of a stream satisfy a predicate.
1529all :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1530all p m = S.all p (toStreamS m)
1532-- | Determine whether any of the elements of a stream satisfy a predicate.
1536any :: Monad m => (a -> Bool) -> SerialT m a -> m Bool
1537any p m = S.any p (toStreamS m)
1539-- | Determines if all elements of a boolean stream are True.
1543and :: Monad m => SerialT m Bool -> m Bool
1546-- | Determines whether at least one element of a boolean stream is True.
1550or :: Monad m => SerialT m Bool -> m Bool
1553-- | Determine the sum of all elements of a stream of numbers. Returns @0@ when
1554-- the stream is empty. Note that this is not numerically stable for floating
1559sum :: (Monad m, Num a) => SerialT m a -> m a
1562-- | Determine the product of all elements of a stream of numbers. Returns @1@
1563-- when the stream is empty.
1566{-# INLINE product #-}
1567product :: (Monad m, Num a) => SerialT m a -> m a
1568product = foldl' (*) 1
1570-- | Fold a stream of monoid elements by appending them.
1573{-# INLINE mconcat #-}
1574mconcat :: (Monad m, Monoid a) => SerialT m a -> m a
1575mconcat = foldr mappend mempty
1579-- minimum = 'minimumBy' compare
1582-- Determine the minimum element in a stream.
1585{-# INLINE minimum #-}
1586minimum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1587minimum m = S.minimum (toStreamS m)
1589-- | Determine the minimum element in a stream using the supplied comparison
1593{-# INLINE minimumBy #-}
1594minimumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1595minimumBy cmp m = S.minimumBy cmp (toStreamS m)
1599-- maximum = 'maximumBy' compare
1602-- Determine the maximum element in a stream.
1605{-# INLINE maximum #-}
1606maximum :: (Monad m, Ord a) => SerialT m a -> m (Maybe a)
1609-- | Determine the maximum element in a stream using the supplied comparison
1613{-# INLINE maximumBy #-}
1614maximumBy :: Monad m => (a -> a -> Ordering) -> SerialT m a -> m (Maybe a)
1615maximumBy cmp m = S.maximumBy cmp (toStreamS m)
1617-- | Lookup the element at the given index.
1621(!!) :: Monad m => SerialT m a -> Int -> m (Maybe a)
1622m !! i = toStreamS m S.!! i
1624-- | In a stream of (key-value) pairs @(a, b)@, return the value @b@ of the
1625-- first pair where the key equals the given value @a@.
1627-- > lookup = snd <$> find ((==) . fst)
1630{-# INLINE lookup #-}
1631lookup :: (Monad m, Eq a) => a -> SerialT m (a, b) -> m (Maybe b)
1632lookup a m = S.lookup a (toStreamS m)
1634-- | Like 'findM' but with a non-monadic predicate.
1636-- > find p = findM (return . p)
1640find :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe a)
1641find p m = S.find p (toStreamS m)
1643-- | Returns the first element that satisfies the given predicate.
1647findM :: Monad m => (a -> m Bool) -> SerialT m a -> m (Maybe a)
1648findM p m = S.findM p (toStreamS m)
1650-- | Find all the indices where the element in the stream satisfies the given
1654{-# INLINE findIndices #-}
1655findIndices :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m Int
1656findIndices p m = fromStreamS $ S.findIndices p (toStreamS m)
1658-- | Returns the first index that satisfies the given predicate.
1661{-# INLINE findIndex #-}
1662findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
1663findIndex p = head . findIndices p
1665-- | Find all the indices where the value of the element in the stream is equal
1666-- to the given value.
1669{-# INLINE elemIndices #-}
1670elemIndices :: (IsStream t, Eq a, Monad m) => a -> t m a -> t m Int
1671elemIndices a = findIndices (==a)
1673-- | Returns the first index where a given value is found in the stream.
1675-- > elemIndex a = findIndex (== a)
1678{-# INLINE elemIndex #-}
1679elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
1680elemIndex a = findIndex (== a)
1682------------------------------------------------------------------------------
1684------------------------------------------------------------------------------
1686-- | Returns 'True' if the first stream is the same as or a prefix of the
1687-- second. A stream is a prefix of itself.
1690-- > S.isPrefixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1695{-# INLINE isPrefixOf #-}
1696isPrefixOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1697isPrefixOf m1 m2 = D.isPrefixOf (toStreamD m1) (toStreamD m2)
1699-- Note: isPrefixOf uses the prefix stream only once. In contrast, isSuffixOf
1700-- may use the suffix stream many times. To run in optimal memory we do not
1701-- want to buffer the suffix stream in memory therefore we need an ability to
1702-- clone (or consume it multiple times) the suffix stream without any side
1703-- effects so that multiple potential suffix matches can proceed in parallel
1704-- without buffering the suffix stream. For example, we may create the suffix
1705-- stream from a file handle, however, if we evaluate the stream multiple
1706-- times, once for each match, we will need a different file handle each time
1707-- which may exhaust the file descriptors. Instead, we want to share the same
1708-- underlying file descriptor, use pread on it to generate the stream and clone
1709-- the stream for each match. Therefore the suffix stream should be built in
1710-- such a way that it can be consumed multiple times without any problems.
1712-- XXX Can be implemented with better space/time complexity.
1713-- Space: @O(n)@ worst case where @n@ is the length of the suffix.
1715-- | Returns 'True' if the first stream is a suffix of the second. A stream is
1716-- considered a suffix of itself.
1719-- > S.isSuffixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1723-- Space: @O(n)@, buffers entire input stream and the suffix.
1727-- /Suboptimal/ - Help wanted.
1729{-# INLINE isSuffixOf #-}
1730isSuffixOf :: (Monad m, Eq a) => SerialT m a -> SerialT m a -> m Bool
1731isSuffixOf suffix stream = reverse suffix `isPrefixOf` reverse stream
1733-- | Returns 'True' if the first stream is an infix of the second. A stream is
1734-- considered an infix of itself.
1737-- > S.isInfixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
1741-- Space: @O(n)@ worst case where @n@ is the length of the infix.
1745-- /Requires 'Storable' constraint/ - Help wanted.
1747{-# INLINE isInfixOf #-}
1748isInfixOf :: (MonadIO m, Eq a, Enum a, Storable a)
1749 => SerialT m a -> SerialT m a -> m Bool
1750isInfixOf infx stream = do
1751 arr <- fold A.write infx
1752 -- XXX can use breakOnSeq instead (when available)
1753 r <- null $ drop 1 $ splitOnSeq arr FL.drain stream
1756-- | Returns 'True' if all the elements of the first stream occur, in order, in
1757-- the second stream. The elements do not have to occur consecutively. A stream
1758-- is a subsequence of itself.
1761-- > S.isSubsequenceOf (S.fromList "hlo") (S.fromList "hello" :: SerialT IO Char)
1766{-# INLINE isSubsequenceOf #-}
1767isSubsequenceOf :: (Eq a, IsStream t, Monad m) => t m a -> t m a -> m Bool
1768isSubsequenceOf m1 m2 = D.isSubsequenceOf (toStreamD m1) (toStreamD m2)
1770-- | Strip prefix if present and tell whether it was stripped or not. Returns
1771-- 'Nothing' if the stream does not start with the given prefix, stripped
1772-- stream otherwise. Returns @Just nil@ when the prefix is the same as the
1778{-# INLINE stripPrefix #-}
1780 :: (Eq a, IsStream t, Monad m)
1781 => t m a -> t m a -> m (Maybe (t m a))
1782stripPrefix m1 m2 = fmap fromStreamD <$>
1783 D.stripPrefix (toStreamD m1) (toStreamD m2)
1785-- Note: If we want to return a Maybe value to know whether the
1786-- suffix/infix was present or not along with the stripped stream then
1787-- we need to buffer the whole input stream.
1789-- | Drops the given suffix from a stream. Returns 'Nothing' if the stream does
1790-- not end with the given suffix. Returns @Just nil@ when the suffix is the
1791-- same as the stream.
1793-- It may be more efficient to convert the stream to an Array and use
1794-- stripSuffix on that especially if the elements have a Storable or Prim
1797-- Space: @O(n)@, buffers the entire input stream as well as the suffix
1800{-# INLINE stripSuffix #-}
1803 => SerialT m a -> SerialT m a -> m (Maybe (SerialT m a))
1804stripSuffix m1 m2 = fmap reverse <$> stripPrefix (reverse m1) (reverse m2)
1806-- | Drop prefix from the input stream if present.
1810-- /Unimplemented/ - Help wanted.
1811{-# INLINE dropPrefix #-}
1813 -- (Eq a, IsStream t, Monad m) =>
1814 t m a -> t m a -> t m a
1815dropPrefix = error "Not implemented yet!"
1817-- | Drop all matching infix from the input stream if present. Infix stream
1818-- may be consumed multiple times.
1820-- Space: @O(n)@ where n is the length of the infix.
1822-- /Unimplemented/ - Help wanted.
1823{-# INLINE dropInfix #-}
1825 -- (Eq a, IsStream t, Monad m) =>
1826 t m a -> t m a -> t m a
1827dropInfix = error "Not implemented yet!"
1829-- | Drop suffix from the input stream if present. Suffix stream may be
1830-- consumed multiple times.
1832-- Space: @O(n)@ where n is the length of the suffix.
1834-- /Unimplemented/ - Help wanted.
1835{-# INLINE dropSuffix #-}
1837 -- (Eq a, IsStream t, Monad m) =>
1838 t m a -> t m a -> t m a
1839dropSuffix = error "Not implemented yet!"
1841------------------------------------------------------------------------------
1843------------------------------------------------------------------------------
1845-- XXX this can utilize parallel mapping if we implement it as drain . mapM
1847-- > mapM_ = drain . mapM
1849-- Apply a monadic action to each element of the stream and discard the output
1850-- of the action. This is not really a pure transformation operation but a
1851-- transformation followed by fold.
1855mapM_ :: Monad m => (a -> m b) -> SerialT m a -> m ()
1856mapM_ f m = S.mapM_ f $ toStreamS m
1858------------------------------------------------------------------------------
1860------------------------------------------------------------------------------
1864-- toList = S.foldr (:) []
1867-- Convert a stream into a list in the underlying monad. The list can be
1868-- consumed lazily in a lazy monad (e.g. 'Identity'). In a strict monad (e.g.
1869-- IO) the whole list is generated and buffered before it can be consumed.
1871-- /Warning!/ working on large lists accumulated as buffers in memory could be
1872-- very inefficient, consider using "Streamly.Array" instead.
1875{-# INLINE toList #-}
1876toList :: Monad m => SerialT m a -> m [a]
1881-- toListRev = S.foldl' (flip (:)) []
1884-- Convert a stream into a list in reverse order in the underlying monad.
1886-- /Warning!/ working on large lists accumulated as buffers in memory could be
1887-- very inefficient, consider using "Streamly.Array" instead.
1890{-# INLINE toListRev #-}
1891toListRev :: Monad m => SerialT m a -> m [a]
1892toListRev = D.toListRev . toStreamD
1896-- toHandle h = S.mapM_ $ hPutStrLn h
1899-- Write a stream of Strings to an IO Handle.
1902{-# DEPRECATED toHandle
1903 "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}
1904toHandle :: MonadIO m => IO.Handle -> SerialT m String -> m ()
1908 let stop = return ()
1909 single a = liftIO (IO.hPutStrLn h a)
1910 yieldk a r = liftIO (IO.hPutStrLn h a) >> go r
1911 in K.foldStream defState yieldk single stop m1
1913-- XXX rename these to write/writeRev to make the naming consistent with folds
1916-- | A fold that buffers its input to a pure stream.
1918-- /Warning!/ working on large streams accumulated as buffers in memory could
1919-- be very inefficient, consider using "Streamly.Array" instead.
1922{-# INLINE toStream #-}
1923toStream :: Monad m => Fold m a (SerialT Identity a)
1924toStream = Fold (\f x -> return $ f . (x `K.cons`))
1926 (return . ($ K.nil))
1928-- This is more efficient than 'toStream'. toStream is exactly the same as
1929-- reversing the stream after toStreamRev.
1931-- | Buffers the input stream to a pure stream in the reverse order of the
1934-- /Warning!/ working on large streams accumulated as buffers in memory could
1935-- be very inefficient, consider using "Streamly.Array" instead.
1939-- xn : ... : x2 : x1 : []
1940{-# INLINABLE toStreamRev #-}
1941toStreamRev :: Monad m => Fold m a (SerialT Identity a)
1942toStreamRev = Fold (\xs x -> return $ x `K.cons` xs) (return K.nil) return
1944-- | Convert a stream to a pure stream.
1947-- toPure = foldr cons nil
1952{-# INLINE toPure #-}
1953toPure :: Monad m => SerialT m a -> m (SerialT Identity a)
1954toPure = foldr K.cons K.nil
1956-- | Convert a stream to a pure stream in reverse order.
1959-- toPureRev = foldl' (flip cons) nil
1964{-# INLINE toPureRev #-}
1965toPureRev :: Monad m => SerialT m a -> m (SerialT Identity a)
1966toPureRev = foldl' (flip K.cons) K.nil
1968------------------------------------------------------------------------------
1969-- Concurrent Application
1970------------------------------------------------------------------------------
1978-- | Parallel transform application operator; applies a stream transformation
1979-- function @t m a -> t m b@ to a stream @t m a@ concurrently; the input stream
1980-- is evaluated asynchronously in an independent thread yielding elements to a
1981-- buffer and the transformation function runs in another thread consuming the
1982-- input from the buffer. '|$' is just like regular function application
1983-- operator '$' except that it is concurrent.
1985-- If you read the signature as @(t m a -> t m b) -> (t m a -> t m b)@ you can
1986-- look at it as a transformation that converts a transform function to a
1987-- buffered concurrent transform function.
1989-- The following code prints a value every second even though each stage adds a
1995-- S.mapM (\\x -> threadDelay 1000000 >> print x)
1996-- |$ S.repeatM (threadDelay 1000000 >> return 1)
2003(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> (t m a -> t m b)
2004-- (|$) f = f . Async.mkAsync
2005(|$) f = f . D.mkParallel
2011{-# INLINE applyAsync #-}
2012applyAsync :: (IsStream t, MonadAsync m)
2013 => (t m a -> t m b) -> (t m a -> t m b)
2016-- | Parallel reverse function application operator for streams; just like the
2017-- regular reverse function application operator '&' except that it is
2022-- S.repeatM (threadDelay 1000000 >> return 1)
2023-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
2030(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
2033-- | Parallel fold application operator; applies a fold function @t m a -> m b@
2034-- to a stream @t m a@ concurrently; The the input stream is evaluated
2035-- asynchronously in an independent thread yielding elements to a buffer and
2036-- the folding action runs in another thread consuming the input from the
2039-- If you read the signature as @(t m a -> m b) -> (t m a -> m b)@ you can look
2040-- at it as a transformation that converts a fold function to a buffered
2041-- concurrent fold function.
2043-- The @.@ at the end of the operator is a mnemonic for termination of the
2047-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
2048-- |$. S.repeatM (threadDelay 1000000 >> return 1)
2055(|$.) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
2056-- (|$.) f = f . Async.mkAsync
2057(|$.) f = f . D.mkParallel
2063{-# INLINE foldAsync #-}
2064foldAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> (t m a -> m b)
2067-- | Parallel reverse function application operator for applying a run or fold
2068-- functions to a stream. Just like '|$.' except that the operands are reversed.
2071-- S.repeatM (threadDelay 1000000 >> return 1)
2072-- |&. S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
2079(|&.) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
2082------------------------------------------------------------------------------
2083-- General Transformation
2084------------------------------------------------------------------------------
2086-- | Use a 'Pipe' to transform a stream.
2090{-# INLINE transform #-}
2091transform :: (IsStream t, Monad m) => Pipe m a b -> t m a -> t m b
2092transform pipe xs = fromStreamD $ D.transform pipe (toStreamD xs)
2094------------------------------------------------------------------------------
2095-- Transformation by Folding (Scans)
2096------------------------------------------------------------------------------
2098-- XXX It may be useful to have a version of scan where we can keep the
2099-- accumulator independent of the value emitted. So that we do not necessarily
2100-- have to keep a value in the accumulator which we are not using. We can pass
2101-- an extraction function that will take the accumulator and the current value
2102-- of the element and emit the next value in the stream. That will also make it
2103-- possible to modify the accumulator after using it. In fact, the step function
2104-- can return new accumulator and the value to be emitted. The signature would
2105-- be more like mapAccumL. Or we can change the signature of scanx to
2108-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
2109-- user supplied extraction function (the third argument) at each step. This is
2110-- designed to work with the @foldl@ library. The suffix @x@ is a mnemonic for
2113-- /Since: 0.7.0 (Monad m constraint)/
2116{-# DEPRECATED scanx "Please use scanl followed by map instead." #-}
2118scanx :: (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
2121-- XXX this needs to be concurrent
2122-- | Like 'scanl'' but with a monadic fold function.
2125{-# INLINE scanlM' #-}
2126scanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
2127scanlM' step begin m = fromStreamD $ D.scanlM' step begin $ toStreamD m
2129-- | @scanlMAfter' accumulate initial done stream@ is like 'scanlM'' except
2130-- that it provides an additional @done@ function to be applied on the
2131-- accumulator when the stream stops. The result of @done@ is also emitted in
2134-- This function can be used to allocate a resource in the beginning of the
2135-- scan and release it when the stream ends or to flush the internal state of
2136-- the scan at the end.
2140{-# INLINE scanlMAfter' #-}
2141scanlMAfter' :: (IsStream t, Monad m)
2142 => (b -> a -> m b) -> m b -> (b -> m b) -> t m a -> t m b
2143scanlMAfter' step initial done stream =
2144 fromStreamD $ D.scanlMAfter' step initial done $ toStreamD stream
2146-- | Strict left scan. Like 'map', 'scanl'' too is a one to one transformation,
2147-- however it adds an extra element.
2150-- > S.toList $ S.scanl' (+) 0 $ fromList [1,2,3,4]
2155-- > S.toList $ S.scanl' (flip (:)) [] $ S.fromList [1,2,3,4]
2156-- [[],[1],[2,1],[3,2,1],[4,3,2,1]]
2159-- The output of 'scanl'' is the initial value of the accumulator followed by
2160-- all the intermediate steps and the final result of 'foldl''.
2162-- By streaming the accumulated state after each fold step, we can share the
2163-- state across multiple stages of stream composition. Each stage can modify or
2164-- extend the state, do some processing with it and emit it for the next stage,
2165-- thus modularizing the stream processing. This can be useful in
2166-- stateful or event-driven programming.
2168-- Consider the following monolithic example, computing the sum and the product
2169-- of the elements in a stream in one go using a @foldl'@:
2172-- > S.foldl' (\\(s, p) x -> (s + x, p * x)) (0,1) $ S.fromList \[1,2,3,4]
2176-- Using @scanl'@ we can make it modular by computing the sum in the first
2177-- stage and passing it down to the next stage for computing the product:
2180-- > S.foldl' (\\(_, p) (s, x) -> (s, p * x)) (0,1)
2181-- $ S.scanl' (\\(s, _) x -> (s + x, x)) (0,1)
2182-- $ S.fromList \[1,2,3,4]
2186-- IMPORTANT: 'scanl'' evaluates the accumulator to WHNF. To avoid building
2187-- lazy expressions inside the accumulator, it is recommended that a strict
2188-- data structure is used for accumulator.
2191{-# INLINE scanl' #-}
2192scanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
2193scanl' step z m = fromStreamS $ S.scanl' step z $ toStreamS m
2195-- | Like 'scanl'' but does not stream the initial value of the accumulator.
2197-- > postscanl' f z xs = S.drop 1 $ S.scanl' f z xs
2200{-# INLINE postscanl' #-}
2201postscanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
2202postscanl' step z m = fromStreamD $ D.postscanl' step z $ toStreamD m
2204-- XXX this needs to be concurrent
2205-- | Like 'postscanl'' but with a monadic step function.
2208{-# INLINE postscanlM' #-}
2209postscanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> t m b
2210postscanlM' step z m = fromStreamD $ D.postscanlM' step z $ toStreamD m
2212-- XXX prescanl does not sound very useful, enable only if there is a
2213-- compelling use case.
2215-- | Like scanl' but does not stream the final value of the accumulator.
2218{-# INLINE prescanl' #-}
2219prescanl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> t m b
2220prescanl' step z m = fromStreamD $ D.prescanl' step z $ toStreamD m
2222-- XXX this needs to be concurrent
2223-- | Like postscanl' but with a monadic step function.
2226{-# INLINE prescanlM' #-}
2227prescanlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> m b -> t m a -> t m b
2228prescanlM' step z m = fromStreamD $ D.prescanlM' step z $ toStreamD m
2230-- XXX this needs to be concurrent
2231-- | Like 'scanl1'' but with a monadic step function.
2234{-# INLINE scanl1M' #-}
2235scanl1M' :: (IsStream t, Monad m) => (a -> a -> m a) -> t m a -> t m a
2236scanl1M' step m = fromStreamD $ D.scanl1M' step $ toStreamD m
2238-- | Like 'scanl'' but for a non-empty stream. The first element of the stream
2239-- is used as the initial value of the accumulator. Does nothing if the stream
2243-- > S.toList $ S.scanl1 (+) $ fromList [1,2,3,4]
2248{-# INLINE scanl1' #-}
2249scanl1' :: (IsStream t, Monad m) => (a -> a -> a) -> t m a -> t m a
2250scanl1' step m = fromStreamD $ D.scanl1' step $ toStreamD m
2252------------------------------------------------------------------------------
2253-- Scanning with a Fold
2254------------------------------------------------------------------------------
2256-- | Scan a stream using the given monadic fold.
2260scan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
2261scan (Fold step begin done) = P.scanlMx' step begin done
2263-- | Postscan a stream using the given monadic fold.
2266{-# INLINE postscan #-}
2267postscan :: (IsStream t, Monad m) => Fold m a b -> t m a -> t m b
2268postscan (Fold step begin done) = P.postscanlMx' step begin done
2270------------------------------------------------------------------------------
2271-- Stateful Transformations
2272------------------------------------------------------------------------------
2274-- XXX this is not a one-to-one map so calling it map may not be right.
2275-- We can perhaps call it zipWithTail or rollWith.
2277-- | Apply a function on every two successive elements of a stream. If the
2278-- stream consists of a single element the output is an empty stream.
2280-- This is the stream equivalent of the list idiom @zipWith f xs (tail xs)@.
2284{-# INLINE rollingMap #-}
2285rollingMap :: (IsStream t, Monad m) => (a -> a -> b) -> t m a -> t m b
2286rollingMap f m = fromStreamD $ D.rollingMap f $ toStreamD m
2288-- | Like 'rollingMap' but with an effectful map function.
2292{-# INLINE rollingMapM #-}
2293rollingMapM :: (IsStream t, Monad m) => (a -> a -> m b) -> t m a -> t m b
2294rollingMapM f m = fromStreamD $ D.rollingMapM f $ toStreamD m
2296------------------------------------------------------------------------------
2297-- Transformation by Filtering
2298------------------------------------------------------------------------------
2300-- | Include only those elements that pass a predicate.
2303{-# INLINE filter #-}
2304#if __GLASGOW_HASKELL__ != 802
2305-- GHC 8.2.2 crashes with this code, when used with "stack"
2306filter :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2307filter p m = fromStreamS $ S.filter p $ toStreamS m
2309filter :: IsStream t => (a -> Bool) -> t m a -> t m a
2313-- | Same as 'filter' but with a monadic predicate.
2316{-# INLINE filterM #-}
2317filterM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2318filterM p m = fromStreamD $ D.filterM p $ toStreamD m
2320-- | Drop repeated elements that are adjacent to each other.
2324uniq :: (Eq a, IsStream t, Monad m) => t m a -> t m a
2325uniq = fromStreamD . D.uniq . toStreamD
2327-- | Ensures that all the elements of the stream are identical and then returns
2328-- that unique element.
2332the :: (Eq a, Monad m) => SerialT m a -> m (Maybe a)
2333the m = S.the (toStreamS m)
2335-- | Take first 'n' elements from the stream and discard the rest.
2339take :: (IsStream t, Monad m) => Int -> t m a -> t m a
2340take n m = fromStreamS $ S.take n $ toStreamS
2341 (maxYields (Just (fromIntegral n)) m)
2343-- | End the stream as soon as the predicate fails on an element.
2346{-# INLINE takeWhile #-}
2347takeWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2348takeWhile p m = fromStreamS $ S.takeWhile p $ toStreamS m
2350-- | Same as 'takeWhile' but with a monadic predicate.
2353{-# INLINE takeWhileM #-}
2354takeWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2355takeWhileM p m = fromStreamD $ D.takeWhileM p $ toStreamD m
2357-- | @takeByTime duration@ yields stream elements upto specified time
2358-- @duration@. The duration starts when the stream is evaluated for the first
2359-- time, before the first element is yielded. The time duration is checked
2360-- before generating each element, if the duration has expired the stream
2363-- The total time taken in executing the stream is guaranteed to be /at least/
2364-- @duration@, however, because the duration is checked before generating an
2365-- element, the upper bound is indeterminate and depends on the time taken in
2366-- generating and processing the last element.
2368-- No element is yielded if the duration is zero. At least one element is
2369-- yielded if the duration is non-zero.
2373{-# INLINE takeByTime #-}
2374takeByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
2375takeByTime d = fromStreamD . D.takeByTime d . toStreamD
2377-- | Discard first 'n' elements from the stream and take the rest.
2381drop :: (IsStream t, Monad m) => Int -> t m a -> t m a
2382drop n m = fromStreamS $ S.drop n $ toStreamS m
2384-- | Drop elements in the stream as long as the predicate succeeds and then
2385-- take the rest of the stream.
2388{-# INLINE dropWhile #-}
2389dropWhile :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> t m a
2390dropWhile p m = fromStreamS $ S.dropWhile p $ toStreamS m
2392-- | Same as 'dropWhile' but with a monadic predicate.
2395{-# INLINE dropWhileM #-}
2396dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
2397dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
2399-- | @dropByTime duration@ drops stream elements until specified @duration@ has
2400-- passed. The duration begins when the stream is evaluated for the first
2401-- time. The time duration is checked /after/ generating a stream element, the
2402-- element is yielded if the duration has expired otherwise it is dropped.
2404-- The time elapsed before starting to generate the first element is /at most/
2405-- @duration@, however, because the duration expiry is checked after the
2406-- element is generated, the lower bound is indeterminate and depends on the
2407-- time taken in generating an element.
2409-- All elements are yielded if the duration is zero.
2413{-# INLINE dropByTime #-}
2414dropByTime ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
2415dropByTime d = fromStreamD . D.dropByTime d . toStreamD
2417------------------------------------------------------------------------------
2418-- Transformation by Mapping
2419------------------------------------------------------------------------------
2423-- mapM f = sequence . map f
2426-- Apply a monadic function to each element of the stream and replace it with
2427-- the output of the resulting action.
2430-- > drain $ S.mapM putStr $ S.fromList ["a", "b", "c"]
2433-- drain $ S.replicateM 10 (return 1)
2434-- & (serially . S.mapM (\\x -> threadDelay 1000000 >> print x))
2436-- drain $ S.replicateM 10 (return 1)
2437-- & (asyncly . S.mapM (\\x -> threadDelay 1000000 >> print x))
2440-- /Concurrent (do not use with 'parallely' on infinite streams)/
2443{-# INLINE_EARLY mapM #-}
2444mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
2447{-# RULES "mapM serial" mapM = mapMSerial #-}
2448{-# INLINE mapMSerial #-}
2449mapMSerial :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
2450mapMSerial = Serial.mapM
2454-- sequence = mapM id
2457-- Replace the elements of a stream of monadic actions with the outputs of
2461-- > drain $ S.sequence $ S.fromList [putStr "a", putStr "b", putStrLn "c"]
2464-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
2465-- & (serially . S.sequence)
2467-- drain $ S.replicateM 10 (return $ threadDelay 1000000 >> print 1)
2468-- & (asyncly . S.sequence)
2471-- /Concurrent (do not use with 'parallely' on infinite streams)/
2474{-# INLINE sequence #-}
2475sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
2476sequence m = fromStreamS $ S.sequence (toStreamS m)
2478------------------------------------------------------------------------------
2479-- Transformation by Map and Filter
2480------------------------------------------------------------------------------
2482-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
2483-- elements, and return a stream of values extracted from 'Just'.
2488-- mapMaybe f = S.map 'fromJust' . S.filter 'isJust' . S.map f
2492{-# INLINE mapMaybe #-}
2493mapMaybe :: (IsStream t, Monad m) => (a -> Maybe b) -> t m a -> t m b
2494mapMaybe f m = fromStreamS $ S.mapMaybe f $ toStreamS m
2496-- | Like 'mapMaybe' but maps a monadic function.
2501-- mapMaybeM f = S.map 'fromJust' . S.filter 'isJust' . S.mapM f
2504-- /Concurrent (do not use with 'parallely' on infinite streams)/
2507{-# INLINE_EARLY mapMaybeM #-}
2508mapMaybeM :: (IsStream t, MonadAsync m, Functor (t m))
2509 => (a -> m (Maybe b)) -> t m a -> t m b
2510mapMaybeM f = fmap fromJust . filter isJust . K.mapM f
2512{-# RULES "mapMaybeM serial" mapMaybeM = mapMaybeMSerial #-}
2513{-# INLINE mapMaybeMSerial #-}
2514mapMaybeMSerial :: Monad m => (a -> m (Maybe b)) -> SerialT m a -> SerialT m b
2515mapMaybeMSerial f m = fromStreamD $ D.mapMaybeM f $ toStreamD m
2517------------------------------------------------------------------------------
2518-- Transformation by Reordering
2519------------------------------------------------------------------------------
2521-- XXX Use a compact region list to temporarily store the list, in both reverse
2522-- as well as in reverse'.
2524-- /Note:/ 'reverse'' is much faster than this, use that when performance
2527-- > reverse = S.foldlT (flip S.cons) S.nil
2529-- | Returns the elements of the stream in reverse order. The stream must be
2530-- finite. Note that this necessarily buffers the entire stream in memory.
2532-- /Since 0.7.0 (Monad m constraint)/
2535{-# INLINE reverse #-}
2536reverse :: (IsStream t, Monad m) => t m a -> t m a
2537reverse s = fromStreamS $ S.reverse $ toStreamS s
2539-- | Like 'reverse' but several times faster, requires a 'Storable' instance.
2542{-# INLINE reverse' #-}
2543reverse' :: (IsStream t, MonadIO m, Storable a) => t m a -> t m a
2544reverse' s = fromStreamD $ D.reverse' $ toStreamD s
2546------------------------------------------------------------------------------
2547-- Transformation by Inserting
2548------------------------------------------------------------------------------
2550-- intersperseM = intersperseBySpan 1
2552-- | Insert an effect and its output before consuming an element of a stream
2553-- except the first one.
2556-- >>> S.toList $ S.trace putChar $ S.intersperseM (putChar '.' >> return ',') $ S.fromList "hello"
2557-- > h.,e.,l.,l.,o"h,e,l,l,o"
2561{-# INLINE intersperseM #-}
2562intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2563intersperseM m = fromStreamS . S.intersperseM m . toStreamS
2565-- | Insert a side effect before consuming an element of a stream except the
2569-- >>> S.drain $ S.trace putChar $ S.intersperseM_ (putChar '.') $ S.fromList "hello"
2574{-# INLINE intersperseM_ #-}
2575intersperseM_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2576intersperseM_ m = fromStreamD . D.intersperseM_ m . toStreamD
2578-- | Insert a pure value between successive elements of a stream.
2581-- > S.toList $ S.intersperse ',' $ S.fromList "hello"
2586{-# INLINE intersperse #-}
2587intersperse :: (IsStream t, MonadAsync m) => a -> t m a -> t m a
2588intersperse a = fromStreamS . S.intersperse a . toStreamS
2590-- | Insert an effect and its output after consuming an element of a stream.
2593-- >>> S.toList $ S.trace putChar $ S.intersperseSuffix (putChar '.' >> return ',') $ S.fromList "hello"
2594-- > h.,e.,l.,l.,o.,"h,e,l,l,o,"
2598{-# INLINE intersperseSuffix #-}
2599intersperseSuffix :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a
2600intersperseSuffix m = fromStreamD . D.intersperseSuffix m . toStreamD
2602-- | Insert a side effect after consuming an element of a stream.
2605-- > S.mapM_ putChar $ S.intersperseSuffix_ (threadDelay 1000000) $ S.fromList "hello"
2610{-# INLINE intersperseSuffix_ #-}
2611intersperseSuffix_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
2612intersperseSuffix_ m = fromStreamD . D.intersperseSuffix_ m . toStreamD
2614-- | Insert a side effect before consuming an element of a stream.
2617-- >>> S.toList $ S.trace putChar $ S.interspersePrefix_ (putChar '.' >> return ',') $ S.fromList "hello"
2618-- > .h.e.l.l.o"hello"
2621-- See also: 'trace_'
2627{-# INLINE interspersePrefix_ #-}
2628interspersePrefix_ :: (IsStream t, MonadAsync m) => m b -> t m a -> t m a
2629interspersePrefix_ m = mapM (\x -> void m >> return x)
2631-- Note: delay must be serial.
2633-- | Introduce a delay of specified seconds before consuming an element of the
2634-- stream except the first one.
2637-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2638-- > (AbsTime (TimeSpec {sec = 2502706, nsec = 751137000}),1)
2639-- > (AbsTime (TimeSpec {sec = 2502707, nsec = 743535000}),2)
2640-- > (AbsTime (TimeSpec {sec = 2502708, nsec = 749758000}),3)
2646delay :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2647delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000
2649-- Note: delay must be serial.
2651-- | Introduce a delay of specified seconds after consuming an element of a
2655-- >>> S.mapM_ print $ S.timestamped $ S.delayPost 1 $ S.enumerateFromTo 1 3
2656-- > (AbsTime (TimeSpec {sec = 2502826, nsec = 119030000}),1)
2657-- > (AbsTime (TimeSpec {sec = 2502827, nsec = 111393000}),2)
2658-- > (AbsTime (TimeSpec {sec = 2502828, nsec = 112221000}),3)
2663{-# INLINE delayPost #-}
2664delayPost :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2665delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000
2667-- Note: delay must be serial, that's why 'trace_' is used.
2669-- | Introduce a delay of specified seconds before consuming an element of a
2673-- >>> S.mapM_ print $ S.timestamped $ S.delayPre 1 $ S.enumerateFromTo 1 3
2674-- > (AbsTime (TimeSpec {sec = 2502207, nsec = 533177000}),1)
2675-- > (AbsTime (TimeSpec {sec = 2502208, nsec = 530859000}),2)
2676-- > (AbsTime (TimeSpec {sec = 2502209, nsec = 531619000}),3)
2681{-# INLINE delayPre #-}
2682delayPre :: (IsStream t, MonadIO m) => Double -> t m a -> t m a
2683delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000
2685-- | Like 'intersperseSuffix' but intersperses an effectful action into the
2686-- input stream after every @n@ elements and after the last element.
2689-- > S.toList $ S.intersperseSuffixBySpan 2 (return ',') $ S.fromList "hello"
2695{-# INLINE intersperseSuffixBySpan #-}
2696intersperseSuffixBySpan :: (IsStream t, MonadAsync m)
2697 => Int -> m a -> t m a -> t m a
2698intersperseSuffixBySpan n eff =
2699 fromStreamD . D.intersperseSuffixBySpan n eff . toStreamD
2702-- | Intersperse a monadic action into the input stream after every @n@
2706-- > S.toList $ S.intersperseBySpan 2 (return ',') $ S.fromList "hello"
2711{-# INLINE intersperseBySpan #-}
2712intersperseBySpan :: IsStream t => Int -> m a -> t m a -> t m a
2713intersperseBySpan _n _f _xs = undefined
2716-- | Intersperse a monadic action into the input stream after every @n@
2720-- > S.drain $ S.interjectSuffix 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"
2725{-# INLINE interjectSuffix #-}
2727 :: (IsStream t, MonadAsync m)
2728 => Double -> m a -> t m a -> t m a
2729interjectSuffix n f xs = xs `Par.parallelFst` repeatM timed
2730 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
2732-- | @insertBy cmp elem stream@ inserts @elem@ before the first element in
2733-- @stream@ that is less than @elem@ when compared using @cmp@.
2736-- insertBy cmp x = 'mergeBy' cmp ('yield' x)
2740-- > S.toList $ S.insertBy compare 2 $ S.fromList [1,3,5]
2745{-# INLINE insertBy #-}
2747 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a
2748insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)
2750------------------------------------------------------------------------------
2752------------------------------------------------------------------------------
2754-- | Deletes the first occurrence of the element in the stream that satisfies
2755-- the given equality predicate.
2758-- > S.toList $ S.deleteBy (==) 3 $ S.fromList [1,3,3,5]
2763{-# INLINE deleteBy #-}
2764deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a
2765deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)
2767------------------------------------------------------------------------------
2769------------------------------------------------------------------------------
2772-- > indexed = S.postscanl' (\(i, _) x -> (i + 1, x)) (-1,undefined)
2773-- > indexed = S.zipWith (,) (S.enumerateFrom 0)
2775-- Pair each element in a stream with its index, starting from index 0.
2778-- > S.toList $ S.indexed $ S.fromList "hello"
2779-- [(0,'h'),(1,'e'),(2,'l'),(3,'l'),(4,'o')]
2783{-# INLINE indexed #-}
2784indexed :: (IsStream t, Monad m) => t m a -> t m (Int, a)
2785indexed = fromStreamD . D.indexed . toStreamD
2788-- > indexedR n = S.postscanl' (\(i, _) x -> (i - 1, x)) (n + 1,undefined)
2789-- > indexedR n = S.zipWith (,) (S.enumerateFromThen n (n - 1))
2791-- Pair each element in a stream with its index, starting from the
2792-- given index @n@ and counting down.
2795-- > S.toList $ S.indexedR 10 $ S.fromList "hello"
2796-- [(10,'h'),(9,'e'),(8,'l'),(7,'l'),(6,'o')]
2800{-# INLINE indexedR #-}
2801indexedR :: (IsStream t, Monad m) => Int -> t m a -> t m (Int, a)
2802indexedR n = fromStreamD . D.indexedR n . toStreamD
2804-- Note: The timestamp stream must be the second stream in the zip so that the
2805-- timestamp is generated after generating the stream element and not before.
2806-- If we do not do that then the following example will generate the same
2807-- timestamp for first two elements:
2809-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
2811-- | Pair each element in a stream with an absolute timestamp, using a clock of
2812-- specified granularity. The timestamp is generated just before the element
2816-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2821{-# INLINE timestampWith #-}
2822timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
2823 => Double -> t m a -> t m (AbsTime, a)
2824timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
2826-- TBD: check performance vs a custom implementation without using zipWith.
2828-- | Pair each element in a stream with an absolute timestamp, using a clock of
2829-- 10 ms granularity. The timestamp is generated just before the element is
2833-- >>> S.mapM_ print $ S.timestamped $ S.delay 1 $ S.enumerateFromTo 1 3
2834-- (AbsTime (TimeSpec {sec = 2460689, nsec = 641121000}),1)
2835-- (AbsTime (TimeSpec {sec = 2460690, nsec = 639334000}),2)
2836-- (AbsTime (TimeSpec {sec = 2460691, nsec = 644479000}),3)
2841{-# INLINE timestamped #-}
2842timestamped :: (IsStream t, MonadAsync m, Functor (t m))
2843 => t m a -> t m (AbsTime, a)
2844timestamped = timestampWith 0.01
2846-- | Pair each element in a stream with relative times starting from 0, using a
2847-- clock with the specified granularity. The time is measured just before the
2848-- element is consumed.
2851-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
2856{-# INLINE timeIndexWith #-}
2857timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
2858 => Double -> t m a -> t m (RelTime64, a)
2859timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
2861-- | Pair each element in a stream with relative times starting from 0, using a
2862-- 10 ms granularity clock. The time is measured just before the element is
2866-- >>> S.mapM_ print $ S.timeIndexed $ S.delay 1 $ S.enumerateFromTo 1 3
2867-- (RelTime64 (NanoSecond64 0),1)
2868-- (RelTime64 (NanoSecond64 996239000),2)
2869-- (RelTime64 (NanoSecond64 1996712000),3)
2874{-# INLINE timeIndexed #-}
2875timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
2876 => t m a -> t m (RelTime64, a)
2877timeIndexed = timeIndexWith 0.01
2879------------------------------------------------------------------------------
2881------------------------------------------------------------------------------
2883-- | Compare two streams for equality using an equality function.
2886{-# INLINABLE eqBy #-}
2887eqBy :: (IsStream t, Monad m) => (a -> b -> Bool) -> t m a -> t m b -> m Bool
2890-- | Compare two streams lexicographically using a comparison function.
2893{-# INLINABLE cmpBy #-}
2895 :: (IsStream t, Monad m)
2896 => (a -> b -> Ordering) -> t m a -> t m b -> m Ordering
2899------------------------------------------------------------------------------
2901------------------------------------------------------------------------------
2903-- | Merge two streams using a comparison function. The head elements of both
2904-- the streams are compared and the smaller of the two elements is emitted, if
2905-- both elements are equal then the element from the first stream is used
2908-- If the streams are sorted in ascending order, the resulting stream would
2909-- also remain sorted in ascending order.
2912-- > S.toList $ S.mergeBy compare (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2917{-# INLINABLE mergeBy #-}
2919 (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2920mergeBy f m1 m2 = fromStreamS $ S.mergeBy f (toStreamS m1) (toStreamS m2)
2922-- | Like 'mergeBy' but with a monadic comparison function.
2924-- Merge two streams randomly:
2927-- > randomly _ _ = randomIO >>= \x -> return $ if x then LT else GT
2928-- > S.toList $ S.mergeByM randomly (S.fromList [1,1,1,1]) (S.fromList [2,2,2,2])
2932-- Merge two streams in a proportion of 2:1:
2935-- proportionately m n = do
2936-- ref <- newIORef $ cycle $ concat [replicate m LT, replicate n GT]
2937-- return $ \\_ _ -> do
2938-- r <- readIORef ref
2939-- writeIORef ref $ tail r
2943-- f <- proportionately 2 1
2944-- xs <- S.toList $ S.mergeByM f (S.fromList [1,1,1,1,1,1]) (S.fromList [2,2,2])
2948-- [1,1,2,1,1,2,1,1,2]
2952{-# INLINABLE mergeByM #-}
2954 :: (IsStream t, Monad m)
2955 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2956mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
2959-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
2960{-# INLINABLE mergeEndByAny #-}
2962 :: (IsStream t, Monad m)
2963 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2964mergeEndByAny f m1 m2 = fromStreamD $
2965 D.mergeEndByAny f (toStreamD m1) (toStreamD m2)
2967-- Like 'mergeByM' but stops merging as soon as the first stream stops.
2968{-# INLINABLE mergeEndByFirst #-}
2970 :: (IsStream t, Monad m)
2971 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
2972mergeEndByFirst f m1 m2 = fromStreamS $
2973 D.mergeEndByFirst f (toStreamD m1) (toStreamD m2)
2976-- Holding this back for now, we may want to use the name "merge" differently
2978-- | Same as @'mergeBy' 'compare'@.
2981-- > S.toList $ S.merge (S.fromList [1,3,5]) (S.fromList [2,4,6,8])
2986{-# INLINABLE merge #-}
2988 (IsStream t, Monad m, Ord a) => t m a -> t m a -> t m a
2989merge = mergeBy compare
2992-- | Like 'mergeBy' but merges concurrently (i.e. both the elements being
2993-- merged are generated concurrently).
2996{-# INLINE mergeAsyncBy #-}
2997mergeAsyncBy :: (IsStream t, MonadAsync m)
2998 => (a -> a -> Ordering) -> t m a -> t m a -> t m a
2999mergeAsyncBy f = mergeAsyncByM (\a b -> return $ f a b)
3001-- | Like 'mergeByM' but merges concurrently (i.e. both the elements being
3002-- merged are generated concurrently).
3005{-# INLINE mergeAsyncByM #-}
3006mergeAsyncByM :: (IsStream t, MonadAsync m)
3007 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
3008mergeAsyncByM f m1 m2 = fromStreamD $
3009 D.mergeByM f (D.mkParallelD $ toStreamD m1) (D.mkParallelD $ toStreamD m2)
3011------------------------------------------------------------------------------
3013------------------------------------------------------------------------------
3015-- | @concatMapWith merge map stream@ is a two dimensional looping combinator.
3016-- The first argument specifies a merge or concat function that is used to
3017-- merge the streams generated by applying the second argument i.e. the @map@
3018-- function to each element of the input stream. The concat function could be
3019-- 'serial', 'parallel', 'async', 'ahead' or any other zip or merge function
3020-- and the second argument could be any stream generation function using a
3023-- /Compare 'foldMapWith'/
3026{-# INLINE concatMapWith #-}
3029 => (forall c. t m c -> t m c -> t m c)
3033concatMapWith = K.concatMapBy
3035-- | Map a stream producing function on each element of the stream and then
3036-- flatten the results into a single stream.
3039-- concatMap = 'concatMapWith' 'Serial.serial'
3040-- concatMap f = 'concatMapM' (return . f)
3041-- concatMap f = 'concatUnfold' (UF.lmap f UF.fromStream)
3045{-# INLINE concatMap #-}
3046concatMap ::(IsStream t, Monad m) => (a -> t m b) -> t m a -> t m b
3047concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)
3049-- | Flatten a stream of streams to a single stream.
3052-- concat = concatMap id
3056{-# INLINE concat #-}
3057concat :: (IsStream t, Monad m) => t m (t m a) -> t m a
3058concat = concatMap id
3060-- | Append the outputs of two streams, yielding all the elements from the
3061-- first stream and then yielding all the elements from the second stream.
3063-- IMPORTANT NOTE: This could be 100x faster than @serial/<>@ for appending a
3064-- few (say 100) streams because it can fuse via stream fusion. However, it
3065-- does not scale for a large number of streams (say 1000s) and becomes
3066-- qudartically slow. Therefore use this for custom appending of a few streams
3067-- but use 'concatMap' or 'concatMapWith serial' for appending @n@ streams or
3068-- infinite containers of streams.
3071{-# INLINE append #-}
3072append ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3073append m1 m2 = fromStreamD $ D.append (toStreamD m1) (toStreamD m2)
3075-- XXX Same as 'wSerial'. We should perhaps rename wSerial to interleave.
3076-- XXX Document the interleaving behavior of side effects in all the
3077-- interleaving combinators.
3078-- XXX Write time-domain equivalents of these. In the time domain we can
3079-- interleave two streams such that the value of second stream is always taken
3080-- from its last value even if no new value is being yielded, like
3081-- zipWithLatest. It would be something like interleaveWithLatest.
3083-- | Interleaves the outputs of two streams, yielding elements from each stream
3084-- alternately, starting from the first stream. If any of the streams finishes
3085-- early the other stream continues alone until it too finishes.
3087-- >>> :set -XOverloadedStrings
3088-- >>> interleave "ab" ",,,," :: SerialT Identity Char
3090-- >>> interleave "abcd" ",," :: SerialT Identity Char
3093-- 'interleave' is dual to 'interleaveMin', it can be called @interleaveMax@.
3095-- Do not use at scale in concatMapWith.
3098{-# INLINE interleave #-}
3099interleave ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3100interleave m1 m2 = fromStreamD $ D.interleave (toStreamD m1) (toStreamD m2)
3102-- | Interleaves the outputs of two streams, yielding elements from each stream
3103-- alternately, starting from the first stream. As soon as the first stream
3104-- finishes, the output stops, discarding the remaining part of the second
3105-- stream. In this case, the last element in the resulting stream would be from
3106-- the second stream. If the second stream finishes early then the first stream
3107-- still continues to yield elements until it finishes.
3109-- >>> :set -XOverloadedStrings
3110-- >>> interleaveSuffix "abc" ",,,," :: SerialT Identity Char
3112-- >>> interleaveSuffix "abc" "," :: SerialT Identity Char
3115-- 'interleaveSuffix' is a dual of 'interleaveInfix'.
3117-- Do not use at scale in concatMapWith.
3120{-# INLINE interleaveSuffix #-}
3121interleaveSuffix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3122interleaveSuffix m1 m2 =
3123 fromStreamD $ D.interleaveSuffix (toStreamD m1) (toStreamD m2)
3125-- | Interleaves the outputs of two streams, yielding elements from each stream
3126-- alternately, starting from the first stream and ending at the first stream.
3127-- If the second stream is longer than the first, elements from the second
3128-- stream are infixed with elements from the first stream. If the first stream
3129-- is longer then it continues yielding elements even after the second stream
3132-- >>> :set -XOverloadedStrings
3133-- >>> interleaveInfix "abc" ",,,," :: SerialT Identity Char
3135-- >>> interleaveInfix "abc" "," :: SerialT Identity Char
3138-- 'interleaveInfix' is a dual of 'interleaveSuffix'.
3140-- Do not use at scale in concatMapWith.
3143{-# INLINE interleaveInfix #-}
3144interleaveInfix ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3145interleaveInfix m1 m2 =
3146 fromStreamD $ D.interleaveInfix (toStreamD m1) (toStreamD m2)
3148-- | Interleaves the outputs of two streams, yielding elements from each stream
3149-- alternately, starting from the first stream. The output stops as soon as any
3150-- of the two streams finishes, discarding the remaining part of the other
3151-- stream. The last element of the resulting stream would be from the longer
3154-- >>> :set -XOverloadedStrings
3155-- >>> interleaveMin "ab" ",,,," :: SerialT Identity Char
3157-- >>> interleaveMin "abcd" ",," :: SerialT Identity Char
3160-- 'interleaveMin' is dual to 'interleave'.
3162-- Do not use at scale in concatMapWith.
3165{-# INLINE interleaveMin #-}
3166interleaveMin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3167interleaveMin m1 m2 = fromStreamD $ D.interleaveMin (toStreamD m1) (toStreamD m2)
3169-- | Schedule the execution of two streams in a fair round-robin manner,
3170-- executing each stream once, alternately. Execution of a stream may not
3171-- necessarily result in an output, a stream may chose to @Skip@ producing an
3172-- element until later giving the other stream a chance to run. Therefore, this
3173-- combinator fairly interleaves the execution of two streams rather than
3174-- fairly interleaving the output of the two streams. This can be useful in
3175-- co-operative multitasking without using explicit threads. This can be used
3176-- as an alternative to `async`.
3178-- Do not use at scale in concatMapWith.
3181{-# INLINE roundrobin #-}
3182roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b
3183roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)
3185-- | Map a stream producing monadic function on each element of the stream
3186-- and then flatten the results into a single stream. Since the stream
3187-- generation function is monadic, unlike 'concatMap', it can produce an
3188-- effect at the beginning of each iteration of the inner loop.
3191{-# INLINE concatMapM #-}
3192concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b
3193concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)
3195-- | Given a stream value in the underlying monad, lift and join the underlying
3196-- monad with the stream monad.
3198-- Compare with 'concat' and 'sequence'.
3202{-# INLINE concatM #-}
3203concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
3204concatM generator = concatMapM (\() -> generator) (yield ())
3206-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike
3207-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and
3208-- therefore provide many times better performance.
3211{-# INLINE concatUnfold #-}
3212concatUnfold ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b
3213concatUnfold u m = fromStreamD $ D.concatMapU u (toStreamD m)
3215-- | Like 'concatUnfold' but interleaves the streams in the same way as
3216-- 'interleave' behaves instead of appending them.
3219{-# INLINE concatUnfoldInterleave #-}
3220concatUnfoldInterleave ::(IsStream t, Monad m)
3221 => Unfold m a b -> t m a -> t m b
3222concatUnfoldInterleave u m =
3223 fromStreamD $ D.concatUnfoldInterleave u (toStreamD m)
3225-- | Like 'concatUnfold' but executes the streams in the same way as
3229{-# INLINE concatUnfoldRoundrobin #-}
3230concatUnfoldRoundrobin ::(IsStream t, Monad m)
3231 => Unfold m a b -> t m a -> t m b
3232concatUnfoldRoundrobin u m =
3233 fromStreamD $ D.concatUnfoldRoundrobin u (toStreamD m)
3235-- XXX we can swap the order of arguments to gintercalate so that the
3236-- definition of concatUnfold becomes simpler? The first stream should be
3237-- infixed inside the second one. However, if we change the order in
3238-- "interleave" as well similarly, then that will make it a bit unintuitive.
3240-- > concatUnfold unf str =
3241-- > gintercalate unf str (UF.nilM (\_ -> return ())) (repeat ())
3243-- | 'interleaveInfix' followed by unfold and concat.
3246{-# INLINE gintercalate #-}
3248 :: (IsStream t, Monad m)
3249 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
3250gintercalate unf1 str1 unf2 str2 =
3251 D.fromStreamD $ D.gintercalate
3252 unf1 (D.toStreamD str1)
3253 unf2 (D.toStreamD str2)
3255-- XXX The order of arguments in "intercalate" is consistent with the list
3256-- intercalate but inconsistent with gintercalate and other stream interleaving
3257-- combinators. We can change the order of the arguments in other combinators
3258-- but then 'interleave' combinator may become a bit unintuitive because we
3259-- will be starting with the second stream.
3261-- > intercalate seed unf str = gintercalate unf str unf (repeatM seed)
3262-- > intercalate a unf str = concatUnfold unf $ intersperse a str
3264-- | 'intersperse' followed by unfold and concat.
3266-- > unwords = intercalate " " UF.fromList
3268-- >>> intercalate " " UF.fromList ["abc", "def", "ghi"]
3272{-# INLINE intercalate #-}
3273intercalate :: (IsStream t, Monad m)
3274 => b -> Unfold m b c -> t m b -> t m c
3275intercalate seed unf str = D.fromStreamD $
3276 D.concatMapU unf $ D.intersperse seed (toStreamD str)
3278-- > interpose x unf str = gintercalate unf str UF.identity (repeat x)
3280-- | Unfold the elements of a stream, intersperse the given element between the
3281-- unfolded streams and then concat them into a single stream.
3283-- > unwords = S.interpose ' '
3286{-# INLINE interpose #-}
3287interpose :: (IsStream t, Monad m)
3288 => c -> Unfold m b c -> t m b -> t m c
3289interpose x unf str =
3290 D.fromStreamD $ D.interpose (return x) unf (D.toStreamD str)
3292-- | 'interleaveSuffix' followed by unfold and concat.
3295{-# INLINE gintercalateSuffix #-}
3297 :: (IsStream t, Monad m)
3298 => Unfold m a c -> t m a -> Unfold m b c -> t m b -> t m c
3299gintercalateSuffix unf1 str1 unf2 str2 =
3300 D.fromStreamD $ D.gintercalateSuffix
3301 unf1 (D.toStreamD str1)
3302 unf2 (D.toStreamD str2)
3304-- > intercalateSuffix seed unf str = gintercalateSuffix unf str unf (repeatM seed)
3305-- > intercalateSuffix a unf str = concatUnfold unf $ intersperseSuffix a str
3307-- | 'intersperseSuffix' followed by unfold and concat.
3309-- > unlines = intercalateSuffix "\n" UF.fromList
3311-- >>> intercalate "\n" UF.fromList ["abc", "def", "ghi"]
3312-- > "abc\ndef\nghi\n"
3315{-# INLINE intercalateSuffix #-}
3316intercalateSuffix :: (IsStream t, Monad m)
3317 => b -> Unfold m b c -> t m b -> t m c
3318intercalateSuffix seed unf str = fromStreamD $ D.concatMapU unf
3319 $ D.intersperseSuffix (return seed) (D.toStreamD str)
3321-- interposeSuffix x unf str = gintercalateSuffix unf str UF.identity (repeat x)
3323-- | Unfold the elements of a stream, append the given element after each
3324-- unfolded stream and then concat them into a single stream.
3326-- > unlines = S.interposeSuffix '\n'
3329{-# INLINE interposeSuffix #-}
3330interposeSuffix :: (IsStream t, Monad m)
3331 => c -> Unfold m b c -> t m b -> t m c
3332interposeSuffix x unf str =
3333 D.fromStreamD $ D.interposeSuffix (return x) unf (D.toStreamD str)
3335------------------------------------------------------------------------------
3337------------------------------------------------------------------------------
3339-- | Like 'iterateM' but using a stream generator function.
3343{-# INLINE concatMapIterateWith #-}
3346 => (forall c. t m c -> t m c -> t m c)
3350concatMapIterateWith combine f = concatMapWith combine go
3352 go x = yield x `combine` concatMapWith combine go (f x)
3354-- concatMapIterateLeftsWith
3356-- | Traverse a forest with recursive tree structures whose non-leaf nodes are
3357-- of type @a@ and leaf nodes are of type @b@, flattening all the trees into
3358-- streams and combining the streams into a single stream consisting of both
3359-- leaf and non-leaf nodes.
3361-- 'concatMapTreeWith' is a generalization of 'concatMap', using a recursive
3362-- feedback loop to append the non-leaf nodes back to the input stream enabling
3363-- recursive traversal. 'concatMap' flattens a single level nesting whereas
3364-- 'concatMapTreeWith' flattens a recursively nested structure.
3366-- Traversing a directory tree recursively is a canonical use case of
3367-- 'concatMapTreeWith'.
3370-- concatMapTreeWith combine f xs = concatMapIterateWith combine g xs
3372-- g (Left tree) = f tree
3373-- g (Right leaf) = nil
3378{-# INLINE concatMapTreeWith #-}
3381 => (forall c. t m c -> t m c -> t m c)
3382 -> (a -> t m (Either a b))
3383 -> t m (Either a b) -- Should be t m a?
3385concatMapTreeWith combine f = concatMapWith combine go
3387 go (Left tree) = yield (Left tree) `combine` concatMapWith combine go (f tree)
3388 go (Right leaf) = yield $ Right leaf
3391-- | Like concatMapTreeWith but produces only stream of leaf elements.
3392-- On an either stream, iterate lefts but yield only rights.
3394-- concatMapEitherYieldRightsWith combine f xs =
3395-- catRights $ concatMapTreeWith combine f xs
3397{-# INLINE concatMapEitherYieldRightsWith #-}
3398concatMapEitherYieldRightsWith :: (IsStream t, MonadAsync m)
3399 => _ -> (a -> t m (Either a b)) -> t m (Either a b) -> t m b
3400concatMapEitherYieldRightsWith combine f xs = undefined
3404{-# INLINE concatUnfoldTree #-}
3405concatUnfoldTree :: (IsStream t, MonadAsync m)
3406 => Unfold m a (Either a b) -> t m (Either a b) -> t m (Either a b)
3407concatUnfoldTree unf xs = undefined
3410------------------------------------------------------------------------------
3412------------------------------------------------------------------------------
3414-- We can perhaps even implement the SVar using this. The stream we are mapping
3415-- on is the work queue. When evaluated it results in either a leaf element to
3416-- yield or a tail stream to queue back to the work queue.
3418-- | Flatten a stream with a feedback loop back into the input.
3420-- For example, exceptions generated by the output stream can be fed back to
3421-- the input to take any corrective action. The corrective action may be to
3422-- retry the action or do nothing or log the errors. For the retry case we need
3427{-# INLINE concatMapLoopWith #-}
3429 :: (IsStream t, MonadAsync m)
3430 => (forall x. t m x -> t m x -> t m x)
3431 -> (a -> t m (Either b c))
3432 -> (b -> t m a) -- ^ feedback function to feed @b@ back into input
3435concatMapLoopWith combine f fb xs =
3436 concatMapWith combine go $ concatMapWith combine f xs
3438 go (Left b) = concatMapLoopWith combine f fb $ fb b
3439 go (Right c) = yield c
3441-- | Concat a stream of trees, generating only leaves.
3443-- Compare with 'concatMapTreeWith'. While the latter returns all nodes in the
3444-- tree, this one returns only the leaves.
3446-- Traversing a directory tree recursively and yielding on the files is a
3447-- canonical use case of 'concatMapTreeYieldLeavesWith'.
3450-- concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield
3455{-# INLINE concatMapTreeYieldLeavesWith #-}
3456concatMapTreeYieldLeavesWith
3457 :: (IsStream t, MonadAsync m)
3458 => (forall x. t m x -> t m x -> t m x)
3459 -> (a -> t m (Either a b))
3462concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield
3464------------------------------------------------------------------------------
3466------------------------------------------------------------------------------
3468-- Splitting operations that take a predicate and a Fold can be
3469-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
3470-- can be expressed using parseMany when used with an appropriate Parser.
3472-- XXX We need takeGE/takeBetween to implement "some" using "many".
3474-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
3477-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
3480-- >>> S.toList $ S.parseMany (PR.take 2 $ PR.fromFold FL.sum) $ S.fromList [1..10]
3483-- >>> S.toList $ S.parseMany (PR.line FL.toList) $ S.fromList "hello\nworld"
3484-- > ["hello\n","world"]
3488{-# INLINE parseMany #-}
3490 :: (IsStream t, MonadThrow m)
3495 D.fromStreamD $ D.parseMany (PRD.fromParserK p) (D.toStreamD m)
3497-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
3498-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
3499-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
3500-- output of @test@ is discarded and the output of @collect@ is emitted in the
3501-- output stream. The parser fails if @collect@ fails.
3505{-# INLINE parseManyTill #-}
3507 -- (IsStream t, MonadThrow m) =>
3512parseManyTill = undefined
3514-- | Iterate a parser generating function on a stream. The initial value @b@ is
3515-- used to generate the first parser, the parser is applied on the stream and
3516-- the result is used to generate the next parser and so on.
3518-- >>> S.toList $ S.map getSum $ S.parseIterate (\b -> PR.take 2 (FL.mconcatTo b)) 0 $ S.map Sum $ S.fromList [1..10]
3519-- > [3,10,21,36,55,55]
3521-- This is the streaming equivalent of monad like sequenced application of
3522-- parsers where next parser is dependent on the previous parser.
3526{-# INLINE parseIterate #-}
3528 :: (IsStream t, MonadThrow m)
3529 => (b -> Parser m a b)
3533parseIterate f i m = D.fromStreamD $
3534 D.parseIterate (PRD.fromParserK . f) i (D.toStreamD m)
3536------------------------------------------------------------------------------
3537-- Grouping/Splitting
3538------------------------------------------------------------------------------
3540------------------------------------------------------------------------------
3541-- Grouping without looking at elements
3542------------------------------------------------------------------------------
3544------------------------------------------------------------------------------
3546------------------------------------------------------------------------------
3549------------------------------------------------------------------------------
3551------------------------------------------------------------------------------
3553------------------------------------------------------------------------------
3554-- Generalized grouping
3555------------------------------------------------------------------------------
3557-- This combinator is the most general grouping combinator and can be used to
3558-- implement all other grouping combinators.
3560-- XXX check if this can implement the splitOn combinator i.e. we can slide in
3561-- new elements, slide out old elements and incrementally compute the hash.
3562-- Also, can we implement the windowed classification combinators using this?
3564-- In fact this is a parse. Instead of using a special return value in the fold
3565-- we are using a mapping function.
3567-- Note that 'scanl'' (usually followed by a map to extract the desired value
3568-- from the accumulator) can be used to realize many implementations e.g. a
3569-- sliding window implementation. A scan followed by a mapMaybe is also a good
3570-- pattern to express many problems where we want to emit a filtered output and
3571-- not emit an output on every input.
3573-- Passing on of the initial accumulator value to the next fold is equivalent
3574-- to returning the leftover concept.
3577-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
3578-- @splitter@ is applied on the accumulator of the fold every time an item is
3579-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
3580-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
3581-- in the output stream and the initial value of the accumulator for the next
3582-- group's fold. This allows us to control whether to start fresh for the next
3583-- fold or to continue from the previous fold's output.
3585{-# INLINE groupScan #-}
3587 :: (IsStream t, Monad m)
3588 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
3589groupScan split fold m = undefined
3592-- | Group the input stream into groups of @n@ elements each and then fold each
3593-- group using the provided fold function.
3595-- >> S.toList $ S.chunksOf 2 FL.sum (S.enumerateFromTo 1 10)
3598-- This can be considered as an n-fold version of 'ltake' where we apply
3599-- 'ltake' repeatedly on the leftover stream until the stream exhausts.
3602{-# INLINE chunksOf #-}
3604 :: (IsStream t, Monad m)
3605 => Int -> Fold m a b -> t m a -> t m b
3606chunksOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
3611{-# INLINE chunksOf2 #-}
3613 :: (IsStream t, Monad m)
3614 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
3615chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
3617-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
3618-- @n@ elements each.
3620-- Same as the following but may be more efficient:
3622-- > arraysOf n = S.chunksOf n (A.writeN n)
3625{-# INLINE arraysOf #-}
3626arraysOf :: (IsStream t, MonadIO m, Storable a)
3627 => Int -> t m a -> t m (Array a)
3628arraysOf n = chunksOf n (writeNUnsafe n)
3630-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
3631-- XXX add this example after fixing the serial stream rate control
3632-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
3635-- | Group the input stream into windows of @n@ second each and then fold each
3636-- group using the provided fold function.
3639{-# INLINE intervalsOf #-}
3641 :: (IsStream t, MonadAsync m)
3642 => Double -> Fold m a b -> t m a -> t m b
3644 splitWithSuffix isNothing (FL.lcatMaybes f)
3645 (interjectSuffix n (return Nothing) (Serial.map Just xs))
3647------------------------------------------------------------------------------
3649------------------------------------------------------------------------------
3651-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
3652-- first group, if @a \`cmp` b@ is 'True' then @b@ is also assigned to the same
3653-- group. If @a \`cmp` c@ is 'True' then @c@ is also assigned to the same
3654-- group and so on. When the comparison fails a new group is started. Each
3655-- group is folded using the fold @f@ and the result of the fold is emitted in
3656-- the output stream.
3658-- >>> S.toList $ S.groupsBy (>) FL.toList $ S.fromList [1,3,7,0,2,5]
3659-- > [[1,3,7],[0,2,5]]
3662{-# INLINE groupsBy #-}
3664 :: (IsStream t, Monad m)
3669groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
3671-- | Unlike @groupsBy@ this function performs a rolling comparison of two
3672-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
3673-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
3674-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
3675-- 'True' then @c@ is also assigned to the same group and so on. When the
3676-- comparison fails a new group is started. Each group is folded using the fold
3679-- >>> S.toList $ S.groupsByRolling (\a b -> a + 1 == b) FL.toList $ S.fromList [1,2,3,7,8,9]
3680-- > [[1,2,3],[7,8,9]]
3683{-# INLINE groupsByRolling #-}
3685 :: (IsStream t, Monad m)
3690groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
3693-- > groups = groupsBy (==)
3694-- > groups = groupsByRolling (==)
3696-- Groups contiguous spans of equal elements together in individual groups.
3698-- >>> S.toList $ S.groups FL.toList $ S.fromList [1,1,2,2]
3702groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
3703groups = groupsBy (==)
3705------------------------------------------------------------------------------
3706-- N-ary split on a predicate
3707------------------------------------------------------------------------------
3709-- TODO: Use a Splitter configuration similar to the "split" package to make it
3710-- possible to express all splitting combinations. In general, we can have
3711-- infix/suffix/prefix/condensing of separators, dropping both leading/trailing
3712-- separators. We can have a single split operation taking the splitter config
3715-- | Split on an infixed separator element, dropping the separator. Splits the
3716-- stream on separator elements determined by the supplied predicate, separator
3717-- is considered as infixed between two segments, if one side of the separator
3718-- is missing then it is parsed as an empty stream. The supplied 'Fold' is
3719-- applied on the split segments. With '-' representing non-separator elements
3720-- and '.' as separator, 'splitOn' splits as follows:
3723-- "--.--" => "--" "--"
3728-- @splitOn (== x)@ is an inverse of @intercalate (S.yield x)@
3730-- Let's use the following definition for illustration:
3732-- > splitOn' p xs = S.toList $ S.splitOn p (FL.toList) (S.fromList xs)
3734-- >>> splitOn' (== '.') ""
3737-- >>> splitOn' (== '.') "."
3740-- >>> splitOn' (== '.') ".a"
3743-- >>> splitOn' (== '.') "a."
3746-- >>> splitOn' (== '.') "a.b"
3749-- >>> splitOn' (== '.') "a..b"
3754-- This can be considered as an n-fold version of 'breakOn' where we apply
3755-- 'breakOn' successively on the input stream, dropping the first element
3756-- of the second segment after each break.
3758{-# INLINE splitOn #-}
3760 :: (IsStream t, Monad m)
3761 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3762splitOn predicate f m =
3763 D.fromStreamD $ D.splitBy predicate f (D.toStreamD m)
3765-- | Like 'splitOn' but the separator is considered as suffixed to the segments
3766-- in the stream. A missing suffix at the end is allowed. A separator at the
3767-- beginning is parsed as empty segment. With '-' representing elements and
3768-- '.' as separator, 'splitOnSuffix' splits as follows:
3771-- "--.--." => "--" "--"
3772-- "--.--" => "--" "--"
3776-- > splitOnSuffix' p xs = S.toList $ S.splitSuffixBy p (FL.toList) (S.fromList xs)
3778-- >>> splitOnSuffix' (== '.') ""
3781-- >>> splitOnSuffix' (== '.') "."
3784-- >>> splitOnSuffix' (== '.') "a"
3787-- >>> splitOnSuffix' (== '.') ".a"
3790-- >>> splitOnSuffix' (== '.') "a."
3793-- >>> splitOnSuffix' (== '.') "a.b"
3796-- >>> splitOnSuffix' (== '.') "a.b."
3799-- >>> splitOnSuffix' (== '.') "a..b.."
3802-- > lines = splitOnSuffix (== '\n')
3806-- This can be considered as an n-fold version of 'breakPost' where we apply
3807-- 'breakPost' successively on the input stream, dropping the first element
3808-- of the second segment after each break.
3810{-# INLINE splitOnSuffix #-}
3812 :: (IsStream t, Monad m)
3813 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3814splitOnSuffix predicate f m =
3815 D.fromStreamD $ D.splitSuffixBy predicate f (D.toStreamD m)
3817-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
3818-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
3819-- @["a","b"]@. In other words, its like parsing words from whitespace
3822-- > wordsBy' p xs = S.toList $ S.wordsBy p (FL.toList) (S.fromList xs)
3824-- >>> wordsBy' (== ',') ""
3827-- >>> wordsBy' (== ',') ","
3830-- >>> wordsBy' (== ',') ",a,,b,"
3833-- > words = wordsBy isSpace
3837-- It is equivalent to splitting in any of the infix/prefix/suffix styles
3838-- followed by removal of empty segments.
3839{-# INLINE wordsBy #-}
3841 :: (IsStream t, Monad m)
3842 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3843wordsBy predicate f m =
3844 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
3846-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
3849-- > splitWithSuffix' p xs = S.toList $ S.splitWithSuffix p (FL.toList) (S.fromList xs)
3851-- >>> splitWithSuffix' (== '.') ""
3854-- >>> splitWithSuffix' (== '.') "."
3857-- >>> splitWithSuffix' (== '.') "a"
3860-- >>> splitWithSuffix' (== '.') ".a"
3863-- >>> splitWithSuffix' (== '.') "a."
3866-- >>> splitWithSuffix' (== '.') "a.b"
3869-- >>> splitWithSuffix' (== '.') "a.b."
3872-- >>> splitWithSuffix' (== '.') "a..b.."
3873-- > ["a.",".","b.","."]
3877-- This can be considered as an n-fold version of 'breakPost' where we apply
3878-- 'breakPost' successively on the input stream.
3880{-# INLINE splitWithSuffix #-}
3882 :: (IsStream t, Monad m)
3883 => (a -> Bool) -> Fold m a b -> t m a -> t m b
3884splitWithSuffix predicate f m =
3885 D.fromStreamD $ D.splitSuffixBy' predicate f (D.toStreamD m)
3887------------------------------------------------------------------------------
3888-- Split on a delimiter sequence
3889------------------------------------------------------------------------------
3891-- Int list examples for splitOn:
3893-- >>> splitList [] [1,2,3,3,4]
3894-- > [[1],[2],[3],[3],[4]]
3896-- >>> splitList [5] [1,2,3,3,4]
3899-- >>> splitList [1] [1,2,3,3,4]
3902-- >>> splitList [4] [1,2,3,3,4]
3905-- >>> splitList [2] [1,2,3,3,4]
3908-- >>> splitList [3] [1,2,3,3,4]
3911-- >>> splitList [3,3] [1,2,3,3,4]
3914-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
3917-- | Like 'splitOn' but the separator is a sequence of elements instead of a
3920-- For illustration, let's define a function that operates on pure lists:
3923-- splitOnSeq' pat xs = S.toList $ S.splitOnSeq (A.fromList pat) (FL.toList) (S.fromList xs)
3926-- >>> splitOnSeq' "" "hello"
3927-- > ["h","e","l","l","o"]
3929-- >>> splitOnSeq' "hello" ""
3932-- >>> splitOnSeq' "hello" "hello"
3935-- >>> splitOnSeq' "x" "hello"
3938-- >>> splitOnSeq' "h" "hello"
3941-- >>> splitOnSeq' "o" "hello"
3944-- >>> splitOnSeq' "e" "hello"
3947-- >>> splitOnSeq' "l" "hello"
3950-- >>> splitOnSeq' "ll" "hello"
3953-- 'splitOnSeq' is an inverse of 'intercalate'. The following law always holds:
3955-- > intercalate . splitOn == id
3957-- The following law holds when the separator is non-empty and contains none of
3958-- the elements present in the input lists:
3960-- > splitOn . intercalate == id
3964-- XXX We can use a polymorphic vector implemented by Array# to represent the
3965-- sequence, that way we can avoid the Storable constraint. If we still need
3966-- Storable Array for performance, we can use a separate splitOnArray API for
3967-- that. We can also have an API where the sequence itself is a lazy stream, so
3968-- that we can search files in files for example.
3969{-# INLINE splitOnSeq #-}
3971 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
3972 => Array a -> Fold m a b -> t m a -> t m b
3973splitOnSeq patt f m = D.fromStreamD $ D.splitOn patt f (D.toStreamD m)
3976-- This can be implemented easily using Rabin Karp
3977-- | Split on any one of the given patterns.
3978{-# INLINE splitOnAny #-}
3980 :: (IsStream t, Monad m, Storable a, Integral a)
3981 => [Array a] -> Fold m a b -> t m a -> t m b
3982splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
3985-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
3986-- of a predicate for a single element.
3988-- > splitSuffixOn_ pat xs = S.toList $ S.splitSuffixOn (A.fromList pat) (FL.toList) (S.fromList xs)
3990-- >>> splitSuffixOn_ "." ""
3993-- >>> splitSuffixOn_ "." "."
3996-- >>> splitSuffixOn_ "." "a"
3999-- >>> splitSuffixOn_ "." ".a"
4002-- >>> splitSuffixOn_ "." "a."
4005-- >>> splitSuffixOn_ "." "a.b"
4008-- >>> splitSuffixOn_ "." "a.b."
4011-- >>> splitSuffixOn_ "." "a..b.."
4014-- > lines = splitSuffixOn "\n"
4017{-# INLINE splitOnSuffixSeq #-}
4019 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
4020 => Array a -> Fold m a b -> t m a -> t m b
4021splitOnSuffixSeq patt f m =
4022 D.fromStreamD $ D.splitSuffixOn False patt f (D.toStreamD m)
4025-- | Like 'splitOn' but drops any empty splits.
4027{-# INLINE wordsOn #-}
4029 :: (IsStream t, Monad m, Storable a, Eq a)
4030 => Array a -> Fold m a b -> t m a -> t m b
4031wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
4034-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
4036-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
4038-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
4040-- >>> splitOn'_ "" "hello"
4041-- > ["h","","e","","l","","l","","o"]
4043-- >>> splitOn'_ "hello" ""
4046-- >>> splitOn'_ "hello" "hello"
4049-- >>> splitOn'_ "x" "hello"
4052-- >>> splitOn'_ "h" "hello"
4055-- >>> splitOn'_ "o" "hello"
4058-- >>> splitOn'_ "e" "hello"
4061-- >>> splitOn'_ "l" "hello"
4062-- > ["he","l","","l","o"]
4064-- >>> splitOn'_ "ll" "hello"
4068{-# INLINE splitBySeq #-}
4070 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
4071 => Array a -> Fold m a b -> t m a -> t m b
4072splitBySeq patt f m =
4073 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
4075-- | Like 'splitSuffixOn' but keeps the suffix intact in the splits.
4077-- > splitSuffixOn'_ pat xs = S.toList $ FL.splitSuffixOn' (A.fromList pat) (FL.toList) (S.fromList xs)
4079-- >>> splitSuffixOn'_ "." ""
4082-- >>> splitSuffixOn'_ "." "."
4085-- >>> splitSuffixOn'_ "." "a"
4088-- >>> splitSuffixOn'_ "." ".a"
4091-- >>> splitSuffixOn'_ "." "a."
4094-- >>> splitSuffixOn'_ "." "a.b"
4097-- >>> splitSuffixOn'_ "." "a.b."
4100-- >>> splitSuffixOn'_ "." "a..b.."
4101-- > ["a.",".","b.","."]
4104{-# INLINE splitWithSuffixSeq #-}
4106 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
4107 => Array a -> Fold m a b -> t m a -> t m b
4108splitWithSuffixSeq patt f m =
4109 D.fromStreamD $ D.splitSuffixOn True patt f (D.toStreamD m)
4112-- This can be implemented easily using Rabin Karp
4113-- | Split post any one of the given patterns.
4114{-# INLINE splitSuffixOnAny #-}
4116 :: (IsStream t, Monad m, Storable a, Integral a)
4117 => [Array a] -> Fold m a b -> t m a -> t m b
4118splitSuffixOnAny subseq f m = undefined
4119 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
4122------------------------------------------------------------------------------
4124------------------------------------------------------------------------------
4126-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
4127-- an input stream @t m (f a)@ using the @splitter@ function. Container
4128-- elements @f a@ are collected until a split occurs, then all the elements
4129-- before the split are joined using the @joiner@ function.
4131-- For example, if we have a stream of @Array Word8@, we may want to split the
4132-- stream into arrays representing lines separated by '\n' byte such that the
4133-- resulting stream after a split would be one array for each line.
4135-- CAUTION! This is not a true streaming function as the container size after
4136-- the split and merge may not be bounded.
4139{-# INLINE splitInnerBy #-}
4141 :: (IsStream t, Monad m)
4142 => (f a -> m (f a, Maybe (f a))) -- splitter
4143 -> (f a -> f a -> m (f a)) -- joiner
4146splitInnerBy splitter joiner xs =
4147 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
4149-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
4153{-# INLINE splitInnerBySuffix #-}
4155 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
4156 => (f a -> m (f a, Maybe (f a))) -- splitter
4157 -> (f a -> f a -> m (f a)) -- joiner
4160splitInnerBySuffix splitter joiner xs =
4161 D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
4163------------------------------------------------------------------------------
4164-- Reorder in sequence
4165------------------------------------------------------------------------------
4168-- Buffer until the next element in sequence arrives. The function argument
4169-- determines the difference in sequence numbers. This could be useful in
4170-- implementing sequenced streams, for example, TCP reassembly.
4171{-# INLINE reassembleBy #-}
4173 :: (IsStream t, Monad m)
4178reassembleBy = undefined
4181------------------------------------------------------------------------------
4183------------------------------------------------------------------------------
4185-- | Tap the data flowing through a stream into a 'Fold'. For example, you may
4186-- add a tap to log the contents flowing through the stream. The fold is used
4187-- only for effects, its result is discarded.
4192-- -----stream m a ---------------stream m a-----
4197-- > S.drain $ S.tap (FL.drainBy print) (S.enumerateFromTo 1 2)
4202-- Compare with 'trace'.
4206tap :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> t m a
4207tap f xs = D.fromStreamD $ D.tap f (D.toStreamD xs)
4209-- | @tapOffsetEvery offset n@ taps every @n@th element in the stream
4210-- starting at @offset@. @offset@ can be between @0@ and @n - 1@. Offset 0
4211-- means start at the first element in the stream. If the offset is outside
4212-- this range then @offset `mod` n@ is used as offset.
4215-- >>> S.drain $ S.tapOffsetEvery 0 2 (FL.mapM print FL.toList) $ S.enumerateFromTo 0 10
4221{-# INLINE tapOffsetEvery #-}
4222tapOffsetEvery :: (IsStream t, Monad m)
4223 => Int -> Int -> FL.Fold m a b -> t m a -> t m a
4224tapOffsetEvery offset n f xs =
4225 D.fromStreamD $ D.tapOffsetEvery offset n f (D.toStreamD xs)
4227-- | Redirect a copy of the stream to a supplied fold and run it concurrently
4228-- in an independent thread. The fold may buffer some elements. The buffer size
4229-- is determined by the prevailing 'maxBuffer' setting.
4234-- -----stream m a ---------------stream m a-----
4239-- > S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
4244-- Exceptions from the concurrently running fold are propagated to the current
4245-- computation. Note that, because of buffering in the fold, exceptions may be
4246-- delayed and may not correspond to the current element being processed in the
4247-- parent stream, but we guarantee that before the parent stream stops the tap
4248-- finishes and all exceptions from it are drained.
4251-- Compare with 'tap'.
4254{-# INLINE tapAsync #-}
4255tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a
4256tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)
4258-- | @pollCounts predicate transform fold stream@ counts those elements in the
4259-- stream that pass the @predicate@. The resulting count stream is sent to
4260-- another thread which transforms it using @transform@ and then folds it using
4261-- @fold@. The thread is automatically cleaned up if the stream stops or
4262-- aborts due to exception.
4264-- For example, to print the count of elements processed every second:
4267-- > S.drain $ S.pollCounts (const True) (S.rollingMap (-) . S.delayPost 1) (FL.drainBy print)
4268-- $ S.enumerateFrom 0
4271-- Note: This may not work correctly on 32-bit machines.
4275{-# INLINE pollCounts #-}
4277 (IsStream t, MonadAsync m)
4279 -> (t m Int -> t m Int)
4283pollCounts predicate transf f xs =
4285 $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f
4288-- | Calls the supplied function with the number of elements consumed
4289-- every @n@ seconds. The given function is run in a separate thread
4290-- until the end of the stream. In case there is an exception in the
4291-- stream the thread is killed during the next major GC.
4293-- Note: The action is not guaranteed to run if the main thread exits.
4296-- > delay n = threadDelay (round $ n * 1000000) >> return n
4297-- > S.drain $ S.tapRate 2 (\\n -> print $ show n ++ " elements processed") (delay 1 S.|: delay 0.5 S.|: delay 0.5 S.|: S.nil)
4298-- 2 elements processed
4299-- 1 elements processed
4302-- Note: This may not work correctly on 32-bit machines.
4305{-# INLINE tapRate #-}
4307 (IsStream t, MonadAsync m, MonadCatch m)
4312tapRate n f xs = D.fromStreamD $ D.tapRate n f $ D.toStreamD xs
4314-- | Apply a monadic function to each element flowing through the stream and
4315-- discard the results.
4318-- > S.drain $ S.trace print (S.enumerateFromTo 1 2)
4323-- Compare with 'tap'.
4327trace :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m a
4328trace f = mapM (\x -> void (f x) >> return x)
4330-- | Perform a side effect before yielding each element of the stream and
4331-- discard the results.
4334-- > S.drain $ S.trace_ (print "got here") (S.enumerateFromTo 1 2)
4339-- See also: 'trace', 'interspersePrefix_'
4342{-# INLINE trace_ #-}
4343trace_ :: (IsStream t, Monad m) => m b -> t m a -> t m a
4344trace_ eff = Serial.mapM (\x -> eff >> return x)
4346------------------------------------------------------------------------------
4347-- Windowed classification
4348------------------------------------------------------------------------------
4350-- We divide the stream into windows or chunks in space or time and each window
4351-- can be associated with a key, all events associated with a particular key in
4352-- the window can be folded to a single result. The stream can be split into
4353-- windows by size or by using a split predicate on the elements in the stream.
4354-- For example, when we receive a closing flag, we can close the window.
4356-- A "chunk" is a space window and a "session" is a time window. Are there any
4357-- other better short words to describe them. An alternative is to use
4358-- "swindow" and "twindow". Another word for "session" could be "spell".
4360-- TODO: To mark the position in space or time we can have Indexed or
4361-- TimeStamped types. That can make it easy to deal with the position indices
4364------------------------------------------------------------------------------
4365-- Keyed Sliding Windows
4366------------------------------------------------------------------------------
4369{-# INLINABLE classifySlidingChunks #-}
4370classifySlidingChunks
4371 :: (IsStream t, MonadAsync m, Ord k)
4372 => Int -- ^ window size
4373 -> Int -- ^ window slide
4374 -> Fold m a b -- ^ Fold to be applied to window events
4375 -> t m (k, a, Bool) -- ^ window key, data, close event
4377classifySlidingChunks wsize wslide (Fold step initial extract) str
4380-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
4381-- slide the send window when an ack is received and we slide the receive
4382-- window when a sequence is complete. Sliding is stateful in case of TCP,
4383-- sliding releases the send buffer or makes data available to the user from
4384-- the receive buffer.
4385{-# INLINABLE classifySlidingSessions #-}
4386classifySlidingSessions
4387 :: (IsStream t, MonadAsync m, Ord k)
4388 => Double -- ^ timer tick in seconds
4389 -> Double -- ^ time window size
4390 -> Double -- ^ window slide
4391 -> Fold m a b -- ^ Fold to be applied to window events
4392 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
4394classifySlidingSessions tick interval slide (Fold step initial extract) str
4398------------------------------------------------------------------------------
4399-- Sliding Window Buffers
4400------------------------------------------------------------------------------
4402-- These buffered versions could be faster than concurrent incremental folds of
4403-- all overlapping windows as in many cases we may not need all the values to
4404-- compute the fold, we can just compute the result using the old value and new
4405-- value. However, we may need the buffer once in a while, for example for
4406-- string search we usually compute the hash incrementally but when the hash
4407-- matches the hash of the pattern we need to compare the whole string.
4409-- XXX we should be able to implement sequence based splitting combinators
4410-- using this combinator.
4413-- | Buffer n elements of the input in a ring buffer. When t new elements are
4414-- collected, slide the window to remove the same number of oldest elements,
4415-- insert the new elements, and apply an incremental fold on the sliding
4416-- window, supplying the outgoing elements, the new ring buffer as arguments.
4418 :: (IsStream t, Monad m, Ord a, Storable a)
4419 => Int -- window size
4420 -> Int -- window slide
4421 -> Fold m (Ring a, Array a) b
4424slidingChunkBuffer = undefined
4426-- Buffer n seconds worth of stream elements of the input in a radix tree.
4427-- Every t seconds, remove the items that are older than n seconds, and apply
4428-- an incremental fold on the sliding window, supplying the outgoing elements,
4429-- and the new radix tree buffer as arguments.
4431 :: (IsStream t, Monad m, Ord a, Storable a)
4432 => Int -- window size
4434 -> Fold m (RTree a, Array a) b
4437slidingSessionBuffer = undefined
4440------------------------------------------------------------------------------
4441-- Keyed Session Windows
4442------------------------------------------------------------------------------
4445-- | Keyed variable size space windows. Close the window if we do not receive a
4446-- window event in the next "spaceout" elements.
4447{-# INLINABLE classifyChunksBy #-}
4449 :: (IsStream t, MonadAsync m, Ord k)
4450 => Int -- ^ window spaceout (spread)
4451 -> Bool -- ^ reset the spaceout when a chunk window element is received
4452 -> Fold m a b -- ^ Fold to be applied to chunk window elements
4453 -> t m (k, a, Bool) -- ^ chunk key, data, last element
4455classifyChunksBy spanout reset (Fold step initial extract) str = undefined
4457-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
4458-- received within the chunk size window. The chunk gets closed only if no
4459-- element is received within the chunk window.
4461{-# INLINABLE classifyKeepAliveChunks #-}
4462classifyKeepAliveChunks
4463 :: (IsStream t, MonadAsync m, Ord k)
4464 => Int -- ^ window spaceout (spread)
4465 -> Fold m a b -- ^ Fold to be applied to chunk window elements
4466 -> t m (k, a, Bool) -- ^ chunk key, data, last element
4468classifyKeepAliveChunks spanout = classifyChunksBy spanout True
4471#if __GLASGOW_HASKELL__ < 800
4475data SessionState t m k a b = SessionState
4476 { sessionCurTime :: !AbsTime -- ^ time since last event
4477 , sessionEventTime :: !AbsTime -- ^ time as per last event
4478 , sessionCount :: !Int -- ^ total number sessions in progress
4479 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
4480 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
4481 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
4486-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
4487-- events in an input event stream into sessions based on a session key. Each
4488-- element in the stream is an event consisting of a triple @(session key,
4489-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
4490-- the session. All the events belonging to a session are folded using the
4491-- fold @f@ until the fold returns a 'Left' result or a timeout has occurred.
4492-- The session key and the result of the fold are emitted in the output stream
4493-- when the session is purged.
4495-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
4496-- seconds, measured from the @timestamp@ of the first event in that session.
4497-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
4498-- every event received in the session.
4500-- @timestamp@ in an event characterizes the time when the input event was
4501-- generated, this is an absolute time measured from some @Epoch@. The notion
4502-- of current time is maintained by a monotonic event time clock using the
4503-- timestamps seen in the input stream. The latest timestamp seen till now is
4504-- used as the base for the current time. When no new events are seen, a timer
4505-- is started with a tick duration specified by @tick@. This timer is used to
4506-- detect session timeouts in the absence of new events.
4508-- The predicate @pred@ is invoked with the current session count, if it
4509-- returns 'True' a session is ejected from the session cache before inserting
4510-- a new session. This could be useful to alert or eject sessions when the
4511-- number of sessions becomes too high.
4516-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
4517{-# INLINABLE classifySessionsBy #-}
4519 :: (IsStream t, MonadAsync m, Ord k)
4520 => Double -- ^ timer tick in seconds
4521 -> Double -- ^ session timeout in seconds
4522 -> Bool -- ^ reset the timeout when an event is received
4523 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
4524 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4525 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4526 -> t m (k, b) -- ^ session key, fold result
4527classifySessionsBy tick tmout reset ejectPred
4528 (Fold step initial extract) str =
4529 concatMap sessionOutputStream $
4530 scanlMAfter' sstep (return szero) flush stream
4534 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
4535 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
4536 szero = SessionState
4537 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
4538 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
4540 , sessionTimerHeap = H.empty
4541 , sessionKeyValueMap = Map.empty
4542 , sessionOutputStream = K.nil
4545 -- We can eject sessions based on the current session count to limit
4546 -- memory consumption. There are two possible strategies:
4548 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
4549 -- threshold even before timeout, effectively reduce the timeout.
4550 -- 2) Drop creation of new sessions but keep accepting new events for the
4553 -- We use the first strategy as of now.
4555 -- Got a new stream input element
4556 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
4557 -- XXX we should use a heap in pinned memory to scale it to a large
4560 -- To detect session inactivity we keep a timestamp of the latest event
4561 -- in the Map along with the fold result. When we purge the session
4562 -- from the heap we match the timestamp in the heap with the timestamp
4563 -- in the Map, if the latest timestamp is newer and has not expired we
4564 -- reinsert the key in the heap.
4566 -- XXX if the key is an Int, we can also use an IntMap for slightly
4567 -- better performance.
4569 let curTime = max sessionEventTime timestamp
4573 Just (Tuple' _ acc) -> return acc
4574 new <- step old value
4575 return $ Tuple' timestamp new
4576 mOld = Map.lookup key sessionKeyValueMap
4578 acc@(Tuple' _ fres) <- accumulate mOld
4582 -- deleting a key from the heap is expensive, so we never
4583 -- delete a key from heap, we just purge it from the Map and it
4584 -- gets purged from the heap on timeout. We just need an extra
4585 -- lookup in the Map when the key is purged from the heap, that
4586 -- should not be expensive.
4588 let (mp, cnt) = case mOld of
4589 Nothing -> (sessionKeyValueMap, sessionCount)
4590 Just _ -> (Map.delete key sessionKeyValueMap
4593 { sessionCurTime = curTime
4594 , sessionEventTime = curTime
4595 , sessionCount = cnt
4596 , sessionKeyValueMap = mp
4597 , sessionOutputStream = yield (key, x)
4600 (hp1, mp1, out1, cnt1) <- do
4601 let vars = (sessionTimerHeap, sessionKeyValueMap,
4602 K.nil, sessionCount)
4604 -- inserting new entry
4606 -- Eject a session from heap and map is needed
4607 eject <- ejectPred sessionCount
4608 (hp, mp, out, cnt) <-
4613 -- Insert the new session in heap
4614 let expiry = addToAbsTime timestamp timeoutMs
4615 hp' = H.insert (Entry expiry key) hp
4616 in return (hp', mp, out, cnt + 1)
4617 -- updating old entry
4618 Just _ -> return vars
4620 let mp2 = Map.insert key acc mp1
4621 return $ SessionState
4622 { sessionCurTime = curTime
4623 , sessionEventTime = curTime
4624 , sessionCount = cnt1
4625 , sessionTimerHeap = hp1
4626 , sessionKeyValueMap = mp2
4627 , sessionOutputStream = out1
4630 -- Got a timer tick event
4631 sstep sessionState@SessionState{..} Nothing =
4632 let curTime = addToAbsTime sessionCurTime tickMs
4633 in ejectExpired sessionState curTime
4635 flush session@SessionState{..} = do
4636 (hp', mp', out, count) <-
4639 , sessionKeyValueMap
4644 { sessionCount = count
4645 , sessionTimerHeap = hp'
4646 , sessionKeyValueMap = mp'
4647 , sessionOutputStream = out
4655 -- delete from map and output the fold accumulator
4656 ejectEntry hp mp out cnt acc key = do
4658 let out1 = (key, fromEither sess) `K.cons` out
4659 let mp1 = Map.delete key mp
4660 return (hp, mp1, out1, cnt - 1)
4662 ejectAll (hp, mp, out, !cnt) = do
4663 let hres = H.uncons hp
4665 Just (Entry _ key, hp1) -> do
4666 r <- case Map.lookup key mp of
4667 Nothing -> return (hp1, mp, out, cnt)
4668 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
4671 assert (Map.null mp) (return ())
4672 return (hp, mp, out, cnt)
4674 ejectOne (hp, mp, out, !cnt) = do
4675 let hres = H.uncons hp
4677 Just (Entry expiry key, hp1) ->
4678 case Map.lookup key mp of
4679 Nothing -> ejectOne (hp1, mp, out, cnt)
4680 Just (Tuple' latestTS acc) -> do
4681 let expiry1 = addToAbsTime latestTS timeoutMs
4682 if not reset || expiry1 <= expiry
4683 then ejectEntry hp1 mp out cnt acc key
4685 -- reset the session timeout and continue
4686 let hp2 = H.insert (Entry expiry1 key) hp1
4687 in ejectOne (hp2, mp, out, cnt)
4689 assert (Map.null mp) (return ())
4690 return (hp, mp, out, cnt)
4692 ejectExpired session@SessionState{..} curTime = do
4693 (hp', mp', out, count) <-
4694 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
4696 { sessionCurTime = curTime
4697 , sessionCount = count
4698 , sessionTimerHeap = hp'
4699 , sessionKeyValueMap = mp'
4700 , sessionOutputStream = out
4705 ejectLoop hp mp out !cnt = do
4706 let hres = H.uncons hp
4708 Just (Entry expiry key, hp1) -> do
4710 if curTime >= expiry
4711 then return (True, False)
4717 case Map.lookup key mp of
4718 Nothing -> ejectLoop hp1 mp out cnt
4719 Just (Tuple' latestTS acc) -> do
4720 let expiry1 = addToAbsTime latestTS timeoutMs
4721 if expiry1 <= curTime || not reset || force
4723 (hp2,mp1,out1,cnt1) <-
4724 ejectEntry hp1 mp out cnt acc key
4725 ejectLoop hp2 mp1 out1 cnt1
4727 -- reset the session timeout and continue
4728 let hp2 = H.insert (Entry expiry1 key) hp1
4729 in ejectLoop hp2 mp out cnt
4730 else return (hp, mp, out, cnt)
4732 assert (Map.null mp) (return ())
4733 return (hp, mp, out, cnt)
4735 -- merge timer events in the stream
4736 stream = Serial.map Just str `Par.parallelFst` repeatM timer
4738 liftIO $ threadDelay (round $ tick * 1000000)
4741-- | Like 'classifySessionsOf' but the session is kept alive if an event is
4742-- received within the session window. The session times out and gets closed
4743-- only if no event is received within the specified session window size.
4745-- If the ejection predicate returns 'True', the session that was idle for
4746-- the longest time is ejected before inserting a new session.
4749-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
4754{-# INLINABLE classifyKeepAliveSessions #-}
4755classifyKeepAliveSessions
4756 :: (IsStream t, MonadAsync m, Ord k)
4757 => Double -- ^ session inactive timeout
4758 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
4759 -> Fold m a (Either b b) -- ^ Fold to be applied to session payload data
4760 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4762classifyKeepAliveSessions tmout =
4763 classifySessionsBy 1 tmout True
4765------------------------------------------------------------------------------
4766-- Keyed tumbling windows
4767------------------------------------------------------------------------------
4769-- Tumbling windows is a special case of sliding windows where the window slide
4770-- is the same as the window size. Or it can be a special case of session
4771-- windows where the reset flag is set to False.
4773-- XXX instead of using the early termination flag in the stream, we can use an
4774-- early terminating fold instead.
4777-- | Split the stream into fixed size chunks of specified size. Within each
4778-- such chunk fold the elements in buckets identified by the keys. A particular
4779-- bucket fold can be terminated early if a closing flag is encountered in an
4780-- element for that key.
4783{-# INLINABLE classifyChunksOf #-}
4785 :: (IsStream t, MonadAsync m, Ord k)
4786 => Int -- ^ window size
4787 -> Fold m a b -- ^ Fold to be applied to window events
4788 -> t m (k, a, Bool) -- ^ window key, data, close event
4790classifyChunksOf wsize = classifyChunksBy wsize False
4793-- | Split the stream into fixed size time windows of specified interval in
4794-- seconds. Within each such window, fold the elements in sessions identified
4795-- by the session keys. The fold result is emitted in the output stream if the
4796-- fold returns a 'Left' result or if the time window ends.
4798-- Session @timestamp@ in the input stream is an absolute time from some epoch,
4799-- characterizing the time when the input element was generated. To detect
4800-- session window end, a monotonic event time clock is maintained synced with
4801-- the timestamps with a clock resolution of 1 second.
4803-- If the ejection predicate returns 'True', the session with the longest
4804-- lifetime is ejected before inserting a new session.
4807-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
4812-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
4813-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
4816-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
4821{-# INLINABLE classifySessionsOf #-}
4823 :: (IsStream t, MonadAsync m, Ord k)
4824 => Double -- ^ time window size
4825 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
4826 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4827 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
4829classifySessionsOf interval =
4830 classifySessionsBy 1 interval False
4832------------------------------------------------------------------------------
4834------------------------------------------------------------------------------
4836-- | Run a side effect before the stream yields its first element.
4839{-# INLINE before #-}
4840before :: (IsStream t, Monad m) => m b -> t m a -> t m a
4841before action xs = D.fromStreamD $ D.before action $ D.toStreamD xs
4843-- | Run a side effect whenever the stream stops normally.
4845-- Prefer 'afterIO' over this as the @after@ action in this combinator is not
4846-- executed if the unfold is partially evaluated lazily and then garbage
4851after :: (IsStream t, Monad m) => m b -> t m a -> t m a
4852after action xs = D.fromStreamD $ D.after action $ D.toStreamD xs
4854-- | Run a side effect whenever the stream stops normally
4855-- or is garbage collected after a partial lazy evaluation.
4859{-# INLINE afterIO #-}
4860afterIO :: (IsStream t, MonadIO m, MonadBaseControl IO m) => m b -> t m a -> t m a
4861afterIO action xs = D.fromStreamD $ D.afterIO action $ D.toStreamD xs
4863-- | Run a side effect whenever the stream aborts due to an exception.
4866{-# INLINE onException #-}
4867onException :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
4868onException action xs = D.fromStreamD $ D.onException action $ D.toStreamD xs
4870-- | Run a side effect whenever the stream stops normally or aborts due to an
4873-- Prefer 'finallyIO' over this as the @after@ action in this combinator is not
4874-- executed if the unfold is partially evaluated lazily and then garbage
4878{-# INLINE finally #-}
4879finally :: (IsStream t, MonadCatch m) => m b -> t m a -> t m a
4880finally action xs = D.fromStreamD $ D.finally action $ D.toStreamD xs
4882-- | Run a side effect whenever the stream stops normally, aborts due to an
4883-- exception or if it is garbage collected after a partial lazy evaluation.
4887{-# INLINE finallyIO #-}
4888finallyIO :: (IsStream t, MonadAsync m, MonadCatch m) => m b -> t m a -> t m a
4889finallyIO action xs = D.fromStreamD $ D.finallyIO action $ D.toStreamD xs
4891-- | Run the first action before the stream starts and remember its output,
4892-- generate a stream using the output, run the second action using the
4893-- remembered value as an argument whenever the stream ends normally or due to
4896-- Prefer 'bracketIO' over this as the @after@ action in this combinator is not
4897-- executed if the unfold is partially evaluated lazily and then garbage
4901{-# INLINE bracket #-}
4902bracket :: (IsStream t, MonadCatch m)
4903 => m b -> (b -> m c) -> (b -> t m a) -> t m a
4904bracket bef aft bet = D.fromStreamD $
4905 D.bracket bef aft (toStreamD . bet)
4907-- | Run the first action before the stream starts and remember its output,
4908-- generate a stream using the output, run the second action using the
4909-- remembered value as an argument whenever the stream ends normally, due to
4910-- an exception or if it is garbage collected after a partial lazy evaluation.
4914{-# INLINE bracketIO #-}
4915bracketIO :: (IsStream t, MonadAsync m, MonadCatch m)
4916 => m b -> (b -> m c) -> (b -> t m a) -> t m a
4917bracketIO bef aft bet = D.fromStreamD $
4918 D.bracketIO bef aft (toStreamD . bet)
4920-- | When evaluating a stream if an exception occurs, stream evaluation aborts
4921-- and the specified exception handler is run with the exception as argument.
4924{-# INLINE handle #-}
4925handle :: (IsStream t, MonadCatch m, Exception e)
4926 => (e -> t m a) -> t m a -> t m a
4928 D.fromStreamD $ D.handle (D.toStreamD . handler) $ D.toStreamD xs
4930------------------------------------------------------------------------------
4931-- Generalize the underlying monad
4932------------------------------------------------------------------------------
4934-- | Transform the inner monad of a stream using a natural transformation.
4939hoist :: (Monad m, Monad n)
4940 => (forall x. m x -> n x) -> SerialT m a -> SerialT n a
4941hoist f xs = fromStreamS $ S.hoist f (toStreamS xs)
4943-- | Generalize the inner monad of the stream from 'Identity' to any monad.
4947{-# INLINE generally #-}
4948generally :: (IsStream t, Monad m) => t Identity a -> t m a
4949generally xs = fromStreamS $ S.hoist (return . runIdentity) (toStreamS xs)
4951------------------------------------------------------------------------------
4952-- Add and remove a monad transformer
4953------------------------------------------------------------------------------
4955-- | Lift the inner monad of a stream using a monad transformer.
4959{-# INLINE liftInner #-}
4960liftInner :: (Monad m, IsStream t, MonadTrans tr, Monad (tr m))
4961 => t m a -> t (tr m) a
4962liftInner xs = fromStreamD $ D.liftInner (toStreamD xs)
4964-- | Evaluate the inner monad of a stream as 'ReaderT'.
4968{-# INLINE runReaderT #-}
4969runReaderT :: (IsStream t, Monad m) => s -> t (ReaderT s m) a -> t m a
4970runReaderT s xs = fromStreamD $ D.runReaderT s (toStreamD xs)
4972-- | Evaluate the inner monad of a stream as 'StateT'.
4974-- This is supported only for 'SerialT' as concurrent state updation may not be
4979{-# INLINE evalStateT #-}
4980evalStateT :: Monad m => s -> SerialT (StateT s m) a -> SerialT m a
4981evalStateT s xs = fromStreamD $ D.evalStateT s (toStreamD xs)
4983-- | Run a stateful (StateT) stream transformation using a given state.
4985-- This is supported only for 'SerialT' as concurrent state updation may not be
4990{-# INLINE usingStateT #-}
4994 -> (SerialT (StateT s m) a -> SerialT (StateT s m) a)
4997usingStateT s f xs = evalStateT s $ f $ liftInner xs
4999-- | Evaluate the inner monad of a stream as 'StateT' and emit the resulting
5000-- state and value pair after each step.
5002-- This is supported only for 'SerialT' as concurrent state updation may not be
5007{-# INLINE runStateT #-}
5008runStateT :: Monad m => s -> SerialT (StateT s m) a -> SerialT m (s, a)
5009runStateT s xs = fromStreamD $ D.runStateT s (toStreamD xs)
5011-- | Run a stream transformation using a given environment.
5015{-# INLINE usingReaderT #-}
5017 :: (Monad m, IsStream t)
5019 -> (t (ReaderT r m) a -> t (ReaderT r m) a)
5022usingReaderT r f xs = runReaderT r $ f $ liftInner 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