1{-# LANGUAGE TypeFamilies #-}
2{-# LANGUAGE UndecidableInstances #-}
5-- Module : Streamly.Internal.Data.Stream.Serial
6-- Copyright : (c) 2017 Harendra Kumar
9-- Maintainer : streamly@composewell.com
10-- Stability : experimental
14module Streamly.Internal.Data.Stream.Serial
16 -- * Serial appending stream
22 -- * Serial interleaving stream
45import Control.Applicative (liftA2)
46import Control.DeepSeq (NFData(..))
47#if MIN_VERSION_deepseq(1,4,3)
48import Control.DeepSeq (NFData1(..))
50import Control.Monad.Base (MonadBase(..), liftBaseDefault)
51import Control.Monad.Catch (MonadThrow, throwM)
52-- import Control.Monad.Error.Class (MonadError(..))
53import Control.Monad.IO.Class (MonadIO(..))
54import Control.Monad.Reader.Class (MonadReader(..))
55import Control.Monad.State.Class (MonadState(..))
56import Control.Monad.Trans.Class (MonadTrans(lift))
57import Data.Foldable (Foldable(foldl'), fold)
58import Data.Functor.Identity (Identity(..), runIdentity)
59import Data.Maybe (fromMaybe)
60import Data.Semigroup (Endo(..))
61#if __GLASGOW_HASKELL__ < 808
62import Data.Semigroup (Semigroup(..))
64import GHC.Exts (IsList(..), IsString(..))
65import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
67import Prelude hiding (map, mapM, errorWithoutStackTrace)
69import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
70import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStream,
72import Streamly.Internal.Data.Strict (Maybe'(..), toMaybe)
73import qualified Streamly.Internal.Data.Stream.Prelude as P
74import qualified Streamly.Internal.Data.Stream.StreamK as K
75import qualified Streamly.Internal.Data.Stream.StreamD as D
77#include "Instances.hs"
80------------------------------------------------------------------------------
82------------------------------------------------------------------------------
84-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
85-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
86-- first until it exhausts and then stream @b@ is evaluated. In other words,
87-- the elements of stream @b@ are appended to the elements of stream @a@. This
88-- operation can be used to fold an infinite lazy container of streams.
92-- import qualified "Streamly.Prelude" as S
94-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
100-- The 'Monad' instance runs the /monadic continuation/ for each
101-- element of the stream, serially.
104-- main = S.drain . 'serially' $ do
105-- x <- return 1 \<\> return 2
113-- 'SerialT' nests streams serially in a depth first manner.
116-- main = S.drain . 'serially' $ do
117-- x <- return 1 \<\> return 2
118-- y <- return 3 \<\> return 4
119-- S.yieldM $ print (x, y)
128-- We call the monadic code being run for each element of the stream a monadic
129-- continuation. In imperative paradigm we can think of this composition as
130-- nested @for@ loops and the monadic continuation is the body of the loop. The
131-- loop iterates for all elements of the stream.
133-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
134-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
135-- can contain effectful actions while lists are pure.
137-- In the code above, the 'serially' combinator can be omitted as the default
138-- stream type is 'SerialT'.
141newtype SerialT m a = SerialT {getSerialT :: Stream m a}
142 -- XXX when deriving do we inherit an INLINE?
143 deriving (Semigroup, Monoid, MonadTrans)
145-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
149type Serial = SerialT IO
153{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
154type StreamT = SerialT
156-- | Fix the type of a polymorphic stream as 'SerialT'.
159serially :: IsStream t => SerialT m a -> t m a
162{-# INLINE consMSerial #-}
163{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
164consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
165consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
167instance IsStream SerialT where
168 toStream = getSerialT
173------------------------------------------------------------------------------
175------------------------------------------------------------------------------
177instance Monad m => Monad (SerialT m) where
181 (>>=) = K.bindWith K.serial
186 -- StreamD based implementation
187 -- return = SerialT . D.fromStreamD . D.yield
188 -- m >>= f = D.fromStreamD $ D.concatMap (\a -> D.toStreamD (f a)) (D.toStreamD m)
190------------------------------------------------------------------------------
192------------------------------------------------------------------------------
195mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
196mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
206-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
212map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
213map f = mapM (return . f)
215{-# INLINE apSerial #-}
216apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
217apSerial (SerialT m1) (SerialT m2) =
218 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
220{-# INLINE apSequence #-}
221apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
222apSequence (SerialT m1) (SerialT m2) =
223 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
225{-# INLINE apDiscardSnd #-}
226apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
227apDiscardSnd (SerialT m1) (SerialT m2) =
228 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
230-- Note: we need to define all the typeclass operations because we want to
232instance Monad m => Applicative (SerialT m) where
234 pure = SerialT . K.yield
238 -- (<*>) = K.apSerial
240#if MIN_VERSION_base(4,10,0)
241 {-# INLINE liftA2 #-}
242 liftA2 f x = (<*>) (fmap f x)
247 -- (*>) = K.apSerialDiscardFst
251 -- (<*) = K.apSerialDiscardSnd
253MONAD_COMMON_INSTANCES(SerialT,)
254LIST_INSTANCES(SerialT)
255NFDATA1_INSTANCE(SerialT)
256FOLDABLE_INSTANCE(SerialT)
257TRAVERSABLE_INSTANCE(SerialT)
259------------------------------------------------------------------------------
261------------------------------------------------------------------------------
263-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
264-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
265-- first to produce the first element of the combined stream and then stream
266-- @b@ is evaluated to produce the next element of the combined stream, and
267-- then we go back to evaluating stream @a@ and so on. In other words, the
268-- elements of stream @a@ are interleaved with the elements of stream @b@.
270-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
271-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
272-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
273-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
274-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
275-- leftmost stream gets the same scheduling priority as the rest of the
276-- streams taken together. The same is true for each subexpression on the right.
278-- Note that this operation cannot be used to fold a container of infinite
279-- streams as the state that it needs to maintain is proportional to the number
282-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
283-- contrast to the depth wise scheduling behavior of 'SerialT'.
287-- import qualified "Streamly.Prelude" as S
289-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
295-- Similarly, the 'Monad' instance interleaves the iterations of the
296-- inner and the outer loop, nesting loops in a breadth first manner.
300-- main = S.drain . 'wSerially' $ do
301-- x <- return 1 \<\> return 2
302-- y <- return 3 \<\> return 4
303-- S.yieldM $ print (x, y)
313newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
314 deriving (MonadTrans)
316-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
317-- documentation for more details.
320type WSerial = WSerialT IO
324{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
325type InterleavedT = WSerialT
327-- | Fix the type of a polymorphic stream as 'WSerialT'.
330wSerially :: IsStream t => WSerialT m a -> t m a
333-- | Same as 'wSerially'.
336{-# DEPRECATED interleaving "Please use wSerially instead." #-}
337interleaving :: IsStream t => WSerialT m a -> t m a
338interleaving = wSerially
340consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
341consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
343instance IsStream WSerialT where
344 toStream = getWSerialT
345 fromStream = WSerialT
348 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
349 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
353 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
354 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
357------------------------------------------------------------------------------
359------------------------------------------------------------------------------
361-- Additionally we can have m elements yield from the first stream and n
362-- elements yielding from the second stream. We can also have time slicing
363-- variants of positional interleaving, e.g. run first stream for m seconds and
364-- run the second stream for n seconds.
366-- Similar combinators can be implemented using WAhead style.
368-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
369-- Interleaves two streams, yielding one element from each stream alternately.
370-- When one stream stops the rest of the other stream is used in the output
374{-# INLINE wSerial #-}
375wSerial :: IsStream t => t m a -> t m a -> t m a
376wSerial m1 m2 = mkStream $ \st yld sng stp -> do
377 let stop = foldStream st yld sng stp m2
379 yieldk a r = yld a (wSerial m2 r)
380 foldStream st yieldk single stop m1
382-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
385{-# INLINE wSerialFst #-}
386wSerialFst :: IsStream t => t m a -> t m a -> t m a
387wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
388 let yieldFirst a r = yld a (yieldSecond r m2)
389 in foldStream st yieldFirst sng stp m1
393 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
394 let stop = foldStream st yld sng stp s1
396 yieldk a r = yld a (wSerial s1 r)
397 in foldStream st yieldk single stop s2
399-- | Like `wSerial` but stops interleaving as soon as any of the two streams
403{-# INLINE wSerialMin #-}
404wSerialMin :: IsStream t => t m a -> t m a -> t m a
405wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
408 yieldk a r = yld a (wSerial m2 r)
409 foldStream st yieldk single stop m1
411instance Semigroup (WSerialT m a) where
416-- | Same as 'wSerial'.
419{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
421(<=>) :: IsStream t => t m a -> t m a -> t m a
424------------------------------------------------------------------------------
426------------------------------------------------------------------------------
428instance Monoid (WSerialT m a) where
432{-# INLINE apWSerial #-}
433apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
434apWSerial (WSerialT m1) (WSerialT m2) =
435 let f x1 = K.concatMapBy wSerial (pure . x1) m2
436 in WSerialT $ K.concatMapBy wSerial f m1
438instance Monad m => Applicative (WSerialT m) where
440 pure = WSerialT . K.yield
444------------------------------------------------------------------------------
446------------------------------------------------------------------------------
448instance Monad m => Monad (WSerialT m) where
451 (>>=) = K.bindWith wSerial
453------------------------------------------------------------------------------
455------------------------------------------------------------------------------
457MONAD_COMMON_INSTANCES(WSerialT,)
458LIST_INSTANCES(WSerialT)
459NFDATA1_INSTANCE(WSerialT)
460FOLDABLE_INSTANCE(WSerialT)
461TRAVERSABLE_INSTANCE(WSerialT)
463------------------------------------------------------------------------------
465------------------------------------------------------------------------------
467-- | Build a stream by unfolding a /monadic/ step function starting from a
468-- seed. The step function returns the next element in the stream and the next
469-- seed value. When it is done it returns 'Nothing' and the stream ends. For
475-- then return Nothing
476-- else print b >> return (Just (b, b + 1))
477-- in drain $ unfoldrM f 0
488{-# INLINE unfoldrM #-}
489unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
490unfoldrM step seed = D.fromStreamD (D.unfoldrM step seed)
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