@@ -1,3 +1,5 @@
1+{-# LANGUAGE UndecidableInstances #-}
2-- Module : Streamly.Internal.Data.Stream.Serial
4-- Module : Streamly.Internal.Data.Stream.Serial
3-- Copyright : (c) 2017 Composewell Technologies
5-- Copyright : (c) 2017 Composewell Technologies
@@ -10,6 +12,7 @@
10-- To run examples in this module:
12-- To run examples in this module:
12-- >>> import qualified Streamly.Data.Stream as Stream
14-- >>> import qualified Streamly.Data.Stream as Stream
15+-- >>> import qualified Streamly.Prelude as IsStream
14module Streamly.Internal.Data.Stream.Serial
17module Streamly.Internal.Data.Stream.Serial
@@ -18,6 +21,14 @@ module Streamly.Internal.Data.Stream.Serial
24+ -- * Serial interleaving stream
21 -- * Construction
32 -- * Construction
22 , Stream.cons
33 , Stream.cons
23 , Stream.consM
34 , Stream.consM
@@ -34,15 +45,42 @@ module Streamly.Internal.Data.Stream.Serial
48+import Control.Applicative (liftA2)
49+import Control.DeepSeq (NFData(..), NFData1(..))
50+import Control.Monad.Base (MonadBase(..), liftBaseDefault)
51+import Control.Monad.Catch (MonadThrow, throwM)
52+import Control.Monad.IO.Class (MonadIO(..))
53+import Control.Monad.Reader.Class (MonadReader(..))
54+import Control.Monad.State.Class (MonadState(..))
55+import Control.Monad.Trans.Class (MonadTrans(lift))
56+import Data.Foldable (Foldable(foldl'), fold)
57+import Data.Functor.Identity (Identity(..), runIdentity)
58+import Data.Maybe (fromMaybe)
59+import Data.Semigroup (Endo(..))
37−import GHC.Exts (IsList(..))
60+import GHC.Exts (IsList(..), IsString(..), oneShot)
62+ ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
63+ , readListPrecDefault)
64+import Streamly.Internal.BaseCompat ((#.))
65+import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
38import Streamly.Data.Stream (Stream)
66import Streamly.Data.Stream (Stream)
40import qualified Streamly.Data.Stream as Stream
68import qualified Streamly.Data.Stream as Stream
69+import qualified Streamly.Internal.Data.Stream.Common as P
70+import qualified Streamly.Internal.Data.Stream.StreamD as D
71+ (fromStreamK, toStreamK, mapM)
72+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
73+ (Stream, cons, consM, nil, concatMapWith, fromPure, bindWith
74+ , withLocal, interleave, interleaveFst, interleaveMin)
42−import Prelude hiding (map)
76+import Prelude hiding (map, mapM, repeat, filter)
78+#include "Instances.hs"
45-- >>> import qualified Streamly.Data.Stream as Stream
82-- >>> import qualified Streamly.Data.Stream as Stream
83+-- >>> import qualified Streamly.Prelude as IsStream
47------------------------------------------------------------------------------
85------------------------------------------------------------------------------
@@ -82,3 +120,146 @@ serial = (<>)
82{-# INLINE map #-}
120{-# INLINE map #-}
83map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b
121map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b
84map f = Stream.mapM (return . f)
122map f = Stream.mapM (return . f)
124+------------------------------------------------------------------------------
126+------------------------------------------------------------------------------
128+-- | For 'WSerialT' streams:
131+-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
132+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
135+-- Note that '<>' is associative only if we disregard the ordering of elements
136+-- in the resulting stream.
138+-- A single 'Monad' bind behaves like a @for@ loop:
141+-- IsStream.toList $ IsStream.fromWSerial $ do
142+-- x <- IsStream.fromList [1,2] -- foreach x in stream
147+-- Nested monad binds behave like interleaved nested @for@ loops:
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
155+-- [(1,3),(2,3),(1,4),(2,4)]
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
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)]
165+-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
166+-- contrast to the depth wise scheduling behavior of 'Stream'.
168+-- /Since: 0.2.0 ("Streamly")/
171+newtype 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.
177+-- /Since: 0.2.0 ("Streamly")/
180+type WSerial = WSerialT IO
182+{-# INLINE consMWSerial #-}
183+{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
184+consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
185+consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
187+------------------------------------------------------------------------------
189+------------------------------------------------------------------------------
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.
197+-- This gives exponential priority to earlier streams than the ones joining
198+-- later. Because of exponential weighting it can be used with 'concatMapWith'.
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.
213+{-# INLINE wSerial #-}
214+wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
215+wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleave m1 m2
217+{-# INLINE wSerialFst #-}
218+wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
219+wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveFst m1 m2
221+{-# INLINE wSerialMin #-}
222+wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
223+wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveMin m1 m2
225+instance Semigroup (WSerialT m a) where
228+------------------------------------------------------------------------------
230+------------------------------------------------------------------------------
232+instance Monoid (WSerialT m a) where
233+ mempty = WSerialT K.nil
236+{-# INLINE apWSerial #-}
237+apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
238+apWSerial (WSerialT m1) (WSerialT m2) =
239+ let f x1 = K.concatMapWith K.interleave (pure . x1) m2
240+ in WSerialT $ K.concatMapWith K.interleave f m1
242+instance Monad m => Applicative (WSerialT m) where
244+ pure = WSerialT . K.fromPure
245+ {-# INLINE (<*>) #-}
248+------------------------------------------------------------------------------
250+------------------------------------------------------------------------------
252+instance Monad m => Monad (WSerialT m) where
254+ {-# INLINE (>>=) #-}
255+ (>>=) (WSerialT m) f = WSerialT $ K.bindWith K.interleave m (getWSerialT . f)
257+------------------------------------------------------------------------------
259+------------------------------------------------------------------------------
261+MONAD_COMMON_INSTANCES(WSerialT,)
262+LIST_INSTANCES(WSerialT)
263+NFDATA1_INSTANCE(WSerialT)
264+FOLDABLE_INSTANCE(WSerialT)
265+TRAVERSABLE_INSTANCE(WSerialT)