1{-# LANGUAGE UndecidableInstances #-}
4-- Module : Streamly.Internal.Data.Stream.Serial
5-- Copyright : (c) 2017 Harendra Kumar
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
13module Streamly.Internal.Data.Stream.Serial
15 -- * Serial appending stream
21 -- * Serial interleaving stream
44import Control.Applicative (liftA2)
45import Control.DeepSeq (NFData(..))
46#if MIN_VERSION_deepseq(1,4,3)
47import Control.DeepSeq (NFData1(..))
49import Control.Monad.Base (MonadBase(..), liftBaseDefault)
50import Control.Monad.Catch (MonadThrow, throwM)
51import Control.Monad.IO.Class (MonadIO(..))
52import Control.Monad.Reader.Class (MonadReader(..))
53import Control.Monad.State.Class (MonadState(..))
54import Control.Monad.Trans.Class (MonadTrans(lift))
55import Data.Foldable (Foldable(foldl'), fold)
56import Data.Functor.Identity (Identity(..), runIdentity)
57import Data.Maybe (fromMaybe)
58import Data.Semigroup (Endo(..))
59#if __GLASGOW_HASKELL__ < 808
60import Data.Semigroup (Semigroup(..))
62import GHC.Exts (IsList(..), IsString(..))
64 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
65 , readListPrecDefault)
66import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
67import Streamly.Internal.Data.Stream.StreamK
68 (IsStream(..), adapt, Stream, mkStream, foldStream)
69import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
71import qualified Streamly.Internal.Data.Stream.Prelude as P
72import qualified Streamly.Internal.Data.Stream.StreamK as K
73import qualified Streamly.Internal.Data.Stream.StreamD as D
75import Prelude hiding (map, mapM, errorWithoutStackTrace)
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'.
140-- /Since: 0.2.0 ("Streamly")/
143newtype SerialT m a = SerialT {getSerialT :: Stream m a}
144 -- XXX when deriving do we inherit an INLINE?
145 deriving (Semigroup, Monoid, MonadTrans)
147-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
150-- /Since: 0.2.0 ("Streamly")/
153type Serial = SerialT IO
157{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
158type StreamT = SerialT
160-- | Fix the type of a polymorphic stream as 'SerialT'.
162-- /Since: 0.1.0 ("Streamly")/
165serially :: IsStream t => SerialT m a -> t m a
168{-# INLINE consMSerial #-}
169{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
170consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
171consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
173instance IsStream SerialT where
174 toStream = getSerialT
179------------------------------------------------------------------------------
181------------------------------------------------------------------------------
183instance Monad m => Monad (SerialT m) where
187 (>>=) = K.bindWith K.serial
192 -- StreamD based implementation
193 -- return = SerialT . D.fromStreamD . D.yield
194 -- m >>= f = D.fromStreamD $ D.concatMap (\a -> D.toStreamD (f a)) (D.toStreamD m)
196------------------------------------------------------------------------------
198------------------------------------------------------------------------------
201mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
202mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
212-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
218map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
219map f = mapM (return . f)
221{-# INLINE apSerial #-}
222apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
223apSerial (SerialT m1) (SerialT m2) =
224 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
226{-# INLINE apSequence #-}
227apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
228apSequence (SerialT m1) (SerialT m2) =
229 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
231{-# INLINE apDiscardSnd #-}
232apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
233apDiscardSnd (SerialT m1) (SerialT m2) =
234 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
236-- Note: we need to define all the typeclass operations because we want to
238instance Monad m => Applicative (SerialT m) where
240 pure = SerialT . K.yield
244 -- (<*>) = K.apSerial
246#if MIN_VERSION_base(4,10,0)
247 {-# INLINE liftA2 #-}
248 liftA2 f x = (<*>) (fmap f x)
253 -- (*>) = K.apSerialDiscardFst
257 -- (<*) = K.apSerialDiscardSnd
259MONAD_COMMON_INSTANCES(SerialT,)
260LIST_INSTANCES(SerialT)
261NFDATA1_INSTANCE(SerialT)
262FOLDABLE_INSTANCE(SerialT)
263TRAVERSABLE_INSTANCE(SerialT)
265------------------------------------------------------------------------------
267------------------------------------------------------------------------------
269-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
270-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
271-- first to produce the first element of the combined stream and then stream
272-- @b@ is evaluated to produce the next element of the combined stream, and
273-- then we go back to evaluating stream @a@ and so on. In other words, the
274-- elements of stream @a@ are interleaved with the elements of stream @b@.
276-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
277-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
278-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
279-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
280-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
281-- leftmost stream gets the same scheduling priority as the rest of the
282-- streams taken together. The same is true for each subexpression on the right.
284-- Note that this operation cannot be used to fold a container of infinite
285-- streams as the state that it needs to maintain is proportional to the number
288-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
289-- contrast to the depth wise scheduling behavior of 'SerialT'.
293-- import qualified "Streamly.Prelude" as S
295-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
301-- Similarly, the 'Monad' instance interleaves the iterations of the
302-- inner and the outer loop, nesting loops in a breadth first manner.
306-- main = S.drain . 'wSerially' $ do
307-- x <- return 1 \<\> return 2
308-- y <- return 3 \<\> return 4
309-- S.yieldM $ print (x, y)
318-- /Since: 0.2.0 ("Streamly")/
321newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
322 deriving (MonadTrans)
324-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
325-- documentation for more details.
327-- /Since: 0.2.0 ("Streamly")/
330type WSerial = WSerialT IO
334{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
335type InterleavedT = WSerialT
337-- | Fix the type of a polymorphic stream as 'WSerialT'.
339-- /Since: 0.2.0 ("Streamly")/
342wSerially :: IsStream t => WSerialT m a -> t m a
345-- | Same as 'wSerially'.
348{-# DEPRECATED interleaving "Please use wSerially instead." #-}
349interleaving :: IsStream t => WSerialT m a -> t m a
350interleaving = wSerially
352consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
353consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
355instance IsStream WSerialT where
356 toStream = getWSerialT
357 fromStream = WSerialT
360 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
361 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
365 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
366 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
369------------------------------------------------------------------------------
371------------------------------------------------------------------------------
375-- Additionally we can have m elements yield from the first stream and n
376-- elements yielding from the second stream. We can also have time slicing
377-- variants of positional interleaving, e.g. run first stream for m seconds and
378-- run the second stream for n seconds.
380-- Similar combinators can be implemented using WAhead style.
382-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
383-- Interleaves two streams, yielding one element from each stream alternately.
384-- When one stream stops the rest of the other stream is used in the output
389-- /Since: 0.2.0 ("Streamly")/
390{-# INLINE wSerial #-}
391wSerial :: IsStream t => t m a -> t m a -> t m a
392wSerial m1 m2 = mkStream $ \st yld sng stp -> do
393 let stop = foldStream st yld sng stp m2
395 yieldk a r = yld a (wSerial m2 r)
396 foldStream st yieldk single stop m1
398-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
401{-# INLINE wSerialFst #-}
402wSerialFst :: IsStream t => t m a -> t m a -> t m a
403wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
404 let yieldFirst a r = yld a (yieldSecond r m2)
405 in foldStream st yieldFirst sng stp m1
409 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
410 let stop = foldStream st yld sng stp s1
412 yieldk a r = yld a (wSerial s1 r)
413 in foldStream st yieldk single stop s2
415-- | Like `wSerial` but stops interleaving as soon as any of the two streams
419{-# INLINE wSerialMin #-}
420wSerialMin :: IsStream t => t m a -> t m a -> t m a
421wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
424 yieldk a r = yld a (wSerial m2 r)
425 foldStream st yieldk single stop m1
427instance Semigroup (WSerialT m a) where
432-- | Same as 'wSerial'.
435{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
437(<=>) :: IsStream t => t m a -> t m a -> t m a
440------------------------------------------------------------------------------
442------------------------------------------------------------------------------
444instance Monoid (WSerialT m a) where
448{-# INLINE apWSerial #-}
449apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
450apWSerial (WSerialT m1) (WSerialT m2) =
451 let f x1 = K.concatMapBy wSerial (pure . x1) m2
452 in WSerialT $ K.concatMapBy wSerial f m1
454instance Monad m => Applicative (WSerialT m) where
456 pure = WSerialT . K.yield
460------------------------------------------------------------------------------
462------------------------------------------------------------------------------
464instance Monad m => Monad (WSerialT m) where
467 (>>=) = K.bindWith wSerial
469------------------------------------------------------------------------------
471------------------------------------------------------------------------------
473MONAD_COMMON_INSTANCES(WSerialT,)
474LIST_INSTANCES(WSerialT)
475NFDATA1_INSTANCE(WSerialT)
476FOLDABLE_INSTANCE(WSerialT)
477TRAVERSABLE_INSTANCE(WSerialT)
479------------------------------------------------------------------------------
481------------------------------------------------------------------------------
483-- | Build a stream by unfolding a /monadic/ step function starting from a
484-- seed. The step function returns the next element in the stream and the next
485-- seed value. When it is done it returns 'Nothing' and the stream ends. For
491-- then return Nothing
492-- else print b >> return (Just (b, b + 1))
493-- in drain $ unfoldrM f 0
504{-# INLINE unfoldrM #-}
505unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
506unfoldrM 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