1{-# LANGUAGE UndecidableInstances #-}
4-- Module : Streamly.Internal.Data.Stream.Serial
5-- Copyright : (c) 2017 Composewell Technologies
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.Type
68 (IsStream(..), adapt, Stream, mkStream, foldStream)
69import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
71import qualified Streamly.Internal.Data.Stream.Prelude as P
72 (cmpBy, foldl', foldr, eqBy, fromList, toList)
73import qualified Streamly.Internal.Data.Stream.StreamK as K (withLocal)
74import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
75import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D (unfoldrM)
76import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
78import Prelude hiding (map, mapM, errorWithoutStackTrace)
80#include "Instances.hs"
83------------------------------------------------------------------------------
85------------------------------------------------------------------------------
87-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
88-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
89-- first until it exhausts and then stream @b@ is evaluated. In other words,
90-- the elements of stream @b@ are appended to the elements of stream @a@. This
91-- operation can be used to fold an infinite lazy container of streams.
95-- import qualified "Streamly.Prelude" as S
97-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
103-- The 'Monad' instance runs the /monadic continuation/ for each
104-- element of the stream, serially.
107-- main = S.drain . 'serially' $ do
108-- x <- return 1 \<\> return 2
116-- 'SerialT' nests streams serially in a depth first manner.
119-- main = S.drain . 'serially' $ do
120-- x <- return 1 \<\> return 2
121-- y <- return 3 \<\> return 4
122-- S.yieldM $ print (x, y)
131-- We call the monadic code being run for each element of the stream a monadic
132-- continuation. In imperative paradigm we can think of this composition as
133-- nested @for@ loops and the monadic continuation is the body of the loop. The
134-- loop iterates for all elements of the stream.
136-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
137-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
138-- can contain effectful actions while lists are pure.
140-- In the code above, the 'serially' combinator can be omitted as the default
141-- stream type is 'SerialT'.
143-- /Since: 0.2.0 ("Streamly")/
146newtype SerialT m a = SerialT {getSerialT :: Stream m a}
147 -- XXX when deriving do we inherit an INLINE?
148 deriving (Semigroup, Monoid, MonadTrans)
150-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
153-- /Since: 0.2.0 ("Streamly")/
156type Serial = SerialT IO
160{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
161type StreamT = SerialT
163-- | Fix the type of a polymorphic stream as 'SerialT'.
165-- /Since: 0.1.0 ("Streamly")/
168serially :: IsStream t => SerialT m a -> t m a
171{-# INLINE consMSerial #-}
172{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
173consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
174consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
176instance IsStream SerialT where
177 toStream = getSerialT
182------------------------------------------------------------------------------
184------------------------------------------------------------------------------
186instance Monad m => Monad (SerialT m) where
189 -- Benchmarks better with StreamD bind and pure:
190 -- toList, filterAllout, *>, *<, >> (~2x)
192 -- pure = SerialT . D.fromStreamD . D.yield
193 -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)
195 -- Benchmarks better with CPS bind and pure:
197 -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)
200 (>>=) = K.bindWith K.serial
205------------------------------------------------------------------------------
207------------------------------------------------------------------------------
210mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
211mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
221-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
227map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
228map f = mapM (return . f)
230{-# INLINE apSerial #-}
231apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
232apSerial (SerialT m1) (SerialT m2) =
233 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
235{-# INLINE apSequence #-}
236apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
237apSequence (SerialT m1) (SerialT m2) =
238 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
240{-# INLINE apDiscardSnd #-}
241apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
242apDiscardSnd (SerialT m1) (SerialT m2) =
243 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
245-- Note: we need to define all the typeclass operations because we want to
247instance Monad m => Applicative (SerialT m) where
249 pure = SerialT . K.yield
253 -- (<*>) = K.apSerial
255#if MIN_VERSION_base(4,10,0)
256 {-# INLINE liftA2 #-}
257 liftA2 f x = (<*>) (fmap f x)
262 -- (*>) = K.apSerialDiscardFst
266 -- (<*) = K.apSerialDiscardSnd
268MONAD_COMMON_INSTANCES(SerialT,)
269LIST_INSTANCES(SerialT)
270NFDATA1_INSTANCE(SerialT)
271FOLDABLE_INSTANCE(SerialT)
272TRAVERSABLE_INSTANCE(SerialT)
274------------------------------------------------------------------------------
276------------------------------------------------------------------------------
278-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
279-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
280-- first to produce the first element of the combined stream and then stream
281-- @b@ is evaluated to produce the next element of the combined stream, and
282-- then we go back to evaluating stream @a@ and so on. In other words, the
283-- elements of stream @a@ are interleaved with the elements of stream @b@.
285-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
286-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
287-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
288-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
289-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
290-- leftmost stream gets the same scheduling priority as the rest of the
291-- streams taken together. The same is true for each subexpression on the right.
293-- Note that this operation cannot be used to fold a container of infinite
294-- streams as the state that it needs to maintain is proportional to the number
297-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
298-- contrast to the depth wise scheduling behavior of 'SerialT'.
302-- import qualified "Streamly.Prelude" as S
304-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
310-- Similarly, the 'Monad' instance interleaves the iterations of the
311-- inner and the outer loop, nesting loops in a breadth first manner.
315-- main = S.drain . 'wSerially' $ do
316-- x <- return 1 \<\> return 2
317-- y <- return 3 \<\> return 4
318-- S.yieldM $ print (x, y)
327-- /Since: 0.2.0 ("Streamly")/
330newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
331 deriving (MonadTrans)
333-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
334-- documentation for more details.
336-- /Since: 0.2.0 ("Streamly")/
339type WSerial = WSerialT IO
343{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
344type InterleavedT = WSerialT
346-- | Fix the type of a polymorphic stream as 'WSerialT'.
348-- /Since: 0.2.0 ("Streamly")/
351wSerially :: IsStream t => WSerialT m a -> t m a
354-- | Same as 'wSerially'.
357{-# DEPRECATED interleaving "Please use wSerially instead." #-}
358interleaving :: IsStream t => WSerialT m a -> t m a
359interleaving = wSerially
361consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
362consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
364instance IsStream WSerialT where
365 toStream = getWSerialT
366 fromStream = WSerialT
369 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
370 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
374 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
375 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
378------------------------------------------------------------------------------
380------------------------------------------------------------------------------
384-- Additionally we can have m elements yield from the first stream and n
385-- elements yielding from the second stream. We can also have time slicing
386-- variants of positional interleaving, e.g. run first stream for m seconds and
387-- run the second stream for n seconds.
389-- Similar combinators can be implemented using WAhead style.
391-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
392-- Interleaves two streams, yielding one element from each stream alternately.
393-- When one stream stops the rest of the other stream is used in the output
398-- /Since: 0.2.0 ("Streamly")/
399{-# INLINE wSerial #-}
400wSerial :: IsStream t => t m a -> t m a -> t m a
401wSerial m1 m2 = mkStream $ \st yld sng stp -> do
402 let stop = foldStream st yld sng stp m2
404 yieldk a r = yld a (wSerial m2 r)
405 foldStream st yieldk single stop m1
407-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
410{-# INLINE wSerialFst #-}
411wSerialFst :: IsStream t => t m a -> t m a -> t m a
412wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
413 let yieldFirst a r = yld a (yieldSecond r m2)
414 in foldStream st yieldFirst sng stp m1
418 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
419 let stop = foldStream st yld sng stp s1
421 yieldk a r = yld a (wSerial s1 r)
422 in foldStream st yieldk single stop s2
424-- | Like `wSerial` but stops interleaving as soon as any of the two streams
428{-# INLINE wSerialMin #-}
429wSerialMin :: IsStream t => t m a -> t m a -> t m a
430wSerialMin m1 m2 = mkStream $ \st yld _ stp -> do
432 -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
433 -- "sng a" to keep the behaviour consistent with the yield continuation.
434 single a = yld a (wSerialMin m2 K.nil)
435 yieldk a r = yld a (wSerialMin m2 r)
436 foldStream st yieldk single stop m1
438instance Semigroup (WSerialT m a) where
443-- | Same as 'wSerial'.
446{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
448(<=>) :: IsStream t => t m a -> t m a -> t m a
451------------------------------------------------------------------------------
453------------------------------------------------------------------------------
455instance Monoid (WSerialT m a) where
459{-# INLINE apWSerial #-}
460apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
461apWSerial (WSerialT m1) (WSerialT m2) =
462 let f x1 = K.concatMapBy wSerial (pure . x1) m2
463 in WSerialT $ K.concatMapBy wSerial f m1
465instance Monad m => Applicative (WSerialT m) where
467 pure = WSerialT . K.yield
471------------------------------------------------------------------------------
473------------------------------------------------------------------------------
475instance Monad m => Monad (WSerialT m) where
478 (>>=) = K.bindWith wSerial
480------------------------------------------------------------------------------
482------------------------------------------------------------------------------
484MONAD_COMMON_INSTANCES(WSerialT,)
485LIST_INSTANCES(WSerialT)
486NFDATA1_INSTANCE(WSerialT)
487FOLDABLE_INSTANCE(WSerialT)
488TRAVERSABLE_INSTANCE(WSerialT)
490------------------------------------------------------------------------------
492------------------------------------------------------------------------------
494-- | Build a stream by unfolding a /monadic/ step function starting from a
495-- seed. The step function returns the next element in the stream and the next
496-- seed value. When it is done it returns 'Nothing' and the stream ends. For
502-- then return Nothing
503-- else print b >> return (Just (b, b + 1))
504-- in drain $ unfoldrM f 0
515{-# INLINE unfoldrM #-}
516unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
517unfoldrM 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