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 (ap)
58import Control.Monad.Base (MonadBase(..), liftBaseDefault)
59import Control.Monad.Catch (MonadThrow, throwM)
60-- import Control.Monad.Error.Class (MonadError(..))
61import Control.Monad.IO.Class (MonadIO(..))
62import Control.Monad.Reader.Class (MonadReader(..))
63import Control.Monad.State.Class (MonadState(..))
64import Control.Monad.Trans.Class (MonadTrans(lift))
65import Data.Functor.Identity (Identity(..), runIdentity)
66import Data.Foldable (fold)
67#if __GLASGOW_HASKELL__ < 808
68import Data.Semigroup (Semigroup(..))
70import GHC.Exts (IsList(..), IsString(..))
71import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
73import Prelude hiding (map, mapM)
75import Streamly.Internal.Data.Stream.StreamK (IsStream(..), adapt, Stream, mkStream,
77import qualified Streamly.Internal.Data.Stream.Prelude as P
78import qualified Streamly.Internal.Data.Stream.StreamK as K
79import qualified Streamly.Internal.Data.Stream.StreamD as D
81#include "Instances.hs"
84------------------------------------------------------------------------------
86------------------------------------------------------------------------------
88-- | The 'Semigroup' operation for 'SerialT' behaves like a regular append
89-- operation. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
90-- first until it exhausts and then stream @b@ is evaluated. In other words,
91-- the elements of stream @b@ are appended to the elements of stream @a@. This
92-- operation can be used to fold an infinite lazy container of streams.
96-- import qualified "Streamly.Prelude" as S
98-- main = (S.toList . 'serially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
104-- The 'Monad' instance runs the /monadic continuation/ for each
105-- element of the stream, serially.
108-- main = S.drain . 'serially' $ do
109-- x <- return 1 \<\> return 2
117-- 'SerialT' nests streams serially in a depth first manner.
120-- main = S.drain . 'serially' $ do
121-- x <- return 1 \<\> return 2
122-- y <- return 3 \<\> return 4
123-- S.yieldM $ print (x, y)
132-- We call the monadic code being run for each element of the stream a monadic
133-- continuation. In imperative paradigm we can think of this composition as
134-- nested @for@ loops and the monadic continuation is the body of the loop. The
135-- loop iterates for all elements of the stream.
137-- Note that the behavior and semantics of 'SerialT', including 'Semigroup'
138-- and 'Monad' instances are exactly like Haskell lists except that 'SerialT'
139-- can contain effectful actions while lists are pure.
141-- In the code above, the 'serially' combinator can be omitted as the default
142-- stream type is 'SerialT'.
145newtype SerialT m a = SerialT {getSerialT :: Stream m a}
146 deriving (Semigroup, Monoid, MonadTrans)
148-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
152type Serial = SerialT IO
156{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
157type StreamT = SerialT
159-- | Fix the type of a polymorphic stream as 'SerialT'.
162serially :: IsStream t => SerialT m a -> t m a
165{-# INLINE consMSerial #-}
166{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
167consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
168consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
170instance IsStream SerialT where
171 toStream = getSerialT
176------------------------------------------------------------------------------
178------------------------------------------------------------------------------
180instance Monad m => Monad (SerialT m) where
183 (>>=) = 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------------------------------------------------------------------------------
193{-# INLINE_EARLY mapM #-}
194mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
195mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream 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)
214MONAD_APPLICATIVE_INSTANCE(SerialT,)
215MONAD_COMMON_INSTANCES(SerialT,)
216LIST_INSTANCES(SerialT)
217NFDATA1_INSTANCE(SerialT)
218FOLDABLE_INSTANCE(SerialT)
219TRAVERSABLE_INSTANCE(SerialT)
221------------------------------------------------------------------------------
223------------------------------------------------------------------------------
225-- | The 'Semigroup' operation for 'WSerialT' interleaves the elements from the
226-- two streams. Therefore, when @a <> b@ is evaluated, stream @a@ is evaluated
227-- first to produce the first element of the combined stream and then stream
228-- @b@ is evaluated to produce the next element of the combined stream, and
229-- then we go back to evaluating stream @a@ and so on. In other words, the
230-- elements of stream @a@ are interleaved with the elements of stream @b@.
232-- Note that when multiple actions are combined like @a <> b <> c ... <> z@ we
233-- interleave them in a binary fashion i.e. @a@ and @b@ are interleaved with
234-- each other and the result is interleaved with @c@ and so on. This will not
235-- act as a true round-robin scheduling across all the streams. Note that this
236-- operation cannot be used to fold a container of infinite streams as the
237-- state that it needs to maintain is proportional to the number of streams.
241-- import qualified "Streamly.Prelude" as S
243-- main = (S.toList . 'wSerially' $ (S.fromList [1,2]) \<\> (S.fromList [3,4])) >>= print
249-- Similarly, the 'Monad' instance interleaves the iterations of the
250-- inner and the outer loop, nesting loops in a breadth first manner.
254-- main = S.drain . 'wSerially' $ do
255-- x <- return 1 \<\> return 2
256-- y <- return 3 \<\> return 4
257-- S.yieldM $ print (x, y)
267newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
268 deriving (MonadTrans)
270-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
271-- documentation for more details.
274type WSerial = WSerialT IO
278{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
279type InterleavedT = WSerialT
281-- | Fix the type of a polymorphic stream as 'WSerialT'.
284wSerially :: IsStream t => WSerialT m a -> t m a
287-- | Same as 'wSerially'.
290{-# DEPRECATED interleaving "Please use wSerially instead." #-}
291interleaving :: IsStream t => WSerialT m a -> t m a
292interleaving = wSerially
294consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
295consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
297instance IsStream WSerialT where
298 toStream = getWSerialT
299 fromStream = WSerialT
302 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
303 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
307 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
308 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
311------------------------------------------------------------------------------
313------------------------------------------------------------------------------
315-- Additionally we can have m elements yield from the first stream and n
316-- elements yielding from the second stream. We can also have time slicing
317-- variants of positional interleaving, e.g. run first stream for m seconds and
318-- run the second stream for n seconds.
320-- Similar combinators can be implemented using WAhead style.
322-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
323-- Interleaves two streams, yielding one element from each stream alternately.
324-- When one stream stops the rest of the other stream is used in the output
328{-# INLINE wSerial #-}
329wSerial :: IsStream t => t m a -> t m a -> t m a
330wSerial m1 m2 = mkStream $ \st yld sng stp -> do
331 let stop = foldStream st yld sng stp m2
333 yieldk a r = yld a (wSerial m2 r)
334 foldStream st yieldk single stop m1
336-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
339{-# INLINE wSerialFst #-}
340wSerialFst :: IsStream t => t m a -> t m a -> t m a
341wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
342 let yieldFirst a r = yld a (yieldSecond r m2)
343 in foldStream st yieldFirst sng stp m1
347 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
348 let stop = foldStream st yld sng stp s1
350 yieldk a r = yld a (wSerial s1 r)
351 in foldStream st yieldk single stop s2
353-- | Like `wSerial` but stops interleaving as soon as any of the two streams
357{-# INLINE wSerialMin #-}
358wSerialMin :: IsStream t => t m a -> t m a -> t m a
359wSerialMin m1 m2 = mkStream $ \st yld sng stp -> do
362 yieldk a r = yld a (wSerial m2 r)
363 foldStream st yieldk single stop m1
365instance Semigroup (WSerialT m a) where
370-- | Same as 'wSerial'.
373{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
375(<=>) :: IsStream t => t m a -> t m a -> t m a
378------------------------------------------------------------------------------
380------------------------------------------------------------------------------
382instance Monoid (WSerialT m a) where
386------------------------------------------------------------------------------
388------------------------------------------------------------------------------
390instance Monad m => Monad (WSerialT m) where
393 (>>=) = K.bindWith wSerial
395------------------------------------------------------------------------------
397------------------------------------------------------------------------------
399MONAD_APPLICATIVE_INSTANCE(WSerialT,)
400MONAD_COMMON_INSTANCES(WSerialT,)
401LIST_INSTANCES(WSerialT)
402NFDATA1_INSTANCE(WSerialT)
403FOLDABLE_INSTANCE(WSerialT)
404TRAVERSABLE_INSTANCE(WSerialT)
406------------------------------------------------------------------------------
408------------------------------------------------------------------------------
410-- | Build a stream by unfolding a /monadic/ step function starting from a
411-- seed. The step function returns the next element in the stream and the next
412-- seed value. When it is done it returns 'Nothing' and the stream ends. For
418-- then return Nothing
419-- else print b >> return (Just (b, b + 1))
420-- in drain $ unfoldrM f 0
431{-# INLINE unfoldrM #-}
432unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
433unfoldrM 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