git.iko.soy/streamly
on master help sign in
Add stream types specialized to IO monad
12b970cHarendra Kumar
Changelog.md+1
compare with master
unifiedsplit
@@ -69,6 +69,7 @@
69 * `foldlM'` strict left fold with a monadic fold function
69 * `foldlM'` strict left fold with a monadic fold function
70 * `splice` run two streams serially one after the other
70 * `splice` run two streams serially one after the other
71 * `parallel` run two streams in parallel (replaces `<|>`)
71 * `parallel` run two streams in parallel (replaces `<|>`)
72+* Add simpler stream types that are specialized to the IO monad
72
73
73## 0.1.2
74## 0.1.2
74
75
examples/loops.hs+15−15
@@ -3,37 +3,37 @@ import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
3import Streamly.Prelude (nil)
3import Streamly.Prelude (nil)
4
4
5main = do
5main = do
6− liftIO $ hSetBuffering stdout LineBuffering
6+ hSetBuffering stdout LineBuffering
7
7
8 putStrLn $ "\nloopTail:\n"
8 putStrLn $ "\nloopTail:\n"
9 runStream $ do
9 runStream $ do
10 x <- loopTail 0
10 x <- loopTail 0
11− liftIO $ print (x :: Int)
11+ fromIO $ print (x :: Int)
12
12
13 putStrLn $ "\nloopHead:\n"
13 putStrLn $ "\nloopHead:\n"
14 runStream $ do
14 runStream $ do
15 x <- loopHead 0
15 x <- loopHead 0
16− liftIO $ print (x :: Int)
16+ fromIO $ print (x :: Int)
17
17
18 putStrLn $ "\nloopTailA:\n"
18 putStrLn $ "\nloopTailA:\n"
19 runStream $ do
19 runStream $ do
20 x <- loopTailA 0
20 x <- loopTailA 0
21− liftIO $ print (x :: Int)
21+ fromIO $ print (x :: Int)
22
22
23 putStrLn $ "\nloopHeadA:\n"
23 putStrLn $ "\nloopHeadA:\n"
24 runStream $ do
24 runStream $ do
25 x <- loopHeadA 0
25 x <- loopHeadA 0
26− liftIO $ print (x :: Int)
26+ fromIO $ print (x :: Int)
27
27
28 putStrLn $ "\ncosplice:\n"
28 putStrLn $ "\ncosplice:\n"
29 runStream $ do
29 runStream $ do
30 x <- (return 0 <> return 1) `cosplice` (return 100 <> return 101)
30 x <- (return 0 <> return 1) `cosplice` (return 100 <> return 101)
31− liftIO $ print (x :: Int)
31+ fromIO $ print (x :: Int)
32
32
33 putStrLn $ "\nParallel interleave:\n"
33 putStrLn $ "\nParallel interleave:\n"
34 runStream $ do
34 runStream $ do
35 x <- (return 0 <> return 1) `parallel` (return 100 <> return 101)
35 x <- (return 0 <> return 1) `parallel` (return 100 <> return 101)
36− liftIO $ print (x :: Int)
36+ fromIO $ print (x :: Int)
37
37
38 where
38 where
39
39
@@ -47,17 +47,17 @@ main = do
47
47
48 -- Generates a value and then loops. Can be used to generate an infinite
48 -- Generates a value and then loops. Can be used to generate an infinite
49 -- stream. Interleaves the generator and the consumer.
49 -- stream. Interleaves the generator and the consumer.
50− loopTail :: Int -> StreamT IO Int
50+ loopTail :: Int -> Stream Int
51 loopTail x = do
51 loopTail x = do
52− liftIO $ putStrLn "LoopTail..."
52+ fromIO $ putStrLn "LoopTail..."
53 return x <> (if x < 3 then loopTail (x + 1) else nil)
53 return x <> (if x < 3 then loopTail (x + 1) else nil)
54
54
55 -- Loops and then generates a value. The consumer can run only after the
55 -- Loops and then generates a value. The consumer can run only after the
56 -- loop has finished. An infinite generator will not let the consumer run
56 -- loop has finished. An infinite generator will not let the consumer run
57 -- at all.
57 -- at all.
58− loopHead :: Int -> StreamT IO Int
58+ loopHead :: Int -> Stream Int
59 loopHead x = do
59 loopHead x = do
60− liftIO $ putStrLn "LoopHead..."
60+ fromIO $ putStrLn "LoopHead..."
61 (if x < 3 then loopHead (x + 1) else nil) <> return x
61 (if x < 3 then loopHead (x + 1) else nil) <> return x
62
62
63-------------------------------------------------------------------------------
63-------------------------------------------------------------------------------
@@ -69,14 +69,14 @@ main = do
69 -- then the action on the right is also spawned concurrently. In other
69 -- then the action on the right is also spawned concurrently. In other
70 -- words, both actions may run concurrently based on the need.
70 -- words, both actions may run concurrently based on the need.
71
71
72− loopTailA :: Int -> StreamT IO Int
72+ loopTailA :: Int -> Stream Int
73 loopTailA x = do
73 loopTailA x = do
74− liftIO $ putStrLn "LoopTailA..."
74+ fromIO $ putStrLn "LoopTailA..."
75 return x `coparallel` (if x < 3 then loopTailA (x + 1) else nil)
75 return x `coparallel` (if x < 3 then loopTailA (x + 1) else nil)
76
76
77− loopHeadA :: Int -> StreamT IO Int
77+ loopHeadA :: Int -> Stream Int
78 loopHeadA x = do
78 loopHeadA x = do
79− liftIO $ putStrLn "LoopHeadA..."
79+ fromIO $ putStrLn "LoopHeadA..."
80 (if x < 3 then loopHeadA (x + 1) else nil) `coparallel` return x
80 (if x < 3 then loopHeadA (x + 1) else nil) `coparallel` return x
81
81
82-------------------------------------------------------------------------------
82-------------------------------------------------------------------------------
examples/nested-loops.hs+3−4
@@ -1,15 +1,14 @@
1import Control.Concurrent (myThreadId)
1import Control.Concurrent (myThreadId)
2−import Control.Monad.IO.Class (liftIO)
3import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
2import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
4import System.Random (randomIO)
3import System.Random (randomIO)
5import Streamly
4import Streamly
6import Streamly.Prelude (nil)
5import Streamly.Prelude (nil)
7
6
8main = runStream $ do
7main = runStream $ do
9− liftIO $ hSetBuffering stdout LineBuffering
8+ fromIO $ hSetBuffering stdout LineBuffering
10 x <- loop "A " 2
9 x <- loop "A " 2
11 y <- loop "B " 2
10 y <- loop "B " 2
12− liftIO $ myThreadId >>= putStr . show
11+ fromIO $ myThreadId >>= putStr . show
13 >> putStr " "
12 >> putStr " "
14 >> print (x, y)
13 >> print (x, y)
15
14
@@ -17,7 +16,7 @@ main = runStream $ do
17
16
18 loop :: String -> Int -> StreamT IO String
17 loop :: String -> Int -> StreamT IO String
19 loop name n = do
18 loop name n = do
20− rnd <- liftIO (randomIO :: IO Int)
19+ rnd <- fromIO (randomIO :: IO Int)
21 let result = (name ++ show rnd)
20 let result = (name ++ show rnd)
22 repeat = if n > 1 then loop name (n - 1) else nil
21 repeat = if n > 1 then loop name (n - 1) else nil
23 in (return result) `parallel` repeat
22 in (return result) `parallel` repeat
examples/parallel-loops.hs+5−6
@@ -1,20 +1,19 @@
1import Control.Concurrent (myThreadId, threadDelay)
1import Control.Concurrent (myThreadId, threadDelay)
2−import Control.Monad.IO.Class (liftIO)
3import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
2import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
4import System.Random (randomIO)
3import System.Random (randomIO)
5import Streamly
4import Streamly
6
5
7main = runStream $ do
6main = runStream $ do
8− liftIO $ hSetBuffering stdout LineBuffering
7+ fromIO $ hSetBuffering stdout LineBuffering
9 x <- loop "A" `parallel` loop "B"
8 x <- loop "A" `parallel` loop "B"
10− liftIO $ myThreadId >>= putStr . show
9+ fromIO $ myThreadId >>= putStr . show
11 >> putStr " "
10 >> putStr " "
12 >> print x
11 >> print x
13
12
14 where
13 where
15
14
16− loop :: String -> StreamT IO (String, Int)
15+ loop :: String -> Stream (String, Int)
17 loop name = do
16 loop name = do
18− liftIO $ threadDelay 1000000
17+ fromIO $ threadDelay 1000000
19− rnd <- liftIO (randomIO :: IO Int)
18+ rnd <- fromIO (randomIO :: IO Int)
20 return (name, rnd) `parallel` loop name
19 return (name, rnd) `parallel` loop name
src/Streamly.hs+15−3
@@ -16,19 +16,31 @@ module Streamly
16 -- $overview
16 -- $overview
17
17
18 MonadParallel
18 MonadParallel
19+
20+ -- * IO Streams
21+ , Stream
22+ , Costream
23+ , Coparallel
24+ , Parallel
25+ , ZipStream
26+ , ZipParallel
27+ , fromIO
28+ , fromM
29+
30+ -- * Stream transformers
19 , IsStream
31 , IsStream
20
32
21− -- * Serial Streams
33+ -- ** Serial Streams
22 -- $serial
34 -- $serial
23 , StreamT
35 , StreamT
24 , CostreamT
36 , CostreamT
25
37
26− -- * Parallel Streams
38+ -- ** Parallel Streams
27 -- $parallel
39 -- $parallel
28 , CoparallelT
40 , CoparallelT
29 , ParallelT
41 , ParallelT
30
42
31− -- * Zipping Streams
43+ -- ** Zipping Streams
32 -- $zipping
44 -- $zipping
33 , ZipStreamM
45 , ZipStreamM
34 , ZipParallelM
46 , ZipParallelM
src/Streamly/Examples/AcidRainGame.hs+3−3
@@ -14,7 +14,7 @@ import Data.Semigroup (cycle1)
14data Event = Harm Int | Heal Int | Quit deriving (Show)
14data Event = Harm Int | Heal Int | Quit deriving (Show)
15
15
16userAction :: MonadIO m => StreamT m Event
16userAction :: MonadIO m => StreamT m Event
17−userAction = cycle1 $ liftIO askUser
17+userAction = cycle1 $ fromIO askUser
18 where
18 where
19 askUser = do
19 askUser = do
20 command <- getLine
20 command <- getLine
@@ -24,7 +24,7 @@ userAction = cycle1 $ liftIO askUser
24 _ -> putStrLn "What?" >> askUser
24 _ -> putStrLn "What?" >> askUser
25
25
26acidRain :: MonadIO m => StreamT m Event
26acidRain :: MonadIO m => StreamT m Event
27−acidRain = cycle1 $ liftIO (threadDelay 1000000) >> return (Harm 1)
27+acidRain = cycle1 $ fromIO (threadDelay 1000000) >> return (Harm 1)
28
28
29game :: (MonadParallel m, MonadState Int m) => StreamT m ()
29game :: (MonadParallel m, MonadState Int m) => StreamT m ()
30game = do
30game = do
@@ -36,7 +36,7 @@ game = do
36
36
37 h <- get
37 h <- get
38 when (h <= 0) $ fail "You die!"
38 when (h <= 0) $ fail "You die!"
39− liftIO $ putStrLn $ "Health = " ++ show h
39+ fromIO $ putStrLn $ "Health = " ++ show h
40
40
41acidRainGame :: IO ()
41acidRainGame :: IO ()
42acidRainGame = do
42acidRainGame = do
src/Streamly/Examples/CirclingSquare.hs+1−1
@@ -87,4 +87,4 @@ circlingSquare :: IO ()
87circlingSquare = do
87circlingSquare = do
88 sdlInit
88 sdlInit
89 cref <- newIORef (0,0)
89 cref <- newIORef (0,0)
90− runStream $ liftIO (updateController cref) `parallel` liftIO (updateDisplay cref)
90+ runStream $ fromIO (updateController cref) `parallel` fromIO (updateDisplay cref)
src/Streamly/Examples/MergeSortedStreams.hs+10−10
@@ -8,26 +8,26 @@ import Data.List (sort)
8import Streamly
8import Streamly
9import qualified Streamly.Prelude as A
9import qualified Streamly.Prelude as A
10
10
11−getSorted :: MonadIO m => StreamT m Word16
11+getSorted :: Stream Word16
12getSorted = do
12getSorted = do
13− g <- liftIO getStdGen
13+ g <- fromIO getStdGen
14 let ls = take 100000 (randoms g) :: [Word16]
14 let ls = take 100000 (randoms g) :: [Word16]
15− foldMapWith (<>) return (sort ls)
15+ foldMap return (sort ls)
16
16
17+-- | merge two streams generating the elements from each in parallel
17−mergeAsync :: (Ord a, MonadParallel m)
18+mergeAsync :: Ord a => Stream a -> Stream a -> Stream a
18− => StreamT m a -> StreamT m a -> StreamT m a
19mergeAsync a b = do
19mergeAsync a b = do
20− x <- lift $ async a
20+ x <- fromIO $ async a
21− y <- lift $ async b
21+ y <- fromIO $ async b
22 merge x y
22 merge x y
23
23
24−merge :: (Ord a, MonadParallel m) => StreamT m a -> StreamT m a -> StreamT m a
24+merge :: Ord a => Stream a -> Stream a -> Stream a
25merge a b = do
25merge a b = do
26− a1 <- lift $ A.uncons a
26+ a1 <- fromIO $ A.uncons a
27 case a1 of
27 case a1 of
28 Nothing -> b
28 Nothing -> b
29 Just (x, ma) -> do
29 Just (x, ma) -> do
30− b1 <- lift $ A.uncons b
30+ b1 <- fromIO $ A.uncons b
31 case b1 of
31 case b1 of
32 Nothing -> return x <> ma
32 Nothing -> return x <> ma
33 Just (y, mb) ->
33 Just (y, mb) ->
src/Streamly/Examples/SearchEngineQuery.hs+1−1
@@ -14,7 +14,7 @@ searchEngineQuery = do
14
14
15 where
15 where
16 get :: IsStream t => String -> t IO ()
16 get :: IsStream t => String -> t IO ()
17− get s = streamly $ liftIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))
17+ get s = fromIO (httpNoBody (parseRequest_ s) >> putStrLn (show s))
18
18
19 google, bing, duckduckgo :: IsStream t => t IO ()
19 google, bing, duckduckgo :: IsStream t => t IO ()
20 google = get "https://www.google.com/search?q=haskell"
20 google = get "https://www.google.com/search?q=haskell"
src/Streamly/Streams.hs+67−22
@@ -57,7 +57,17 @@ module Streamly.Streams
57 , (<=>) --deprecated
57 , (<=>) --deprecated
58 , (<|) --deprecated
58 , (<|) --deprecated
59
59
60+ -- * IO Streams
61+ , Stream
62+ , Costream
63+ , Coparallel
64+ , Parallel
65+ , ZipStream
66+ , ZipParallel
67+ , fromIO
68+ , fromM
69+
60− -- * Stream Styles
70+ -- * Stream Transformers
61 , StreamT
71 , StreamT
62 , CostreamT
72 , CostreamT
63 , InterleavedT -- deprecated
73 , InterleavedT -- deprecated
@@ -111,7 +121,7 @@ import Control.Monad.State.Class (MonadState(..))
111import Control.Monad.Trans.Class (MonadTrans (lift))
121import Control.Monad.Trans.Class (MonadTrans (lift))
112import Data.Semigroup (Semigroup(..))
122import Data.Semigroup (Semigroup(..))
113import Prelude hiding (zipWith)
123import Prelude hiding (zipWith)
114−import Streamly.Core ( MonadParallel, Stream(Stream)
124+import Streamly.Core ( MonadParallel
115 , singleton
125 , singleton
116 , SVar, SVarStyle(..)
126 , SVar, SVarStyle(..)
117 , SVarTag(..), SVarSched(..))
127 , SVarTag(..), SVarSched(..))
@@ -124,8 +134,8 @@ import qualified Streamly.Core as S
124-- | Class of types that can represent a stream of elements of some type 'a' in
134-- | Class of types that can represent a stream of elements of some type 'a' in
125-- some monad 'm'.
135-- some monad 'm'.
126class IsStream t where
136class IsStream t where
127− toStream :: t m a -> Stream m a
137+ toStream :: t m a -> S.Stream m a
128− fromStream :: Stream m a -> t m a
138+ fromStream :: S.Stream m a -> t m a
129
139
130-- | Same as 'IsStream'.
140-- | Same as 'IsStream'.
131{-# Deprecated Streaming "Please use IsStream instead." #-}
141{-# Deprecated Streaming "Please use IsStream instead." #-}
@@ -195,13 +205,13 @@ streamBuild :: IsStream t
195 -> m r
205 -> m r
196 -> m r)
206 -> m r)
197 -> t m a
207 -> t m a
198−streamBuild k = fromStream $ Stream $ \svr stp sng yld ->
208+streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
199 let yield a r = yld a (toStream r)
209 let yield a r = yld a (toStream r)
200 in k svr yield sng stp
210 in k svr yield sng stp
201
211
202-- | Build a singleton stream from a callback function.
212-- | Build a singleton stream from a callback function.
203fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
213fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
204−fromCallback k = fromStream $ Stream $ \_ _ sng _ -> k sng
214+fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
205
215
206-- | Read an SVar to get a stream.
216-- | Read an SVar to get a stream.
207fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
217fromSVar :: (MonadParallel m, IsStream t) => SVar m a -> t m a
@@ -358,7 +368,7 @@ instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
358-- Note that serial composition can be used to combine an infinite number of
368-- Note that serial composition can be used to combine an infinite number of
359-- streams as it explores only one stream at a time.
369-- streams as it explores only one stream at a time.
360--
370--
361−newtype StreamT m a = StreamT {getStreamT :: Stream m a}
371+newtype StreamT m a = StreamT {getStreamT :: S.Stream m a}
362 deriving (Semigroup, Monoid, Functor, MonadTrans)
372 deriving (Semigroup, Monoid, Functor, MonadTrans)
363
373
364instance IsStream StreamT where
374instance IsStream StreamT where
@@ -382,7 +392,7 @@ splice m1 m2 = fromStream $ S.splice (toStream m1) (toStream m2)
382
392
383instance Monad m => Monad (StreamT m) where
393instance Monad m => Monad (StreamT m) where
384 return = pure
394 return = pure
385− (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp sng yld ->
395+ (StreamT (S.Stream m)) >>= f = StreamT $ S.Stream $ \_ stp sng yld ->
386 let run x = (S.runStream x) Nothing stp sng yld
396 let run x = (S.runStream x) Nothing stp sng yld
387 single a = run $ toStream (f a)
397 single a = run $ toStream (f a)
388 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
398 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
@@ -429,7 +439,7 @@ MONAD_COMMON_INSTANCES(StreamT,)
429-- Note that interleaving composition can only combine a finite number of
439-- Note that interleaving composition can only combine a finite number of
430-- streams as it needs to retain state for each unfinished stream.
440-- streams as it needs to retain state for each unfinished stream.
431--
441--
432−newtype CostreamT m a = CostreamT {getCostreamT :: Stream m a}
442+newtype CostreamT m a = CostreamT {getCostreamT :: S.Stream m a}
433 deriving (Functor, MonadTrans)
443 deriving (Functor, MonadTrans)
434
444
435{-# DEPRECATED InterleavedT "Please use 'CostreamT' instead." #-}
445{-# DEPRECATED InterleavedT "Please use 'CostreamT' instead." #-}
@@ -474,7 +484,7 @@ instance Monoid (CostreamT m a) where
474
484
475instance Monad m => Monad (CostreamT m) where
485instance Monad m => Monad (CostreamT m) where
476 return = pure
486 return = pure
477− (CostreamT (Stream m)) >>= f = CostreamT $ Stream $ \_ stp sng yld ->
487+ (CostreamT (S.Stream m)) >>= f = CostreamT $ S.Stream $ \_ stp sng yld ->
478 let run x = (S.runStream x) Nothing stp sng yld
488 let run x = (S.runStream x) Nothing stp sng yld
479 single a = run $ toStream (f a)
489 single a = run $ toStream (f a)
480 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
490 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
@@ -530,7 +540,7 @@ MONAD_COMMON_INSTANCES(CostreamT,)
530-- Note that this composition can be used to combine infinite number of streams
540-- Note that this composition can be used to combine infinite number of streams
531-- as it explores only a bounded number of streams at a time.
541-- as it explores only a bounded number of streams at a time.
532--
542--
533−newtype CoparallelT m a = CoparallelT {getCoparallelT :: Stream m a}
543+newtype CoparallelT m a = CoparallelT {getCoparallelT :: S.Stream m a}
534 deriving (Functor, MonadTrans)
544 deriving (Functor, MonadTrans)
535
545
536{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
546{-# DEPRECATED AsyncT "Please use 'CoparallelT' instead." #-}
@@ -574,14 +584,14 @@ instance MonadParallel m => Monoid (CoparallelT m a) where
574
584
575{-# INLINE parbind #-}
585{-# INLINE parbind #-}
576parbind
586parbind
577− :: (forall c. Stream m c -> Stream m c -> Stream m c)
587+ :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
578− -> Stream m a
588+ -> S.Stream m a
579− -> (a -> Stream m b)
589+ -> (a -> S.Stream m b)
580− -> Stream m b
590+ -> S.Stream m b
581parbind par m f = go m
591parbind par m f = go m
582 where
592 where
583− go (Stream g) =
593+ go (S.Stream g) =
584− Stream $ \ctx stp sng yld ->
594+ S.Stream $ \ctx stp sng yld ->
585 let run x = (S.runStream x) ctx stp sng yld
595 let run x = (S.runStream x) ctx stp sng yld
586 single a = run $ f a
596 single a = run $ f a
587 yield a r = run $ f a `par` go r
597 yield a r = run $ f a `par` go r
@@ -640,7 +650,7 @@ MONAD_COMMON_INSTANCES(CoparallelT, MONADPARALLEL)
640-- Note that round robin composition can only combine a finite number of
650-- Note that round robin composition can only combine a finite number of
641-- streams as it needs to retain state for each unfinished stream.
651-- streams as it needs to retain state for each unfinished stream.
642--
652--
643−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
653+newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
644 deriving (Functor, MonadTrans)
654 deriving (Functor, MonadTrans)
645
655
646instance IsStream ParallelT where
656instance IsStream ParallelT where
@@ -692,7 +702,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
692zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
702zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
693zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
703zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
694 where
704 where
695− go mx my = Stream $ \_ stp sng yld -> do
705+ go mx my = S.Stream $ \_ stp sng yld -> do
696 let merge a ra =
706 let merge a ra =
697 let single2 b = sng (f a b)
707 let single2 b = sng (f a b)
698 yield2 b rb = yld (f a b) (go ra rb)
708 yield2 b rb = yld (f a b) (go ra rb)
@@ -718,7 +728,7 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
718-- The 'Semigroup' instance of this type works the same way as that of
728-- The 'Semigroup' instance of this type works the same way as that of
719-- 'StreamT'.
729-- 'StreamT'.
720--
730--
721−newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: Stream m a}
731+newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}
722 deriving (Functor, Semigroup, Monoid)
732 deriving (Functor, Semigroup, Monoid)
723
733
724instance Monad m => Applicative (ZipStreamM m) where
734instance Monad m => Applicative (ZipStreamM m) where
@@ -737,7 +747,7 @@ instance IsStream ZipStreamM where
737-- generated concurrently) using a pure zipping function.
747-- generated concurrently) using a pure zipping function.
738zipParallelWith :: (IsStream t, MonadParallel m)
748zipParallelWith :: (IsStream t, MonadParallel m)
739 => (a -> b -> c) -> t m a -> t m b -> t m c
749 => (a -> b -> c) -> t m a -> t m b -> t m c
740−zipParallelWith f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do
750+zipParallelWith f m1 m2 = fromStream $ S.Stream $ \_ stp sng yld -> do
741 ma <- async m1
751 ma <- async m1
742 mb <- async m2
752 mb <- async m2
743 (S.runStream (toStream (zipWith f ma mb))) Nothing stp sng yld
753 (S.runStream (toStream (zipWith f ma mb))) Nothing stp sng yld
@@ -763,7 +773,7 @@ zipAsyncWith = zipParallelWith
763-- The 'Semigroup' instance of this type works the same way as that of
773-- The 'Semigroup' instance of this type works the same way as that of
764-- 'StreamT'.
774-- 'StreamT'.
765--
775--
766−newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: Stream m a}
776+newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: S.Stream m a}
767 deriving (Functor, Semigroup, Monoid)
777 deriving (Functor, Semigroup, Monoid)
768
778
769{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}
779{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}
@@ -863,6 +873,41 @@ runZipStream = runStream . zipStreamly
863runZipAsync :: Monad m => ZipParallelM m a -> m ()
873runZipAsync :: Monad m => ZipParallelM m a -> m ()
864runZipAsync = runStream . zipParallely
874runZipAsync = runStream . zipParallely
865
875
876+------------------------------------------------------------------------------
877+-- IO Streams
878+------------------------------------------------------------------------------
879+
880+-- | A serial IO stream of elements of type @a@. See 'StreamT' documentation
881+-- for more details.
882+type Stream a = StreamT IO a
883+
884+-- | An interleaving serial IO stream of elements of type @a@. See 'CostreamT'
885+-- documentation for more details.
886+type Costream a = CostreamT IO a
887+
888+-- | A demand driven left biased parallelly composing IO stream of elements of
889+-- type @a@. See 'CoparallelT' documentation for more details.
890+type Coparallel a = CoparallelT IO a
891+
892+-- | A round robin parallelly composing IO stream of elements of type @a@.
893+-- See 'ParallelT' documentation for more details.
894+type Parallel a = ParallelT IO a
895+
896+-- | An IO stream whose applicative instance zips streams serially.
897+type ZipStream a = ZipStreamM IO a
898+
899+-- | An IO stream whose applicative instance zips streams parallely.
900+type ZipParallel a = ZipParallelM IO a
901+
902+-- | Lift a value from IO monad to any of the stream types. This works for all
903+-- stream types including those that are not monads e.g. 'ZipStream' and
904+-- 'ZipParallel'.
905+fromIO :: (IsStream t, MonadIO m) => IO a -> t m a
906+fromIO = streamly . liftIO
907+
908+fromM :: (IsStream t, Monad m) => m a -> t m a
909+fromM = streamly . lift
910+
866------------------------------------------------------------------------------
911------------------------------------------------------------------------------
867-- Fold Utilities
912-- Fold Utilities
868------------------------------------------------------------------------------
913------------------------------------------------------------------------------