git.iko.soy/streamly
on master help sign in
split Core.hs into CPS stream and concurrent
9ad5b20Harendra Kumar
src/Streamly/Core.hs → src/Streamly/CPS/Concurrent.hsrenamed+7−154
compare with master
unifiedsplit
@@ -10,7 +10,7 @@
10{-# LANGUAGE UndecidableInstances #-} -- XXX
10{-# LANGUAGE UndecidableInstances #-} -- XXX
11
11
12-- |
12-- |
13−-- Module : Streamly.Core
13+-- Module : Streamly.CPS.Concurrent
14-- Copyright : (c) 2017 Harendra Kumar
14-- Copyright : (c) 2017 Harendra Kumar
15--
15--
16-- License : BSD3
16-- License : BSD3
@@ -19,28 +19,15 @@
19-- Portability : GHC
19-- Portability : GHC
20--
20--
21--
21--
22−module Streamly.Core
22+module Streamly.CPS.Concurrent
23 (
23 (
24− -- * Streams
25− Stream (..)
26−
27− -- * Construction (pure)
28− , nil
29− , cons
30− , singleton
31− , once
32− , repeat
33−
34 -- * Construction (monadic)
24 -- * Construction (monadic)
35− , consM
36− , consMAhead
25+ consMAhead
37 , consMAsync
26 , consMAsync
38 , consMWAsync
27 , consMWAsync
39 , consMParallel
28 , consMParallel
40
29
41 -- * Semigroup Style Composition
30 -- * Semigroup Style Composition
42− , serial
43− , wSerial
44 , ahead
31 , ahead
45 , async
32 , async
46 , wAsync
33 , wAsync
@@ -51,7 +38,6 @@ module Streamly.Core
51 , runWith
38 , runWith
52
39
53 -- * zip
40 -- * zip
54− , zipWith
55 , zipAsyncWith
41 , zipAsyncWith
56
42
57 -- * Concurrent Stream Vars (SVars)
43 -- * Concurrent Stream Vars (SVars)
@@ -61,10 +47,10 @@ module Streamly.Core
61 )
47 )
62where
48where
63
49
50+--
64import Control.Concurrent.MVar (newEmptyMVar)
51import Control.Concurrent.MVar (newEmptyMVar)
65import Control.Monad.Catch (throwM)
52import Control.Monad.Catch (throwM)
66import Control.Monad.IO.Class (MonadIO(..))
53import Control.Monad.IO.Class (MonadIO(..))
67−import Control.Monad.Trans.Class (MonadTrans(lift))
68import Data.Atomics (atomicModifyIORefCAS_)
54import Data.Atomics (atomicModifyIORefCAS_)
69import Data.Concurrent.Queue.MichaelScott
55import Data.Concurrent.Queue.MichaelScott
70 (LinkedQueue, newQ, nullQ)
56 (LinkedQueue, newQ, nullQ)
@@ -72,14 +58,15 @@ import Data.Functor (void)
72import Data.Heap (Heap, Entry(..))
58import Data.Heap (Heap, Entry(..))
73import Data.IORef (IORef, newIORef, readIORef)
59import Data.IORef (IORef, newIORef, readIORef)
74import Data.Maybe (fromJust)
60import Data.Maybe (fromJust)
75−import Data.Semigroup (Semigroup(..))
76import GHC.IO (IO(..))
61import GHC.IO (IO(..))
77import Prelude hiding (repeat, zipWith)
62import Prelude hiding (repeat, zipWith)
78
63
79−import Streamly.SVar
80import qualified Data.Set as S
64import qualified Data.Set as S
81import qualified Data.Heap as H
65import qualified Data.Heap as H
82
66
67+import Streamly.SVar
68+import Streamly.CPS.Stream
69+
83-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
70-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
84-- can keep it on in production to debug problems quickly if and when they
71-- can keep it on in production to debug problems quickly if and when they
85-- happen, but it may result in unexpected output when threads are left hanging
72-- happen, but it may result in unexpected output when threads are left hanging
@@ -90,97 +77,6 @@ import Control.Monad (when)
90import Data.IORef (writeIORef)
77import Data.IORef (writeIORef)
91#endif
78#endif
92
79
93−
94−------------------------------------------------------------------------------
95−-- The stream type
96−------------------------------------------------------------------------------
97−
98−-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
99−-- constructed using actions in monad 'm'. It uses stop, singleton and yield
100−-- continuations equivalent to the following direct style type:
101−--
102−-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
103−--
104−-- To facilitate parallel composition we maintain a local state in an SVar that
105−-- is shared across and is used for synchronization of the streams being
106−-- composed.
107−--
108−-- The singleton case can be expressed in terms of stop and yield but we have
109−-- it as a separate case to optimize composition operations for streams with
110−-- single element. We build singleton streams in the implementation of 'pure'
111−-- for Applicative and Monad, and in 'lift' for MonadTrans.
112−--
113−newtype Stream m a =
114− Stream {
115− runStream :: forall r.
116− Maybe (SVar Stream m a) -- local state
117− -> m r -- stop
118− -> (a -> m r) -- singleton
119− -> (a -> Stream m a -> m r) -- yield
120− -> m r
121− }
122−
123−nil :: Stream m a
124−nil = Stream $ \_ stp _ _ -> stp
125−
126−-- | faster than consM because there is no bind.
127−cons :: a -> Stream m a -> Stream m a
128−cons a r = Stream $ \_ _ _ yld -> yld a r
129−
130−-- | Same as @once . return@ but may be faster because there is no bind
131−singleton :: a -> Stream m a
132−singleton a = Stream $ \_ _ single _ -> single a
133−
134−{-# INLINE once #-}
135−once :: Monad m => m a -> Stream m a
136−once m = Stream $ \_ _ single _ -> m >>= single
137−
138−{-# INLINE consM #-}
139−consM :: Monad m => m a -> Stream m a -> Stream m a
140−consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
141−
142−repeat :: a -> Stream m a
143−repeat a = let x = cons a x in x
144−
145−------------------------------------------------------------------------------
146−-- Semigroup
147−------------------------------------------------------------------------------
148−
149−-- | Concatenates two streams sequentially i.e. the first stream is
150−-- exhausted completely before yielding any element from the second stream.
151−{-# INLINE serial #-}
152−serial :: Stream m a -> Stream m a -> Stream m a
153−serial m1 m2 = go m1
154− where
155− go (Stream m) = Stream $ \_ stp sng yld ->
156− let stop = (runStream m2) Nothing stp sng yld
157− single a = yld a m2
158− yield a r = yld a (go r)
159− in m Nothing stop single yield
160−
161−instance Semigroup (Stream m a) where
162− (<>) = serial
163−
164−------------------------------------------------------------------------------
165−-- Monoid
166−------------------------------------------------------------------------------
167−
168−instance Monoid (Stream m a) where
169− mempty = nil
170− mappend = (<>)
171−
172−------------------------------------------------------------------------------
173−-- Interleave
174−------------------------------------------------------------------------------
175−
176−{-# INLINE wSerial #-}
177−wSerial :: Stream m a -> Stream m a -> Stream m a
178−wSerial m1 m2 = Stream $ \_ stp sng yld -> do
179− let stop = (runStream m2) Nothing stp sng yld
180− single a = yld a m2
181− yield a r = yld a (wSerial m2 r)
182− (runStream m1) Nothing stop single yield
183−
184-------------------------------------------------------------------------------
80-------------------------------------------------------------------------------
185-- Async
81-- Async
186-------------------------------------------------------------------------------
82-------------------------------------------------------------------------------
@@ -599,25 +495,6 @@ parallel = joinStreamVarPar ParallelVar
599consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
495consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
600consMParallel m r = once m `parallel` r
496consMParallel m r = once m `parallel` r
601
497
602−-------------------------------------------------------------------------------
603−-- Functor instace is the same for all types
604−-------------------------------------------------------------------------------
605−
606−instance Monad m => Functor (Stream m) where
607− fmap f m = Stream $ \_ stp sng yld ->
608− let single = sng . f
609− yield a r = yld (f a) (fmap f r)
610− in (runStream m) Nothing stp single yield
611−
612−------------------------------------------------------------------------------
613−-- Alternative & MonadPlus
614−------------------------------------------------------------------------------
615−
616−_alt :: Stream m a -> Stream m a -> Stream m a
617−_alt m1 m2 = Stream $ \_ stp sng yld ->
618− let stop = runStream m2 Nothing stp sng yld
619− in runStream m1 Nothing stop sng yld
620−
621------------------------------------------------------------------------------
498------------------------------------------------------------------------------
622-- Stream to stream function application
499-- Stream to stream function application
623------------------------------------------------------------------------------
500------------------------------------------------------------------------------
@@ -645,23 +522,6 @@ runWith f m = do
645 pushWorkerPar sv (runOne sv m)
522 pushWorkerPar sv (runOne sv m)
646 f $ fromStreamVar sv
523 f $ fromStreamVar sv
647
524
648−------------------------------------------------------------------------------
649−-- Zipping
650−------------------------------------------------------------------------------
651−
652−{-# INLINE zipWith #-}
653−zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
654−zipWith f m1 m2 = go m1 m2
655− where
656− go mx my = Stream $ \_ stp sng yld -> do
657− let merge a ra =
658− let single2 b = sng (f a b)
659− yield2 b rb = yld (f a b) (go ra rb)
660− in (runStream my) Nothing stp single2 yield2
661− let single1 a = merge a nil
662− yield1 a ra = merge a ra
663− (runStream mx) Nothing stp single1 yield1
664−
665{-# INLINE zipAsyncWith #-}
525{-# INLINE zipAsyncWith #-}
666zipAsyncWith :: MonadAsync m
526zipAsyncWith :: MonadAsync m
667 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
527 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
@@ -676,13 +536,6 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
676 mkAsync m = newAsyncVar m
536 mkAsync m = newAsyncVar m
677 >>= return . fromStreamVar
537 >>= return . fromStreamVar
678
538
679−-------------------------------------------------------------------------------
680−-- Transformers
681−-------------------------------------------------------------------------------
682−
683−instance MonadTrans Stream where
684− lift = once
685−
686-------------------------------------------------------------------------------
539-------------------------------------------------------------------------------
687-- SVar creation
540-- SVar creation
688-- This code belongs in SVar.hs but is kept here for perf reasons
541-- This code belongs in SVar.hs but is kept here for perf reasons
src/Streamly/CPS/Stream.hsnew file+183
@@ -0,0 +1,183 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE ConstraintKinds #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE FlexibleInstances #-}
5+{-# LANGUAGE LambdaCase #-}
6+{-# LANGUAGE MagicHash #-}
7+{-# LANGUAGE MultiParamTypeClasses #-}
8+{-# LANGUAGE RankNTypes #-}
9+{-# LANGUAGE UnboxedTuples #-}
10+{-# LANGUAGE UndecidableInstances #-} -- XXX
11+
12+-- |
13+-- Module : Streamly.CPS.Stream
14+-- Copyright : (c) 2017 Harendra Kumar
15+--
16+-- License : BSD3
17+-- Maintainer : harendra.kumar@gmail.com
18+-- Stability : experimental
19+-- Portability : GHC
20+--
21+--
22+module Streamly.CPS.Stream
23+ (
24+ -- * Streams
25+ Stream (..)
26+
27+ -- * Construction (pure)
28+ , nil
29+ , cons
30+ , singleton
31+ , once
32+ , repeat
33+
34+ -- * Construction (monadic)
35+ , consM
36+
37+ -- * Semigroup Style Composition
38+ , serial
39+ , wSerial
40+
41+ -- * zip
42+ , zipWith
43+ )
44+where
45+
46+import Control.Monad.Trans.Class (MonadTrans(lift))
47+import Data.Semigroup (Semigroup(..))
48+import Prelude hiding (repeat, zipWith)
49+
50+import Streamly.SVar
51+
52+------------------------------------------------------------------------------
53+-- The stream type
54+------------------------------------------------------------------------------
55+
56+-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
57+-- constructed using actions in monad 'm'. It uses stop, singleton and yield
58+-- continuations equivalent to the following direct style type:
59+--
60+-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
61+--
62+-- To facilitate parallel composition we maintain a local state in an SVar that
63+-- is shared across and is used for synchronization of the streams being
64+-- composed.
65+--
66+-- The singleton case can be expressed in terms of stop and yield but we have
67+-- it as a separate case to optimize composition operations for streams with
68+-- single element. We build singleton streams in the implementation of 'pure'
69+-- for Applicative and Monad, and in 'lift' for MonadTrans.
70+--
71+newtype Stream m a =
72+ Stream {
73+ runStream :: forall r.
74+ Maybe (SVar Stream m a) -- local state
75+ -> m r -- stop
76+ -> (a -> m r) -- singleton
77+ -> (a -> Stream m a -> m r) -- yield
78+ -> m r
79+ }
80+
81+nil :: Stream m a
82+nil = Stream $ \_ stp _ _ -> stp
83+
84+-- | faster than consM because there is no bind.
85+cons :: a -> Stream m a -> Stream m a
86+cons a r = Stream $ \_ _ _ yld -> yld a r
87+
88+-- | Same as @once . return@ but may be faster because there is no bind
89+singleton :: a -> Stream m a
90+singleton a = Stream $ \_ _ single _ -> single a
91+
92+{-# INLINE once #-}
93+once :: Monad m => m a -> Stream m a
94+once m = Stream $ \_ _ single _ -> m >>= single
95+
96+{-# INLINE consM #-}
97+consM :: Monad m => m a -> Stream m a -> Stream m a
98+consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
99+
100+repeat :: a -> Stream m a
101+repeat a = let x = cons a x in x
102+
103+------------------------------------------------------------------------------
104+-- Semigroup
105+------------------------------------------------------------------------------
106+
107+-- | Concatenates two streams sequentially i.e. the first stream is
108+-- exhausted completely before yielding any element from the second stream.
109+{-# INLINE serial #-}
110+serial :: Stream m a -> Stream m a -> Stream m a
111+serial m1 m2 = go m1
112+ where
113+ go (Stream m) = Stream $ \_ stp sng yld ->
114+ let stop = (runStream m2) Nothing stp sng yld
115+ single a = yld a m2
116+ yield a r = yld a (go r)
117+ in m Nothing stop single yield
118+
119+instance Semigroup (Stream m a) where
120+ (<>) = serial
121+
122+------------------------------------------------------------------------------
123+-- Monoid
124+------------------------------------------------------------------------------
125+
126+instance Monoid (Stream m a) where
127+ mempty = nil
128+ mappend = (<>)
129+
130+------------------------------------------------------------------------------
131+-- Interleave
132+------------------------------------------------------------------------------
133+
134+{-# INLINE wSerial #-}
135+wSerial :: Stream m a -> Stream m a -> Stream m a
136+wSerial m1 m2 = Stream $ \_ stp sng yld -> do
137+ let stop = (runStream m2) Nothing stp sng yld
138+ single a = yld a m2
139+ yield a r = yld a (wSerial m2 r)
140+ (runStream m1) Nothing stop single yield
141+
142+-------------------------------------------------------------------------------
143+-- Functor
144+-------------------------------------------------------------------------------
145+
146+instance Monad m => Functor (Stream m) where
147+ fmap f m = Stream $ \_ stp sng yld ->
148+ let single = sng . f
149+ yield a r = yld (f a) (fmap f r)
150+ in (runStream m) Nothing stp single yield
151+
152+------------------------------------------------------------------------------
153+-- Alternative & MonadPlus
154+------------------------------------------------------------------------------
155+
156+_alt :: Stream m a -> Stream m a -> Stream m a
157+_alt m1 m2 = Stream $ \_ stp sng yld ->
158+ let stop = runStream m2 Nothing stp sng yld
159+ in runStream m1 Nothing stop sng yld
160+
161+------------------------------------------------------------------------------
162+-- Zipping
163+------------------------------------------------------------------------------
164+
165+{-# INLINE zipWith #-}
166+zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
167+zipWith f m1 m2 = go m1 m2
168+ where
169+ go mx my = Stream $ \_ stp sng yld -> do
170+ let merge a ra =
171+ let single2 b = sng (f a b)
172+ yield2 b rb = yld (f a b) (go ra rb)
173+ in (runStream my) Nothing stp single2 yield2
174+ let single1 a = merge a nil
175+ yield1 a ra = merge a ra
176+ (runStream mx) Nothing stp single1 yield1
177+
178+-------------------------------------------------------------------------------
179+-- Transformers
180+-------------------------------------------------------------------------------
181+
182+instance MonadTrans Stream where
183+ lift = once
src/Streamly/Prelude.hs+69−69
@@ -142,8 +142,8 @@ import Prelude hiding (filter, drop, dropWhile, take,
142import qualified Prelude
142import qualified Prelude
143import qualified System.IO as IO
143import qualified System.IO as IO
144
144
145−import qualified Streamly.Core as S
145+import qualified Streamly.CPS.Stream as C
146−import Streamly.Core (Stream(Stream))
146+import qualified Streamly.CPS.Concurrent as CC
147import Streamly.Streams
147import Streamly.Streams
148
148
149------------------------------------------------------------------------------
149------------------------------------------------------------------------------
@@ -171,7 +171,7 @@ import Streamly.Streams
171unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
171unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
172unfoldr step = fromStream . go
172unfoldr step = fromStream . go
173 where
173 where
174− go s = Stream $ \_ stp _ yld ->
174+ go s = C.Stream $ \_ stp _ yld ->
175 case step s of
175 case step s of
176 Nothing -> stp
176 Nothing -> stp
177 Just (a, b) -> yld a (go b)
177 Just (a, b) -> yld a (go b)
@@ -211,12 +211,12 @@ unfoldr step = fromStream . go
211unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
211unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
212unfoldrM step = go
212unfoldrM step = go
213 where
213 where
214− go s = fromStream $ Stream $ \svr stp sng yld -> do
214+ go s = fromStream $ C.Stream $ \svr stp sng yld -> do
215 mayb <- step s
215 mayb <- step s
216 case mayb of
216 case mayb of
217 Nothing -> stp
217 Nothing -> stp
218 Just (a, b) ->
218 Just (a, b) ->
219− S.runStream (toStream (return a |: go b)) svr stp sng yld
219+ C.runStream (toStream (return a |: go b)) svr stp sng yld
220
220
221-- | Construct a stream from a 'Foldable' containing pure values.
221-- | Construct a stream from a 'Foldable' containing pure values.
222--
222--
@@ -258,7 +258,7 @@ each = fromFoldable
258--
258--
259-- @since 0.2.0
259-- @since 0.2.0
260once :: (IsStream t, Monad m) => m a -> t m a
260once :: (IsStream t, Monad m) => m a -> t m a
261−once = fromStream . S.once
261+once = fromStream . C.once
262
262
263-- | Generate a stream by performing a monadic action @n@ times.
263-- | Generate a stream by performing a monadic action @n@ times.
264--
264--
@@ -296,7 +296,7 @@ repeatM = go
296iterate :: IsStream t => (a -> a) -> a -> t m a
296iterate :: IsStream t => (a -> a) -> a -> t m a
297iterate step = fromStream . go
297iterate step = fromStream . go
298 where
298 where
299− go s = S.cons s (go (step s))
299+ go s = C.cons s (go (step s))
300
300
301-- | Iterate a monadic function from a seed value, streaming the results
301-- | Iterate a monadic function from a seed value, streaming the results
302-- forever.
302-- forever.
@@ -320,9 +320,9 @@ iterate step = fromStream . go
320iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
320iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
321iterateM step = go
321iterateM step = go
322 where
322 where
323− go s = fromStream $ Stream $ \svr stp sng yld -> do
323+ go s = fromStream $ C.Stream $ \svr stp sng yld -> do
324 next <- step s
324 next <- step s
325− S.runStream (toStream (return s |: go next)) svr stp sng yld
325+ C.runStream (toStream (return s |: go next)) svr stp sng yld
326
326
327-- | Read lines from an IO Handle into a stream of Strings.
327-- | Read lines from an IO Handle into a stream of Strings.
328--
328--
@@ -330,7 +330,7 @@ iterateM step = go
330fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
330fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
331fromHandle h = fromStream go
331fromHandle h = fromStream go
332 where
332 where
333− go = Stream $ \_ stp _ yld -> do
333+ go = C.Stream $ \_ stp _ yld -> do
334 eof <- liftIO $ IO.hIsEOF h
334 eof <- liftIO $ IO.hIsEOF h
335 if eof
335 if eof
336 then stp
336 then stp
@@ -357,7 +357,7 @@ foldr step acc m = go (toStream m)
357 let stop = return acc
357 let stop = return acc
358 single a = return (step a acc)
358 single a = return (step a acc)
359 yield a r = go r >>= \b -> return (step a b)
359 yield a r = go r >>= \b -> return (step a b)
360− in (S.runStream m1) Nothing stop single yield
360+ in (C.runStream m1) Nothing stop single yield
361
361
362-- | Lazy right fold with a monadic step function. For example, to fold a
362-- | Lazy right fold with a monadic step function. For example, to fold a
363-- stream into a list:
363-- stream into a list:
@@ -376,7 +376,7 @@ foldrM step acc m = go (toStream m)
376 let stop = return acc
376 let stop = return acc
377 single a = step a acc
377 single a = step a acc
378 yield a r = go r >>= step a
378 yield a r = go r >>= step a
379− in (S.runStream m1) Nothing stop single yield
379+ in (C.runStream m1) Nothing stop single yield
380
380
381-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
381-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
382-- user supplied extraction function (the third argument) at each step. This is
382-- user supplied extraction function (the third argument) at each step. This is
@@ -389,12 +389,12 @@ scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
389scanx step begin done m =
389scanx step begin done m =
390 cons (done begin) $ fromStream $ go (toStream m) begin
390 cons (done begin) $ fromStream $ go (toStream m) begin
391 where
391 where
392− go m1 !acc = Stream $ \_ stp sng yld ->
392+ go m1 !acc = C.Stream $ \_ stp sng yld ->
393 let single a = sng (done $ step acc a)
393 let single a = sng (done $ step acc a)
394 yield a r =
394 yield a r =
395 let s = step acc a
395 let s = step acc a
396 in yld (done s) (go r s)
396 in yld (done s) (go r s)
397− in S.runStream m1 Nothing stp single yield
397+ in C.runStream m1 Nothing stp single yield
398
398
399-- |
399-- |
400-- @since 0.1.1
400-- @since 0.1.1
@@ -425,19 +425,19 @@ foldx step begin done m = get $ go (toStream m) begin
425 {-# NOINLINE get #-}
425 {-# NOINLINE get #-}
426 get m1 =
426 get m1 =
427 let single = return . done
427 let single = return . done
428− in (S.runStream m1) Nothing undefined single undefined
428+ in (C.runStream m1) Nothing undefined single undefined
429
429
430 -- Note, this can be implemented by making a recursive call to "go",
430 -- Note, this can be implemented by making a recursive call to "go",
431 -- however that is more expensive because of unnecessary recursion
431 -- however that is more expensive because of unnecessary recursion
432 -- that cannot be tail call optimized. Unfolding recursion explicitly via
432 -- that cannot be tail call optimized. Unfolding recursion explicitly via
433 -- continuations is much more efficient.
433 -- continuations is much more efficient.
434− go m1 !acc = Stream $ \_ _ sng yld ->
434+ go m1 !acc = C.Stream $ \_ _ sng yld ->
435 let stop = sng acc
435 let stop = sng acc
436 single a = sng $ step acc a
436 single a = sng $ step acc a
437 yield a r =
437 yield a r =
438 let stream = go r (step acc a)
438 let stream = go r (step acc a)
439− in (S.runStream stream) Nothing undefined sng yld
439+ in (C.runStream stream) Nothing undefined sng yld
440− in (S.runStream m1) Nothing stop single yield
440+ in (C.runStream m1) Nothing stop single yield
441
441
442-- |
442-- |
443-- @since 0.1.0
443-- @since 0.1.0
@@ -463,7 +463,7 @@ foldxM step begin done m = go begin (toStream m)
463 let stop = acc >>= done
463 let stop = acc >>= done
464 single a = acc >>= \b -> step b a >>= done
464 single a = acc >>= \b -> step b a >>= done
465 yield a r = acc >>= \b -> go (step b a) r
465 yield a r = acc >>= \b -> go (step b a) r
466− in (S.runStream m1) Nothing stop single yield
466+ in (C.runStream m1) Nothing stop single yield
467
467
468-- |
468-- |
469-- @since 0.1.0
469-- @since 0.1.0
@@ -487,7 +487,7 @@ uncons m =
487 let stop = return Nothing
487 let stop = return Nothing
488 single a = return (Just (a, nil))
488 single a = return (Just (a, nil))
489 yield a r = return (Just (a, fromStream r))
489 yield a r = return (Just (a, fromStream r))
490− in (S.runStream (toStream m)) Nothing stop single yield
490+ in (C.runStream (toStream m)) Nothing stop single yield
491
491
492-- | Write a stream of Strings to an IO Handle.
492-- | Write a stream of Strings to an IO Handle.
493--
493--
@@ -499,7 +499,7 @@ toHandle h m = go (toStream m)
499 let stop = return ()
499 let stop = return ()
500 single a = liftIO (IO.hPutStrLn h a)
500 single a = liftIO (IO.hPutStrLn h a)
501 yield a r = liftIO (IO.hPutStrLn h a) >> go r
501 yield a r = liftIO (IO.hPutStrLn h a) >> go r
502− in (S.runStream m1) Nothing stop single yield
502+ in (C.runStream m1) Nothing stop single yield
503
503
504------------------------------------------------------------------------------
504------------------------------------------------------------------------------
505-- Special folds
505-- Special folds
@@ -519,9 +519,9 @@ toList = foldrM (\a xs -> return (a : xs)) []
519take :: IsStream t => Int -> t m a -> t m a
519take :: IsStream t => Int -> t m a -> t m a
520take n m = fromStream $ go n (toStream m)
520take n m = fromStream $ go n (toStream m)
521 where
521 where
522− go n1 m1 = Stream $ \_ stp sng yld ->
522+ go n1 m1 = C.Stream $ \_ stp sng yld ->
523 let yield a r = yld a (go (n1 - 1) r)
523 let yield a r = yld a (go (n1 - 1) r)
524− in if n1 <= 0 then stp else (S.runStream m1) Nothing stp sng yield
524+ in if n1 <= 0 then stp else (C.runStream m1) Nothing stp sng yield
525
525
526-- | Include only those elements that pass a predicate.
526-- | Include only those elements that pass a predicate.
527--
527--
@@ -530,12 +530,12 @@ take n m = fromStream $ go n (toStream m)
530filter :: IsStream t => (a -> Bool) -> t m a -> t m a
530filter :: IsStream t => (a -> Bool) -> t m a -> t m a
531filter p m = fromStream $ go (toStream m)
531filter p m = fromStream $ go (toStream m)
532 where
532 where
533− go m1 = Stream $ \_ stp sng yld ->
533+ go m1 = C.Stream $ \_ stp sng yld ->
534 let single a | p a = sng a
534 let single a | p a = sng a
535 | otherwise = stp
535 | otherwise = stp
536 yield a r | p a = yld a (go r)
536 yield a r | p a = yld a (go r)
537− | otherwise = (S.runStream r) Nothing stp single yield
537+ | otherwise = (C.runStream r) Nothing stp single yield
538− in (S.runStream m1) Nothing stp single yield
538+ in (C.runStream m1) Nothing stp single yield
539
539
540-- | End the stream as soon as the predicate fails on an element.
540-- | End the stream as soon as the predicate fails on an element.
541--
541--
@@ -544,12 +544,12 @@ filter p m = fromStream $ go (toStream m)
544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
545takeWhile p m = fromStream $ go (toStream m)
545takeWhile p m = fromStream $ go (toStream m)
546 where
546 where
547− go m1 = Stream $ \_ stp sng yld ->
547+ go m1 = C.Stream $ \_ stp sng yld ->
548 let single a | p a = sng a
548 let single a | p a = sng a
549 | otherwise = stp
549 | otherwise = stp
550 yield a r | p a = yld a (go r)
550 yield a r | p a = yld a (go r)
551 | otherwise = stp
551 | otherwise = stp
552− in (S.runStream m1) Nothing stp single yield
552+ in (C.runStream m1) Nothing stp single yield
553
553
554-- | Discard first 'n' elements from the stream and take the rest.
554-- | Discard first 'n' elements from the stream and take the rest.
555--
555--
@@ -557,13 +557,13 @@ takeWhile p m = fromStream $ go (toStream m)
557drop :: IsStream t => Int -> t m a -> t m a
557drop :: IsStream t => Int -> t m a -> t m a
558drop n m = fromStream $ go n (toStream m)
558drop n m = fromStream $ go n (toStream m)
559 where
559 where
560− go n1 m1 = Stream $ \_ stp sng yld ->
560+ go n1 m1 = C.Stream $ \_ stp sng yld ->
561 let single _ = stp
561 let single _ = stp
562− yield _ r = (S.runStream $ go (n1 - 1) r) Nothing stp sng yld
562+ yield _ r = (C.runStream $ go (n1 - 1) r) Nothing stp sng yld
563 -- Somehow "<=" check performs better than a ">"
563 -- Somehow "<=" check performs better than a ">"
564 in if n1 <= 0
564 in if n1 <= 0
565− then (S.runStream m1) Nothing stp sng yld
565+ then (C.runStream m1) Nothing stp sng yld
566− else (S.runStream m1) Nothing stp single yield
566+ else (C.runStream m1) Nothing stp single yield
567
567
568-- | Drop elements in the stream as long as the predicate succeeds and then
568-- | Drop elements in the stream as long as the predicate succeeds and then
569-- take the rest of the stream.
569-- take the rest of the stream.
@@ -573,12 +573,12 @@ drop n m = fromStream $ go n (toStream m)
573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
574dropWhile p m = fromStream $ go (toStream m)
574dropWhile p m = fromStream $ go (toStream m)
575 where
575 where
576− go m1 = Stream $ \_ stp sng yld ->
576+ go m1 = C.Stream $ \_ stp sng yld ->
577 let single a | p a = stp
577 let single a | p a = stp
578 | otherwise = sng a
578 | otherwise = sng a
579− yield a r | p a = (S.runStream r) Nothing stp single yield
579+ yield a r | p a = (C.runStream r) Nothing stp single yield
580 | otherwise = yld a r
580 | otherwise = yld a r
581− in (S.runStream m1) Nothing stp single yield
581+ in (C.runStream m1) Nothing stp single yield
582
582
583-- | Determine whether all elements of a stream satisfy a predicate.
583-- | Determine whether all elements of a stream satisfy a predicate.
584--
584--
@@ -591,7 +591,7 @@ all p m = go (toStream m)
591 | otherwise = return False
591 | otherwise = return False
592 yield a r | p a = go r
592 yield a r | p a = go r
593 | otherwise = return False
593 | otherwise = return False
594− in (S.runStream m1) Nothing (return True) single yield
594+ in (C.runStream m1) Nothing (return True) single yield
595
595
596-- | Determine whether any of the elements of a stream satisfy a predicate.
596-- | Determine whether any of the elements of a stream satisfy a predicate.
597--
597--
@@ -604,7 +604,7 @@ any p m = go (toStream m)
604 | otherwise = return False
604 | otherwise = return False
605 yield a r | p a = return True
605 yield a r | p a = return True
606 | otherwise = go r
606 | otherwise = go r
607− in (S.runStream m1) Nothing (return False) single yield
607+ in (C.runStream m1) Nothing (return False) single yield
608
608
609-- | Determine the sum of all elements of a stream of numbers
609-- | Determine the sum of all elements of a stream of numbers
610--
610--
@@ -626,7 +626,7 @@ head m =
626 let stop = return Nothing
626 let stop = return Nothing
627 single a = return (Just a)
627 single a = return (Just a)
628 yield a _ = return (Just a)
628 yield a _ = return (Just a)
629− in (S.runStream (toStream m)) Nothing stop single yield
629+ in (C.runStream (toStream m)) Nothing stop single yield
630
630
631-- | Extract all but the first element of the stream, if any.
631-- | Extract all but the first element of the stream, if any.
632--
632--
@@ -636,7 +636,7 @@ tail m =
636 let stop = return Nothing
636 let stop = return Nothing
637 single _ = return $ Just nil
637 single _ = return $ Just nil
638 yield _ r = return $ Just $ fromStream r
638 yield _ r = return $ Just $ fromStream r
639− in (S.runStream (toStream m)) Nothing stop single yield
639+ in (C.runStream (toStream m)) Nothing stop single yield
640
640
641-- | Extract the last element of the stream, if any.
641-- | Extract the last element of the stream, if any.
642--
642--
@@ -653,7 +653,7 @@ null m =
653 let stop = return True
653 let stop = return True
654 single _ = return False
654 single _ = return False
655 yield _ _ = return False
655 yield _ _ = return False
656− in (S.runStream (toStream m)) Nothing stop single yield
656+ in (C.runStream (toStream m)) Nothing stop single yield
657
657
658-- | Determine whether an element is present in the stream.
658-- | Determine whether an element is present in the stream.
659--
659--
@@ -665,7 +665,7 @@ elem e m = go (toStream m)
665 let stop = return False
665 let stop = return False
666 single a = return (a == e)
666 single a = return (a == e)
667 yield a r = if a == e then return True else go r
667 yield a r = if a == e then return True else go r
668− in (S.runStream m1) Nothing stop single yield
668+ in (C.runStream m1) Nothing stop single yield
669
669
670-- | Determine whether an element is not present in the stream.
670-- | Determine whether an element is not present in the stream.
671--
671--
@@ -677,7 +677,7 @@ notElem e m = go (toStream m)
677 let stop = return True
677 let stop = return True
678 single a = return (a /= e)
678 single a = return (a /= e)
679 yield a r = if a == e then return False else go r
679 yield a r = if a == e then return False else go r
680− in (S.runStream m1) Nothing stop single yield
680+ in (C.runStream m1) Nothing stop single yield
681
681
682-- | Determine the length of the stream.
682-- | Determine the length of the stream.
683--
683--
@@ -690,14 +690,14 @@ length = foldl (\n _ -> n + 1) 0 id
690--
690--
691-- @since 0.1.1
691-- @since 0.1.1
692reverse :: (IsStream t) => t m a -> t m a
692reverse :: (IsStream t) => t m a -> t m a
693−reverse m = fromStream $ go S.nil (toStream m)
693+reverse m = fromStream $ go C.nil (toStream m)
694 where
694 where
695− go rev rest = Stream $ \_ stp sng yld ->
695+ go rev rest = C.Stream $ \_ stp sng yld ->
696− let run x = S.runStream x Nothing stp sng yld
696+ let run x = C.runStream x Nothing stp sng yld
697 stop = run rev
697 stop = run rev
698− single a = run $ a `S.cons` rev
698+ single a = run $ a `C.cons` rev
699− yield a r = run $ go (a `S.cons` rev) r
699+ yield a r = run $ go (a `C.cons` rev) r
700− in S.runStream rest Nothing stop single yield
700+ in C.runStream rest Nothing stop single yield
701
701
702-- XXX replace the recursive "go" with continuation
702-- XXX replace the recursive "go" with continuation
703-- | Determine the minimum element in a stream.
703-- | Determine the minimum element in a stream.
@@ -710,7 +710,7 @@ minimum m = go Nothing (toStream m)
710 let stop = return res
710 let stop = return res
711 single a = return $ min_ a res
711 single a = return $ min_ a res
712 yield a r = go (min_ a res) r
712 yield a r = go (min_ a res) r
713− in (S.runStream m1) Nothing stop single yield
713+ in (C.runStream m1) Nothing stop single yield
714
714
715 min_ a res = case res of
715 min_ a res = case res of
716 Nothing -> Just a
716 Nothing -> Just a
@@ -727,7 +727,7 @@ maximum m = go Nothing (toStream m)
727 let stop = return res
727 let stop = return res
728 single a = return $ max_ a res
728 single a = return $ max_ a res
729 yield a r = go (max_ a res) r
729 yield a r = go (max_ a res) r
730− in (S.runStream m1) Nothing stop single yield
730+ in (C.runStream m1) Nothing stop single yield
731
731
732 max_ a res = case res of
732 max_ a res = case res of
733 Nothing -> Just a
733 Nothing -> Just a
@@ -755,10 +755,10 @@ maximum m = go Nothing (toStream m)
755mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
755mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
756mapM f m = go (toStream m)
756mapM f m = go (toStream m)
757 where
757 where
758− go m1 = fromStream $ Stream $ \svr stp sng yld ->
758+ go m1 = fromStream $ C.Stream $ \svr stp sng yld ->
759 let single a = f a >>= sng
759 let single a = f a >>= sng
760− yield a r = S.runStream (toStream (f a |: (go r))) svr stp sng yld
760+ yield a r = C.runStream (toStream (f a |: (go r))) svr stp sng yld
761− in (S.runStream m1) Nothing stp single yield
761+ in (C.runStream m1) Nothing stp single yield
762
762
763-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
763-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
764-- elements, and return a stream of values extracted from 'Just'.
764-- elements, and return a stream of values extracted from 'Just'.
@@ -768,14 +768,14 @@ mapM f m = go (toStream m)
768mapMaybe :: (IsStream t) => (a -> Maybe b) -> t m a -> t m b
768mapMaybe :: (IsStream t) => (a -> Maybe b) -> t m a -> t m b
769mapMaybe f m = go (toStream m)
769mapMaybe f m = go (toStream m)
770 where
770 where
771− go m1 = fromStream $ Stream $ \_ stp sng yld ->
771+ go m1 = fromStream $ C.Stream $ \_ stp sng yld ->
772 let single a = case f a of
772 let single a = case f a of
773 Just b -> sng b
773 Just b -> sng b
774 Nothing -> stp
774 Nothing -> stp
775 yield a r = case f a of
775 yield a r = case f a of
776 Just b -> yld b (toStream $ go r)
776 Just b -> yld b (toStream $ go r)
777− Nothing -> (S.runStream r) Nothing stp single yield
777+ Nothing -> (C.runStream r) Nothing stp single yield
778− in (S.runStream m1) Nothing stp single yield
778+ in (C.runStream m1) Nothing stp single yield
779
779
780-- | Like 'mapMaybe' but maps a monadic function.
780-- | Like 'mapMaybe' but maps a monadic function.
781--
781--
@@ -799,7 +799,7 @@ mapM_ f m = go (toStream m)
799 let stop = return ()
799 let stop = return ()
800 single a = void (f a)
800 single a = void (f a)
801 yield a r = f a >> go r
801 yield a r = f a >> go r
802− in (S.runStream m1) Nothing stop single yield
802+ in (C.runStream m1) Nothing stop single yield
803
803
804-- | Reduce a stream of monadic actions to a stream of the output of those
804-- | Reduce a stream of monadic actions to a stream of the output of those
805-- actions.
805-- actions.
@@ -818,10 +818,10 @@ mapM_ f m = go (toStream m)
818sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
818sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
819sequence m = go (toStream m)
819sequence m = go (toStream m)
820 where
820 where
821− go m1 = fromStream $ Stream $ \svr stp sng yld ->
821+ go m1 = fromStream $ C.Stream $ \svr stp sng yld ->
822 let single ma = ma >>= sng
822 let single ma = ma >>= sng
823− yield ma r = S.runStream (toStream $ ma |: go r) svr stp sng yld
823+ yield ma r = C.runStream (toStream $ ma |: go r) svr stp sng yld
824− in (S.runStream m1) Nothing stp single yield
824+ in (C.runStream m1) Nothing stp single yield
825
825
826------------------------------------------------------------------------------
826------------------------------------------------------------------------------
827-- Serially Zipping Streams
827-- Serially Zipping Streams
@@ -832,7 +832,7 @@ sequence m = go (toStream m)
832-- @since 0.1.0
832-- @since 0.1.0
833{-# INLINABLE zipWith #-}
833{-# INLINABLE zipWith #-}
834zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
834zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
835−zipWith f m1 m2 = fromStream $ S.zipWith f (toStream m1) (toStream m2)
835+zipWith f m1 m2 = fromStream $ C.zipWith f (toStream m1) (toStream m2)
836
836
837-- | Zip two streams serially using a monadic zipping function.
837-- | Zip two streams serially using a monadic zipping function.
838--
838--
@@ -840,15 +840,15 @@ zipWith f m1 m2 = fromStream $ S.zipWith f (toStream m1) (toStream m2)
840zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c
840zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c
841zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
841zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
842 where
842 where
843− go mx my = Stream $ \_ stp sng yld -> do
843+ go mx my = C.Stream $ \_ stp sng yld -> do
844 let merge a ra =
844 let merge a ra =
845− let run x = S.runStream x Nothing stp sng yld
845+ let run x = C.runStream x Nothing stp sng yld
846 single2 b = run $ toStream (f a b)
846 single2 b = run $ toStream (f a b)
847 yield2 b rb = run $ toStream (f a b) <> go ra rb
847 yield2 b rb = run $ toStream (f a b) <> go ra rb
848− in (S.runStream my) Nothing stp single2 yield2
848+ in (C.runStream my) Nothing stp single2 yield2
849− let single1 a = merge a S.nil
849+ let single1 a = merge a C.nil
850 yield1 a ra = merge a ra
850 yield1 a ra = merge a ra
851− (S.runStream mx) Nothing stp single1 yield1
851+ (C.runStream mx) Nothing stp single1 yield1
852
852
853------------------------------------------------------------------------------
853------------------------------------------------------------------------------
854-- Parallely Zipping Streams
854-- Parallely Zipping Streams
@@ -861,7 +861,7 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
861zipAsyncWith :: (IsStream t, MonadAsync m)
861zipAsyncWith :: (IsStream t, MonadAsync m)
862 => (a -> b -> c) -> t m a -> t m b -> t m c
862 => (a -> b -> c) -> t m a -> t m b -> t m c
863zipAsyncWith f m1 m2 =
863zipAsyncWith f m1 m2 =
864− fromStream $ S.zipAsyncWith f (toStream m1) (toStream m2)
864+ fromStream $ CC.zipAsyncWith f (toStream m1) (toStream m2)
865
865
866-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
866-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
867-- concurrently) using a monadic zipping function.
867-- concurrently) using a monadic zipping function.
@@ -869,7 +869,7 @@ zipAsyncWith f m1 m2 =
869-- @since 0.1.0
869-- @since 0.1.0
870zipAsyncWithM :: (IsStream t, MonadAsync m)
870zipAsyncWithM :: (IsStream t, MonadAsync m)
871 => (a -> b -> t m c) -> t m a -> t m b -> t m c
871 => (a -> b -> t m c) -> t m a -> t m b -> t m c
872−zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do
872+zipAsyncWithM f m1 m2 = fromStream $ C.Stream $ \_ stp sng yld -> do
873 ma <- mkAsync m1
873 ma <- mkAsync m1
874 mb <- mkAsync m2
874 mb <- mkAsync m2
875− (S.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld
875+ (C.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld
src/Streamly/Streams.hs+70−68
@@ -118,7 +118,9 @@ import Control.Monad.State.Class (MonadState(..))
118import Control.Monad.Trans.Class (MonadTrans (lift))
118import Control.Monad.Trans.Class (MonadTrans (lift))
119import Data.Semigroup (Semigroup(..))
119import Data.Semigroup (Semigroup(..))
120import Streamly.SVar (MonadAsync, SVar, toStreamVar)
120import Streamly.SVar (MonadAsync, SVar, toStreamVar)
121+
121−import qualified Streamly.Core as S
122+import qualified Streamly.CPS.Stream as C
123+import qualified Streamly.CPS.Concurrent as CC
122
124
123------------------------------------------------------------------------------
125------------------------------------------------------------------------------
124-- Types that can behave as a Stream
126-- Types that can behave as a Stream
@@ -132,8 +134,8 @@ infixr 5 |:
132--
134--
133-- @since 0.2.0
135-- @since 0.2.0
134class IsStream t where
136class IsStream t where
135− toStream :: t m a -> S.Stream m a
137+ toStream :: t m a -> C.Stream m a
136− fromStream :: S.Stream m a -> t m a
138+ fromStream :: C.Stream m a -> t m a
137 -- | Constructs a stream by adding a monadic action at the head of an
139 -- | Constructs a stream by adding a monadic action at the head of an
138 -- existing stream. For example:
140 -- existing stream. For example:
139 --
141 --
@@ -190,7 +192,7 @@ type Streaming = IsStream
190--
192--
191-- @since 0.1.0
193-- @since 0.1.0
192nil :: IsStream t => t m a
194nil :: IsStream t => t m a
193−nil = fromStream S.nil
195+nil = fromStream C.nil
194
196
195-- | Constructs a stream by adding a monadic action at the head of an existing
197-- | Constructs a stream by adding a monadic action at the head of an existing
196-- stream. For example:
198-- stream. For example:
@@ -204,7 +206,7 @@ nil = fromStream S.nil
204--
206--
205-- @since 0.2.0
207-- @since 0.2.0
206consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
208consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
207−consMSerial m r = fromStream $ S.consM m (toStream r)
209+consMSerial m r = fromStream $ C.consM m (toStream r)
208
210
209infixr 5 `cons`
211infixr 5 `cons`
210
212
@@ -218,7 +220,7 @@ infixr 5 `cons`
218--
220--
219-- @since 0.1.0
221-- @since 0.1.0
220cons :: IsStream t => a -> t m a -> t m a
222cons :: IsStream t => a -> t m a -> t m a
221−cons a r = fromStream $ S.cons a (toStream r)
223+cons a r = fromStream $ C.cons a (toStream r)
222
224
223infixr 5 .:
225infixr 5 .:
224
226
@@ -239,23 +241,23 @@ infixr 5 .:
239-- remaining stream if any otherwise 'Nothing'. The third parameter is to
241-- remaining stream if any otherwise 'Nothing'. The third parameter is to
240-- represent an "empty" stream.
242-- represent an "empty" stream.
241streamBuild :: IsStream t
243streamBuild :: IsStream t
242− => (forall r. Maybe (SVar S.Stream m a)
244+ => (forall r. Maybe (SVar C.Stream m a)
243 -> (a -> t m a -> m r)
245 -> (a -> t m a -> m r)
244 -> (a -> m r)
246 -> (a -> m r)
245 -> m r
247 -> m r
246 -> m r)
248 -> m r)
247 -> t m a
249 -> t m a
248−streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
250+streamBuild k = fromStream $ C.Stream $ \svr stp sng yld ->
249 let yield a r = yld a (toStream r)
251 let yield a r = yld a (toStream r)
250 in k svr yield sng stp
252 in k svr yield sng stp
251
253
252-- | Build a singleton stream from a callback function.
254-- | Build a singleton stream from a callback function.
253fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
255fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
254−fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
256+fromCallback k = fromStream $ C.Stream $ \_ _ sng _ -> k sng
255
257
256-- | Read an SVar to get a stream.
258-- | Read an SVar to get a stream.
257−fromSVar :: (MonadAsync m, IsStream t) => SVar S.Stream m a -> t m a
259+fromSVar :: (MonadAsync m, IsStream t) => SVar C.Stream m a -> t m a
258−fromSVar sv = fromStream $ S.fromStreamVar sv
260+fromSVar sv = fromStream $ CC.fromStreamVar sv
259
261
260------------------------------------------------------------------------------
262------------------------------------------------------------------------------
261-- Destroying a stream
263-- Destroying a stream
@@ -266,7 +268,7 @@ fromSVar sv = fromStream $ S.fromStreamVar sv
266-- argument is for consuming an "empty" stream that yields nothing.
268-- argument is for consuming an "empty" stream that yields nothing.
267streamFold
269streamFold
268 :: IsStream t
270 :: IsStream t
269− => Maybe (SVar S.Stream m a)
271+ => Maybe (SVar C.Stream m a)
270 -> (a -> t m a -> m r)
272 -> (a -> t m a -> m r)
271 -> (a -> m r)
273 -> (a -> m r)
272 -> m r
274 -> m r
@@ -274,7 +276,7 @@ streamFold
274 -> m r
276 -> m r
275streamFold svr step single blank m =
277streamFold svr step single blank m =
276 let yield a x = step a (fromStream x)
278 let yield a x = step a (fromStream x)
277− in (S.runStream (toStream m)) svr blank single yield
279+ in (C.runStream (toStream m)) svr blank single yield
278
280
279-- | Run a streaming composition, discard the results. By default it interprets
281-- | Run a streaming composition, discard the results. By default it interprets
280-- the stream as 'SerialT', to run other types of streams use the type adapting
282-- the stream as 'SerialT', to run other types of streams use the type adapting
@@ -288,7 +290,7 @@ runStream m = go (toStream m)
288 let stop = return ()
290 let stop = return ()
289 single _ = return ()
291 single _ = return ()
290 yield _ r = go r
292 yield _ r = go r
291− in (S.runStream m1) Nothing stop single yield
293+ in (C.runStream m1) Nothing stop single yield
292
294
293-- | Same as 'runStream'
295-- | Same as 'runStream'
294--
296--
@@ -299,7 +301,7 @@ runStreaming = runStream . adapt
299
301
300-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
302-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
301-- be read back from the SVar using 'fromSVar'.
303-- be read back from the SVar using 'fromSVar'.
302−toSVar :: (IsStream t, MonadAsync m) => SVar S.Stream m a -> t m a -> m ()
304+toSVar :: (IsStream t, MonadAsync m) => SVar C.Stream m a -> t m a -> m ()
303toSVar sv m = toStreamVar sv (toStream m)
305toSVar sv m = toStreamVar sv (toStream m)
304
306
305------------------------------------------------------------------------------
307------------------------------------------------------------------------------
@@ -316,17 +318,17 @@ toSVar sv m = toStreamVar sv (toStream m)
316-- @since 0.2.0
318-- @since 0.2.0
317mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
319mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
318mkAsync m = do
320mkAsync m = do
319− s <- S.mkParallel (toStream m)
321+ s <- CC.mkParallel (toStream m)
320 return $ fromStream s
322 return $ fromStream s
321
323
322{-# INLINE applyWith #-}
324{-# INLINE applyWith #-}
323applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
325applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
324applyWith f x = fromStream $
326applyWith f x = fromStream $
325− S.applyWith (toStream . f . fromStream) (toStream x)
327+ CC.applyWith (toStream . f . fromStream) (toStream x)
326
328
327{-# INLINE runWith #-}
329{-# INLINE runWith #-}
328runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
330runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
329−runWith f x = S.runWith (f . fromStream) (toStream x)
331+runWith f x = CC.runWith (f . fromStream) (toStream x)
330
332
331infixr 0 |$
333infixr 0 |$
332infixr 0 |$.
334infixr 0 |$.
@@ -410,21 +412,21 @@ x |&. f = f |$. x
410-- XXX use template haskell instead and include Monoid and IsStream instances
412-- XXX use template haskell instead and include Monoid and IsStream instances
411-- as well.
413-- as well.
412
414
413−withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
415+withLocal :: MonadReader r m => (r -> r) -> C.Stream m a -> C.Stream m a
414withLocal f m =
416withLocal f m =
415− S.Stream $ \_ stp sng yld ->
417+ C.Stream $ \_ stp sng yld ->
416 let single = local f . sng
418 let single = local f . sng
417 yield a r = local f $ yld a (withLocal f r)
419 yield a r = local f $ yld a (withLocal f r)
418− in (S.runStream m) Nothing (local f stp) single yield
420+ in (C.runStream m) Nothing (local f stp) single yield
419
421
420{-
422{-
421-- XXX handle and test cross thread state transfer
423-- XXX handle and test cross thread state transfer
422withCatchError
424withCatchError
423 :: MonadError e m
425 :: MonadError e m
424− => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
426+ => C.Stream m a -> (e -> C.Stream m a) -> C.Stream m a
425withCatchError m h =
427withCatchError m h =
426− S.Stream $ \_ stp sng yld ->
428+ C.Stream $ \_ stp sng yld ->
427− let run x = S.runStream x Nothing stp sng yield
429+ let run x = C.runStream x Nothing stp sng yield
428 handle r = r `catchError` \e -> run $ h e
430 handle r = r `catchError` \e -> run $ h e
429 yield a r = yld a (withCatchError r h)
431 yield a r = yld a (withCatchError r h)
430 in handle $ run m
432 in handle $ run m
@@ -434,7 +436,7 @@ withCatchError m h =
434
436
435#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
437#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
436instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
438instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
437− pure = STREAM . S.singleton; \
439+ pure = STREAM . C.singleton; \
438 (<*>) = ap }
440 (<*>) = ap }
439
441
440#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
442#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
@@ -523,7 +525,7 @@ instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
523-- time.
525-- time.
524--
526--
525-- @since 0.2.0
527-- @since 0.2.0
526−newtype SerialT m a = SerialT {getSerialT :: S.Stream m a}
528+newtype SerialT m a = SerialT {getSerialT :: C.Stream m a}
527 deriving (Semigroup, Monoid, Functor, MonadTrans)
529 deriving (Semigroup, Monoid, Functor, MonadTrans)
528
530
529-- |
531-- |
@@ -556,7 +558,7 @@ instance IsStream SerialT where
556-- @since 0.2.0
558-- @since 0.2.0
557{-# INLINE serial #-}
559{-# INLINE serial #-}
558serial :: IsStream t => t m a -> t m a -> t m a
560serial :: IsStream t => t m a -> t m a -> t m a
559−serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
561+serial m1 m2 = fromStream $ C.serial (toStream m1) (toStream m2)
560
562
561------------------------------------------------------------------------------
563------------------------------------------------------------------------------
562-- Monad
564-- Monad
@@ -564,8 +566,8 @@ serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
564
566
565instance Monad m => Monad (SerialT m) where
567instance Monad m => Monad (SerialT m) where
566 return = pure
568 return = pure
567− (SerialT (S.Stream m)) >>= f = SerialT $ S.Stream $ \_ stp sng yld ->
569+ (SerialT (C.Stream m)) >>= f = SerialT $ C.Stream $ \_ stp sng yld ->
568− let run x = (S.runStream x) Nothing stp sng yld
570+ let run x = (C.runStream x) Nothing stp sng yld
569 single a = run $ toStream (f a)
571 single a = run $ toStream (f a)
570 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
572 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
571 in m Nothing stp single yield
573 in m Nothing stp single yield
@@ -618,7 +620,7 @@ MONAD_COMMON_INSTANCES(SerialT,)
618-- stream.
620-- stream.
619--
621--
620-- @since 0.2.0
622-- @since 0.2.0
621−newtype WSerialT m a = WSerialT {getWSerialT :: S.Stream m a}
623+newtype WSerialT m a = WSerialT {getWSerialT :: C.Stream m a}
622 deriving (Functor, MonadTrans)
624 deriving (Functor, MonadTrans)
623
625
624-- |
626-- |
@@ -650,7 +652,7 @@ instance IsStream WSerialT where
650-- @since 0.2.0
652-- @since 0.2.0
651{-# INLINE wSerial #-}
653{-# INLINE wSerial #-}
652wSerial :: IsStream t => t m a -> t m a -> t m a
654wSerial :: IsStream t => t m a -> t m a -> t m a
653−wSerial m1 m2 = fromStream $ S.wSerial (toStream m1) (toStream m2)
655+wSerial m1 m2 = fromStream $ C.wSerial (toStream m1) (toStream m2)
654
656
655instance Semigroup (WSerialT m a) where
657instance Semigroup (WSerialT m a) where
656 (<>) = wSerial
658 (<>) = wSerial
@@ -679,8 +681,8 @@ instance Monoid (WSerialT m a) where
679
681
680instance Monad m => Monad (WSerialT m) where
682instance Monad m => Monad (WSerialT m) where
681 return = pure
683 return = pure
682− (WSerialT (S.Stream m)) >>= f = WSerialT $ S.Stream $ \_ stp sng yld ->
684+ (WSerialT (C.Stream m)) >>= f = WSerialT $ C.Stream $ \_ stp sng yld ->
683− let run x = (S.runStream x) Nothing stp sng yld
685+ let run x = (C.runStream x) Nothing stp sng yld
684 single a = run $ toStream (f a)
686 single a = run $ toStream (f a)
685 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
687 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
686 in m Nothing stp single yield
688 in m Nothing stp single yield
@@ -740,7 +742,7 @@ MONAD_COMMON_INSTANCES(WSerialT,)
740-- streams at a time.
742-- streams at a time.
741--
743--
742-- @since 0.3.0
744-- @since 0.3.0
743−newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}
745+newtype AheadT m a = AheadT {getAheadT :: C.Stream m a}
744 deriving (Functor, MonadTrans)
746 deriving (Functor, MonadTrans)
745
747
746instance IsStream AheadT where
748instance IsStream AheadT where
@@ -749,7 +751,7 @@ instance IsStream AheadT where
749
751
750 {-# INLINE consM #-}
752 {-# INLINE consM #-}
751 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
753 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
752− consM m r = fromStream $ S.consMAhead m (toStream r)
754+ consM m r = fromStream $ CC.consMAhead m (toStream r)
753
755
754 {-# INLINE (|:) #-}
756 {-# INLINE (|:) #-}
755 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
757 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
@@ -765,7 +767,7 @@ instance IsStream AheadT where
765-- @since 0.3.0
767-- @since 0.3.0
766{-# INLINE ahead #-}
768{-# INLINE ahead #-}
767ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
769ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
768−ahead m1 m2 = fromStream $ S.ahead (toStream m1) (toStream m2)
770+ahead m1 m2 = fromStream $ CC.ahead (toStream m1) (toStream m2)
769
771
770instance MonadAsync m => Semigroup (AheadT m a) where
772instance MonadAsync m => Semigroup (AheadT m a) where
771 (<>) = ahead
773 (<>) = ahead
@@ -785,16 +787,16 @@ instance MonadAsync m => Monoid (AheadT m a) where
785{-# INLINE aheadbind #-}
787{-# INLINE aheadbind #-}
786aheadbind
788aheadbind
787 :: MonadAsync m
789 :: MonadAsync m
788− => S.Stream m a
790+ => C.Stream m a
789− -> (a -> S.Stream m b)
791+ -> (a -> C.Stream m b)
790− -> S.Stream m b
792+ -> C.Stream m b
791aheadbind m f = go m
793aheadbind m f = go m
792 where
794 where
793− go (S.Stream g) =
795+ go (C.Stream g) =
794− S.Stream $ \ctx stp sng yld ->
796+ C.Stream $ \ctx stp sng yld ->
795− let run x = (S.runStream x) ctx stp sng yld
797+ let run x = (C.runStream x) ctx stp sng yld
796 single a = run $ f a
798 single a = run $ f a
797− yield a r = run $ f a `S.ahead` go r
799+ yield a r = run $ f a `CC.ahead` go r
798 in g Nothing stp single yield
800 in g Nothing stp single yield
799
801
800instance MonadAsync m => Monad (AheadT m) where
802instance MonadAsync m => Monad (AheadT m) where
@@ -861,7 +863,7 @@ MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
861-- streams at a time.
863-- streams at a time.
862--
864--
863-- @since 0.1.0
865-- @since 0.1.0
864−newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}
866+newtype AsyncT m a = AsyncT {getAsyncT :: C.Stream m a}
865 deriving (Functor, MonadTrans)
867 deriving (Functor, MonadTrans)
866
868
867instance IsStream AsyncT where
869instance IsStream AsyncT where
@@ -870,7 +872,7 @@ instance IsStream AsyncT where
870
872
871 {-# INLINE consM #-}
873 {-# INLINE consM #-}
872 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
874 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
873− consM m r = fromStream $ S.consMAsync m (toStream r)
875+ consM m r = fromStream $ CC.consMAsync m (toStream r)
874
876
875 {-# INLINE (|:) #-}
877 {-# INLINE (|:) #-}
876 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
878 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
@@ -887,7 +889,7 @@ instance IsStream AsyncT where
887-- @since 0.2.0
889-- @since 0.2.0
888{-# INLINE async #-}
890{-# INLINE async #-}
889async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
891async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
890−async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)
892+async m1 m2 = fromStream $ CC.async (toStream m1) (toStream m2)
891
893
892-- | Same as 'async'.
894-- | Same as 'async'.
893--
895--
@@ -914,22 +916,22 @@ instance MonadAsync m => Monoid (AsyncT m a) where
914
916
915{-# INLINE parbind #-}
917{-# INLINE parbind #-}
916parbind
918parbind
917− :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
919+ :: (forall c. C.Stream m c -> C.Stream m c -> C.Stream m c)
918− -> S.Stream m a
920+ -> C.Stream m a
919− -> (a -> S.Stream m b)
921+ -> (a -> C.Stream m b)
920− -> S.Stream m b
922+ -> C.Stream m b
921parbind par m f = go m
923parbind par m f = go m
922 where
924 where
923− go (S.Stream g) =
925+ go (C.Stream g) =
924− S.Stream $ \ctx stp sng yld ->
926+ C.Stream $ \ctx stp sng yld ->
925− let run x = (S.runStream x) ctx stp sng yld
927+ let run x = (C.runStream x) ctx stp sng yld
926 single a = run $ f a
928 single a = run $ f a
927 yield a r = run $ f a `par` go r
929 yield a r = run $ f a `par` go r
928 in g Nothing stp single yield
930 in g Nothing stp single yield
929
931
930instance MonadAsync m => Monad (AsyncT m) where
932instance MonadAsync m => Monad (AsyncT m) where
931 return = pure
933 return = pure
932− (AsyncT m) >>= f = AsyncT $ parbind S.async m (getAsyncT . f)
934+ (AsyncT m) >>= f = AsyncT $ parbind CC.async m (getAsyncT . f)
933
935
934------------------------------------------------------------------------------
936------------------------------------------------------------------------------
935-- Other instances
937-- Other instances
@@ -989,7 +991,7 @@ MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
989-- stream.
991-- stream.
990--
992--
991-- @since 0.2.0
993-- @since 0.2.0
992−newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}
994+newtype WAsyncT m a = WAsyncT {getWAsyncT :: C.Stream m a}
993 deriving (Functor, MonadTrans)
995 deriving (Functor, MonadTrans)
994
996
995instance IsStream WAsyncT where
997instance IsStream WAsyncT where
@@ -998,7 +1000,7 @@ instance IsStream WAsyncT where
998
1000
999 {-# INLINE consM #-}
1001 {-# INLINE consM #-}
1000 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1002 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1001− consM m r = fromStream $ S.consMWAsync m (toStream r)
1003+ consM m r = fromStream $ CC.consMWAsync m (toStream r)
1002
1004
1003 {-# INLINE (|:) #-}
1005 {-# INLINE (|:) #-}
1004 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1006 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
@@ -1014,7 +1016,7 @@ instance IsStream WAsyncT where
1014-- @since 0.2.0
1016-- @since 0.2.0
1015{-# INLINE wAsync #-}
1017{-# INLINE wAsync #-}
1016wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1018wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1017−wAsync m1 m2 = fromStream $ S.wAsync (toStream m1) (toStream m2)
1019+wAsync m1 m2 = fromStream $ CC.wAsync (toStream m1) (toStream m2)
1018
1020
1019instance MonadAsync m => Semigroup (WAsyncT m a) where
1021instance MonadAsync m => Semigroup (WAsyncT m a) where
1020 (<>) = wAsync
1022 (<>) = wAsync
@@ -1034,7 +1036,7 @@ instance MonadAsync m => Monoid (WAsyncT m a) where
1034instance MonadAsync m => Monad (WAsyncT m) where
1036instance MonadAsync m => Monad (WAsyncT m) where
1035 return = pure
1037 return = pure
1036 (WAsyncT m) >>= f =
1038 (WAsyncT m) >>= f =
1037− WAsyncT $ parbind S.wAsync m (getWAsyncT . f)
1039+ WAsyncT $ parbind CC.wAsync m (getWAsyncT . f)
1038
1040
1039------------------------------------------------------------------------------
1041------------------------------------------------------------------------------
1040-- Other instances
1042-- Other instances
@@ -1132,7 +1134,7 @@ MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
1132-- streams as it needs to retain state for each unfinished stream.
1134-- streams as it needs to retain state for each unfinished stream.
1133--
1135--
1134-- @since 0.1.0
1136-- @since 0.1.0
1135−newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
1137+newtype ParallelT m a = ParallelT {getParallelT :: C.Stream m a}
1136 deriving (Functor, MonadTrans)
1138 deriving (Functor, MonadTrans)
1137
1139
1138instance IsStream ParallelT where
1140instance IsStream ParallelT where
@@ -1141,7 +1143,7 @@ instance IsStream ParallelT where
1141
1143
1142 {-# INLINE consM #-}
1144 {-# INLINE consM #-}
1143 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1145 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1144− consM m r = fromStream $ S.consMParallel m (toStream r)
1146+ consM m r = fromStream $ CC.consMParallel m (toStream r)
1145
1147
1146 {-# INLINE (|:) #-}
1148 {-# INLINE (|:) #-}
1147 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1149 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
@@ -1157,7 +1159,7 @@ instance IsStream ParallelT where
1157-- @since 0.2.0
1159-- @since 0.2.0
1158{-# INLINE parallel #-}
1160{-# INLINE parallel #-}
1159parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1161parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1160−parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
1162+parallel m1 m2 = fromStream $ CC.parallel (toStream m1) (toStream m2)
1161
1163
1162instance MonadAsync m => Semigroup (ParallelT m a) where
1164instance MonadAsync m => Semigroup (ParallelT m a) where
1163 (<>) = parallel
1165 (<>) = parallel
@@ -1176,7 +1178,7 @@ instance MonadAsync m => Monoid (ParallelT m a) where
1176
1178
1177instance MonadAsync m => Monad (ParallelT m) where
1179instance MonadAsync m => Monad (ParallelT m) where
1178 return = pure
1180 return = pure
1179− (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)
1181+ (ParallelT m) >>= f = ParallelT $ parbind CC.parallel m (getParallelT . f)
1180
1182
1181------------------------------------------------------------------------------
1183------------------------------------------------------------------------------
1182-- Other instances
1184-- Other instances
@@ -1207,7 +1209,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1207-- 'SerialT'.
1209-- 'SerialT'.
1208--
1210--
1209-- @since 0.2.0
1211-- @since 0.2.0
1210−newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: S.Stream m a}
1212+newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: C.Stream m a}
1211 deriving (Functor, Semigroup, Monoid)
1213 deriving (Functor, Semigroup, Monoid)
1212
1214
1213-- |
1215-- |
@@ -1230,8 +1232,8 @@ instance IsStream ZipSerialM where
1230 (|:) = consMSerial
1232 (|:) = consMSerial
1231
1233
1232instance Monad m => Applicative (ZipSerialM m) where
1234instance Monad m => Applicative (ZipSerialM m) where
1233− pure = ZipSerialM . S.repeat
1235+ pure = ZipSerialM . C.repeat
1234− m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
1236+ m1 <*> m2 = fromStream $ C.zipWith id (toStream m1) (toStream m2)
1235
1237
1236------------------------------------------------------------------------------
1238------------------------------------------------------------------------------
1237-- Parallely Zipping Streams
1239-- Parallely Zipping Streams
@@ -1254,7 +1256,7 @@ instance Monad m => Applicative (ZipSerialM m) where
1254-- 'SerialT'.
1256-- 'SerialT'.
1255--
1257--
1256-- @since 0.2.0
1258-- @since 0.2.0
1257−newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}
1259+newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: C.Stream m a}
1258 deriving (Functor, Semigroup, Monoid)
1260 deriving (Functor, Semigroup, Monoid)
1259
1261
1260instance IsStream ZipAsyncM where
1262instance IsStream ZipAsyncM where
@@ -1272,8 +1274,8 @@ instance IsStream ZipAsyncM where
1272 (|:) = consMSerial
1274 (|:) = consMSerial
1273
1275
1274instance MonadAsync m => Applicative (ZipAsyncM m) where
1276instance MonadAsync m => Applicative (ZipAsyncM m) where
1275− pure = ZipAsyncM . S.repeat
1277+ pure = ZipAsyncM . C.repeat
1276− m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)
1278+ m1 <*> m2 = fromStream $ CC.zipAsyncWith id (toStream m1) (toStream m2)
1277
1279
1278-------------------------------------------------------------------------------
1280-------------------------------------------------------------------------------
1279-- Type adapting combinators
1281-- Type adapting combinators
streamly.cabal+2−1
@@ -107,7 +107,8 @@ flag examples-sdl
107library
107library
108 hs-source-dirs: src
108 hs-source-dirs: src
109 other-modules: Streamly.SVar
109 other-modules: Streamly.SVar
110− , Streamly.Core
110+ , Streamly.CPS.Stream
111+ , Streamly.CPS.Concurrent
111 , Streamly.Streams
112 , Streamly.Streams
112
113
113 exposed-modules: Streamly.Prelude
114 exposed-modules: Streamly.Prelude