@@ -0,0 +1,249 @@
1+{-# LANGUAGE UndecidableInstances #-}
2+
3+-- |
4+-- Module : Streamly.Internal.Data.Stream.WSerial
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.Prelude as Stream
15+--
16+module Streamly.Internal.Data.Stream.WSerial
17+ (
18+ -- * Serial interleaving stream
19+ WSerialT(..)
20+ , WSerial
21+ , wSerialK
22+ , wSerial
23+ , wSerialFst
24+ , wSerialMin
25+ , consMWSerial
26+ )
27+where
28+
29+import Control.Applicative (liftA2)
30+import Control.DeepSeq (NFData(..))
31+import Control.DeepSeq (NFData1(..))
32+import Control.Monad.Base (MonadBase(..), liftBaseDefault)
33+import Control.Monad.Catch (MonadThrow, throwM)
34+import Control.Monad.IO.Class (MonadIO(..))
35+import Control.Monad.Reader.Class (MonadReader(..))
36+import Control.Monad.State.Class (MonadState(..))
37+import Control.Monad.Trans.Class (MonadTrans(lift))
38+import Data.Foldable (Foldable(foldl'), fold)
39+import Data.Functor.Identity (Identity(..), runIdentity)
40+import Data.Maybe (fromMaybe)
41+import Data.Semigroup (Endo(..))
42+#if __GLASGOW_HASKELL__ < 808
43+import Data.Semigroup (Semigroup(..))
44+#endif
45+import GHC.Exts (IsList(..), IsString(..), oneShot)
46+import Text.Read
47+ ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
48+ , readListPrecDefault)
49+import Streamly.Internal.BaseCompat ((#.))
50+import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
51+import Streamly.Internal.Data.Stream.StreamK.Type
52+ (Stream, mkStream, foldStream)
53+
54+import qualified Streamly.Internal.Data.Stream.Common as P
55+import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
56+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
57+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
58+
59+import Prelude hiding (map, mapM, repeat, filter)
60+
61+#include "Instances.hs"
62+#include "inline.hs"
63+
64+-- $setup
65+-- >>> import qualified Streamly.Prelude as Stream
66+
67+------------------------------------------------------------------------------
68+-- WSerialT
69+------------------------------------------------------------------------------
70+
71+-- | For 'WSerialT' streams:
72+--
73+-- @
74+-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
75+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
76+-- @
77+--
78+-- Note that '<>' is associative only if we disregard the ordering of elements
79+-- in the resulting stream.
80+--
81+-- A single 'Monad' bind behaves like a @for@ loop:
82+--
83+-- >>> :{
84+-- Stream.toList $ Stream.fromWSerial $ do
85+-- x <- Stream.fromList [1,2] -- foreach x in stream
86+-- return x
87+-- :}
88+-- [1,2]
89+--
90+-- Nested monad binds behave like interleaved nested @for@ loops:
91+--
92+-- >>> :{
93+-- Stream.toList $ Stream.fromWSerial $ do
94+-- x <- Stream.fromList [1,2] -- foreach x in stream
95+-- y <- Stream.fromList [3,4] -- foreach y in stream
96+-- return (x, y)
97+-- :}
98+-- [(1,3),(2,3),(1,4),(2,4)]
99+--
100+-- It is a result of interleaving all the nested iterations corresponding to
101+-- element @1@ in the first stream with all the nested iterations of element
102+-- @2@:
103+--
104+-- >>> import Streamly.Prelude (wSerial)
105+-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `Stream.wSerial` Stream.fromList [(2,3),(2,4)]
106+-- [(1,3),(2,3),(1,4),(2,4)]
107+--
108+-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
109+-- contrast to the depth wise scheduling behavior of 'SerialT'.
110+--
111+-- /Since: 0.2.0 ("Streamly")/
112+--
113+-- @since 0.8.0
114+newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
115+ deriving (MonadTrans)
116+
117+-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
118+-- documentation for more details.
119+--
120+-- /Since: 0.2.0 ("Streamly")/
121+--
122+-- @since 0.8.0
123+type WSerial = WSerialT IO
124+
125+{-# INLINE consMWSerial #-}
126+{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
127+consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
128+consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
129+
130+------------------------------------------------------------------------------
131+-- Semigroup
132+------------------------------------------------------------------------------
133+
134+infixr 6 `wSerial`
135+
136+-- Additionally we can have m elements yield from the first stream and n
137+-- elements yielding from the second stream. We can also have time slicing
138+-- variants of positional interleaving, e.g. run first stream for m seconds and
139+-- run the second stream for n seconds.
140+--
141+-- Similar combinators can be implemented using WAhead style.
142+
143+{-# INLINE wSerialK #-}
144+wSerialK :: Stream m a -> Stream m a -> Stream m a
145+wSerialK m1 m2 = mkStream $ \st yld sng stp -> do
146+ let stop = foldStream st yld sng stp m2
147+ single a = yld a m2
148+ yieldk a r = yld a (wSerialK m2 r)
149+ foldStream st yieldk single stop m1
150+
151+-- | Interleaves two streams, yielding one element from each stream
152+-- alternately. When one stream stops the rest of the other stream is used in
153+-- the output stream.
154+
155+-- Scheduling Notes:
156+--
157+-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
158+-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
159+-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
160+-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
161+-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
162+-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
163+-- priority as the rest of the streams taken together. The same is true for
164+-- each subexpression on the right.
165+--
166+{-# INLINE wSerial #-}
167+wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
168+wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialK m1 m2
169+
170+-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
171+--
172+-- @since 0.7.0
173+{-# INLINE wSerialFstK #-}
174+wSerialFstK :: Stream m a -> Stream m a -> Stream m a
175+wSerialFstK m1 m2 = mkStream $ \st yld sng stp -> do
176+ let yieldFirst a r = yld a (yieldSecond r m2)
177+ in foldStream st yieldFirst sng stp m1
178+
179+ where
180+
181+ yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
182+ let stop = foldStream st yld sng stp s1
183+ single a = yld a s1
184+ yieldk a r = yld a (wSerialK s1 r)
185+ in foldStream st yieldk single stop s2
186+
187+{-# INLINE wSerialFst #-}
188+wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
189+wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialFstK m1 m2
190+
191+-- | Like `wSerial` but stops interleaving as soon as any of the two streams
192+-- stops.
193+--
194+-- @since 0.7.0
195+{-# INLINE wSerialMinK #-}
196+wSerialMinK :: Stream m a -> Stream m a -> Stream m a
197+wSerialMinK m1 m2 = mkStream $ \st yld _ stp -> do
198+ let stop = stp
199+ -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
200+ -- "sng a" to keep the behaviour consistent with the yield continuation.
201+ single a = yld a (wSerialMinK m2 K.nil)
202+ yieldk a r = yld a (wSerialMinK m2 r)
203+ foldStream st yieldk single stop m1
204+
205+{-# INLINE wSerialMin #-}
206+wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
207+wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialMinK m1 m2
208+
209+instance Semigroup (WSerialT m a) where
210+ (<>) = wSerial
211+
212+------------------------------------------------------------------------------
213+-- Monoid
214+------------------------------------------------------------------------------
215+
216+instance Monoid (WSerialT m a) where
217+ mempty = WSerialT K.nil
218+ mappend = (<>)
219+
220+{-# INLINE apWSerial #-}
221+apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
222+apWSerial (WSerialT m1) (WSerialT m2) =
223+ let f x1 = K.concatMapWith wSerialK (pure . x1) m2
224+ in WSerialT $ K.concatMapWith wSerialK f m1
225+
226+instance Monad m => Applicative (WSerialT m) where
227+ {-# INLINE pure #-}
228+ pure = WSerialT . K.fromPure
229+ {-# INLINE (<*>) #-}
230+ (<*>) = apWSerial
231+
232+------------------------------------------------------------------------------
233+-- Monad
234+------------------------------------------------------------------------------
235+
236+instance Monad m => Monad (WSerialT m) where
237+ return = pure
238+ {-# INLINE (>>=) #-}
239+ (>>=) (WSerialT m) f = WSerialT $ K.bindWith wSerialK m (getWSerialT . f)
240+
241+------------------------------------------------------------------------------
242+-- Other instances
243+------------------------------------------------------------------------------
244+
245+MONAD_COMMON_INSTANCES(WSerialT,)
246+LIST_INSTANCES(WSerialT)
247+NFDATA1_INSTANCE(WSerialT)
248+FOLDABLE_INSTANCE(WSerialT)
249+TRAVERSABLE_INSTANCE(WSerialT)