@@ -17,27 +17,43 @@
1717--
1818--
1919module Asyncly.Prelude
20− ( Streaming (..)
21− , StreamT
20+ (
21+ -- * Product Style Composition
22+ StreamT
2223 , InterleavedT
2324 , AsyncT
2425 , ParallelT
2526 , ZipStream
2627 , ZipAsync
28+
29+ -- * Stream Type Adapters
2730 , serially
28− , toListSerial
29− , interleaved
30− , toListInterleaved
31+ , interleaving
3132 , asyncly
32− , toListAsync
3333 , parallely
34− , toListParallel
34+ , zipping
35+ , zippingAsync
36+ , adapt
37+
38+ -- * Construction
39+ , unfoldr
40+
41+ -- * Elimination
42+ , uncons
43+
44+ -- * Special folds
45+ , take
46+ , drop
47+
48+ -- * Zipping
49+ , zipWith
50+ , zipWithM
51+ , zipAsyncWith
52+ , zipAsyncWithM
3553 )
3654where
3755
38−import Control.Applicative (Alternative (..), liftA2)
39−import Control.Arrow (second)
40−import Data.Semigroup (Semigroup(..), cycle1)
56+import Control.Applicative (Alternative (..), liftA2)
4157import Control.Monad (MonadPlus(..))
4258import Control.Monad.Base (MonadBase (..))
4359import Control.Monad.Catch (MonadThrow)
@@ -46,11 +62,12 @@ import Control.Monad.IO.Class (MonadIO(..))
4662import Control.Monad.Reader.Class (MonadReader(..))
4763import Control.Monad.State.Class (MonadState(..))
4864import Control.Monad.Trans.Class (MonadTrans)
49−import Prelude hiding (drop, take, zipWith)
50−import Asyncly.AsyncT
65+import Data.Semigroup (Semigroup(..), cycle1)
66+import Prelude hiding (drop, take, zipWith)
67+import Asyncly.AsyncT
5168
5269------------------------------------------------------------------------------
53−-- Streams with different styles of product composition
70+-- StreamT iterates serially in the Monad and Applicative compositions
5471------------------------------------------------------------------------------
5572
5673newtype StreamT m a = StreamT {getStreamT :: Stream m a}
@@ -64,6 +81,15 @@ deriving instance MonadError e m => MonadError e (StreamT m)
6481deriving instance MonadReader r m => MonadReader r (StreamT m)
6582deriving instance MonadState s m => MonadState s (StreamT m)
6683
84+instance Streaming StreamT where
85+ toStream = getStreamT
86+ fromStream = StreamT
87+
88+------------------------------------------------------------------------------
89+-- InterleavedT interleaves the iterations in the Monad and Applicative
90+-- compositions. It does not make much sense for infinite streams.
91+------------------------------------------------------------------------------
92+
6793-- XXX the applicative, functor and num instances may not be correct
6894newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
6995 deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
@@ -76,6 +102,27 @@ deriving instance MonadError e m => MonadError e (InterleavedT m)
76102deriving instance MonadReader r m => MonadReader r (InterleavedT m)
77103deriving instance MonadState s m => MonadState s (InterleavedT m)
78104
105+instance Streaming InterleavedT where
106+ toStream = getInterleavedT
107+ fromStream = InterleavedT
108+
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)
116+ `interleave`
117+ getInterleavedT ((InterleavedT r) >>= f)
118+ in m Nothing stp yield
119+
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+------------------------------------------------------------------------------
125+
79126newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
80127 deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
81128 Floating, MonadTrans)
@@ -89,183 +136,9 @@ deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)
89136deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
90137deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
91138
92−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
93− deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
94− Floating, MonadTrans)
95−
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)
104−
105−infixr 5 <=>
106−
107−class Streaming t where
108− -- sum style operations
109−
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
114−
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
122−
123− -- Conversions
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
130−
131− -- * Elimination
132−
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))
137−
138− -- * Construction
139− unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> t m a
140−
141− -- * Special folds
142− take :: MonadAsync m => Int -> t m a -> t m a
143− drop :: MonadAsync m => Int -> t m a -> t m a
144−
145− -- * Zipping
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
149−
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
154−
155−instance Streaming StreamT where
156− {-# INLINE (<=>) #-}
157− m1 <=> m2 = StreamT $ interleave (getStreamT m1) (getStreamT m2)
158−
159− {-# INLINE (<|) #-}
160− m1 <| m2 = StreamT $ parLeft (getStreamT m1) (getStreamT m2)
161−
162− fromStream = id
163− fromInterleaved = StreamT . getInterleavedT
164− fromAsync = StreamT . getAsyncT
165− fromParallel = StreamT . getParallelT
166− fromZipStream = StreamT . getZipStream
167− fromZipAsync = StreamT . getZipAsync
168−
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)
179−
180−instance Streaming InterleavedT where
181− {-# INLINE (<=>) #-}
182− m1 <=> m2 = InterleavedT $ interleave (getInterleavedT m1) (getInterleavedT m2)
183−
184− {-# INLINE (<|) #-}
185− m1 <| m2 = InterleavedT $ parLeft (getInterleavedT m1) (getInterleavedT m2)
186−
187− fromStream = InterleavedT . getStreamT
188− fromInterleaved = id
189− fromAsync = InterleavedT . getAsyncT
190− fromParallel = InterleavedT . getParallelT
191− fromZipStream = InterleavedT . getZipStream
192− fromZipAsync = InterleavedT . getZipAsync
193−
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)
204−
205139instance Streaming AsyncT where
206− {-# INLINE (<=>) #-}
207− m1 <=> m2 = AsyncT $ interleave (getAsyncT m1) (getAsyncT m2)
208−
209− {-# INLINE (<|) #-}
210− m1 <| m2 = AsyncT $ parLeft (getAsyncT m1) (getAsyncT m2)
211−
212− fromStream = AsyncT . getStreamT
213− fromInterleaved = AsyncT . getInterleavedT
214− fromAsync = id
215− fromParallel = AsyncT . getParallelT
216− fromZipStream = AsyncT . getZipStream
217− fromZipAsync = AsyncT . getZipAsync
218−
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)
229−
230−instance Streaming ParallelT where
231− {-# INLINE (<=>) #-}
232− m1 <=> m2 = ParallelT $ interleave (getParallelT m1) (getParallelT m2)
233−
234− {-# INLINE (<|) #-}
235− m1 <| m2 = ParallelT $ parLeft (getParallelT m1) (getParallelT m2)
236−
237− fromStream = ParallelT . getStreamT
238− fromInterleaved = ParallelT . getInterleavedT
239− fromAsync = ParallelT . getAsyncT
240− fromParallel = id
241− fromZipStream = ParallelT . getZipStream
242− fromZipAsync = ParallelT . getZipAsync
243−
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)
254−
255−------------------------------------------------------------------------------
256−-- Alternative ways to bind
257−------------------------------------------------------------------------------
258−
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)
266− `interleave`
267− getInterleavedT ((InterleavedT r) >>= f)
268− in m Nothing stp yield
140+ toStream = getAsyncT
141+ fromStream = AsyncT
269142
270143{-# INLINE parbind #-}
271144parbind
@@ -290,6 +163,29 @@ instance MonadAsync m => Monad (AsyncT m) where
290163 where g x = getAsyncT (f x)
291164 par = parallel (CtxType Conjunction LIFO)
292165
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+------------------------------------------------------------------------------
171+
172+newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
173+ deriving (Functor, Applicative, Semigroup, Monoid, Num, Fractional,
174+ Floating, MonadTrans)
175+
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)
184+
185+instance Streaming ParallelT where
186+ toStream = getParallelT
187+ fromStream = ParallelT
188+
293189instance MonadAsync m => Monad (ParallelT m) where
294190 -- | Execute a monadic action for each element in the stream, running
295191 -- iterations in a fairly parallel manner, i.e. all iterations are equally
@@ -298,107 +194,37 @@ instance MonadAsync m => Monad (ParallelT m) where
298194 where g x = getParallelT (f x)
299195 par = parallel (CtxType Conjunction FIFO)
300196
301−serially :: Monad m => StreamT m a -> m ()
302−serially = runStreaming . getStreamT
303−
304−toListSerial :: Monad m => StreamT m a -> m [a]
305−toListSerial = toList . getStreamT
306−
307−interleaved :: Monad m => InterleavedT m a -> m ()
308−interleaved = runStreaming . getInterleavedT
309−
310−toListInterleaved :: Monad m => InterleavedT m a -> m [a]
311−toListInterleaved = toList . getInterleavedT
312−
313−asyncly :: Monad m => AsyncT m a -> m ()
314−asyncly = runStreaming . getAsyncT
315−
316−toListAsync :: Monad m => AsyncT m a -> m [a]
317−toListAsync = toList . getAsyncT
318−
319−parallely :: Monad m => ParallelT m a -> m ()
320−parallely = runStreaming . getParallelT
321−
322−toListParallel :: Monad m => ParallelT m a -> m [a]
323−toListParallel = toList . getParallelT
324−
325−------------------------------------------------------------------------------
326−-- Construct and deconstruct
327−------------------------------------------------------------------------------
328−
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
334−
335− where
336−
337− stop = return Nothing
338−
339− {-# INLINE yield #-}
340− yield a Nothing = return (Just (a, empty))
341− yield a (Just x) = return (Just (a, x))
342−
343−-- | Build a Stream by unfolding steps starting from a seed.
344−unfold :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> Stream m a
345−unfold step = go
346− where
347− go s = Stream $ \_ stp yld -> do
348− mayb <- step s
349− case mayb of
350− Nothing -> stp
351− Just (a, b) -> yld a (Just (go b))
352−
353−------------------------------------------------------------------------------
354−-- Special folds
355−------------------------------------------------------------------------------
356−
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))
362− if (n <= 0)
363− then stp
364− else (runStream m) ctx stp yield
365−
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
371− if (n <= 0)
372− then (runStream m) ctx stp yld
373− else (runStream m) ctx stp yield
374−
375197------------------------------------------------------------------------------
376198-- Serially Zipping Streams
377199------------------------------------------------------------------------------
378200
379201-- | Zip two AsyncT streams serially using a monadic zipping function.
380−zippingWithM :: Monad m =>
381− (a -> b -> Stream m c) -> Stream m a -> Stream m b -> Stream m c
382−zippingWithM f m1 m2 = Stream $ \_ stp yld -> do
383− let merge a ra =
384− let yield2 b Nothing = (runStream (f a b)) Nothing stp yld
385− yield2 b (Just rb) =
386− (runStream ((f a b) <> (zippingWithM f ra rb))) Nothing stp yld
387− in (runStream m2) Nothing stp yield2
388− let yield1 a Nothing = merge a mempty
389− yield1 a (Just ra) = merge a ra
390− (runStream m1) Nothing stp yield1
202+zipWithM :: Streaming t => (a -> b -> t m c) -> t m a -> t m b -> t m c
203+zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
204+ where
205+ go mx my = Stream $ \_ stp yld -> do
206+ let merge a ra =
207+ let yield2 b Nothing = (runStream (g a b)) Nothing stp yld
208+ yield2 b (Just rb) =
209+ (runStream ((g a b) <> (go ra rb))) Nothing stp yld
210+ in (runStream my) Nothing stp yield2
211+ let yield1 a Nothing = merge a mempty
212+ yield1 a (Just ra) = merge a ra
213+ (runStream mx) Nothing stp yield1
214+ g a b = toStream $ f a b
391215
392216-- | 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
394−zippingWith f m1 m2 = Stream $ \_ stp yld -> do
395− let merge a ra =
396− let yield2 b Nothing = yld (f a b) Nothing
397− yield2 b (Just rb) = yld (f a b) (Just (zippingWith f ra rb))
398− in (runStream m2) Nothing stp yield2
399− let yield1 a Nothing = merge a mempty
400− yield1 a (Just ra) = merge a ra
401− (runStream m1) Nothing stp yield1
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)
219+ where
220+ go mx my = Stream $ \_ stp yld -> do
221+ let merge a ra =
222+ let yield2 b Nothing = yld (f a b) Nothing
223+ yield2 b (Just rb) = yld (f a b) (Just (go ra rb))
224+ in (runStream my) Nothing stp yield2
225+ let yield1 a Nothing = merge a mempty
226+ yield1 a (Just ra) = merge a ra
227+ (runStream mx) Nothing stp yield1
402228
403229-- | Wrapper around AsyncT type with a serial zipping Applicative instance.
404230-- Note that the binary function interleave (\<=>) is a special case of
@@ -410,32 +236,11 @@ newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
410236
411237instance Monad m => Applicative (ZipStream m) where
412238 pure a = ZipStream $ cycle1 (pure a)
413− (ZipStream xs) <*> (ZipStream ys) = ZipStream (zippingWith id xs ys)
239+ (<*>) = zipWith id
414240
415241instance Streaming ZipStream where
416− {-# INLINE (<=>) #-}
417− m1 <=> m2 = ZipStream $ interleave (getZipStream m1) (getZipStream m2)
418−
419− {-# INLINE (<|) #-}
420− m1 <| m2 = ZipStream $ parLeft (getZipStream m1) (getZipStream m2)
421−
422− fromStream = ZipStream . getStreamT
423− fromInterleaved = ZipStream . getInterleavedT
424− fromAsync = ZipStream . getAsyncT
425− fromParallel = ZipStream . getParallelT
426− fromZipStream = ZipStream . getZipStream
427− fromZipAsync = ZipStream . getZipAsync
428−
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)
242+ toStream = getZipStream
243+ fromStream = ZipStream
439244
440245instance (Monad m, Num a) => Num (ZipStream m a) where
441246 fromInteger n = pure (fromInteger n)
@@ -483,21 +288,21 @@ instance (Monad m, Floating a) => Floating (ZipStream m a) where
483288
484289-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
485290-- concurrently) using a monadic zipping function.
486−zippingAsyncWithM :: MonadAsync m
487− => (a -> b -> Stream m c) -> Stream m a -> Stream m b -> Stream m c
488−zippingAsyncWithM f m1 m2 = Stream $ \_ stp yld -> do
291+zipAsyncWithM :: (Streaming t, MonadAsync m)
292+ => (a -> b -> t m c) -> t m a -> t m b -> t m c
293+zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
489294 ma <- async m1
490295 mb <- async m2
491− (runStream (zippingWithM f ma mb)) Nothing stp yld
296+ (runStream (toStream (zipWithM f ma mb))) Nothing stp yld
492297
493298-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
494299-- concurrently) using a pure zipping function.
495−zippingAsyncWith :: MonadAsync m
496− => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
497−zippingAsyncWith f m1 m2 = Stream $ \_ stp yld -> do
300+zipAsyncWith :: (Streaming t, MonadAsync m)
301+ => (a -> b -> c) -> t m a -> t m b -> t m c
302+zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
498303 ma <- async m1
499304 mb <- async m2
500− (runStream (zippingWith f ma mb)) Nothing stp yld
305+ (runStream (toStream (zipWith f ma mb))) Nothing stp yld
501306
502307-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.
503308-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT
@@ -509,32 +314,11 @@ newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
509314
510315instance MonadAsync m => Applicative (ZipAsync m) where
511316 pure a = ZipAsync $ cycle1 (pure a)
512− (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zippingAsyncWith id xs ys)
317+ (<*>) = zipAsyncWith id
513318
514319instance Streaming ZipAsync where
515− {-# INLINE (<=>) #-}
516− m1 <=> m2 = ZipAsync $ interleave (getZipAsync m1) (getZipAsync m2)
517−
518− {-# INLINE (<|) #-}
519− m1 <| m2 = ZipAsync $ parLeft (getZipAsync m1) (getZipAsync m2)
520−
521− fromStream = ZipAsync . getStreamT
522− fromInterleaved = ZipAsync . getInterleavedT
523− fromAsync = ZipAsync . getAsyncT
524− fromParallel = ZipAsync . getParallelT
525− fromZipStream = ZipAsync . getZipStream
526− fromZipAsync = ZipAsync . getZipAsync
527−
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)
320+ toStream = getZipAsync
321+ fromStream = ZipAsync
538322
539323instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
540324 fromInteger n = pure (fromInteger n)
@@ -575,3 +359,88 @@ instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
575359
576360 (**) = liftA2 (**)
577361 logBase = liftA2 logBase
362+
363+-------------------------------------------------------------------------------
364+-- Type adapting combinators
365+-------------------------------------------------------------------------------
366+
367+-- | Adapt one streaming type to another.
368+adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a
369+adapt = fromStream . toStream
370+
371+-- | Interpret an ambiguously typed stream as 'StreamT'.
372+serially :: StreamT m a -> StreamT m a
373+serially x = x
374+
375+-- | Interpret an ambiguously typed stream as 'InterleavedT'.
376+interleaving :: InterleavedT m a -> InterleavedT m a
377+interleaving x = x
378+
379+-- | Interpret an ambiguously typed stream as 'AsyncT'.
380+asyncly :: AsyncT m a -> AsyncT m a
381+asyncly x = x
382+
383+-- | Interpret an ambiguously typed stream as 'ParallelT'.
384+parallely :: ParallelT m a -> ParallelT m a
385+parallely x = x
386+
387+-- | Interpret an ambiguously typed stream as 'ZipStream'.
388+zipping :: ZipStream m a -> ZipStream m a
389+zipping x = x
390+
391+-- | Interpret an ambiguously typed stream as 'ZipAsync'.
392+zippingAsync :: ZipAsync m a -> ZipAsync m a
393+zippingAsync x = x
394+
395+------------------------------------------------------------------------------
396+-- Construct and deconstruct
397+------------------------------------------------------------------------------
398+
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
405+ where
406+ stop = return Nothing
407+
408+ {-# INLINE yield #-}
409+ yield a Nothing = return (Just (a, mempty))
410+ yield a (Just x) = return (Just (a, (fromStream x)))
411+
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
415+ where
416+ go s = Stream $ \_ stp yld -> do
417+ mayb <- step s
418+ case mayb of
419+ Nothing -> stp
420+ Just (a, b) -> yld a (Just (go b))
421+
422+------------------------------------------------------------------------------
423+-- Special folds
424+------------------------------------------------------------------------------
425+
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)
429+ where
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))
433+ if (n1 <= 0)
434+ then stp
435+ else (runStream m1) ctx stp yield
436+
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)
440+ where
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
444+ if (n1 <= 0)
445+ then (runStream m1) ctx stp yld
446+ else (runStream m1) ctx stp yield