Streamly.Internal.Data.Stream.Ahead

To run examples in this module:

>>> import qualified Streamly.Prelude as Stream
>>> import Control.Concurrent (threadDelay)
>>> :{
 delay n = do
     threadDelay (n * 1000000)   -- sleep for n seconds
     putStrLn (show n ++ " sec") -- print "n sec"
     return n                    -- IO Int
:}

Documentation

newtype AheadT m a

For AheadT streams:

(<>) = ahead
(>>=) = flip . concatMapWith ahead

A single Monad bind behaves like a for loop with iterations executed concurrently, ahead of time, producing side effects of iterations out of order, but results in order:

>>> :{
Stream.toList $ Stream.fromAhead $ do
     x <- Stream.fromList [2,1] -- foreach x in stream
     Stream.fromEffect $ delay x
:}
1 sec
2 sec
[2,1]

Nested monad binds behave like nested for loops with nested iterations executed concurrently, ahead of time:

>>> :{
Stream.toList $ Stream.fromAhead $ do
    x <- Stream.fromList [1,2] -- foreach x in stream
    y <- Stream.fromList [2,4] -- foreach y in stream
    Stream.fromEffect $ delay (x + y)
:}
3 sec
4 sec
5 sec
6 sec
[3,5,4,6]

The behavior can be explained as follows. All the iterations corresponding to the element 1 in the first stream constitute one output stream and all the iterations corresponding to 2 constitute another output stream and these two output streams are merged using ahead.

Since: 0.3.0 (Streamly)

Since: 0.8.0

Constructors

AheadT

Fields

• getAheadT :: Stream m a

Instances

IsStream AheadT

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

Methods

toStream :: forall (m :: Type -> TYPE LiftedRep) a. AheadT m a -> Stream m a

fromStream :: forall (m :: Type -> TYPE LiftedRep) a. 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

MonadTrans AheadT

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

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

(MonadReader r m, MonadAsync m) => MonadReader r (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

ask :: AheadT m r

local :: (r -> r) -> AheadT m a -> AheadT m a

reader :: (r -> a) -> AheadT m a

(MonadState s m, MonadAsync m) => MonadState s (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

get :: AheadT m s

put :: s -> AheadT m ()

state :: (s -> (a, s)) -> AheadT m a

(MonadBase b m, Monad m, MonadAsync m) => MonadBase b (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

liftBase :: b α -> AheadT m α

(MonadIO m, MonadAsync m) => MonadIO (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

liftIO :: IO a -> AheadT m a

(Monad m, MonadAsync m) => Applicative (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

pure :: a -> AheadT m a

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

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

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

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

Monad m => Functor (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

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

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

MonadAsync m => Monad (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

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

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

return :: a -> AheadT m a

(MonadThrow m, MonadAsync m) => MonadThrow (AheadT m)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

throwM :: Exception e => e -> AheadT m a

MonadAsync m => Monoid (AheadT m a)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

mempty :: AheadT m a

mappend :: AheadT m a -> AheadT m a -> AheadT m a

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

MonadAsync m => Semigroup (AheadT m a)

Defined in Streamly.Internal.Data.Stream.Ahead

Methods

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

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

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

type Ahead = AheadT IO

A serial IO stream of elements of type a with concurrent lookahead. See AheadT documentation for more details.

Since: 0.3.0 (Streamly)

Since: 0.8.0

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

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

XXX we can implement it more efficienty by directly implementing instead of combining streams using ahead.