git.iko.soy/streamly
on 7c2b9b7 help sign in
streamly/src/Streamly/Streams/Serial.hs
372 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-- |
202-- @
203-- map = fmap
204-- @
205--
206-- Same as 'fmap'.
207--
208-- @
209-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
210-- [2,3,4]
211-- @
212--
213-- @since 0.4.0
214{-# INLINE map #-}
215map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
216map f = mapM (return . f)
218MONAD_APPLICATIVE_INSTANCE(SerialT,)
219MONAD_COMMON_INSTANCES(SerialT,)
221------------------------------------------------------------------------------
222-- WSerialT
223------------------------------------------------------------------------------
225-- | Wide serial composition or serial composition with a breadth first
226-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
227-- the two streams in a breadth first manner. In other words, it interleaves
228-- two streams, yielding one element from each stream alternately.
229--
230-- @
231-- import Streamly
232-- import qualified "Streamly.Prelude" as S
233--
234-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
235-- @
236-- @
237-- [1,3,2,4]
238-- @
239--
240-- Similarly, the 'Monad' instance interleaves the iterations of the
241-- inner and the outer loop, nesting loops in a breadth first manner.
242--
243--
244-- @
245-- main = 'runStream' . 'wSerially' $ do
246-- x <- return 1 \<\> return 2
247-- y <- return 3 \<\> return 4
248-- S.yieldM $ print (x, y)
249-- @
250-- @
251-- (1,3)
252-- (2,3)
253-- (1,4)
254-- (2,4)
255-- @
256--
257-- Note that a serial composition with breadth first traversal can only combine
258-- a finite number of streams as it needs to retain state for each unfinished
259-- stream.
260--
261-- @since 0.2.0
262newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
263 deriving (MonadTrans)
265-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
266-- documentation for more details.
267--
268-- @since 0.2.0
269type WSerial = WSerialT IO
271-- |
272-- @since 0.1.0
273{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
274type InterleavedT = WSerialT
276-- | Fix the type of a polymorphic stream as 'WSerialT'.
277--
278-- @since 0.2.0
279wSerially :: IsStream t => WSerialT m a -> t m a
280wSerially = adapt
282-- | Same as 'wSerially'.
283--
284-- @since 0.1.0
285{-# DEPRECATED interleaving "Please use wSerially instead." #-}
286interleaving :: IsStream t => WSerialT m a -> t m a
287interleaving = wSerially
289instance IsStream WSerialT where
290 toStream = getWSerialT
291 fromStream = WSerialT
293 {-# INLINE consM #-}
294 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
295 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
296 consM m r = fromStream $ K.consMSerial m (toStream r)
298 {-# INLINE (|:) #-}
299 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
300 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
301 m |: r = fromStream $ K.consMSerial m (toStream r)
303------------------------------------------------------------------------------
304-- Semigroup
305------------------------------------------------------------------------------
307{-# INLINE interleave #-}
308interleave :: Stream m a -> Stream m a -> Stream m a
309interleave m1 m2 = Stream $ \st stp sng yld -> do
310 let stop = unStream m2 (rstState st) stp sng yld
311 single a = yld a m2
312 yieldk a r = yld a (interleave m2 r)
313 unStream m1 (rstState st) stop single yieldk
315-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
316-- Interleaves two streams, yielding one element from each stream alternately.
317--
318-- @since 0.2.0
319{-# INLINE wSerial #-}
320wSerial :: IsStream t => t m a -> t m a -> t m a
321wSerial m1 m2 = fromStream $ Stream $ \st stp sng yld ->
322 unStream (interleave (toStream m1) (toStream m2))
323 (rstState st) stp sng yld
325instance Semigroup (WSerialT m a) where
326 (<>) = wSerial
328infixr 5 <=>
330-- | Same as 'wSerial'.
331--
332-- @since 0.1.0
333{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
334{-# INLINE (<=>) #-}
335(<=>) :: IsStream t => t m a -> t m a -> t m a
336(<=>) = wSerial
338------------------------------------------------------------------------------
339-- Monoid
340------------------------------------------------------------------------------
342instance Monoid (WSerialT m a) where
343 mempty = K.nil
344 mappend = (<>)
346------------------------------------------------------------------------------
347-- Monad
348------------------------------------------------------------------------------
350instance Monad m => Monad (WSerialT m) where
351 return = pure
352 (WSerialT (Stream m)) >>= f = WSerialT $ Stream $ \st stp sng yld ->
353 let run x = unStream x (rstState st) stp sng yld
354 single a = run $ toStream (f a)
355 yieldk a r = run $ toStream $ f a <> (fromStream r >>= f)
356 in m (rstState st) stp single yieldk
358------------------------------------------------------------------------------
359-- Other instances
360------------------------------------------------------------------------------
362MONAD_APPLICATIVE_INSTANCE(WSerialT,)
363MONAD_COMMON_INSTANCES(WSerialT,)
365LIST_INSTANCES(SerialT)
366LIST_INSTANCES(WSerialT)
368FOLDABLE_INSTANCE(SerialT)
369FOLDABLE_INSTANCE(WSerialT)
371TRAVERSABLE_INSTANCE(SerialT)
372TRAVERSABLE_INSTANCE(WSerialT)