@@ -0,0 +1,162 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving#-}
3+{-# LANGUAGE UndecidableInstances #-} -- XXX
6+-- Module : Asyncly.Prelude
7+-- Copyright : (c) 2017 Harendra Kumar
10+-- Maintainer : harendra.kumar@gmail.com
11+-- Stability : experimental
15+module Asyncly.Prelude
38+import Control.Applicative (Alternative (..))
39+import Data.Semigroup (Semigroup(..), cycle1)
40+import Prelude hiding (drop, take, zipWith)
43+------------------------------------------------------------------------------
44+-- Construct and deconstruct
45+------------------------------------------------------------------------------
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
55+ stop = return Nothing
58+ yield a Nothing = return (Just (a, empty))
59+ yield a (Just x) = return (Just (a, x))
61+-- | Build a Stream by unfolding steps starting from a seed.
62+unfoldr :: MonadAsync m => (b -> m (Maybe (a, b))) -> b -> AsyncT m a
65+ go s = AsyncT $ \_ stp yld -> do
69+ Just (a, b) -> yld a (Just (go b))
71+------------------------------------------------------------------------------
73+------------------------------------------------------------------------------
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))
82+ else (runAsyncT m) ctx stp yield
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
90+ then (runAsyncT m) ctx stp yld
91+ else (runAsyncT m) ctx stp yield
93+------------------------------------------------------------------------------
95+------------------------------------------------------------------------------
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
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
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
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
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.
126+-- > f <$> ZipSerial xs1 <*> ... <*> ZipSerial xsN
127+newtype ZipSerial m a = ZipSerial {getZipSerial :: AsyncT m a}
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)
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
141+ (runAsyncT (zipWithM f ma mb)) Nothing stp yld
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
150+ (runAsyncT (zipWith f ma mb)) Nothing stp yld
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.
156+-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN
157+newtype ZipAsync m a = ZipAsync {getZipAsync :: AsyncT m a}
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)