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.Foldable (Foldable(foldl'), fold)
65import Data.Functor.Identity (Identity(..), runIdentity)
66import Data.Maybe (fromMaybe)
67import Data.Semigroup (Endo(..))
68#if __GLASGOW_HASKELL__ < 808
69import Data.Semigroup (Semigroup(..))
71import GHC.Exts (IsList(..), IsString(..))
72import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
74import Prelude hiding (map, mapM, errorWithoutStackTrace)
76import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
77import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStream,
79import Streamly.Internal.Data.Strict (Maybe'(..), toMaybe)
80import qualified Streamly.Internal.Data.Stream.Prelude as P
81import qualified Streamly.Internal.Data.Stream.StreamK as K
82import qualified Streamly.Internal.Data.Stream.StreamD as D
84#include "Instances.hs"
87------------------------------------------------------------------------------
89------------------------------------------------------------------------------
91-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
92-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
93-- first until it exhausts and then stream @b@ is evaluated. In other words,
94-- the elements of stream @b@ are appended to the elements of stream @a@. This
95-- operation can be used to fold an infinite lazy container of streams.
99-- import qualified "Streamly.Prelude" as S
101-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
107-- The 'Monad' instance runs the /monadic continuation/ for each
108-- element of the stream, serially.
111-- main = S.drain . 'serially' $ do
112-- x <- return 1 \<\> return 2
120-- 'SerialT' nests streams serially in a depth first manner.
123-- main = S.drain . 'serially' $ do
124-- x <- return 1 \<\> return 2
125-- y <- return 3 \<\> return 4
126-- S.yieldM $ print (x, y)
135-- We call the monadic code being run for each element of the stream a monadic
136-- continuation. In imperative paradigm we can think of this composition as
137-- nested @for@ loops and the monadic continuation is the body of the loop. The
138-- loop iterates for all elements of the stream.
140-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
141-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
142-- can contain effectful actions while lists are pure.
144-- In the code above, the 'serially' combinator can be omitted as the default
145-- stream type is 'SerialT'.
148newtype SerialT m a = SerialT {getSerialT :: Stream m a}
149 -- XXX when deriving do we inherit an INLINE?
150 deriving (Semigroup, Monoid, MonadTrans)
152-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
156type Serial = SerialT IO
160{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
161type StreamT = SerialT
163-- | Fix the type of a polymorphic stream as 'SerialT'.
166serially :: IsStream t => SerialT m a -> t m a
169{-# INLINE consMSerial #-}
170{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
171consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
172consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
174instance IsStream SerialT where
175 toStream = getSerialT
180------------------------------------------------------------------------------
182------------------------------------------------------------------------------
184instance Monad m => Monad (SerialT m) where
188 (>>=) = K.bindWith K.serial
193 -- StreamD based implementation
194 -- return = SerialT . D.fromStreamD . D.yield
195 -- m >>= f = D.fromStreamD $ D.concatMap (\a -> D.toStreamD (f a)) (D.toStreamD m)
197------------------------------------------------------------------------------
199------------------------------------------------------------------------------
202mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
203mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
213-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
219map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
220map f = mapM (return . f)
222{-# INLINE apSerial #-}
223apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
224apSerial (SerialT m1) (SerialT m2) =
225 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
227{-# INLINE apSequence #-}
228apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
229apSequence (SerialT m1) (SerialT m2) =
230 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
232{-# INLINE apDiscardSnd #-}
233apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
234apDiscardSnd (SerialT m1) (SerialT m2) =
235 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
237-- Note: we need to define all the typeclass operations because we want to
239instance Monad m => Applicative (SerialT m) where
241 pure = SerialT . K.yield
245 -- (<*>) = K.apSerial
247#if MIN_VERSION_base(4,10,0)
248 {-# INLINE liftA2 #-}
249 liftA2 f x = (<*>) (fmap f x)
254 -- (*>) = K.apSerialDiscardFst
258 -- (<*) = K.apSerialDiscardSnd
260MONAD_COMMON_INSTANCES(SerialT,)
261LIST_INSTANCES(SerialT)
262NFDATA1_INSTANCE(SerialT)
263FOLDABLE_INSTANCE(SerialT)
264TRAVERSABLE_INSTANCE(SerialT)
266------------------------------------------------------------------------------
268------------------------------------------------------------------------------
270-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
271-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
272-- first to produce the first element of the combined stream and then stream
273-- @b@ is evaluated to produce the next element of the combined stream, and
274-- then we go back to evaluating stream @a@ and so on. In other words, the
275-- elements of stream @a@ are interleaved with the elements of stream @b@.
277-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
278-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
279-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
280-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
281-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
282-- leftmost stream gets the same scheduling priority as the rest of the
283-- streams taken together. The same is true for each subexpression on the right.
285-- Note that this operation cannot be used to fold a container of infinite
286-- streams as the state that it needs to maintain is proportional to the number
289-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
290-- contrast to the depth wise scheduling behavior of 'SerialT'.
294-- import qualified "Streamly.Prelude" as S
296-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
302-- Similarly, the 'Monad' instance interleaves the iterations of the
303-- inner and the outer loop, nesting loops in a breadth first manner.
307-- main = S.drain . 'wSerially' $ do
308-- x <- return 1 \<\> return 2
309-- y <- return 3 \<\> return 4
310-- S.yieldM $ print (x, y)
320newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
321 deriving (MonadTrans)
323-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
324-- documentation for more details.
327type WSerial = WSerialT IO
331{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
332type InterleavedT = WSerialT
334-- | Fix the type of a polymorphic stream as 'WSerialT'.
337wSerially :: IsStream t => WSerialT m a -> t m a
340-- | Same as 'wSerially'.
343{-# DEPRECATED interleaving "Please use wSerially instead." #-}
344interleaving :: IsStream t => WSerialT m a -> t m a
345interleaving = wSerially
347consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
348consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
350instance IsStream WSerialT where
351 toStream = getWSerialT
352 fromStream = WSerialT
355 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
356 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
360 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
361 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
364------------------------------------------------------------------------------
366------------------------------------------------------------------------------
368-- Additionally we can have m elements yield from the first stream and n
369-- elements yielding from the second stream. We can also have time slicing
370-- variants of positional interleaving, e.g. run first stream for m seconds and
371-- run the second stream for n seconds.
373-- Similar combinators can be implemented using WAhead style.
375-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
376-- Interleaves two streams, yielding one element from each stream alternately.
377-- When one stream stops the rest of the other stream is used in the output
381{-# INLINE wSerial #-}
382wSerial :: IsStream t => t m a -> t m a -> t m a
383wSerial m1 m2 = mkStream $ \st yld sng stp -> do
384 let stop = foldStream st yld sng stp m2
386 yieldk a r = yld a (wSerial m2 r)
387 foldStream st yieldk single stop m1
389-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
392{-# INLINE wSerialFst #-}
393wSerialFst :: IsStream t => t m a -> t m a -> t m a
394wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
395 let yieldFirst a r = yld a (yieldSecond r m2)
396 in foldStream st yieldFirst sng stp m1
400 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
401 let stop = foldStream st yld sng stp s1
403 yieldk a r = yld a (wSerial s1 r)
404 in foldStream st yieldk single stop s2
406-- | Like `wSerial` but stops interleaving as soon as any of the two streams
410{-# INLINE wSerialMin #-}
411wSerialMin :: IsStream t => t m a -> t m a -> t m a
412wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
415 yieldk a r = yld a (wSerial m2 r)
416 foldStream st yieldk single stop m1
418instance Semigroup (WSerialT m a) where
423-- | Same as 'wSerial'.
426{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
428(<=>) :: IsStream t => t m a -> t m a -> t m a
431------------------------------------------------------------------------------
433------------------------------------------------------------------------------
435instance Monoid (WSerialT m a) where
439{-# INLINE apWSerial #-}
440apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
441apWSerial (WSerialT m1) (WSerialT m2) =
442 let f x1 = K.concatMapBy wSerial (pure . x1) m2
443 in WSerialT $ K.concatMapBy wSerial f m1
445instance Monad m => Applicative (WSerialT m) where
447 pure = WSerialT . K.yield
451------------------------------------------------------------------------------
453------------------------------------------------------------------------------
455instance Monad m => Monad (WSerialT m) where
458 (>>=) = K.bindWith wSerial
460------------------------------------------------------------------------------
462------------------------------------------------------------------------------
464MONAD_COMMON_INSTANCES(WSerialT,)
465LIST_INSTANCES(WSerialT)
466NFDATA1_INSTANCE(WSerialT)
467FOLDABLE_INSTANCE(WSerialT)
468TRAVERSABLE_INSTANCE(WSerialT)
470------------------------------------------------------------------------------
472------------------------------------------------------------------------------
474-- | Build a stream by unfolding a /monadic/ step function starting from a
475-- seed. The step function returns the next element in the stream and the next
476-- seed value. When it is done it returns 'Nothing' and the stream ends. For
482-- then return Nothing
483-- else print b >> return (Just (b, b + 1))
484-- in drain $ unfoldrM f 0
495{-# INLINE unfoldrM #-}
496unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
497unfoldrM 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