git.iko.soy/streamly
on 0839d4f help sign in
streamly/src/Streamly/Streams/Serial.hs
355 lines 11 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE DeriveTraversable #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving#-}
7{-# LANGUAGE InstanceSigs #-}
8{-# LANGUAGE MultiParamTypeClasses #-}
9{-# LANGUAGE TypeFamilies #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- |
13-- Module : Streamly.Streams.Serial
14-- Copyright : (c) 2017 Harendra Kumar
15--
16-- License : BSD3
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
19-- Portability : GHC
20--
21--
22module Streamly.Streams.Serial
23 (
24 -- * Serial appending stream
25 SerialT
26 , StreamT -- deprecated
27 , Serial
28 , serial
29 , serially
31 -- * Serial interleaving stream
32 , WSerialT
33 , InterleavedT -- deprecated
34 , WSerial
35 , wSerial
36 , (<=>) -- deprecated
37 , wSerially
38 , interleaving -- deprecated
40 -- * Transformation
41 , map
42 , mapM
43 )
44where
46import Control.DeepSeq (NFData(..), NFData1(..), rnf1)
47import Control.Monad (ap)
48import Control.Monad.Base (MonadBase(..), liftBaseDefault)
49import Control.Monad.Catch (MonadThrow, throwM)
50-- import Control.Monad.Error.Class (MonadError(..))
51import Control.Monad.IO.Class (MonadIO(..))
52import Control.Monad.Reader.Class (MonadReader(..))
53import Control.Monad.State.Class (MonadState(..))
54import Control.Monad.Trans.Class (MonadTrans(lift))
55import Data.Functor.Identity (Identity, runIdentity)
56import Data.Semigroup (Semigroup(..))
57import GHC.Exts (IsList(..), IsString(..))
58import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
59 readListPrecDefault)
60import Prelude hiding (map, mapM)
62import Streamly.SVar (rstState)
63import Streamly.Streams.StreamK (IsStream(..), adapt, Stream(..))
64import qualified Streamly.Streams.Prelude as P
65import qualified Streamly.Streams.StreamK as K
66import qualified Streamly.Streams.StreamD as D
68#include "Instances.hs"
69#include "inline.hs"
71------------------------------------------------------------------------------
72-- SerialT
73------------------------------------------------------------------------------
75-- | Deep serial composition or serial composition with depth first traversal.
76-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
77-- depth first manner, yielding all elements from the first stream, and then
78-- all elements from the second stream.
79--
80-- @
81-- import Streamly
82-- import qualified "Streamly.Prelude" as S
83--
84-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
85-- @
86-- @
87-- [1,2,3,4]
88-- @
89--
90-- The 'Monad' instance runs the /monadic continuation/ for each
91-- element of the stream, serially.
92--
93-- @
94-- main = 'runStream' . 'serially' $ do
95-- x <- return 1 \<\> return 2
96-- S.yieldM $ print x
97-- @
98-- @
99-- 1
100-- 2
101-- @
102--
103-- 'SerialT' nests streams serially in a depth first manner.
104--
105-- @
106-- main = 'runStream' . 'serially' $ do
107-- x <- return 1 \<\> return 2
108-- y <- return 3 \<\> return 4
109-- S.yieldM $ print (x, y)
110-- @
111-- @
112-- (1,3)
113-- (1,4)
114-- (2,3)
115-- (2,4)
116-- @
117--
118-- This behavior of 'SerialT' is exactly like a list transformer. We call the
119-- monadic code being run for each element of the stream a monadic
120-- continuation. In imperative paradigm we can think of this composition as
121-- nested @for@ loops and the monadic continuation is the body of the loop. The
122-- loop iterates for all elements of the stream.
123--
124-- The 'serially' combinator can be omitted as the default stream type is
125-- 'SerialT'.
126-- Note that serial composition with depth first traversal can be used to
127-- combine an infinite number of streams as it explores only one stream at a
128-- time.
129--
130-- @since 0.2.0
131newtype SerialT m a = SerialT {getSerialT :: Stream m a}
132 deriving (Semigroup, Monoid, MonadTrans, Foldable, Traversable)
134-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
135-- for more details.
136--
137-- @since 0.2.0
138type Serial = SerialT IO
140-- |
141-- @since 0.1.0
142{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
143type StreamT = SerialT
145-- | Fix the type of a polymorphic stream as 'SerialT'.
146--
147-- @since 0.1.0
148serially :: IsStream t => SerialT m a -> t m a
149serially = adapt
151instance IsStream SerialT where
152 toStream = getSerialT
153 fromStream = SerialT
155 {-# INLINE consM #-}
156 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
157 consM :: Monad m => m a -> SerialT m a -> SerialT m a
158 consM m r = fromStream $ K.consMSerial m (toStream r)
160 {-# INLINE (|:) #-}
161 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
162 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
163 m |: r = fromStream $ K.consMSerial m (toStream r)
165------------------------------------------------------------------------------
166-- Semigroup
167------------------------------------------------------------------------------
169-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
170-- Appends two streams sequentially, yielding all elements from the first
171-- stream, and then all elements from the second stream.
172--
173-- @since 0.2.0
174{-# INLINE serial #-}
175serial :: IsStream t => t m a -> t m a -> t m a
176serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
177 unStream (K.serial (toStream m1) (toStream m2))
178 (rstState st) stp sng yld
180------------------------------------------------------------------------------
181-- Monad
182------------------------------------------------------------------------------
184instance Monad m => Monad (SerialT m) where
185 return = pure
186 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \st stp sng yld ->
187 let run x = unStream x (rstState st) stp sng yld
188 single a = run $ toStream (f a)
189 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
190 in m (rstState st) stp single yieldk
192------------------------------------------------------------------------------
193-- Other instances
194------------------------------------------------------------------------------
196{-# INLINE_EARLY mapM #-}
197mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
198mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream m)
200-- | Same as 'fmap'.
201--
202-- @since 0.4.0
203{-# INLINE map #-}
204map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
205map f = mapM (return . f)
207MONAD_APPLICATIVE_INSTANCE(SerialT,)
208MONAD_COMMON_INSTANCES(SerialT,)
210------------------------------------------------------------------------------
211-- WSerialT
212------------------------------------------------------------------------------
214-- | Wide serial composition or serial composition with a breadth first
215-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
216-- the two streams in a breadth first manner. In other words, it interleaves
217-- two streams, yielding one element from each stream alternately.
218--
219-- @
220-- import Streamly
221-- import qualified "Streamly.Prelude" as S
222--
223-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
224-- @
225-- @
226-- [1,3,2,4]
227-- @
228--
229-- Similarly, the 'Monad' instance interleaves the iterations of the
230-- inner and the outer loop, nesting loops in a breadth first manner.
231--
232--
233-- @
234-- main = 'runStream' . 'wSerially' $ do
235-- x <- return 1 \<\> return 2
236-- y <- return 3 \<\> return 4
237-- S.yieldM $ print (x, y)
238-- @
239-- @
240-- (1,3)
241-- (2,3)
242-- (1,4)
243-- (2,4)
244-- @
245--
246-- Note that a serial composition with breadth first traversal can only combine
247-- a finite number of streams as it needs to retain state for each unfinished
248-- stream.
249--
250-- @since 0.2.0
251newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
252 deriving (MonadTrans, Foldable, Traversable)
254-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
255-- documentation for more details.
256--
257-- @since 0.2.0
258type WSerial = WSerialT IO
260-- |
261-- @since 0.1.0
262{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
263type InterleavedT = WSerialT
265-- | Fix the type of a polymorphic stream as 'WSerialT'.
266--
267-- @since 0.2.0
268wSerially :: IsStream t => WSerialT m a -> t m a
269wSerially = adapt
271-- | Same as 'wSerially'.
272--
273-- @since 0.1.0
274{-# DEPRECATED interleaving "Please use wSerially instead." #-}
275interleaving :: IsStream t => WSerialT m a -> t m a
276interleaving = wSerially
278instance IsStream WSerialT where
279 toStream = getWSerialT
280 fromStream = WSerialT
282 {-# INLINE consM #-}
283 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
284 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
285 consM m r = fromStream $ K.consMSerial m (toStream r)
287 {-# INLINE (|:) #-}
288 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
289 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
290 m |: r = fromStream $ K.consMSerial m (toStream r)
292------------------------------------------------------------------------------
293-- Semigroup
294------------------------------------------------------------------------------
296{-# INLINE interleave #-}
297interleave :: Stream m a -> Stream m a -> Stream m a
298interleave m1 m2 = Stream $ \st stp sng yld -> do
299 let stop = unStream m2 (rstState st) stp sng yld
300 single a = yld a m2
301 yieldk a r = yld a (interleave m2 r)
302 unStream m1 (rstState st) stop single yieldk
304-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
305-- Interleaves two streams, yielding one element from each stream alternately.
306--
307-- @since 0.2.0
308{-# INLINE wSerial #-}
309wSerial :: IsStream t => t m a -> t m a -> t m a
310wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
311 unStream (interleave (toStream m1) (toStream m2))
312 (rstState st) stp sng yld
314instance Semigroup (WSerialT m a) where
315 (<>) = wSerial
317infixr 5 <=>
319-- | Same as 'wSerial'.
320--
321-- @since 0.1.0
322{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
323{-# INLINE (<=>) #-}
324(<=>) :: IsStream t => t m a -> t m a -> t m a
325(<=>) = wSerial
327------------------------------------------------------------------------------
328-- Monoid
329------------------------------------------------------------------------------
331instance Monoid (WSerialT m a) where
332 mempty = K.nil
333 mappend = (<>)
335------------------------------------------------------------------------------
336-- Monad
337------------------------------------------------------------------------------
339instance Monad m => Monad (WSerialT m) where
340 return = pure
341 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->
342 let run x = unStream x (rstState st) stp sng yld
343 single a = run $ toStream (f a)
344 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
345 in m (rstState st) stp single yieldk
347------------------------------------------------------------------------------
348-- Other instances
349------------------------------------------------------------------------------
351MONAD_APPLICATIVE_INSTANCE(WSerialT,)
352MONAD_COMMON_INSTANCES(WSerialT,)
354LIST_INSTANCES(SerialT)
355LIST_INSTANCES(WSerialT)