2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE TypeFamilies #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Internal.Data.Stream.Serial
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : streamly@composewell.com
17-- Stability : experimental
21module Streamly.Internal.Data.Stream.Serial
23 -- * Serial appending stream
29 -- * Serial interleaving stream
52import Control.Applicative (liftA2)
53import Control.DeepSeq (NFData(..))
54#if MIN_VERSION_deepseq(1,4,3)
55import Control.DeepSeq (NFData1(..))
57import Control.Monad.Base (MonadBase(..), liftBaseDefault)
58import Control.Monad.Catch (MonadThrow, throwM)
59-- import Control.Monad.Error.Class (MonadError(..))
60import Control.Monad.IO.Class (MonadIO(..))
61import Control.Monad.Reader.Class (MonadReader(..))
62import Control.Monad.State.Class (MonadState(..))
63import Control.Monad.Trans.Class (MonadTrans(lift))
64import Data.Functor.Identity (Identity(..), runIdentity)
65import Data.Foldable (fold)
66#if __GLASGOW_HASKELL__ < 808
67import Data.Semigroup (Semigroup(..))
69import GHC.Exts (IsList(..), IsString(..))
70import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
72import Prelude hiding (map, mapM)
74import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStream,
76import qualified Streamly.Internal.Data.Stream.Prelude as P
77import qualified Streamly.Internal.Data.Stream.StreamK as K
78import qualified Streamly.Internal.Data.Stream.StreamD as D
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'.
144newtype SerialT m a = SerialT {getSerialT :: Stream m a}
145 deriving (Semigroup, Monoid, MonadTrans)
147-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
151type Serial = SerialT IO
155{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
156type StreamT = SerialT
158-- | Fix the type of a polymorphic stream as 'SerialT'.
161serially :: IsStream t => SerialT m a -> t m a
164{-# INLINE consMSerial #-}
165{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
166consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
167consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
169instance IsStream SerialT where
170 toStream = getSerialT
175------------------------------------------------------------------------------
177------------------------------------------------------------------------------
179instance Monad m => Monad (SerialT m) where
182 (>>=) = 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) = 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) = D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
223instance Monad m => Applicative (SerialT m) where
225 pure = SerialT . K.yield
231MONAD_COMMON_INSTANCES(SerialT,)
232LIST_INSTANCES(SerialT)
233NFDATA1_INSTANCE(SerialT)
234FOLDABLE_INSTANCE(SerialT)
235TRAVERSABLE_INSTANCE(SerialT)
237------------------------------------------------------------------------------
239------------------------------------------------------------------------------
241-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
242-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
243-- first to produce the first element of the combined stream and then stream
244-- @b@ is evaluated to produce the next element of the combined stream, and
245-- then we go back to evaluating stream @a@ and so on. In other words, the
246-- elements of stream @a@ are interleaved with the elements of stream @b@.
248-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
249-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
250-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
251-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
252-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
253-- leftmost stream gets the same scheduling priority as the rest of the
254-- streams taken together. The same is true for each subexpression on the right.
256-- Note that this operation cannot be used to fold a container of infinite
257-- streams as the state that it needs to maintain is proportional to the number
260-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
261-- contrast to the depth wise scheduling behavior of 'SerialT'.
265-- import qualified "Streamly.Prelude" as S
267-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
273-- Similarly, the 'Monad' instance interleaves the iterations of the
274-- inner and the outer loop, nesting loops in a breadth first manner.
278-- main = S.drain . 'wSerially' $ do
279-- x <- return 1 \<\> return 2
280-- y <- return 3 \<\> return 4
281-- S.yieldM $ print (x, y)
291newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
292 deriving (MonadTrans)
294-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
295-- documentation for more details.
298type WSerial = WSerialT IO
302{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
303type InterleavedT = WSerialT
305-- | Fix the type of a polymorphic stream as 'WSerialT'.
308wSerially :: IsStream t => WSerialT m a -> t m a
311-- | Same as 'wSerially'.
314{-# DEPRECATED interleaving "Please use wSerially instead." #-}
315interleaving :: IsStream t => WSerialT m a -> t m a
316interleaving = wSerially
318consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
319consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
321instance IsStream WSerialT where
322 toStream = getWSerialT
323 fromStream = WSerialT
326 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
327 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
331 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
332 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
335------------------------------------------------------------------------------
337------------------------------------------------------------------------------
339-- Additionally we can have m elements yield from the first stream and n
340-- elements yielding from the second stream. We can also have time slicing
341-- variants of positional interleaving, e.g. run first stream for m seconds and
342-- run the second stream for n seconds.
344-- Similar combinators can be implemented using WAhead style.
346-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
347-- Interleaves two streams, yielding one element from each stream alternately.
348-- When one stream stops the rest of the other stream is used in the output
352{-# INLINE wSerial #-}
353wSerial :: IsStream t => t m a -> t m a -> t m a
354wSerial m1 m2 = mkStream $ \st yld sng stp -> do
355 let stop = foldStream st yld sng stp m2
357 yieldk a r = yld a (wSerial m2 r)
358 foldStream st yieldk single stop m1
360-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
363{-# INLINE wSerialFst #-}
364wSerialFst :: IsStream t => t m a -> t m a -> t m a
365wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
366 let yieldFirst a r = yld a (yieldSecond r m2)
367 in foldStream st yieldFirst sng stp m1
371 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
372 let stop = foldStream st yld sng stp s1
374 yieldk a r = yld a (wSerial s1 r)
375 in foldStream st yieldk single stop s2
377-- | Like `wSerial` but stops interleaving as soon as any of the two streams
381{-# INLINE wSerialMin #-}
382wSerialMin :: IsStream t => t m a -> t m a -> t m a
383wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
386 yieldk a r = yld a (wSerial m2 r)
387 foldStream st yieldk single stop m1
389instance Semigroup (WSerialT m a) where
394-- | Same as 'wSerial'.
397{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
399(<=>) :: IsStream t => t m a -> t m a -> t m a
402------------------------------------------------------------------------------
404------------------------------------------------------------------------------
406instance Monoid (WSerialT m a) where
410{-# INLINE apWSerial #-}
411apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
412apWSerial (WSerialT m1) (WSerialT m2) =
413 let f x1 = K.concatMapBy wSerial (pure . x1) m2
414 in WSerialT $ K.concatMapBy wSerial f m1
416instance Monad m => Applicative (WSerialT m) where
418 pure = WSerialT . K.yield
422------------------------------------------------------------------------------
424------------------------------------------------------------------------------
426instance Monad m => Monad (WSerialT m) where
429 (>>=) = K.bindWith wSerial
431------------------------------------------------------------------------------
433------------------------------------------------------------------------------
435MONAD_COMMON_INSTANCES(WSerialT,)
436LIST_INSTANCES(WSerialT)
437NFDATA1_INSTANCE(WSerialT)
438FOLDABLE_INSTANCE(WSerialT)
439TRAVERSABLE_INSTANCE(WSerialT)
441------------------------------------------------------------------------------
443------------------------------------------------------------------------------
445-- | Build a stream by unfolding a /monadic/ step function starting from a
446-- seed. The step function returns the next element in the stream and the next
447-- seed value. When it is done it returns 'Nothing' and the stream ends. For
453-- then return Nothing
454-- else print b >> return (Just (b, b + 1))
455-- in drain $ unfoldrM f 0
466{-# INLINE unfoldrM #-}
467unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
468unfoldrM 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