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 deriving (Semigroup, Monoid, MonadTrans)
151-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
155type Serial = SerialT IO
159{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
160type StreamT = SerialT
162-- | Fix the type of a polymorphic stream as 'SerialT'.
165serially :: IsStream t => SerialT m a -> t m a
168{-# INLINE consMSerial #-}
169{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
170consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
171consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
173instance IsStream SerialT where
174 toStream = getSerialT
179------------------------------------------------------------------------------
181------------------------------------------------------------------------------
183instance Monad m => Monad (SerialT m) where
186 (>>=) = K.bindWith K.serial
190 -- StreamD based implementation
191 -- return = SerialT . D.fromStreamD . D.yield
192 -- m >>= f = D.fromStreamD $ D.concatMap (\a -> D.toStreamD (f a)) (D.toStreamD m)
194------------------------------------------------------------------------------
196------------------------------------------------------------------------------
199mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
200mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
210-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
216map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
217map f = mapM (return . f)
219{-# INLINE apSerial #-}
220apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
221apSerial (SerialT m1) (SerialT m2) = D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
223{-# INLINE apSequence #-}
224apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
225apSequence (SerialT m1) (SerialT m2) = D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
227instance Monad m => Applicative (SerialT m) where
229 pure = SerialT . K.yield
235MONAD_COMMON_INSTANCES(SerialT,)
236LIST_INSTANCES(SerialT)
237NFDATA1_INSTANCE(SerialT)
238FOLDABLE_INSTANCE(SerialT)
239TRAVERSABLE_INSTANCE(SerialT)
241------------------------------------------------------------------------------
243------------------------------------------------------------------------------
245-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
246-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
247-- first to produce the first element of the combined stream and then stream
248-- @b@ is evaluated to produce the next element of the combined stream, and
249-- then we go back to evaluating stream @a@ and so on. In other words, the
250-- elements of stream @a@ are interleaved with the elements of stream @b@.
252-- Note that evaluation of @a <> b <> c@ does not schedule @a@, @b@ and @c@
253-- with equal priority. This expression is equivalent to @a <> (b <> c)@,
254-- therefore, it fairly interleaves @a@ with the result of @b <> c@. For
255-- example, @S.fromList [1,2] <> S.fromList [3,4] <> S.fromList [5,6] ::
256-- WSerialT Identity Int@ would result in [1,3,2,5,4,6]. In other words, the
257-- leftmost stream gets the same scheduling priority as the rest of the
258-- streams taken together. The same is true for each subexpression on the right.
260-- Note that this operation cannot be used to fold a container of infinite
261-- streams as the state that it needs to maintain is proportional to the number
264-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
265-- contrast to the depth wise scheduling behavior of 'SerialT'.
269-- import qualified "Streamly.Prelude" as S
271-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
277-- Similarly, the 'Monad' instance interleaves the iterations of the
278-- inner and the outer loop, nesting loops in a breadth first manner.
282-- main = S.drain . 'wSerially' $ do
283-- x <- return 1 \<\> return 2
284-- y <- return 3 \<\> return 4
285-- S.yieldM $ print (x, y)
295newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
296 deriving (MonadTrans)
298-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
299-- documentation for more details.
302type WSerial = WSerialT IO
306{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
307type InterleavedT = WSerialT
309-- | Fix the type of a polymorphic stream as 'WSerialT'.
312wSerially :: IsStream t => WSerialT m a -> t m a
315-- | Same as 'wSerially'.
318{-# DEPRECATED interleaving "Please use wSerially instead." #-}
319interleaving :: IsStream t => WSerialT m a -> t m a
320interleaving = wSerially
322consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
323consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
325instance IsStream WSerialT where
326 toStream = getWSerialT
327 fromStream = WSerialT
330 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
331 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
335 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
336 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
339------------------------------------------------------------------------------
341------------------------------------------------------------------------------
343-- Additionally we can have m elements yield from the first stream and n
344-- elements yielding from the second stream. We can also have time slicing
345-- variants of positional interleaving, e.g. run first stream for m seconds and
346-- run the second stream for n seconds.
348-- Similar combinators can be implemented using WAhead style.
350-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
351-- Interleaves two streams, yielding one element from each stream alternately.
352-- When one stream stops the rest of the other stream is used in the output
356{-# INLINE wSerial #-}
357wSerial :: IsStream t => t m a -> t m a -> t m a
358wSerial m1 m2 = mkStream $ \st yld sng stp -> do
359 let stop = foldStream st yld sng stp m2
361 yieldk a r = yld a (wSerial m2 r)
362 foldStream st yieldk single stop m1
364-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
367{-# INLINE wSerialFst #-}
368wSerialFst :: IsStream t => t m a -> t m a -> t m a
369wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
370 let yieldFirst a r = yld a (yieldSecond r m2)
371 in foldStream st yieldFirst sng stp m1
375 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
376 let stop = foldStream st yld sng stp s1
378 yieldk a r = yld a (wSerial s1 r)
379 in foldStream st yieldk single stop s2
381-- | Like `wSerial` but stops interleaving as soon as any of the two streams
385{-# INLINE wSerialMin #-}
386wSerialMin :: IsStream t => t m a -> t m a -> t m a
387wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
390 yieldk a r = yld a (wSerial m2 r)
391 foldStream st yieldk single stop m1
393instance Semigroup (WSerialT m a) where
398-- | Same as 'wSerial'.
401{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
403(<=>) :: IsStream t => t m a -> t m a -> t m a
406------------------------------------------------------------------------------
408------------------------------------------------------------------------------
410instance Monoid (WSerialT m a) where
414{-# INLINE apWSerial #-}
415apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
416apWSerial (WSerialT m1) (WSerialT m2) =
417 let f x1 = K.concatMapBy wSerial (pure . x1) m2
418 in WSerialT $ K.concatMapBy wSerial f m1
420instance Monad m => Applicative (WSerialT m) where
422 pure = WSerialT . K.yield
426------------------------------------------------------------------------------
428------------------------------------------------------------------------------
430instance Monad m => Monad (WSerialT m) where
433 (>>=) = K.bindWith wSerial
435------------------------------------------------------------------------------
437------------------------------------------------------------------------------
439MONAD_COMMON_INSTANCES(WSerialT,)
440LIST_INSTANCES(WSerialT)
441NFDATA1_INSTANCE(WSerialT)
442FOLDABLE_INSTANCE(WSerialT)
443TRAVERSABLE_INSTANCE(WSerialT)
445------------------------------------------------------------------------------
447------------------------------------------------------------------------------
449-- | Build a stream by unfolding a /monadic/ step function starting from a
450-- seed. The step function returns the next element in the stream and the next
451-- seed value. When it is done it returns 'Nothing' and the stream ends. For
457-- then return Nothing
458-- else print b >> return (Just (b, b + 1))
459-- in drain $ unfoldrM f 0
470{-# INLINE unfoldrM #-}
471unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
472unfoldrM 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