git.iko.soy/streamly
on dbcf536 help sign in
streamly/src/Streamly/Streams/Serial.hs
362 lines 11 kB history raw
1{-# LANGUAGE CPP #-}
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
11-- |
12-- Module : Streamly.Streams.Serial
13-- Copyright : (c) 2017 Harendra Kumar
14--
15-- License : BSD3
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
18-- Portability : GHC
19--
20--
21module Streamly.Streams.Serial
22 (
23 -- * Serial appending stream
24 SerialT
25 , StreamT -- deprecated
26 , Serial
27 , serial
28 , serially
30 -- * Serial interleaving stream
31 , WSerialT
32 , InterleavedT -- deprecated
33 , WSerial
34 , wSerial
35 , (<=>) -- deprecated
36 , wSerially
37 , interleaving -- deprecated
39 -- * Transformation
40 , map
41 , mapM
42 )
43where
45import Control.Applicative (liftA2)
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.Foldable (fold)
57import Data.Semigroup (Semigroup(..))
58import GHC.Exts (IsList(..), IsString(..))
59import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
60 readListPrecDefault)
61import Prelude hiding (map, mapM)
63import Streamly.SVar (rstState)
64import Streamly.Streams.StreamK (IsStream(..), adapt, Stream(..))
65import qualified Streamly.Streams.Prelude as P
66import qualified Streamly.Streams.StreamK as K
67import qualified Streamly.Streams.StreamD as D
69#include "Instances.hs"
70#include "inline.hs"
72------------------------------------------------------------------------------
73-- SerialT
74------------------------------------------------------------------------------
76-- | Deep serial composition or serial composition with depth first traversal.
77-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
78-- depth first manner, yielding all elements from the first stream, and then
79-- all elements from the second stream.
80--
81-- @
82-- import Streamly
83-- import qualified "Streamly.Prelude" as S
84--
85-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
86-- @
87-- @
88-- [1,2,3,4]
89-- @
90--
91-- The 'Monad' instance runs the /monadic continuation/ for each
92-- element of the stream, serially.
93--
94-- @
95-- main = 'runStream' . 'serially' $ do
96-- x <- return 1 \<\> return 2
97-- S.yieldM $ print x
98-- @
99-- @
100-- 1
101-- 2
102-- @
103--
104-- 'SerialT' nests streams serially in a depth first manner.
105--
106-- @
107-- main = 'runStream' . 'serially' $ do
108-- x <- return 1 \<\> return 2
109-- y <- return 3 \<\> return 4
110-- S.yieldM $ print (x, y)
111-- @
112-- @
113-- (1,3)
114-- (1,4)
115-- (2,3)
116-- (2,4)
117-- @
118--
119-- This behavior of 'SerialT' is exactly like a list transformer. We call the
120-- monadic code being run for each element of the stream a monadic
121-- continuation. In imperative paradigm we can think of this composition as
122-- nested @for@ loops and the monadic continuation is the body of the loop. The
123-- loop iterates for all elements of the stream.
124--
125-- The 'serially' combinator can be omitted as the default stream type is
126-- 'SerialT'.
127-- Note that serial composition with depth first traversal can be used to
128-- combine an infinite number of streams as it explores only one stream at a
129-- time.
130--
131-- @since 0.2.0
132newtype SerialT m a = SerialT {getSerialT :: Stream m a}
133 deriving (Semigroup, Monoid, MonadTrans)
135-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
136-- for more details.
137--
138-- @since 0.2.0
139type Serial = SerialT IO
141-- |
142-- @since 0.1.0
143{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
144type StreamT = SerialT
146-- | Fix the type of a polymorphic stream as 'SerialT'.
147--
148-- @since 0.1.0
149serially :: IsStream t => SerialT m a -> t m a
150serially = adapt
152instance IsStream SerialT where
153 toStream = getSerialT
154 fromStream = SerialT
156 {-# INLINE consM #-}
157 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
158 consM :: Monad m => m a -> SerialT m a -> SerialT m a
159 consM m r = fromStream $ K.consMSerial m (toStream r)
161 {-# INLINE (|:) #-}
162 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
163 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
164 m |: r = fromStream $ K.consMSerial m (toStream r)
166------------------------------------------------------------------------------
167-- Semigroup
168------------------------------------------------------------------------------
170-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
171-- Appends two streams sequentially, yielding all elements from the first
172-- stream, and then all elements from the second stream.
173--
174-- @since 0.2.0
175{-# INLINE serial #-}
176serial :: IsStream t => t m a -> t m a -> t m a
177serial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
178 unStream (K.serial (toStream m1) (toStream m2))
179 (rstState st) stp sng yld
181------------------------------------------------------------------------------
182-- Monad
183------------------------------------------------------------------------------
185instance Monad m => Monad (SerialT m) where
186 return = pure
187 (SerialT (Stream m)) >>= f = SerialT $ Stream $ \st stp sng yld ->
188 let run x = unStream x (rstState st) stp sng yld
189 single a = run $ toStream (f a)
190 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
191 in m (rstState st) stp single yieldk
193------------------------------------------------------------------------------
194-- Other instances
195------------------------------------------------------------------------------
197{-# INLINE_EARLY mapM #-}
198mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
199mapM f m = fromStream $ D.toStreamK $ D.mapM f $ D.fromStreamK (toStream m)
201-- | Same as 'fmap'.
202--
203-- @since 0.4.0
204{-# INLINE map #-}
205map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
206map f = mapM (return . f)
208MONAD_APPLICATIVE_INSTANCE(SerialT,)
209MONAD_COMMON_INSTANCES(SerialT,)
211------------------------------------------------------------------------------
212-- WSerialT
213------------------------------------------------------------------------------
215-- | Wide serial composition or serial composition with a breadth first
216-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
217-- the two streams in a breadth first manner. In other words, it interleaves
218-- two streams, yielding one element from each stream alternately.
219--
220-- @
221-- import Streamly
222-- import qualified "Streamly.Prelude" as S
223--
224-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
225-- @
226-- @
227-- [1,3,2,4]
228-- @
229--
230-- Similarly, the 'Monad' instance interleaves the iterations of the
231-- inner and the outer loop, nesting loops in a breadth first manner.
232--
233--
234-- @
235-- main = 'runStream' . 'wSerially' $ do
236-- x <- return 1 \<\> return 2
237-- y <- return 3 \<\> return 4
238-- S.yieldM $ print (x, y)
239-- @
240-- @
241-- (1,3)
242-- (2,3)
243-- (1,4)
244-- (2,4)
245-- @
246--
247-- Note that a serial composition with breadth first traversal can only combine
248-- a finite number of streams as it needs to retain state for each unfinished
249-- stream.
250--
251-- @since 0.2.0
252newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
253 deriving (MonadTrans)
255-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
256-- documentation for more details.
257--
258-- @since 0.2.0
259type WSerial = WSerialT IO
261-- |
262-- @since 0.1.0
263{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
264type InterleavedT = WSerialT
266-- | Fix the type of a polymorphic stream as 'WSerialT'.
267--
268-- @since 0.2.0
269wSerially :: IsStream t => WSerialT m a -> t m a
270wSerially = adapt
272-- | Same as 'wSerially'.
273--
274-- @since 0.1.0
275{-# DEPRECATED interleaving "Please use wSerially instead." #-}
276interleaving :: IsStream t => WSerialT m a -> t m a
277interleaving = wSerially
279instance IsStream WSerialT where
280 toStream = getWSerialT
281 fromStream = WSerialT
283 {-# INLINE consM #-}
284 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
285 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
286 consM m r = fromStream $ K.consMSerial m (toStream r)
288 {-# INLINE (|:) #-}
289 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
290 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
291 m |: r = fromStream $ K.consMSerial m (toStream r)
293------------------------------------------------------------------------------
294-- Semigroup
295------------------------------------------------------------------------------
297{-# INLINE interleave #-}
298interleave :: Stream m a -> Stream m a -> Stream m a
299interleave m1 m2 = Stream $ \st stp sng yld -> do
300 let stop = unStream m2 (rstState st) stp sng yld
301 single a = yld a m2
302 yieldk a r = yld a (interleave m2 r)
303 unStream m1 (rstState st) stop single yieldk
305-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
306-- Interleaves two streams, yielding one element from each stream alternately.
307--
308-- @since 0.2.0
309{-# INLINE wSerial #-}
310wSerial :: IsStream t => t m a -> t m a -> t m a
311wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
312 unStream (interleave (toStream m1) (toStream m2))
313 (rstState st) stp sng yld
315instance Semigroup (WSerialT m a) where
316 (<>) = wSerial
318infixr 5 <=>
320-- | Same as 'wSerial'.
321--
322-- @since 0.1.0
323{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
324{-# INLINE (<=>) #-}
325(<=>) :: IsStream t => t m a -> t m a -> t m a
326(<=>) = wSerial
328------------------------------------------------------------------------------
329-- Monoid
330------------------------------------------------------------------------------
332instance Monoid (WSerialT m a) where
333 mempty = K.nil
334 mappend = (<>)
336------------------------------------------------------------------------------
337-- Monad
338------------------------------------------------------------------------------
340instance Monad m => Monad (WSerialT m) where
341 return = pure
342 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->
343 let run x = unStream x (rstState st) stp sng yld
344 single a = run $ toStream (f a)
345 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
346 in m (rstState st) stp single yieldk
348------------------------------------------------------------------------------
349-- Other instances
350------------------------------------------------------------------------------
352MONAD_APPLICATIVE_INSTANCE(WSerialT,)
353MONAD_COMMON_INSTANCES(WSerialT,)
355LIST_INSTANCES(SerialT)
356LIST_INSTANCES(WSerialT)
358FOLDABLE_INSTANCE(SerialT)
359FOLDABLE_INSTANCE(WSerialT)
361TRAVERSABLE_INSTANCE(SerialT)
362TRAVERSABLE_INSTANCE(WSerialT)