composewell / streamly / 3152

{-# OPTIONS_GHC -Wno-orphans  #-}

#include "inline.hs"

-- |

-- Module      : Streamly.Internal.Data.Stream.IsStream

-- Copyright   : (c) 2017 Harendra Kumar

--

-- License     : BSD3

-- Maintainer  : streamly@composewell.com

-- Stability   : experimental

-- Portability : GHC

--

-- This is an Internal module consisting of released, unreleased and

-- unimplemented APIs. For stable and released APIs please see

-- "Streamly.Prelude" module. This module provides documentation only for the

-- unreleased and unimplemented APIs. For documentation on released APIs please

-- see "Streamly.Prelude" module.

module Streamly.Internal.Data.Stream.IsStream

    (

    -- * Construction

    -- ** Primitives

      K.nil

    , K.nilM

    , K.cons

    , (K..:)

    , consM

    , (|:)

    -- ** From Values

    , yield

    , yieldM

    , repeat

    , repeatM

    , replicate

    , replicateM

    -- ** Enumeration

    , Enumerable (..)

    , enumerate

    , enumerateTo

    -- ** Time Enumeration

    , times

    , absTimes

    , relTimes

    , durations

    , ticks

    , timeout

    , currentTime

    -- ** From Generators

    , unfoldr

    , unfoldrM

    , unfold

    , unfold0

    , fromIndices

    , fromIndicesM

    -- ** Iteration

    , iterate

    , iterateM

    -- ** Cyclic Elements

    , K.mfix

    -- ** From Containers

    , P.fromList

    , fromListM

    , K.fromFoldable

    , fromFoldableM

    , fromPrimIORef

    , fromCallback

    -- * Elimination

    -- ** Deconstruction

    , uncons

    , tail

    , init

    -- ** Folding

    -- ** Right Folds

    , foldrM

    , foldrS

    , foldrT

    , foldr

    -- ** Left Folds

    , foldl'

    , foldl1'

    , foldlM'

    -- ** Composable Left Folds

    , fold

    , parse

    , parseK

    , parseD

    -- ** Concurrent Folds

    , foldAsync

    , (|$.)

    , (|&.)

    -- ** Full Folds

    -- -- ** To Summary (Full Folds)

    , drain

    , last

    , length

    , sum

    , product

    , mconcat

    -- -- ** To Summary (Maybe) (Full Folds)

    , maximumBy

    , maximum

    , minimumBy

    , minimum

    , the

    -- ** Lazy Folds

    -- -- ** To Containers (Full Folds)

    , toList

    , toListRev

    , toPure

    , toPureRev

    -- ** Composable Left Folds

    , toStream    -- XXX rename to write?

    , toStreamRev -- XXX rename to writeRev?

    -- ** Partial Folds

    -- -- ** To Elements (Partial Folds)

    , drainN

    , drainWhile

    -- -- | Folds that extract selected elements of a stream or their properties.

    , (!!)

    , head

    , headElse

    , findM

    , find

    , lookup

    , findIndex

    , elemIndex

    -- -- ** To Boolean (Partial Folds)

    , null

    , elem

    , notElem

    , all

    , any

    , and

    , or

    -- ** Multi-Stream folds

    -- Full equivalence

    , eqBy

    , cmpBy

    -- finding subsequences

    , isPrefixOf

    , isSuffixOf

    , isInfixOf

    , isSubsequenceOf

    -- trimming sequences

    , stripPrefix

    , stripSuffix

    -- , stripInfix

    , dropPrefix

    , dropInfix

    , dropSuffix

    -- * Transformation

    , transform

    -- ** Mapping

    , Serial.map

    , sequence

    , mapM

    , smapM

    -- $smapM_Notes

    -- ** Special Maps

    , mapM_

    , trace

    , trace_

    , tap

    , tapOffsetEvery

    , tapAsync

    , tapRate

    , pollCounts

    -- ** Scanning

    -- ** Left scans

    , scanl'

    , scanlM'

    , scanlMAfter'

    , postscanl'

    , postscanlM'

    , prescanl'

    , prescanlM'

    , scanl1'

    , scanl1M'

    -- ** Scan Using Fold

    , scan

    , postscan

    -- XXX Once we have pipes the contravariant transformations can be

    -- represented by attaching pipes before a transformation.

    --

    -- , lscanl'

    -- , lscanlM'

    -- , lscanl1'

    -- , lscanl1M'

    --

    -- , lpostscanl'

    -- , lpostscanlM'

    -- , lprescanl'

    -- , lprescanlM'

    -- ** Concurrent Transformation

    , D.mkParallel

    -- Par.mkParallel

    , applyAsync

    , (|$)

    , (|&)

    -- ** Filtering

    , filter

    , filterM

    -- ** Deleting Elements

    , deleteBy

    , uniq

    -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path

    -- , uniqOn -- to remove duplicate sequences

    -- , pruneBy -- dropAround + uniqBy - like words

    -- ** Inserting Elements

    , insertBy

    , intersperseM

    , intersperseM_

    , intersperse

    , intersperseSuffix

    , intersperseSuffix_

    , interspersePrefix_

    , intersperseSuffixBySpan

    -- , intersperseBySpan

    -- , intersperseByIndices -- using an index function/stream

    -- time domain intersperse

    -- , intersperseByTime

    -- , intersperseByEvent

    , interjectSuffix

    , delay

    , delayPost

    , delayPre

    -- ** Indexing

    , indexed

    , indexedR

    , timestamped

    , timeIndexed

    -- , timestampedR -- timer

    -- ** Reordering

    , reverse

    , reverse'

    -- ** Parsing

    , parseMany

    , parseManyD

    , parseManyTill

    , parseIterate

    -- ** Trimming

    , take

    -- , takeGE

    -- , takeBetween

    , takeByTime

    -- , takeEnd

    , takeWhile

    , takeWhileM

    -- , takeWhileEnd

    , drop

    , dropByTime

    -- , dropEnd

    , dropWhile

    , dropWhileM

    -- , dropWhileEnd

    -- , dropAround

    -- ** Breaking

    -- Nary

    , chunksOf

    , chunksOf2

    , arraysOf

    , intervalsOf

    -- ** Searching

    -- -- *** Searching Elements

    , findIndices

    , elemIndices

    -- -- *** Searching Sequences

    -- , seqIndices -- search a sequence in the stream

    -- -- *** Searching Multiple Sequences

    -- , seqIndicesAny -- search any of the given sequence in the stream

    -- -- -- ** Searching Streams

    -- -- | Finding a stream within another stream.

    -- ** Splitting

    -- | Streams can be sliced into segments in space or in time. We use the

    -- term @chunk@ to refer to a spatial length of the stream (spatial window)

    -- and the term @session@ to refer to a length in time (time window).

    -- -- *** Using Element Separators

    , splitOn

    , splitOnSuffix

    -- , splitOnPrefix

    -- , splitBy

    , splitWithSuffix

    -- , splitByPrefix

    , wordsBy -- stripAndCompactBy

    -- -- *** Splitting By Sequences

    , splitOnSeq

    , splitOnSuffixSeq

    -- , splitOnPrefixSeq

    -- Keeping the delimiters

    , splitBySeq

    , splitWithSuffixSeq

    -- , splitByPrefixSeq

    -- , wordsBySeq

    -- Splitting using multiple sequence separators

    -- , splitOnAnySeq

    -- , splitOnAnySuffixSeq

    -- , splitOnAnyPrefixSeq

    -- -- *** Splitting By Streams

    -- -- | Splitting a stream using another stream as separator.

    -- Nested splitting

    , splitInnerBy

    , splitInnerBySuffix

    -- ** Grouping

    -- In imperative terms, grouped folding can be considered as a nested loop

    -- where we loop over the stream to group elements and then loop over

    -- individual groups to fold them to a single value that is yielded in the

    -- output stream.

    -- , groupScan

    , groups

    , groupsBy

    , groupsByRolling

    -- ** Group map

    , rollingMapM

    , rollingMap

    -- * Windowed Classification

    -- | Split the stream into windows or chunks in space or time. Each window

    -- can be associated with a key, all events associated with a particular

    -- key in the window can be folded to a single result. The stream is split

    -- into windows of specified size, the window can be terminated early if

    -- the closing flag is specified in the input stream.

    --

    -- The term "chunk" is used for a space window and the term "session" is

    -- used for a time window.

    -- ** Tumbling Windows

    -- | A new window starts after the previous window is finished.

    -- , classifyChunksOf

    , classifySessionsBy

    , classifySessionsOf

    -- ** Keep Alive Windows

    -- | The window size is extended if an event arrives within the specified

    -- window size. This can represent sessions with idle or inactive timeout.

    -- , classifyKeepAliveChunks

    , classifyKeepAliveSessions

    {-

    -- ** Sliding Windows

    -- | A new window starts after the specified slide from the previous

    -- window. Therefore windows can overlap.

    , classifySlidingChunks

    , classifySlidingSessions

    -}

    -- ** Sliding Window Buffers

    -- , slidingChunkBuffer

    -- , slidingSessionBuffer

    -- * Combining Streams

    -- ** Appending

    , append

    -- ** Interleaving

    , interleave

    , interleaveMin

    , interleaveSuffix

    , interleaveInfix

    , Serial.wSerialFst

    , Serial.wSerialMin

    -- ** Scheduling

    , roundrobin

    -- ** Parallel

    , Par.parallelFst

    , Par.parallelMin

    -- ** Merging

    -- , merge

    , mergeBy

    , mergeByM

    , mergeAsyncBy

    , mergeAsyncByM

    -- ** Zipping

    , Z.zipWith

    , Z.zipWithM

    , Z.zipAsyncWith

    , Z.zipAsyncWithM

    -- ** Flattening a Container of Streams

    , foldWith

    , foldMapWith

    , forEachWith

    -- ** Flattening a Stream of Streams

    , concat

    , concatM

    , concatMap

    , concatMapM

    , concatMapWith

    , concatSmapMWith

    -- , bindWith

    -- ** Flattening Using Unfolds

    , concatUnfold

    , concatUnfoldInterleave

    , concatUnfoldRoundrobin

    -- ** Flattening a Tree of Streams

    , iterateMapWith

    -- ** Flattening a Graph of Streams

    , iterateSmapMWith

    -- ** Inserting Streams in Streams

    , gintercalate

    , gintercalateSuffix

    , intercalate

    , intercalateSuffix

    , interpose

    , interposeSuffix

    -- , interposeBy

    -- * Exceptions

    , before

    , after_

    , after

    , bracket_

    , bracket

    , onException

    , finally_

    , finally

    , handle

    -- * Generalize Inner Monad

    , hoist

    , generally

    -- * Transform Inner Monad

    , liftInner

    , usingReaderT

    , runReaderT

    , evalStateT

    , usingStateT

    , runStateT

    -- * Maybe Streams

    , catMaybes

    , mapMaybe

    , mapMaybeM

    -- * Either Streams

    , lefts

    , rights

    , iterateMapLeftsWith

    -- * Diagnostics

    , inspectMode

    -- * Deprecated

    , K.once

    , each

    , scanx

    , foldx

    , foldxM

    , foldr1

    , runStream

    , runN

    , runWhile

    , fromHandle

    , toHandle

    )

where

import Control.Concurrent (threadDelay)

import Control.Exception (Exception, assert)

import Control.Monad (void)

import Control.Monad.Catch (MonadCatch, MonadThrow)

import Control.Monad.IO.Class (MonadIO(..))

import Control.Monad.Trans.Reader (ReaderT)

import Control.Monad.Trans.State.Strict (StateT)

import Control.Monad.Trans.Class (MonadTrans(..))

import Control.Monad.Trans.Control (MonadBaseControl)

import Data.Either (isLeft, isRight)

import Data.Functor.Identity (Identity (..))

import Data.Kind (Type)

import Data.Heap (Entry(..))

import Data.Maybe (isJust, fromJust, isNothing)

import Data.Void (Void)

import Foreign.Storable (Storable)

import Streamly.Internal.BaseCompat (fromLeft, fromRight)

import Streamly.Internal.Data.Stream.Enumeration

       (Enumerable(..), enumerate, enumerateTo)

import Streamly.Internal.Data.Fold.Types (Fold (..), Fold2 (..))

import Streamly.Internal.Data.Parser (Parser (..))

import Streamly.Internal.Data.Unfold.Types (Unfold)

import Streamly.Internal.Memory.Array.Types (Array, writeNUnsafe)

import Streamly.Internal.Data.SVar (MonadAsync, defState, Rate)

import Streamly.Internal.Data.Stream.Combinators (inspectMode, maxYields)

import Streamly.Internal.Data.Stream.Prelude

       (fromStreamS, toStreamS, foldWith, foldMapWith, forEachWith)

import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)

import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM))

import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)

import Streamly.Internal.Data.Stream.Zip (ZipSerialM)

import Streamly.Internal.Data.Pipe.Types (Pipe (..))

import Streamly.Internal.Data.Time.Units

       ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime

       , toAbsTime, TimeUnit64, RelTime64, addToAbsTime64)

import Streamly.Internal.Data.IORef.Prim (Prim, IORef)

import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

import qualified Streamly.Internal.Memory.Array as A

import qualified Streamly.Data.Fold as FL

import qualified Streamly.Internal.Data.Fold.Types as FL

import qualified Streamly.Internal.Data.Stream.Prelude as P

import qualified Streamly.Internal.Data.Stream.StreamK as K

import qualified Streamly.Internal.Data.Stream.StreamD as D

import qualified Streamly.Internal.Data.Stream.Serial as Serial

import qualified Streamly.Internal.Data.Stream.Parallel as Par

import qualified Streamly.Internal.Data.Stream.Zip as Z

import qualified Streamly.Internal.Data.Parser.ParserK.Types as PRK

import qualified Streamly.Internal.Data.Parser.ParserD as PRD

import qualified Data.Heap as H

import qualified Data.Map.Strict as Map

import qualified Prelude

import qualified System.IO as IO

#ifdef USE_STREAMK_ONLY

import qualified Streamly.Internal.Data.Stream.StreamK as S

#else

import qualified Streamly.Internal.Data.Stream.StreamD as S

#endif

import Prelude hiding

       ( filter, drop, dropWhile, take, takeWhile, zipWith, foldr

       , foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem

       , notElem, maximum, minimum, head, last, tail, length, null

       , reverse, iterate, init, and, or, lookup, foldr1, (!!)

       , scanl, scanl1, replicate, concatMap, span, splitAt, break

       , repeat, concat, mconcat)

------------------------------------------------------------------------------

-- Deconstruction

------------------------------------------------------------------------------

-- | Decompose a stream into its head and tail. If the stream is empty, returns

-- 'Nothing'. If the stream is non-empty, returns @Just (a, ma)@, where @a@ is

-- the head of the stream and @ma@ its tail.

--

-- This is a brute force primitive. Avoid using it as long as possible, use it

-- when no other combinator can do the job. This can be used to do pretty much

-- anything in an imperative manner, as it just breaks down the stream into

-- individual elements and we can loop over them as we deem fit. For example,

-- this can be used to convert a streamly stream into other stream types.

--

-- @since 0.1.0

{-# INLINE uncons #-}

uncons :: (IsStream t, Monad m) => SerialT m a -> m (Maybe (a, t m a))

------------------------------------------------------------------------------

-- Generation by Unfolding

------------------------------------------------------------------------------

-- |

-- @

-- unfoldr step s =

--     case step s of

--         Nothing -> 'K.nil'

--         Just (a, b) -> a \`cons` unfoldr step b

-- @

--

-- Build a stream by unfolding a /pure/ step function @step@ starting from a

-- seed @s@.  The step function returns the next element in the stream and the

-- next seed value. When it is done it returns 'Nothing' and the stream ends.

-- For example,

--

-- @

-- let f b =

--         if b > 3

--         then Nothing

--         else Just (b, b + 1)

-- in toList $ unfoldr f 0

-- @

-- @

-- [0,1,2,3]

-- @

--

-- @since 0.1.0

{-# INLINE_EARLY unfoldr #-}

unfoldr :: (Monad m, IsStream t) => (b -> Maybe (a, b)) -> b -> t m a

{-# RULES "unfoldr fallback to StreamK" [1]

    forall a b. S.toStreamK (S.unfoldr a b) = K.unfoldr a b #-}

-- | Build a stream by unfolding a /monadic/ step function starting from a

-- seed.  The step function returns the next element in the stream and the next

-- seed value. When it is done it returns 'Nothing' and the stream ends. For

-- example,

--

-- @

-- let f b =

--         if b > 3

--         then return Nothing

--         else print b >> return (Just (b, b + 1))

-- in drain $ unfoldrM f 0

-- @

-- @

--  0

--  1

--  2

--  3

-- @

-- When run concurrently, the next unfold step can run concurrently with the

-- processing of the output of the previous step.  Note that more than one step

-- cannot run concurrently as the next step depends on the output of the

-- previous step.

--

-- @

-- (asyncly $ S.unfoldrM (\\n -> liftIO (threadDelay 1000000) >> return (Just (n, n + 1))) 0)

--     & S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()

-- @

--

-- /Concurrent/

--

-- /Since: 0.1.0/

{-# INLINE_EARLY unfoldrM #-}

unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a

{-# RULES "unfoldrM serial" unfoldrM = unfoldrMSerial #-}

{-# INLINE_EARLY unfoldrMSerial #-}

unfoldrMSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> SerialT m a

{-# RULES "unfoldrM wSerial" unfoldrM = unfoldrMWSerial #-}

{-# INLINE_EARLY unfoldrMWSerial #-}

unfoldrMWSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> WSerialT m a

{-# RULES "unfoldrM zipSerial" unfoldrM = unfoldrMZipSerial #-}

{-# INLINE_EARLY unfoldrMZipSerial #-}

unfoldrMZipSerial :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> ZipSerialM m a

-- | Convert an 'Unfold' into a stream by supplying it an input seed.

--

-- >>> unfold (UF.replicateM 10) (putStrLn "hello")

--

-- /Since: 0.7.0/

{-# INLINE unfold #-}

unfold :: (IsStream t, Monad m) => Unfold m a b -> a -> t m b

-- | Convert an 'Unfold' with a closed input end into a stream.

--

-- /Internal/

{-# INLINE unfold0 #-}

unfold0 :: (IsStream t, Monad m) => Unfold m Void b -> t m b

------------------------------------------------------------------------------

-- Specialized Generation

------------------------------------------------------------------------------

-- Faster than yieldM because there is no bind.

--

-- |

-- @

-- yield a = a \`cons` nil

-- @

--

-- Create a singleton stream from a pure value.

--

-- The following holds in monadic streams, but not in Zip streams:

--

-- @

-- yield = pure

-- yield = yieldM . pure

-- @

--

-- In Zip applicative streams 'yield' is not the same as 'pure' because in that

-- case 'pure' is equivalent to 'repeat' instead. 'yield' and 'pure' are

-- equally efficient, in other cases 'yield' may be slightly more efficient

-- than the other equivalent definitions.

--

-- @since 0.4.0

{-# INLINE yield #-}

yield :: IsStream t => a -> t m a

-- |

-- @

-- yieldM m = m \`consM` nil

-- @

--

-- Create a singleton stream from a monadic action.

--

-- @

-- > toList $ yieldM getLine

-- hello

-- ["hello"]

-- @

--

-- @since 0.4.0

{-# INLINE yieldM #-}

yieldM :: (Monad m, IsStream t) => m a -> t m a

-- |

-- @

-- fromIndices f = let g i = f i \`cons` g (i + 1) in g 0

-- @

--

-- Generate an infinite stream, whose values are the output of a function @f@

-- applied on the corresponding index.  Index starts at 0.

--

-- @

-- > S.toList $ S.take 5 $ S.fromIndices id

-- [0,1,2,3,4]

-- @

--

-- @since 0.6.0

{-# INLINE fromIndices #-}

fromIndices :: (IsStream t, Monad m) => (Int -> a) -> t m a

--

-- |

-- @

-- fromIndicesM f = let g i = f i \`consM` g (i + 1) in g 0

-- @

--

-- Generate an infinite stream, whose values are the output of a monadic

-- function @f@ applied on the corresponding index. Index starts at 0.

--

-- /Concurrent/

--

-- @since 0.6.0

{-# INLINE_EARLY fromIndicesM #-}

fromIndicesM :: (IsStream t, MonadAsync m) => (Int -> m a) -> t m a

{-# RULES "fromIndicesM serial" fromIndicesM = fromIndicesMSerial #-}

{-# INLINE fromIndicesMSerial #-}

fromIndicesMSerial :: MonadAsync m => (Int -> m a) -> SerialT m a

-- |

-- @

-- replicateM = take n . repeatM

-- @

--

-- Generate a stream by performing a monadic action @n@ times. Same as:

--

-- @

-- drain $ serially $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)

-- drain $ asyncly  $ S.replicateM 10 $ (threadDelay 1000000 >> print 1)

-- @

--

-- /Concurrent/

--

-- @since 0.1.1

{-# INLINE_EARLY replicateM #-}

replicateM :: (IsStream t, MonadAsync m) => Int -> m a -> t m a

{-# RULES "replicateM serial" replicateM = replicateMSerial #-}

{-# INLINE replicateMSerial #-}

replicateMSerial :: MonadAsync m => Int -> m a -> SerialT m a

-- |

-- @

-- replicate = take n . repeat

-- @

--

-- Generate a stream of length @n@ by repeating a value @n@ times.

--

-- @since 0.6.0

{-# INLINE_NORMAL replicate #-}

replicate :: (IsStream t, Monad m) => Int -> a -> t m a

-- |

-- Generate an infinite stream by repeating a pure value.

--

-- @since 0.4.0

{-# INLINE_NORMAL repeat #-}

repeat :: (IsStream t, Monad m) => a -> t m a

-- |

-- @

-- repeatM = fix . consM

-- repeatM = cycle1 . yieldM

-- @

--

-- Generate a stream by repeatedly executing a monadic action forever.

--

-- @

-- drain $ serially $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)

-- drain $ asyncly  $ S.take 10 $ S.repeatM $ (threadDelay 1000000 >> print 1)

-- @

--

-- /Concurrent, infinite (do not use with 'parallely')/

--

-- @since 0.2.0

{-# INLINE_EARLY repeatM #-}

repeatM :: (IsStream t, MonadAsync m) => m a -> t m a

{-# RULES "repeatM serial" repeatM = repeatMSerial #-}

{-# INLINE repeatMSerial #-}

repeatMSerial :: MonadAsync m => m a -> SerialT m a

-- |

-- @

-- iterate f x = x \`cons` iterate f x

-- @

--

-- Generate an infinite stream with @x@ as the first element and each

-- successive element derived by applying the function @f@ on the previous

-- element.

--

-- @

-- > S.toList $ S.take 5 $ S.iterate (+1) 1

-- [1,2,3,4,5]

-- @

--

-- @since 0.1.2

{-# INLINE_NORMAL iterate #-}

iterate :: (IsStream t, Monad m) => (a -> a) -> a -> t m a

-- |

-- @

-- iterateM f m = m >>= \a -> return a \`consM` iterateM f (f a)

-- @

--

-- Generate an infinite stream with the first element generated by the action

-- @m@ and each successive element derived by applying the monadic function

-- @f@ on the previous element.

--

-- When run concurrently, the next iteration can run concurrently with the

-- processing of the previous iteration. Note that more than one iteration

-- cannot run concurrently as the next iteration depends on the output of the

-- previous iteration.

--

-- @

-- drain $ serially $ S.take 10 $ S.iterateM

--      (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)

--

-- drain $ asyncly  $ S.take 10 $ S.iterateM

--      (\\x -> threadDelay 1000000 >> print x >> return (x + 1)) (return 0)

-- @

--

-- /Concurrent/

--

-- /Since: 0.7.0 (signature change)/

--

-- /Since: 0.1.2/

{-# INLINE_EARLY iterateM #-}

iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> m a -> t m a

{-# RULES "iterateM serial" iterateM = iterateMSerial #-}

{-# INLINE iterateMSerial #-}

iterateMSerial :: MonadAsync m => (a -> m a) -> m a -> SerialT m a

------------------------------------------------------------------------------

-- Conversions

------------------------------------------------------------------------------

-- |

-- @

-- fromListM = 'Prelude.foldr' 'K.consM' 'K.nil'

-- @

--

-- Construct a stream from a list of monadic actions. This is more efficient

-- than 'fromFoldableM' for serial streams.

--

-- @since 0.4.0

{-# INLINE_EARLY fromListM #-}

fromListM :: (MonadAsync m, IsStream t) => [m a] -> t m a

{-# RULES "fromListM fallback to StreamK" [1]

    forall a. D.toStreamK (D.fromListM a) = fromFoldableM a #-}

-- |

-- @

-- fromFoldableM = 'Prelude.foldr' 'consM' 'K.nil'

-- @

--

-- Construct a stream from a 'Foldable' containing monadic actions.

--

-- @

-- drain $ serially $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)

-- drain $ asyncly  $ S.fromFoldableM $ replicateM 10 (threadDelay 1000000 >> print 1)

-- @

--

-- /Concurrent (do not use with 'parallely' on infinite containers)/

--

-- @since 0.3.0

{-# INLINE fromFoldableM #-}

fromFoldableM :: (IsStream t, MonadAsync m, Foldable f) => f (m a) -> t m a

-- | Same as 'fromFoldable'.

--

-- @since 0.1.0

{-# DEPRECATED each "Please use fromFoldable instead." #-}

{-# INLINE each #-}

each :: (IsStream t, Foldable f) => f a -> t m a

-- | Read lines from an IO Handle into a stream of Strings.

--

-- @since 0.1.0

{-# DEPRECATED fromHandle

   "Please use Streamly.FileSystem.Handle module (see the changelog)" #-}

fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String

  where

-- | Construct a stream by reading a 'Prim' 'IORef' repeatedly.

--

-- /Internal/

--

{-# INLINE fromPrimIORef #-}

fromPrimIORef :: (IsStream t, MonadIO m, Prim a) => IORef a -> t m a

-- | Takes a callback setter function and provides it with a callback.  The

-- callback when invoked adds a value at the tail of the stream. Returns a

-- stream of values generated by the callback.

--

-- /Internal/

--

{-# INLINE fromCallback #-}

fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a

------------------------------------------------------------------------------

-- Time related

------------------------------------------------------------------------------

-- | @timesWith g@ returns a stream of time value tuples. The first component

-- of the tuple is an absolute time reference (epoch) denoting the start of the

-- stream and the second component is a time relative to the reference.

--

-- The argument @g@ specifies the granularity of the relative time in seconds.

-- A lower granularity clock gives higher precision but is more expensive in

-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.

--

-- @

-- >>> S.mapM_ (\x -> print x >> threadDelay 1000000) $ S.timesWith 0.01

-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 0))

-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1002028000))

-- > (AbsTime (TimeSpec {sec = 2496295, nsec = 536223000}),RelTime64 (NanoSecond64 1996656000))

-- @

--

-- Note: This API is not safe on 32-bit machines.

--

-- /Internal/

--

{-# INLINE timesWith #-}

timesWith :: (IsStream t, MonadAsync m) => Double -> t m (AbsTime, RelTime64)

-- | @times@ returns a stream of time value tuples with clock of 10 ms

-- granularity. The first component of the tuple is an absolute time reference