@@ -17,27 +17,43 @@
19module Asyncly.Prelude
19module Asyncly.Prelude
20− ( Streaming (..)
20+ (
21+ -- * Product Style Composition
22 , InterleavedT
23 , InterleavedT
24 , ParallelT
25 , ParallelT
25 , ZipStream
26 , ZipStream
26 , ZipAsync
27 , ZipAsync
29+ -- * Stream Type Adapters
27 , serially
30 , serially
28− , toListSerial
31+ , interleaving
33 , parallely
33 , parallely
34− , toListParallel
34+ , zipping
38−import Control.Applicative (Alternative (..), liftA2)
56+import Control.Applicative (Alternative (..), liftA2)
39−import Control.Arrow (second)
40−import Data.Semigroup (Semigroup(..), cycle1)
41import Control.Monad (MonadPlus(..))
57import Control.Monad (MonadPlus(..))
42import Control.Monad.Base (MonadBase (..))
58import Control.Monad.Base (MonadBase (..))
43import Control.Monad.Catch (MonadThrow)
59import Control.Monad.Catch (MonadThrow)
@@ -46,11 +62,12 @@ import Control.Monad.IO.Class (MonadIO(..))
46import Control.Monad.Reader.Class (MonadReader(..))
62import Control.Monad.Reader.Class (MonadReader(..))
47import Control.Monad.State.Class (MonadState(..))
63import Control.Monad.State.Class (MonadState(..))
48import Control.Monad.Trans.Class (MonadTrans)
64import Control.Monad.Trans.Class (MonadTrans)
65+import Data.Semigroup (Semigroup(..), cycle1)
49−import Prelude hiding (drop, take, zipWith)
66+import Prelude hiding (drop, take, zipWith)
50−import Asyncly.AsyncT
67+import Asyncly.AsyncT
52------------------------------------------------------------------------------
69------------------------------------------------------------------------------
53−-- Streams with different styles of product composition
70+-- StreamT iterates serially in the Monad and Applicative compositions
54------------------------------------------------------------------------------
71------------------------------------------------------------------------------
56newtype StreamT m a = StreamT {getStreamT :: Stream m a}
73newtype StreamT m a = StreamT {getStreamT :: Stream m a}
@@ -64,6 +81,15 @@ deriving instance MonadError e m => MonadError e (StreamT m)
64deriving instance MonadReader r m => MonadReader r (StreamT m)
81deriving instance MonadReader r m => MonadReader r (StreamT m)
65deriving instance MonadState s m => MonadState s (StreamT m)
82deriving instance MonadState s m => MonadState s (StreamT m)
84+instance Streaming StreamT where
85+ toStream = getStreamT
88+------------------------------------------------------------------------------
89+-- InterleavedT interleaves the iterations in the Monad and Applicative
90+-- compositions. It does not make much sense for infinite streams.
91+------------------------------------------------------------------------------
67-- XXX the applicative, functor and num instances may not be correct
93-- XXX the applicative, functor and num instances may not be correct
68newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
94newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
69 deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
95 deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
@@ -76,6 +102,27 @@ deriving instance MonadError e m => MonadError e (InterleavedT m)
76deriving instance MonadReader r m => MonadReader r (InterleavedT m)
102deriving instance MonadReader r m => MonadReader r (InterleavedT m)
77deriving instance MonadState s m => MonadState s (InterleavedT m)
103deriving instance MonadState s m => MonadState s (InterleavedT m)
105+instance Streaming InterleavedT where
106+ toStream = getInterleavedT
107+ fromStream = InterleavedT
109+instance Monad m => Monad (InterleavedT m) where
110+ -- | Execute a monadic action for each element in the stream, in a fairly
111+ -- interleaved manner i.e. iterations yield alternately.
112+ (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
113+ let run x = (runStream x) Nothing stp yld
114+ yield a Nothing = run $ getInterleavedT (f a)
115+ yield a (Just r) = run $ getInterleavedT (f a)
117+ getInterleavedT ((InterleavedT r) >>= f)
118+ in m Nothing stp yield
120+------------------------------------------------------------------------------
121+-- AsyncT runs the iterations asynchronously in the Monad and Applicative
122+-- compositions. More iterations are started in parallel on demand. It can work
123+-- with infinite streams.
124+------------------------------------------------------------------------------
79newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
126newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
80 deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
127 deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
81 Floating, MonadTrans)
128 Floating, MonadTrans)
@@ -89,183 +136,9 @@ deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)
89deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
136deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
90deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
137deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
92−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
93− deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
94− Floating, MonadTrans)
96−deriving instance MonadAsync m => MonadIO (ParallelT m)
97−deriving instance MonadAsync m => MonadThrow (ParallelT m)
98−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
99−deriving instance MonadAsync m => Alternative (ParallelT m)
100−deriving instance MonadAsync m => MonadPlus (ParallelT m)
101−deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
102−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
103−deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
107−class Streaming t where
108− -- sum style operations
110− -- | Sequential interleaved composition, similar to '<>' except that it
111− -- fairly interleaves the two 'AsyncT' streams, yielding an element from
112− -- each stream alternately. Should be used only on finite streams.
113− (<=>) :: t m a -> t m a -> t m a
115− -- | Parallel composition similar to '<|>' except that it is left-biased
116− -- instead of being fair. Action on the left is likely to be given a
117− -- chance to execute before the action on the right. If the left action
118− -- keeps yielding results without blocking it continues running until it
119− -- finishes and only then the right action runs. Unlike '<|>' it can be
120− -- used on infinite streams.
121− (<|) :: MonadAsync m => t m a -> t m a -> t m a
124− fromStream :: StreamT m a -> t m a
125− fromInterleaved :: InterleavedT m a -> t m a
126− fromAsync :: AsyncT m a -> t m a
127− fromParallel :: ParallelT m a -> t m a
128− fromZipStream :: ZipStream m a -> t m a
129− fromZipAsync :: ZipAsync m a -> t m a
133− -- | Decompose a stream into its head and tail. If the stream is empty,
134− -- returns 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)',
135− -- where 'a' is the head of the stream and 'ma' its tail.
136− uncons :: MonadAsync m => t m a -> m (Maybe (a, t m a))
139− unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> t m a
142− take :: MonadAsync m => Int -> t m a -> t m a
143− drop :: MonadAsync m => Int -> t m a -> t m a
146− zipWith :: Monad m => (a -> b -> c) -> t m a -> t m b -> t m c
147− zipWithM :: Monad m =>
148− (a -> b -> t m c) -> t m a -> t m b -> t m c
150− zipAsyncWithM :: MonadAsync m
151− => (a -> b -> t m c) -> t m a -> t m b -> t m c
152− zipAsyncWith :: MonadAsync m
153− => (a -> b -> c) -> t m a -> t m b -> t m c
155−instance Streaming StreamT where
156− {-# INLINE (<=>) #-}
157− m1 <=> m2 = StreamT $ interleave (getStreamT m1) (getStreamT m2)
160− m1 <| m2 = StreamT $ parLeft (getStreamT m1) (getStreamT m2)
163− fromInterleaved = StreamT . getInterleavedT
164− fromAsync = StreamT . getAsyncT
165− fromParallel = StreamT . getParallelT
166− fromZipStream = StreamT . getZipStream
167− fromZipAsync = StreamT . getZipAsync
169− uncons (StreamT s) = fmap (fmap (second StreamT)) (next s)
170− unfoldr step seed = StreamT $ unfold step seed
171− take n (StreamT s) = StreamT $ taking n s
172− drop n (StreamT s) = StreamT $ dropping n s
173− zipWith f (StreamT s1) (StreamT s2) = StreamT $ zippingWith f s1 s2
174− zipWithM f (StreamT s1) (StreamT s2) = StreamT $ zippingWithM g s1 s2
175− where g a b = getStreamT (f a b)
176− zipAsyncWith f (StreamT s1) (StreamT s2) = StreamT $ zippingAsyncWith f s1 s2
177− zipAsyncWithM f (StreamT s1) (StreamT s2) = StreamT $ zippingAsyncWithM g s1 s2
178− where g a b = getStreamT (f a b)
180−instance Streaming InterleavedT where
181− {-# INLINE (<=>) #-}
182− m1 <=> m2 = InterleavedT $ interleave (getInterleavedT m1) (getInterleavedT m2)
185− m1 <| m2 = InterleavedT $ parLeft (getInterleavedT m1) (getInterleavedT m2)
187− fromStream = InterleavedT . getStreamT
188− fromInterleaved = id
189− fromAsync = InterleavedT . getAsyncT
190− fromParallel = InterleavedT . getParallelT
191− fromZipStream = InterleavedT . getZipStream
192− fromZipAsync = InterleavedT . getZipAsync
194− uncons (InterleavedT s) = fmap (fmap (second InterleavedT)) (next s)
195− unfoldr step seed = InterleavedT $ unfold step seed
196− take n (InterleavedT s) = InterleavedT $ taking n s
197− drop n (InterleavedT s) = InterleavedT $ dropping n s
198− zipWith f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingWith f s1 s2
199− zipWithM f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingWithM g s1 s2
200− where g a b = getInterleavedT (f a b)
201− zipAsyncWith f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingAsyncWith f s1 s2
202− zipAsyncWithM f (InterleavedT s1) (InterleavedT s2) = InterleavedT $ zippingAsyncWithM g s1 s2
203− where g a b = getInterleavedT (f a b)
205instance Streaming AsyncT where
139instance Streaming AsyncT where
206− {-# INLINE (<=>) #-}
140+ toStream = getAsyncT
207− m1 <=> m2 = AsyncT $ interleave (getAsyncT m1) (getAsyncT m2)
210− m1 <| m2 = AsyncT $ parLeft (getAsyncT m1) (getAsyncT m2)
212− fromStream = AsyncT . getStreamT
141+ fromStream = AsyncT
213− fromInterleaved = AsyncT . getInterleavedT
215− fromParallel = AsyncT . getParallelT
216− fromZipStream = AsyncT . getZipStream
217− fromZipAsync = AsyncT . getZipAsync
219− uncons (AsyncT s) = fmap (fmap (second AsyncT)) (next s)
220− unfoldr step seed = AsyncT $ unfold step seed
221− take n (AsyncT s) = AsyncT $ taking n s
222− drop n (AsyncT s) = AsyncT $ dropping n s
223− zipWith f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingWith f s1 s2
224− zipWithM f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingWithM g s1 s2
225− where g a b = getAsyncT (f a b)
226− zipAsyncWith f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingAsyncWith f s1 s2
227− zipAsyncWithM f (AsyncT s1) (AsyncT s2) = AsyncT $ zippingAsyncWithM g s1 s2
228− where g a b = getAsyncT (f a b)
230−instance Streaming ParallelT where
231− {-# INLINE (<=>) #-}
232− m1 <=> m2 = ParallelT $ interleave (getParallelT m1) (getParallelT m2)
235− m1 <| m2 = ParallelT $ parLeft (getParallelT m1) (getParallelT m2)
237− fromStream = ParallelT . getStreamT
238− fromInterleaved = ParallelT . getInterleavedT
239− fromAsync = ParallelT . getAsyncT
241− fromZipStream = ParallelT . getZipStream
242− fromZipAsync = ParallelT . getZipAsync
244− uncons (ParallelT s) = fmap (fmap (second ParallelT)) (next s)
245− unfoldr step seed = ParallelT $ unfold step seed
246− take n (ParallelT s) = ParallelT $ taking n s
247− drop n (ParallelT s) = ParallelT $ dropping n s
248− zipWith f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingWith f s1 s2
249− zipWithM f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingWithM g s1 s2
250− where g a b = getParallelT (f a b)
251− zipAsyncWith f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingAsyncWith f s1 s2
252− zipAsyncWithM f (ParallelT s1) (ParallelT s2) = ParallelT $ zippingAsyncWithM g s1 s2
253− where g a b = getParallelT (f a b)
255−------------------------------------------------------------------------------
256−-- Alternative ways to bind
257−------------------------------------------------------------------------------
259−instance Monad m => Monad (InterleavedT m) where
260− -- | Execute a monadic action for each element in the stream, in a fairly
261− -- interleaved manner i.e. iterations yield alternately.
262− (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
263− let run x = (runStream x) Nothing stp yld
264− yield a Nothing = run $ getInterleavedT (f a)
265− yield a (Just r) = run $ getInterleavedT (f a)
267− getInterleavedT ((InterleavedT r) >>= f)
268− in m Nothing stp yield
270{-# INLINE parbind #-}
143{-# INLINE parbind #-}
@@ -290,6 +163,29 @@ instance MonadAsync m => Monad (AsyncT m) where
290 where g x = getAsyncT (f x)
163 where g x = getAsyncT (f x)
291 par = parallel (CtxType Conjunction LIFO)
164 par = parallel (CtxType Conjunction LIFO)
166+------------------------------------------------------------------------------
167+-- ParallelT runs all iterations in parallel in the Monad and Applicative
168+-- compositions. Iterations are scheduled in a round-robin fashion. It cannot
169+-- work well when the streams being composed are infinite.
170+------------------------------------------------------------------------------
172+newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
173+ deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
174+ Floating, MonadTrans)
176+deriving instance MonadAsync m => MonadIO (ParallelT m)
177+deriving instance MonadAsync m => MonadThrow (ParallelT m)
178+deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
179+deriving instance MonadAsync m => Alternative (ParallelT m)
180+deriving instance MonadAsync m => MonadPlus (ParallelT m)
181+deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
182+deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
183+deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
185+instance Streaming ParallelT where
186+ toStream = getParallelT
187+ fromStream = ParallelT
293instance MonadAsync m => Monad (ParallelT m) where
189instance MonadAsync m => Monad (ParallelT m) where
294 -- | Execute a monadic action for each element in the stream, running
190 -- | Execute a monadic action for each element in the stream, running
295 -- iterations in a fairly parallel manner, i.e. all iterations are equally
191 -- iterations in a fairly parallel manner, i.e. all iterations are equally
@@ -298,107 +194,37 @@ instance MonadAsync m => Monad (ParallelT m) where
298 where g x = getParallelT (f x)
194 where g x = getParallelT (f x)
299 par = parallel (CtxType Conjunction FIFO)
195 par = parallel (CtxType Conjunction FIFO)
301−serially :: Monad m => StreamT m a -> m ()
302−serially = runStreaming . getStreamT
304−toListSerial :: Monad m => StreamT m a -> m [a]
305−toListSerial = toList . getStreamT
307−interleaved :: Monad m => InterleavedT m a -> m ()
308−interleaved = runStreaming . getInterleavedT
310−toListInterleaved :: Monad m => InterleavedT m a -> m [a]
311−toListInterleaved = toList . getInterleavedT
313−asyncly :: Monad m => AsyncT m a -> m ()
314−asyncly = runStreaming . getAsyncT
316−toListAsync :: Monad m => AsyncT m a -> m [a]
317−toListAsync = toList . getAsyncT
319−parallely :: Monad m => ParallelT m a -> m ()
320−parallely = runStreaming . getParallelT
322−toListParallel :: Monad m => ParallelT m a -> m [a]
323−toListParallel = toList . getParallelT
325−------------------------------------------------------------------------------
326−-- Construct and deconstruct
327−------------------------------------------------------------------------------
329−-- | Decompose a stream into its head and tail. If the stream is empty, returns
330−-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is
331−-- the head of the stream and 'ma' its tail.
332−next :: MonadAsync m => Stream m a -> m (Maybe (a, Stream m a))
333−next m = (runStream m) Nothing stop yield
337− stop = return Nothing
339− {-# INLINE yield #-}
340− yield a Nothing = return (Just (a, empty))
341− yield a (Just x) = return (Just (a, x))
343−-- | Build a Stream by unfolding steps starting from a seed.
344−unfold :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> Stream m a
347− go s = Stream $ \_ stp yld -> do
351− Just (a, b) -> yld a (Just (go b))
353−------------------------------------------------------------------------------
355−------------------------------------------------------------------------------
357−-- | Take first 'n' elements from the stream and discard the rest.
358−taking :: MonadAsync m => Int -> Stream m a -> Stream m a
359−taking n m = Stream $ \ctx stp yld -> do
360− let yield a Nothing = yld a Nothing
361− yield a (Just x) = yld a (Just (taking (n - 1) x))
364− else (runStream m) ctx stp yield
366−-- | Discard first 'n' elements from the stream and take the rest.
367−dropping :: MonadAsync m => Int -> Stream m a -> Stream m a
368−dropping n m = Stream $ \ctx stp yld -> do
369− let yield _ Nothing = stp
370− yield _ (Just x) = (runStream $ dropping (n - 1) x) ctx stp yld
372− then (runStream m) ctx stp yld
373− else (runStream m) ctx stp yield
375------------------------------------------------------------------------------
197------------------------------------------------------------------------------
376-- Serially Zipping Streams
198-- Serially Zipping Streams
377------------------------------------------------------------------------------
199------------------------------------------------------------------------------
379-- | Zip two AsyncT streams serially using a monadic zipping function.
201-- | Zip two AsyncT streams serially using a monadic zipping function.
380−zippingWithM :: Monad m =>
202+zipWithM :: Streaming t => (a -> b -> t m c) -> t m a -> t m b -> t m c
381− (a -> b -> Stream m c) -> Stream m a -> Stream m b -> Stream m c
203+zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
382−zippingWithM f m1 m2 = Stream $ \_ stp yld -> do
205+ go mx my = Stream $ \_ stp yld -> do
383− let merge a ra =
206+ let merge a ra =
384− let yield2 b Nothing = (runStream (f a b)) Nothing stp yld
207+ let yield2 b Nothing = (runStream (g a b)) Nothing stp yld
385− yield2 b (Just rb) =
208+ yield2 b (Just rb) =
386− (runStream ((f a b) <> (zippingWithM f ra rb))) Nothing stp yld
209+ (runStream ((g a b) <> (go ra rb))) Nothing stp yld
387− in (runStream m2) Nothing stp yield2
210+ in (runStream my) Nothing stp yield2
388− let yield1 a Nothing = merge a mempty
211+ let yield1 a Nothing = merge a mempty
389− yield1 a (Just ra) = merge a ra
212+ yield1 a (Just ra) = merge a ra
390− (runStream m1) Nothing stp yield1
213+ (runStream mx) Nothing stp yield1
214+ g a b = toStream $ f a b
392-- | Zip two AsyncT streams serially using a pure zipping function.
216-- | Zip two AsyncT streams serially using a pure zipping function.
393−zippingWith :: Monad m => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
217+zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c
218+zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
394−zippingWith f m1 m2 = Stream $ \_ stp yld -> do
220+ go mx my = Stream $ \_ stp yld -> do
395− let merge a ra =
221+ let merge a ra =
396− let yield2 b Nothing = yld (f a b) Nothing
222+ let yield2 b Nothing = yld (f a b) Nothing
397− yield2 b (Just rb) = yld (f a b) (Just (zippingWith f ra rb))
223+ yield2 b (Just rb) = yld (f a b) (Just (go ra rb))
398− in (runStream m2) Nothing stp yield2
224+ in (runStream my) Nothing stp yield2
399− let yield1 a Nothing = merge a mempty
225+ let yield1 a Nothing = merge a mempty
400− yield1 a (Just ra) = merge a ra
226+ yield1 a (Just ra) = merge a ra
401− (runStream m1) Nothing stp yield1
227+ (runStream mx) Nothing stp yield1
403-- | Wrapper around AsyncT type with a serial zipping Applicative instance.
229-- | Wrapper around AsyncT type with a serial zipping Applicative instance.
404-- Note that the binary function interleave (\<=>) is a special case of
230-- Note that the binary function interleave (\<=>) is a special case of
@@ -410,32 +236,11 @@ newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
411instance Monad m => Applicative (ZipStream m) where
237instance Monad m => Applicative (ZipStream m) where
412 pure a = ZipStream $ cycle1 (pure a)
238 pure a = ZipStream $ cycle1 (pure a)
413− (ZipStream xs) <*> (ZipStream ys) = ZipStream (zippingWith id xs ys)
239+ (<*>) = zipWith id
415instance Streaming ZipStream where
241instance Streaming ZipStream where
416− {-# INLINE (<=>) #-}
242+ toStream = getZipStream
417− m1 <=> m2 = ZipStream $ interleave (getZipStream m1) (getZipStream m2)
420− m1 <| m2 = ZipStream $ parLeft (getZipStream m1) (getZipStream m2)
422− fromStream = ZipStream . getStreamT
243+ fromStream = ZipStream
423− fromInterleaved = ZipStream . getInterleavedT
424− fromAsync = ZipStream . getAsyncT
425− fromParallel = ZipStream . getParallelT
426− fromZipStream = ZipStream . getZipStream
427− fromZipAsync = ZipStream . getZipAsync
429− uncons (ZipStream s) = fmap (fmap (second ZipStream)) (next s)
430− unfoldr step seed = ZipStream $ unfold step seed
431− take n (ZipStream s) = ZipStream $ taking n s
432− drop n (ZipStream s) = ZipStream $ dropping n s
433− zipWith f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingWith f s1 s2
434− zipWithM f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingWithM g s1 s2
435− where g a b = getZipStream (f a b)
436− zipAsyncWith f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingAsyncWith f s1 s2
437− zipAsyncWithM f (ZipStream s1) (ZipStream s2) = ZipStream $ zippingAsyncWithM g s1 s2
438− where g a b = getZipStream (f a b)
440instance (Monad m, Num a) => Num (ZipStream m a) where
245instance (Monad m, Num a) => Num (ZipStream m a) where
441 fromInteger n = pure (fromInteger n)
246 fromInteger n = pure (fromInteger n)
@@ -483,21 +288,21 @@ instance (Monad m, Floating a) => Floating (ZipStream m a) where
484-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
289-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
485-- concurrently) using a monadic zipping function.
290-- concurrently) using a monadic zipping function.
486−zippingAsyncWithM :: MonadAsync m
291+zipAsyncWithM :: (Streaming t, MonadAsync m)
487− => (a -> b -> Stream m c) -> Stream m a -> Stream m b -> Stream m c
292+ => (a -> b -> t m c) -> t m a -> t m b -> t m c
488−zippingAsyncWithM f m1 m2 = Stream $ \_ stp yld -> do
293+zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
489 ma <- async m1
294 ma <- async m1
490 mb <- async m2
295 mb <- async m2
491− (runStream (zippingWithM f ma mb)) Nothing stp yld
296+ (runStream (toStream (zipWithM f ma mb))) Nothing stp yld
493-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
298-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
494-- concurrently) using a pure zipping function.
299-- concurrently) using a pure zipping function.
495−zippingAsyncWith :: MonadAsync m
300+zipAsyncWith :: (Streaming t, MonadAsync m)
496− => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
301+ => (a -> b -> c) -> t m a -> t m b -> t m c
497−zippingAsyncWith f m1 m2 = Stream $ \_ stp yld -> do
302+zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
498 ma <- async m1
303 ma <- async m1
499 mb <- async m2
304 mb <- async m2
500− (runStream (zippingWith f ma mb)) Nothing stp yld
305+ (runStream (toStream (zipWith f ma mb))) Nothing stp yld
502-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.
307-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.
503-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT
308-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT
@@ -509,32 +314,11 @@ newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
510instance MonadAsync m => Applicative (ZipAsync m) where
315instance MonadAsync m => Applicative (ZipAsync m) where
511 pure a = ZipAsync $ cycle1 (pure a)
316 pure a = ZipAsync $ cycle1 (pure a)
512− (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zippingAsyncWith id xs ys)
317+ (<*>) = zipAsyncWith id
514instance Streaming ZipAsync where
319instance Streaming ZipAsync where
515− {-# INLINE (<=>) #-}
320+ toStream = getZipAsync
516− m1 <=> m2 = ZipAsync $ interleave (getZipAsync m1) (getZipAsync m2)
519− m1 <| m2 = ZipAsync $ parLeft (getZipAsync m1) (getZipAsync m2)
521− fromStream = ZipAsync . getStreamT
321+ fromStream = ZipAsync
522− fromInterleaved = ZipAsync . getInterleavedT
523− fromAsync = ZipAsync . getAsyncT
524− fromParallel = ZipAsync . getParallelT
525− fromZipStream = ZipAsync . getZipStream
526− fromZipAsync = ZipAsync . getZipAsync
528− uncons (ZipAsync s) = fmap (fmap (second ZipAsync)) (next s)
529− unfoldr step seed = ZipAsync $ unfold step seed
530− take n (ZipAsync s) = ZipAsync $ taking n s
531− drop n (ZipAsync s) = ZipAsync $ dropping n s
532− zipWith f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingWith f s1 s2
533− zipWithM f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingWithM g s1 s2
534− where g a b = getZipAsync (f a b)
535− zipAsyncWith f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingAsyncWith f s1 s2
536− zipAsyncWithM f (ZipAsync s1) (ZipAsync s2) = ZipAsync $ zippingAsyncWithM g s1 s2
537− where g a b = getZipAsync (f a b)
539instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
323instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
540 fromInteger n = pure (fromInteger n)
324 fromInteger n = pure (fromInteger n)
@@ -575,3 +359,88 @@ instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
576 (**) = liftA2 (**)
360 (**) = liftA2 (**)
577 logBase = liftA2 logBase
361 logBase = liftA2 logBase
363+-------------------------------------------------------------------------------
364+-- Type adapting combinators
365+-------------------------------------------------------------------------------
367+-- | Adapt one streaming type to another.
368+adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a
369+adapt = fromStream . toStream
371+-- | Interpret an ambiguously typed stream as 'StreamT'.
372+serially :: StreamT m a -> StreamT m a
375+-- | Interpret an ambiguously typed stream as 'InterleavedT'.
376+interleaving :: InterleavedT m a -> InterleavedT m a
379+-- | Interpret an ambiguously typed stream as 'AsyncT'.
380+asyncly :: AsyncT m a -> AsyncT m a
383+-- | Interpret an ambiguously typed stream as 'ParallelT'.
384+parallely :: ParallelT m a -> ParallelT m a
387+-- | Interpret an ambiguously typed stream as 'ZipStream'.
388+zipping :: ZipStream m a -> ZipStream m a
391+-- | Interpret an ambiguously typed stream as 'ZipAsync'.
392+zippingAsync :: ZipAsync m a -> ZipAsync m a
395+------------------------------------------------------------------------------
396+-- Construct and deconstruct
397+------------------------------------------------------------------------------
399+-- | Decompose a stream into its head and tail. If the stream is empty, returns
400+-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is
401+-- the head of the stream and 'ma' its tail.
402+uncons :: (Streaming t, Monad m, Monoid (t m a))
403+ => t m a -> m (Maybe (a, t m a))
404+uncons m = (runStream (toStream m)) Nothing stop yield
406+ stop = return Nothing
408+ {-# INLINE yield #-}
409+ yield a Nothing = return (Just (a, mempty))
410+ yield a (Just x) = return (Just (a, (fromStream x)))
412+-- | Build a Stream by unfolding steps starting from a seed.
413+unfoldr :: (Streaming t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
414+unfoldr step = fromStream . go
416+ go s = Stream $ \_ stp yld -> do
420+ Just (a, b) -> yld a (Just (go b))
422+------------------------------------------------------------------------------
424+------------------------------------------------------------------------------
426+-- | Take first 'n' elements from the stream and discard the rest.
427+take :: Streaming t => Int -> t m a -> t m a
428+take n m = fromStream $ go n (toStream m)
430+ go n1 m1 = Stream $ \ctx stp yld -> do
431+ let yield a Nothing = yld a Nothing
432+ yield a (Just x) = yld a (Just (go (n1 - 1) x))
435+ else (runStream m1) ctx stp yield
437+-- | Discard first 'n' elements from the stream and take the rest.
438+drop :: Streaming t => Int -> t m a -> t m a
439+drop n m = fromStream $ go n (toStream m)
441+ go n1 m1 = Stream $ \ctx stp yld -> do
442+ let yield _ Nothing = stp
443+ yield _ (Just x) = (runStream $ go (n1 - 1) x) ctx stp yld
445+ then (runStream m1) ctx stp yld
446+ else (runStream m1) ctx stp yield