git.iko.soy/streamly
on 2ea9fb0 help sign in
streamly/src/Streamly/Internal/Data/Stream/Serial.hs
265 lines 8.3 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.Serial
5-- Copyright : (c) 2017 Composewell Technologies
6--
7-- License : BSD3
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
10-- Portability : GHC
11--
12-- To run examples in this module:
13--
14-- >>> import qualified Streamly.Data.Stream as Stream
15-- >>> import qualified Streamly.Prelude as IsStream
16--
17module Streamly.Internal.Data.Stream.Serial
18 (
19 -- * Serial appending stream
20 SerialT
21 , Serial
22 , serial
24 -- * Serial interleaving stream
25 , WSerialT(..)
26 , WSerial
27 , wSerial
28 , wSerialFst
29 , wSerialMin
30 , consMWSerial
32 -- * Construction
33 , Stream.cons
34 , Stream.consM
35 , Stream.repeat
36 , Stream.unfoldrM
37 , fromList
39 -- * Elimination
40 , toList
42 -- * Transformation
43 , map
44 , Stream.mapM
45 )
46where
48import Control.Applicative (liftA2)
49import Control.DeepSeq (NFData(..), NFData1(..))
50import Control.Monad.Base (MonadBase(..), liftBaseDefault)
51import Control.Monad.Catch (MonadThrow, throwM)
52import Control.Monad.IO.Class (MonadIO(..))
53import Control.Monad.Reader.Class (MonadReader(..))
54import Control.Monad.State.Class (MonadState(..))
55import Control.Monad.Trans.Class (MonadTrans(lift))
56import Data.Foldable (Foldable(foldl'), fold)
57import Data.Functor.Identity (Identity(..), runIdentity)
58import Data.Maybe (fromMaybe)
59import Data.Semigroup (Endo(..))
60import GHC.Exts (IsList(..), IsString(..), oneShot)
61import Text.Read
62 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
63 , readListPrecDefault)
64import Streamly.Internal.BaseCompat ((#.))
65import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
66import Streamly.Data.Stream (Stream)
68import qualified Streamly.Data.Stream as Stream
69import qualified Streamly.Internal.Data.Stream.Common as P
70import qualified Streamly.Internal.Data.Stream.StreamD as D
71 (fromStreamK, toStreamK, mapM)
72import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
73 (Stream, cons, consM, nil, concatMapWith, fromPure, bindWith
74 , withLocal, interleave, interleaveFst, interleaveMin)
76import Prelude hiding (map, mapM, repeat, filter)
78#include "Instances.hs"
79#include "inline.hs"
81-- $setup
82-- >>> import qualified Streamly.Data.Stream as Stream
83-- >>> import qualified Streamly.Prelude as IsStream
85------------------------------------------------------------------------------
86-- SerialT
87------------------------------------------------------------------------------
89type SerialT = Stream
91-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
92-- for more details.
93--
94-- /Since: 0.2.0 ("Streamly")/
95--
96-- @since 0.8.0
97type Serial = SerialT IO
99------------------------------------------------------------------------------
100-- Combining
101------------------------------------------------------------------------------
103{-# INLINE serial #-}
104serial :: SerialT m a -> SerialT m a -> SerialT m a
105serial = (<>)
107-- |
108-- @
109-- map = fmap
110-- @
111--
112-- Same as 'fmap'.
113--
114-- @
115-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
116-- [2,3,4]
117-- @
118--
119-- @since 0.4.0
120{-# INLINE map #-}
121map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b
122map f = Stream.mapM (return . f)
124------------------------------------------------------------------------------
125-- WSerialT
126------------------------------------------------------------------------------
128-- | For 'WSerialT' streams:
129--
130-- @
131-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
132-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
133-- @
134--
135-- Note that '<>' is associative only if we disregard the ordering of elements
136-- in the resulting stream.
137--
138-- A single 'Monad' bind behaves like a @for@ loop:
139--
140-- >>> :{
141-- IsStream.toList $ IsStream.fromWSerial $ do
142-- x <- IsStream.fromList [1,2] -- foreach x in stream
143-- return x
144-- :}
145-- [1,2]
146--
147-- Nested monad binds behave like interleaved nested @for@ loops:
148--
149-- >>> :{
150-- IsStream.toList $ IsStream.fromWSerial $ do
151-- x <- IsStream.fromList [1,2] -- foreach x in stream
152-- y <- IsStream.fromList [3,4] -- foreach y in stream
153-- return (x, y)
154-- :}
155-- [(1,3),(2,3),(1,4),(2,4)]
156--
157-- It is a result of interleaving all the nested iterations corresponding to
158-- element @1@ in the first stream with all the nested iterations of element
159-- @2@:
160--
161-- >>> import Streamly.Prelude (wSerial)
162-- >>> IsStream.toList $ IsStream.fromList [(1,3),(1,4)] `IsStream.wSerial` IsStream.fromList [(2,3),(2,4)]
163-- [(1,3),(2,3),(1,4),(2,4)]
164--
165-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
166-- contrast to the depth wise scheduling behavior of 'Stream'.
167--
168-- /Since: 0.2.0 ("Streamly")/
169--
170-- @since 0.8.0
171newtype WSerialT m a = WSerialT {getWSerialT :: K.Stream m a}
172 deriving (MonadTrans)
174-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
175-- documentation for more details.
176--
177-- /Since: 0.2.0 ("Streamly")/
178--
179-- @since 0.8.0
180type WSerial = WSerialT IO
182{-# INLINE consMWSerial #-}
183{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
184consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
185consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
187------------------------------------------------------------------------------
188-- Semigroup
189------------------------------------------------------------------------------
191infixr 6 `wSerial`
193-- | Interleaves two streams, yielding one element from each stream
194-- alternately. When one stream stops the rest of the other stream is used in
195-- the output stream.
196--
197-- This gives exponential priority to earlier streams than the ones joining
198-- later. Because of exponential weighting it can be used with 'concatMapWith'.
199--
200-- /Not fused/
202-- NOTE:
203--
204-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
205-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
206-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
207-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
208-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
209-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
210-- priority as the rest of the streams taken together. The same is true for
211-- each subexpression on the right.
212--
213{-# INLINE wSerial #-}
214wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
215wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleave m1 m2
217{-# INLINE wSerialFst #-}
218wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
219wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveFst m1 m2
221{-# INLINE wSerialMin #-}
222wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
223wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveMin m1 m2
225instance Semigroup (WSerialT m a) where
226 (<>) = wSerial
228------------------------------------------------------------------------------
229-- Monoid
230------------------------------------------------------------------------------
232instance Monoid (WSerialT m a) where
233 mempty = WSerialT K.nil
234 mappend = (<>)
236{-# INLINE apWSerial #-}
237apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
238apWSerial (WSerialT m1) (WSerialT m2) =
239 let f x1 = K.concatMapWith K.interleave (pure . x1) m2
240 in WSerialT $ K.concatMapWith K.interleave f m1
242instance Monad m => Applicative (WSerialT m) where
243 {-# INLINE pure #-}
244 pure = WSerialT . K.fromPure
245 {-# INLINE (<*>) #-}
246 (<*>) = apWSerial
248------------------------------------------------------------------------------
249-- Monad
250------------------------------------------------------------------------------
252instance Monad m => Monad (WSerialT m) where
253 return = pure
254 {-# INLINE (>>=) #-}
255 (>>=) (WSerialT m) f = WSerialT $ K.bindWith K.interleave m (getWSerialT . f)
257------------------------------------------------------------------------------
258-- Other instances
259------------------------------------------------------------------------------
261MONAD_COMMON_INSTANCES(WSerialT,)
262LIST_INSTANCES(WSerialT)
263NFDATA1_INSTANCE(WSerialT)
264FOLDABLE_INSTANCE(WSerialT)
265TRAVERSABLE_INSTANCE(WSerialT)