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'.
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------------------------------------------------------------------------------
363-- Additionally we can have m elements yield from the first stream and n
364-- elements yielding from the second stream. We can also have time slicing
365-- variants of positional interleaving, e.g. run first stream for m seconds and
366-- run the second stream for n seconds.
368-- Similar combinators can be implemented using WAhead style.
370-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
371-- Interleaves two streams, yielding one element from each stream alternately.
372-- When one stream stops the rest of the other stream is used in the output
376{-# INLINE wSerial #-}
377wSerial :: IsStream t => t m a -> t m a -> t m a
378wSerial m1 m2 = mkStream $ \st yld sng stp -> do
379 let stop = foldStream st yld sng stp m2
381 yieldk a r = yld a (wSerial m2 r)
382 foldStream st yieldk single stop m1
384-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
387{-# INLINE wSerialFst #-}
388wSerialFst :: IsStream t => t m a -> t m a -> t m a
389wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
390 let yieldFirst a r = yld a (yieldSecond r m2)
391 in foldStream st yieldFirst sng stp m1
395 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
396 let stop = foldStream st yld sng stp s1
398 yieldk a r = yld a (wSerial s1 r)
399 in foldStream st yieldk single stop s2
401-- | Like `wSerial` but stops interleaving as soon as any of the two streams
405{-# INLINE wSerialMin #-}
406wSerialMin :: IsStream t => t m a -> t m a -> t m a
407wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
410 yieldk a r = yld a (wSerial m2 r)
411 foldStream st yieldk single stop m1
413instance Semigroup (WSerialT m a) where
418-- | Same as 'wSerial'.
421{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
423(<=>) :: IsStream t => t m a -> t m a -> t m a
426------------------------------------------------------------------------------
428------------------------------------------------------------------------------
430instance Monoid (WSerialT m a) where
434{-# INLINE apWSerial #-}
435apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
436apWSerial (WSerialT m1) (WSerialT m2) =
437 let f x1 = K.concatMapBy wSerial (pure . x1) m2
438 in WSerialT $ K.concatMapBy wSerial f m1
440instance Monad m => Applicative (WSerialT m) where
442 pure = WSerialT . K.yield
446------------------------------------------------------------------------------
448------------------------------------------------------------------------------
450instance Monad m => Monad (WSerialT m) where
453 (>>=) = K.bindWith wSerial
455------------------------------------------------------------------------------
457------------------------------------------------------------------------------
459MONAD_COMMON_INSTANCES(WSerialT,)
460LIST_INSTANCES(WSerialT)
461NFDATA1_INSTANCE(WSerialT)
462FOLDABLE_INSTANCE(WSerialT)
463TRAVERSABLE_INSTANCE(WSerialT)
465------------------------------------------------------------------------------
467------------------------------------------------------------------------------
469-- | Build a stream by unfolding a /monadic/ step function starting from a
470-- seed. The step function returns the next element in the stream and the next
471-- seed value. When it is done it returns 'Nothing' and the stream ends. For
477-- then return Nothing
478-- else print b >> return (Just (b, b + 1))
479-- in drain $ unfoldrM f 0
490{-# INLINE unfoldrM #-}
491unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
492unfoldrM 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