@@ -0,0 +1,162 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving#-}
3+{-# LANGUAGE UndecidableInstances #-} -- XXX
4+
5+-- |
6+-- Module : Asyncly.Prelude
7+-- Copyright : (c) 2017 Harendra Kumar
8+--
9+-- License : BSD3
10+-- Maintainer : harendra.kumar@gmail.com
11+-- Stability : experimental
12+-- Portability : GHC
13+--
14+--
15+module Asyncly.Prelude
16+ (
17+ -- * Construction
18+ unfoldr
19+
20+ -- * Elimination
21+ , uncons
22+
23+ -- * Special folds
24+ , take
25+ , drop
26+
27+ -- * Zipping
28+ , ZipSerial (..)
29+ , zipWith
30+ , zipWithM
31+
32+ , ZipAsync (..)
33+ , zipAsyncWith
34+ , zipAsyncWithM
35+ )
36+where
37+
38+import Control.Applicative (Alternative (..))
39+import Data.Semigroup (Semigroup(..), cycle1)
40+import Prelude hiding (drop, take, zipWith)
41+import Asyncly.AsyncT
42+
43+------------------------------------------------------------------------------
44+-- Construct and deconstruct
45+------------------------------------------------------------------------------
46+
47+-- | Decompose a stream into its head and tail. If the stream is empty, returns
48+-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is
49+-- the head of the stream and 'ma' its tail.
50+uncons :: MonadAsync m => AsyncT m a -> m (Maybe (a, AsyncT m a))
51+uncons m = (runAsyncT m) Nothing stop yield
52+
53+ where
54+
55+ stop = return Nothing
56+
57+ {-# INLINE yield #-}
58+ yield a Nothing = return (Just (a, empty))
59+ yield a (Just x) = return (Just (a, x))
60+
61+-- | Build a Stream by unfolding steps starting from a seed.
62+unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> AsyncT m a
63+unfoldr step = go
64+ where
65+ go s = AsyncT $ \_ stp yld -> do
66+ mayb <- step s
67+ case mayb of
68+ Nothing -> stp
69+ Just (a, b) -> yld a (Just (go b))
70+
71+------------------------------------------------------------------------------
72+-- Special folds
73+------------------------------------------------------------------------------
74+
75+-- | Take first 'n' elements from the stream and discard the rest.
76+take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
77+take n m = AsyncT $ \ctx stp yld -> do
78+ let yield a Nothing = yld a Nothing
79+ yield a (Just x) = yld a (Just (take (n - 1) x))
80+ if (n <= 0)
81+ then stp
82+ else (runAsyncT m) ctx stp yield
83+
84+-- | Discard first 'n' elements from the stream and take the rest.
85+drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
86+drop n m = AsyncT $ \ctx stp yld -> do
87+ let yield _ Nothing = stp
88+ yield _ (Just x) = (runAsyncT $ drop (n - 1) x) ctx stp yld
89+ if (n <= 0)
90+ then (runAsyncT m) ctx stp yld
91+ else (runAsyncT m) ctx stp yield
92+
93+------------------------------------------------------------------------------
94+-- Zipping Streams
95+------------------------------------------------------------------------------
96+
97+-- | Zip two AsyncT streams serially using a monadic zipping function.
98+zipWithM :: MonadAsync m =>
99+ (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
100+zipWithM f m1 m2 = AsyncT $ \_ stp yld -> do
101+ let merge a ra =
102+ let yield2 b Nothing = (runAsyncT (f a b)) Nothing stp yld
103+ yield2 b (Just rb) =
104+ (runAsyncT ((f a b) <> (zipWithM f ra rb))) Nothing stp yld
105+ in (runAsyncT m2) Nothing stp yield2
106+ let yield1 a Nothing = merge a empty
107+ yield1 a (Just ra) = merge a ra
108+ (runAsyncT m1) Nothing stp yield1
109+
110+-- | Zip two AsyncT streams serially using a pure zipping function.
111+zipWith :: MonadAsync m
112+ => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
113+zipWith f m1 m2 = AsyncT $ \_ stp yld -> do
114+ let merge a ra =
115+ let yield2 b Nothing = yld (f a b) Nothing
116+ yield2 b (Just rb) = yld (f a b) (Just (zipWith f ra rb))
117+ in (runAsyncT m2) Nothing stp yield2
118+ let yield1 a Nothing = merge a empty
119+ yield1 a (Just ra) = merge a ra
120+ (runAsyncT m1) Nothing stp yield1
121+
122+-- | Wrapper around AsyncT type with a serial zipping Applicative instance.
123+-- Note that the binary function interleave (\<=>) is a special case of
124+-- ZipSerial Applicative.
125+--
126+-- > f <$> ZipSerial xs1 <*> ... <*> ZipSerial xsN
127+newtype ZipSerial m a = ZipSerial {getZipSerial :: AsyncT m a}
128+ deriving (Functor)
129+
130+instance MonadAsync m => Applicative (ZipSerial m) where
131+ pure a = ZipSerial $ cycle1 (pure a)
132+ (ZipSerial xs) <*> (ZipSerial ys) = ZipSerial (zipWith id xs ys)
133+
134+-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
135+-- concurrently) using a monadic zipping function.
136+zipAsyncWithM :: MonadAsync m
137+ => (a -> b -> AsyncT m c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
138+zipAsyncWithM f m1 m2 = AsyncT $ \_ stp yld -> do
139+ ma <- async m1
140+ mb <- async m2
141+ (runAsyncT (zipWithM f ma mb)) Nothing stp yld
142+
143+-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
144+-- concurrently) using a pure zipping function.
145+zipAsyncWith :: MonadAsync m
146+ => (a -> b -> c) -> AsyncT m a -> AsyncT m b -> AsyncT m c
147+zipAsyncWith f m1 m2 = AsyncT $ \_ stp yld -> do
148+ ma <- async m1
149+ mb <- async m2
150+ (runAsyncT (zipWith f ma mb)) Nothing stp yld
151+
152+-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.
153+-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT
154+-- is a special case of ZipAsync Applicative.
155+--
156+-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN
157+newtype ZipAsync m a = ZipAsync {getZipAsync :: AsyncT m a}
158+ deriving (Functor)
159+
160+instance MonadAsync m => Applicative (ZipAsync m) where
161+ pure a = ZipAsync $ cycle1 (pure a)
162+ (ZipAsync xs) <*> (ZipAsync ys) = ZipAsync (zipAsyncWith id xs ys)