Streamly.Internal.Data.Stream.StreamK

Continuation passing style (CPS) stream implementation. The symbol K below denotes a function as well as a Kontinuation.

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

A class for streams

class (forall m a. MonadAsync m => Semigroup (t m a), forall m a. MonadAsync m => Monoid (t m a), forall m. Monad m => Functor (t m), forall m. MonadAsync m => Applicative (t m)) => IsStream t where

Class of types that can represent a stream of elements of some type a in some monad m.

Since: 0.2.0

Methods

toStream :: t m a -> Stream m a

fromStream :: Stream m a -> t m a

consM :: MonadAsync m => m a -> t m a -> t m a infixr 5

Constructs a stream by adding a monadic action at the head of an existing stream. For example:

> toList $ getLine `consM` getLine `consM` nil
hello
world
["hello","world"]

Concurrent (do not use parallely to construct infinite streams)

Since: 0.2.0

(|:) :: MonadAsync m => m a -> t m a -> t m a infixr 5

Operator equivalent of consM. We can read it as "parallel colon" to remember that | comes before :.

> toList $ getLine |: getLine |: nil
hello
world
["hello","world"]

let delay = threadDelay 1000000 >> print 1
drain $ serially  $ delay |: delay |: delay |: nil
drain $ parallely $ delay |: delay |: delay |: nil

Concurrent (do not use parallely to construct infinite streams)

Since: 0.2.0

Instances

IsStream Stream

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

toStream :: Stream m a -> Stream m a

fromStream :: Stream m a -> Stream m a

consM :: MonadAsync m => m a -> Stream m a -> Stream m a

(|:) :: MonadAsync m => m a -> Stream m a -> Stream m a

IsStream ZipAsyncM

Defined in Streamly.Internal.Data.Stream.Zip

Methods

toStream :: ZipAsyncM m a -> Stream m a

fromStream :: Stream m a -> ZipAsyncM m a

consM :: MonadAsync m => m a -> ZipAsyncM m a -> ZipAsyncM m a

(|:) :: MonadAsync m => m a -> ZipAsyncM m a -> ZipAsyncM m a

IsStream ZipSerialM

Defined in Streamly.Internal.Data.Stream.Zip

Methods

toStream :: ZipSerialM m a -> Stream m a

fromStream :: Stream m a -> ZipSerialM m a

consM :: MonadAsync m => m a -> ZipSerialM m a -> ZipSerialM m a

(|:) :: MonadAsync m => m a -> ZipSerialM m a -> ZipSerialM m a

IsStream WSerialT

Defined in Streamly.Internal.Data.Stream.Serial

Methods

toStream :: WSerialT m a -> Stream m a

fromStream :: Stream m a -> WSerialT m a

consM :: MonadAsync m => m a -> WSerialT m a -> WSerialT m a

(|:) :: MonadAsync m => m a -> WSerialT m a -> WSerialT m a

IsStream SerialT

Defined in Streamly.Internal.Data.Stream.Serial

Methods

toStream :: SerialT m a -> Stream m a

fromStream :: Stream m a -> SerialT m a

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

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

IsStream ParallelT

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

toStream :: ParallelT m a -> Stream m a

fromStream :: Stream m a -> ParallelT m a

consM :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a

(|:) :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a

IsStream WAsyncT

Defined in Streamly.Internal.Data.Stream.Async

Methods

toStream :: WAsyncT m a -> Stream m a

fromStream :: Stream m a -> WAsyncT m a

consM :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a

(|:) :: MonadAsync m => m a -> WAsyncT m a -> WAsyncT m a

IsStream AsyncT

Defined in Streamly.Internal.Data.Stream.Async

Methods

toStream :: AsyncT m a -> Stream m a

fromStream :: Stream m a -> AsyncT m a

consM :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a

(|:) :: MonadAsync m => m a -> AsyncT m a -> AsyncT m a

IsStream AheadT

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

toStream :: AheadT m a -> Stream m a

fromStream :: Stream m a -> AheadT m a

consM :: MonadAsync m => m a -> AheadT m a -> AheadT m a

(|:) :: MonadAsync m => m a -> AheadT m a -> AheadT m a

adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a

Adapt any specific stream type to any other specific stream type.

Since: 0.1.0

The stream type

newtype Stream m a

The type Stream m a represents a monadic stream of values of type a constructed using actions in monad m. It uses stop, singleton and yield continuations equivalent to the following direct style type:

data Stream m a = Stop | Singleton a | Yield a (Stream m a)

To facilitate parallel composition we maintain a local state in an SVar that is shared across and is used for synchronization of the streams being composed.

The singleton case can be expressed in terms of stop and yield but we have it as a separate case to optimize composition operations for streams with single element. We build singleton streams in the implementation of pure for Applicative and Monad, and in lift for MonadTrans.

XXX remove the Stream type parameter from State as it is always constant. We can remove it from SVar as well

Constructors

MkStream (forall r. State Stream m a -> (a -> Stream m a -> m r) -> (a -> m r) -> m r -> m r)

Instances

MonadTrans Stream

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

lift :: Monad m => m a -> Stream m a

IsStream Stream

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

toStream :: Stream m a -> Stream m a

fromStream :: Stream m a -> Stream m a

consM :: MonadAsync m => m a -> Stream m a -> Stream m a

(|:) :: MonadAsync m => m a -> Stream m a -> Stream m a

Monad m => Monad (Stream m)

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

(>>=) :: Stream m a -> (a -> Stream m b) -> Stream m b

(>>) :: Stream m a -> Stream m b -> Stream m b

return :: a -> Stream m a

fail :: String -> Stream m a

Monad m => Functor (Stream m)

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

fmap :: (a -> b) -> Stream m a -> Stream m b

(<$) :: a -> Stream m b -> Stream m a

Monad m => Applicative (Stream m)

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

pure :: a -> Stream m a

(<*>) :: Stream m (a -> b) -> Stream m a -> Stream m b

liftA2 :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

(*>) :: Stream m a -> Stream m b -> Stream m b

(<*) :: Stream m a -> Stream m b -> Stream m a

Semigroup (Stream m a)

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

(<>) :: Stream m a -> Stream m a -> Stream m a

sconcat :: NonEmpty (Stream m a) -> Stream m a

stimes :: Integral b => b -> Stream m a -> Stream m a

Monoid (Stream m a)

Defined in Streamly.Internal.Data.Stream.StreamK.Type

Methods

mempty :: Stream m a

mappend :: Stream m a -> Stream m a -> Stream m a

mconcat :: [Stream m a] -> Stream m a

Construction Primitives

mkStream :: IsStream t => (forall r. State Stream m a -> (a -> t m a -> m r) -> (a -> m r) -> m r -> m r) -> t m a

Build a stream from an SVar, a stop continuation, a singleton stream continuation and a yield continuation.

nil :: IsStream t => t m a

An empty stream.

> toList nil
[]

Since: 0.1.0

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

An empty stream producing a side effect.

> toList (nilM (print "nil"))
"nil"
[]

Internal

cons :: IsStream t => a -> t m a -> t m a infixr 5

Construct a stream by adding a pure value at the head of an existing stream. For serial streams this is the same as (return a) `consM` r but more efficient. For concurrent streams this is not concurrent whereas consM is concurrent. For example:

> toList $ 1 `cons` 2 `cons` 3 `cons` nil
[1,2,3]

Since: 0.1.0

(.:) :: IsStream t => a -> t m a -> t m a infixr 5

Operator equivalent of cons.

> toList $ 1 .: 2 .: 3 .: nil
[1,2,3]

Since: 0.1.1

Elimination Primitives

foldStream :: IsStream t => State Stream m a -> (a -> t m a -> m r) -> (a -> m r) -> m r -> t m a -> m r

Fold a stream by providing a State, stop continuation, a singleton continuation and a yield continuation. The stream will not use the SVar passed via State.

foldStreamShared :: IsStream t => State Stream m a -> (a -> t m a -> m r) -> (a -> m r) -> m r -> t m a -> m r

Fold a stream by providing an SVar, a stop continuation, a singleton continuation and a yield continuation. The stream would share the current SVar passed via the State.

Transformation Primitives

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

Detach a stream from an SVar

Deconstruction

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

Generation

Unfolds

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

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

Specialized Generation

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

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

repeatM = fix . cons
repeatM = cycle1 . yield

Generate an infinite stream by repeating a monadic value.

Internal

replicate :: IsStream t => Int -> a -> t m a

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

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

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

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

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

Conversions

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

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

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

fromFoldable = foldr cons nil

Construct a stream from a Foldable containing pure values:

Since: 0.2.0

fromList :: IsStream t => [a] -> t m a

fromStreamK :: IsStream t => Stream m a -> t m a

foldr/build

foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b

Lazy right associative fold to a stream.

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

buildS :: IsStream t => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a

buildM :: (IsStream t, MonadAsync m) => (forall r. (a -> t m a -> m r) -> (a -> m r) -> m r -> m r) -> t m a

augmentS :: IsStream t => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a

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

Elimination

General Folds

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

Lazy right associative fold.

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

foldrM :: IsStream t => (a -> m b -> m b) -> m b -> t m a -> m b

Lazy right fold with a monadic step function.

foldrT :: (IsStream t, Monad m, Monad (s m), MonadTrans s) => (a -> s m b -> s m b) -> s m b -> t m a -> s m b

Right associative fold to an arbitrary transformer monad.

foldl' :: (IsStream t, Monad m) => (b -> a -> b) -> b -> t m a -> m b

Strict left associative fold.

foldlM' :: (IsStream t, Monad m) => (b -> a -> m b) -> b -> t m a -> m b

Like foldl' but with a monadic step function.

foldlS :: IsStream t => (t m b -> a -> t m b) -> t m b -> t m a -> t m b

Lazy left fold to a stream.

foldlT :: (IsStream t, Monad m, Monad (s m), MonadTrans s) => (s m b -> a -> s m b) -> s m b -> t m a -> s m b

Lazy left fold to an arbitrary transformer monad.

foldlx' :: forall t m a b x. (IsStream t, Monad m) => (x -> a -> x) -> x -> (x -> b) -> t m a -> m b

Strict left fold with an extraction function. Like the standard strict left fold, but applies a user supplied extraction function (the third argument) to the folded value at the end. This is designed to work with the foldl library. The suffix x is a mnemonic for extraction.

Note that the accumulator is always evaluated including the initial value.

foldlMx' :: (IsStream t, Monad m) => (x -> a -> m x) -> m x -> (x -> m b) -> t m a -> m b

Like foldx, but with a monadic step function.

Specialized Folds

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

drain = foldl' (\_ _ -> ()) ()
drain = mapM_ (\_ -> return ())

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

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

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

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

elem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

notElem :: (IsStream t, Monad m, Eq a) => a -> t m a -> m Bool

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

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

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

Extract the last element of the stream, if any.

minimum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

minimumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)

maximum :: (IsStream t, Monad m, Ord a) => t m a -> m (Maybe a)

maximumBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> m (Maybe a)

findIndices :: IsStream t => (a -> Bool) -> t m a -> t m Int

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

findM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> m (Maybe a)

find :: (IsStream t, Monad m) => (a -> Bool) -> t m a -> m (Maybe a)

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

Map and Fold

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

Apply a monadic action to each element of the stream and discard the output of the action.

Conversions

toList :: (IsStream t, Monad m) => t m a -> m [a]

toStreamK :: Stream m a -> Stream m a

hoist :: (IsStream t, Monad m, Monad n) => (forall x. m x -> n x) -> t m a -> t n a

Transformation

By folding (scans)

scanl' :: IsStream t => (b -> a -> b) -> b -> t m a -> t m b

scanlx' :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b

Filtering

filter :: IsStream t => (a -> Bool) -> t m a -> t m a

take :: IsStream t => Int -> t m a -> t m a

takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

drop :: IsStream t => Int -> t m a -> t m a

dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a

Mapping

map :: IsStream t => (a -> b) -> t m a -> t m b

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

mapMSerial :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b

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

Inserting

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

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

insertBy :: IsStream t => (a -> a -> Ordering) -> a -> t m a -> t m a

Deleting

deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a

Reordering

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

Map and Filter

mapMaybe :: IsStream t => (a -> Maybe b) -> t m a -> t m b

Zipping

zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

Zip two streams serially using a pure zipping function.

Since: 0.1.0

zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

Zip two streams serially using a monadic zipping function.

Since: 0.1.0

Merging

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

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

Nesting

concatMapBy :: IsStream t => (forall c. t m c -> t m c -> t m c) -> (a -> t m b) -> t m a -> t m b

Perform a concatMap using a specified concat strategy. The first argument specifies a merge or concat function that is used to merge the streams generated by the map function. For example, the concat function could be serial, parallel, async, ahead or any other zip or merge function.

Since: 0.7.0

concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b

bindWith :: IsStream t => (forall c. t m c -> t m c -> t m c) -> t m a -> (a -> t m b) -> t m b

Transformation comprehensions

the :: (Eq a, IsStream t, Monad m) => t m a -> m (Maybe a)

Semigroup Style Composition

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

Polymorphic version of the Semigroup operation <> of SerialT. Appends two streams sequentially, yielding all elements from the first stream, and then all elements from the second stream.

Since: 0.2.0

Utilities

consMStream :: Monad m => m a -> Stream m a -> Stream m a

withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a

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

Iterate a lazy function f of the shape `m a -> t m a` until it gets fully defined i.e. becomes independent of its argument action, then return the resulting value of the function (`t m a`).

It can be used to construct a stream that uses a cyclic definition. For example:

import Streamly.Internal.Prelude as S
import System.IO.Unsafe (unsafeInterleaveIO)

main = do
    S.mapM_ print $ S.mfix $ x -> do
      a <- S.fromList [1,2]
      b <- S.fromListM [return 3, unsafeInterleaveIO (fmap fst x)]
      return (a, b)

Note that the function f must be lazy in its argument, that's why we use unsafeInterleaveIO because IO monad is strict.

Deprecated

type Streaming = IsStream

Deprecated: Please use IsStream instead.

Same as IsStream.

Since: 0.1.0

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

Deprecated: Please use yieldM instead.

Same as yieldM

Since: 0.2.0