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)
51-- import Control.Monad.Error.Class (MonadError(..))
52import Control.Monad.IO.Class (MonadIO(..))
53import Control.Monad.Reader.Class (MonadReader(..))
54import Control.Monad.State.Class (MonadState(..))
55import Control.Monad.Trans.Class (MonadTrans(lift))
56import Data.Foldable (Foldable(foldl'), fold)
57import Data.Functor.Identity (Identity(..), runIdentity)
58import Data.Maybe (fromMaybe)
59import Data.Semigroup (Endo(..))
60#if __GLASGOW_HASKELL__ < 808
61import Data.Semigroup (Semigroup(..))
63import GHC.Exts (IsList(..), IsString(..))
64import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
66import Prelude hiding (map, mapM, errorWithoutStackTrace)
68import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
69import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStream,
71import Streamly.Internal.Data.Strict (Maybe'(..), toMaybe)
72import qualified Streamly.Internal.Data.Stream.Prelude as P
73import qualified Streamly.Internal.Data.Stream.StreamK as K
74import qualified Streamly.Internal.Data.Stream.StreamD as D
76#include "Instances.hs"
79------------------------------------------------------------------------------
81------------------------------------------------------------------------------
83-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
84-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
85-- first until it exhausts and then stream @b@ is evaluated. In other words,
86-- the elements of stream @b@ are appended to the elements of stream @a@. This
87-- operation can be used to fold an infinite lazy container of streams.
91-- import qualified "Streamly.Prelude" as S
93-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
99-- The 'Monad' instance runs the /monadic continuation/ for each
100-- element of the stream, serially.
103-- main = S.drain . 'serially' $ do
104-- x <- return 1 \<\> return 2
112-- 'SerialT' nests streams serially in a depth first manner.
115-- main = S.drain . 'serially' $ do
116-- x <- return 1 \<\> return 2
117-- y <- return 3 \<\> return 4
118-- S.yieldM $ print (x, y)
127-- We call the monadic code being run for each element of the stream a monadic
128-- continuation. In imperative paradigm we can think of this composition as
129-- nested @for@ loops and the monadic continuation is the body of the loop. The
130-- loop iterates for all elements of the stream.
132-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
133-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
134-- can contain effectful actions while lists are pure.
136-- In the code above, the 'serially' combinator can be omitted as the default
137-- stream type is 'SerialT'.
140newtype SerialT m a = SerialT {getSerialT :: Stream m a}
141 -- XXX when deriving do we inherit an INLINE?
142 deriving (Semigroup, Monoid, MonadTrans)
144-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
148type Serial = SerialT IO
152{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
153type StreamT = SerialT
155-- | Fix the type of a polymorphic stream as 'SerialT'.
158serially :: IsStream t => SerialT m a -> t m a
161{-# INLINE consMSerial #-}
162{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
163consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
164consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
166instance IsStream SerialT where
167 toStream = getSerialT
172------------------------------------------------------------------------------
174------------------------------------------------------------------------------
176instance Monad m => Monad (SerialT m) where
180 (>>=) = K.bindWith K.serial
185 -- StreamD based implementation
186 -- return = SerialT . D.fromStreamD . D.yield
187 -- m >>= f = D.fromStreamD $ D.concatMap (\a -> D.toStreamD (f a)) (D.toStreamD m)
189------------------------------------------------------------------------------
191------------------------------------------------------------------------------
194mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
195mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
205-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
211map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
212map f = mapM (return . f)
214{-# INLINE apSerial #-}
215apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
216apSerial (SerialT m1) (SerialT m2) =
217 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
219{-# INLINE apSequence #-}
220apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
221apSequence (SerialT m1) (SerialT m2) =
222 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
224{-# INLINE apDiscardSnd #-}
225apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
226apDiscardSnd (SerialT m1) (SerialT m2) =
227 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
229-- Note: we need to define all the typeclass operations because we want to
231instance Monad m => Applicative (SerialT m) where
233 pure = SerialT . K.yield
237 -- (<*>) = K.apSerial
239#if MIN_VERSION_base(4,10,0)
240 {-# INLINE liftA2 #-}
241 liftA2 f x = (<*>) (fmap f x)
246 -- (*>) = K.apSerialDiscardFst
250 -- (<*) = K.apSerialDiscardSnd
252MONAD_COMMON_INSTANCES(SerialT,)
253LIST_INSTANCES(SerialT)
254NFDATA1_INSTANCE(SerialT)
255FOLDABLE_INSTANCE(SerialT)
256TRAVERSABLE_INSTANCE(SerialT)
258------------------------------------------------------------------------------
260------------------------------------------------------------------------------
262-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
263-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
264-- first to produce the first element of the combined stream and then stream
265-- @b@ is evaluated to produce the next element of the combined stream, and
266-- then we go back to evaluating stream @a@ and so on. In other words, the
267-- elements of stream @a@ are interleaved with the elements of stream @b@.
269-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
270-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
271-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
272-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
273-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
274-- leftmost stream gets the same scheduling priority as the rest of the
275-- streams taken together. The same is true for each subexpression on the right.
277-- Note that this operation cannot be used to fold a container of infinite
278-- streams as the state that it needs to maintain is proportional to the number
281-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
282-- contrast to the depth wise scheduling behavior of 'SerialT'.
286-- import qualified "Streamly.Prelude" as S
288-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
294-- Similarly, the 'Monad' instance interleaves the iterations of the
295-- inner and the outer loop, nesting loops in a breadth first manner.
299-- main = S.drain . 'wSerially' $ do
300-- x <- return 1 \<\> return 2
301-- y <- return 3 \<\> return 4
302-- S.yieldM $ print (x, y)
312newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
313 deriving (MonadTrans)
315-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
316-- documentation for more details.
319type WSerial = WSerialT IO
323{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
324type InterleavedT = WSerialT
326-- | Fix the type of a polymorphic stream as 'WSerialT'.
329wSerially :: IsStream t => WSerialT m a -> t m a
332-- | Same as 'wSerially'.
335{-# DEPRECATED interleaving "Please use wSerially instead." #-}
336interleaving :: IsStream t => WSerialT m a -> t m a
337interleaving = wSerially
339consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
340consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
342instance IsStream WSerialT where
343 toStream = getWSerialT
344 fromStream = WSerialT
347 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
348 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
352 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
353 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
356------------------------------------------------------------------------------
358------------------------------------------------------------------------------
360-- Additionally we can have m elements yield from the first stream and n
361-- elements yielding from the second stream. We can also have time slicing
362-- variants of positional interleaving, e.g. run first stream for m seconds and
363-- run the second stream for n seconds.
365-- Similar combinators can be implemented using WAhead style.
367-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
368-- Interleaves two streams, yielding one element from each stream alternately.
369-- When one stream stops the rest of the other stream is used in the output
373{-# INLINE wSerial #-}
374wSerial :: IsStream t => t m a -> t m a -> t m a
375wSerial m1 m2 = mkStream $ \st yld sng stp -> do
376 let stop = foldStream st yld sng stp m2
378 yieldk a r = yld a (wSerial m2 r)
379 foldStream st yieldk single stop m1
381-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
384{-# INLINE wSerialFst #-}
385wSerialFst :: IsStream t => t m a -> t m a -> t m a
386wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
387 let yieldFirst a r = yld a (yieldSecond r m2)
388 in foldStream st yieldFirst sng stp m1
392 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
393 let stop = foldStream st yld sng stp s1
395 yieldk a r = yld a (wSerial s1 r)
396 in foldStream st yieldk single stop s2
398-- | Like `wSerial` but stops interleaving as soon as any of the two streams
402{-# INLINE wSerialMin #-}
403wSerialMin :: IsStream t => t m a -> t m a -> t m a
404wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
407 yieldk a r = yld a (wSerial m2 r)
408 foldStream st yieldk single stop m1
410instance Semigroup (WSerialT m a) where
415-- | Same as 'wSerial'.
418{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
420(<=>) :: IsStream t => t m a -> t m a -> t m a
423------------------------------------------------------------------------------
425------------------------------------------------------------------------------
427instance Monoid (WSerialT m a) where
431{-# INLINE apWSerial #-}
432apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
433apWSerial (WSerialT m1) (WSerialT m2) =
434 let f x1 = K.concatMapBy wSerial (pure . x1) m2
435 in WSerialT $ K.concatMapBy wSerial f m1
437instance Monad m => Applicative (WSerialT m) where
439 pure = WSerialT . K.yield
443------------------------------------------------------------------------------
445------------------------------------------------------------------------------
447instance Monad m => Monad (WSerialT m) where
450 (>>=) = K.bindWith wSerial
452------------------------------------------------------------------------------
454------------------------------------------------------------------------------
456MONAD_COMMON_INSTANCES(WSerialT,)
457LIST_INSTANCES(WSerialT)
458NFDATA1_INSTANCE(WSerialT)
459FOLDABLE_INSTANCE(WSerialT)
460TRAVERSABLE_INSTANCE(WSerialT)
462------------------------------------------------------------------------------
464------------------------------------------------------------------------------
466-- | Build a stream by unfolding a /monadic/ step function starting from a
467-- seed. The step function returns the next element in the stream and the next
468-- seed value. When it is done it returns 'Nothing' and the stream ends. For
474-- then return Nothing
475-- else print b >> return (Just (b, b + 1))
476-- in drain $ unfoldrM f 0
487{-# INLINE unfoldrM #-}
488unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
489unfoldrM 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