Streamly.Internal.Data.Stream.StreamK.Type

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

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.

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

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

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 => 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

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

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

toStreamK :: Stream m a -> Stream m a

fromStreamK :: Stream m a -> Stream m a

foldr/build

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

foldStream :: State Stream m a -> (a -> Stream m a -> m r) -> (a -> m r) -> m r -> Stream 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 :: State Stream m a -> (a -> Stream m a -> m r) -> (a -> m r) -> m r -> Stream 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.

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

Lazy right fold with a monadic step function.

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

Lazy right associative fold to a stream.

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

Fold sharing the SVar state within the reconstructed stream

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

build :: forall m a. (forall b. (a -> b -> b) -> b -> b) -> Stream m a

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

buildM :: Monad m => (forall r. (a -> Stream m a -> m r) -> (a -> m r) -> m r -> m r) -> Stream m a

buildSM :: Monad m => ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a) -> Stream m a

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

augmentSM :: Monad m => ((m a -> Stream m a -> Stream m a) -> Stream m a -> Stream m a) -> Stream m a -> Stream m a

Construction

fromStopK :: StopK m -> Stream m a

Make an empty stream from a stop function.

fromYieldK :: YieldK m a -> Stream m a

Make a singleton stream from a callback function. The callback function calls the one-shot yield continuation to yield an element.

consK :: YieldK m a -> Stream m a -> Stream m a

Add a yield function at the head of the stream.

cons :: a -> Stream m a -> Stream 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

(.:) :: a -> Stream m a -> Stream m a infixr 5

Operator equivalent of cons.

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

Since: 0.1.1

consM :: Monad m => m a -> Stream m a -> Stream m a infixr 5

consMBy :: Monad m => (Stream m a -> Stream m a -> Stream m a) -> m a -> Stream m a -> Stream m a

nil :: Stream m a

An empty stream.

> toList nil
[]

Since: 0.1.0

nilM :: Applicative m => m b -> Stream m a

An empty stream producing a side effect.

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

Pre-release

Generation

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

fromPure :: a -> Stream m a

unfoldr :: (b -> Maybe (a, b)) -> b -> Stream m a

unfoldrMWith :: Monad m => (m a -> Stream m a -> Stream m a) -> (b -> m (Maybe (a, b))) -> b -> Stream m a

repeat :: a -> Stream m a

Generate an infinite stream by repeating a pure value.

Pre-release

repeatMWith :: (m a -> t m a -> t m a) -> m a -> t m a

Like repeatM but takes a stream cons operation to combine the actions in a stream specific manner. A serial cons would repeat the values serially while an async cons would repeat concurrently.

Pre-release

replicateMWith :: (m a -> Stream m a -> Stream m a) -> Int -> m a -> Stream m a

fromIndicesMWith :: (m a -> Stream m a -> Stream m a) -> (Int -> m a) -> Stream m a

iterateMWith :: Monad m => (m a -> Stream m a -> Stream m a) -> (a -> m a) -> m a -> Stream m a

fromFoldable :: Foldable f => f a -> Stream m a

fromFoldable = foldr cons nil

Construct a stream from a Foldable containing pure values:

Since: 0.2.0

fromFoldableM :: (Foldable f, Monad m) => f (m a) -> Stream m a

mfix :: Monad m => (m a -> Stream m a) -> Stream m a

Elimination

uncons :: Applicative m => Stream m a -> m (Maybe (a, Stream m a))

foldl' :: Monad m => (b -> a -> b) -> b -> Stream m a -> m b

Strict left associative fold.

foldlx' :: forall m a b x. Monad m => (x -> a -> x) -> x -> (x -> b) -> Stream 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.

drain :: Monad m => Stream m a -> m ()

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

null :: Monad m => Stream m a -> m Bool

tail :: Applicative m => Stream m a -> m (Maybe (Stream m a))

init :: Applicative m => Stream m a -> m (Maybe (Stream m a))

Transformation

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

serial :: Stream m a -> Stream m a -> Stream m a infixr 6

Appends two streams sequentially, yielding all elements from the first stream, and then all elements from the second stream.

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

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

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

unShare :: Stream m a -> Stream m a

Detach a stream from an SVar

concatMapWith :: (Stream m b -> Stream m b -> Stream m b) -> (a -> Stream m b) -> Stream m a -> Stream 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 :: (a -> Stream m b) -> Stream m a -> Stream m b

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

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

See concatPairsWith for documentation.

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

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

apSerialDiscardFst :: Stream m a -> Stream m b -> Stream m b

apSerialDiscardSnd :: Stream m a -> Stream m b -> Stream m a

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

Lazy left fold to a stream.

reverse :: Stream m a -> Stream m a

Reader

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