git.iko.soy/streamly
on c3dcabc help sign in
streamly/src/Streamly/Internal/Data/Stream/Zip.hs
258 lines 7.8 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.Zip
5-- Copyright : (c) 2017 Harendra Kumar
6--
7-- License : BSD3
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
10-- Portability : GHC
11--
12--
13module Streamly.Internal.Data.Stream.Zip
14 (
15 ZipSerialM
16 , ZipSerial
17 , zipSerially
19 , ZipAsyncM
20 , ZipAsync
21 , zipAsyncly
23 , zipWith
24 , zipWithM
25 , zipAsyncWith
26 , zipAsyncWithM
28 -- * Deprecated
29 , ZipStream
30 , zipping
31 , zippingAsync
32 )
33where
35import Control.Applicative (liftA2)
36import Control.DeepSeq (NFData(..))
37#if MIN_VERSION_deepseq(1,4,3)
38import Control.DeepSeq (NFData1(..))
39#endif
40import Data.Foldable (Foldable(foldl'), fold)
41import Data.Functor.Identity (Identity(..), runIdentity)
42import Data.Maybe (fromMaybe)
43import Data.Semigroup (Endo(..))
44#if __GLASGOW_HASKELL__ < 808
45import Data.Semigroup (Semigroup(..))
46#endif
47import GHC.Exts (IsList(..), IsString(..))
48import Text.Read
49 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
50 , readListPrecDefault)
51import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
52import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)
53import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
54import Streamly.Internal.Data.SVar (MonadAsync)
56import qualified Streamly.Internal.Data.Stream.Prelude as P
57import qualified Streamly.Internal.Data.Stream.StreamK as K
58import qualified Streamly.Internal.Data.Stream.StreamD as D
59#ifdef USE_STREAMK_ONLY
60import qualified Streamly.Internal.Data.Stream.StreamK as S
61#else
62import qualified Streamly.Internal.Data.Stream.StreamD as S
63#endif
65import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
67#include "Instances.hs"
69-- | Like 'zipWith' but using a monadic zipping function.
70--
71-- @since 0.4.0
72{-# INLINABLE zipWithM #-}
73zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
74zipWithM f m1 m2 = P.fromStreamS $ S.zipWithM f (P.toStreamS m1) (P.toStreamS m2)
76-- | Zip two streams serially using a pure zipping function.
77--
78-- @
79-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
80-- [5,7,9]
81-- @
82--
83-- @since 0.1.0
84{-# INLINABLE zipWith #-}
85zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
86zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)
88------------------------------------------------------------------------------
89-- Parallel Zipping
90------------------------------------------------------------------------------
92-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
93-- are generated concurrently.
94--
95-- @since 0.4.0
96{-# INLINE zipAsyncWithM #-}
97zipAsyncWithM :: (IsStream t, MonadAsync m)
98 => (a -> b -> m c) -> t m a -> t m b -> t m c
99zipAsyncWithM f m1 m2 = D.fromStreamD $
100 D.zipWithM f (D.mkParallelD $ D.toStreamD m1)
101 (D.mkParallelD $ D.toStreamD m2)
103-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
104-- are generated concurrently.
105--
106-- @since 0.1.0
107{-# INLINE zipAsyncWith #-}
108zipAsyncWith :: (IsStream t, MonadAsync m)
109 => (a -> b -> c) -> t m a -> t m b -> t m c
110zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
112------------------------------------------------------------------------------
113-- Serially Zipping Streams
114------------------------------------------------------------------------------
116-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
117-- i.e. it produces one element from each stream serially and then zips all
118-- those elements.
119--
120-- @
121-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
122-- where s1 = fromFoldable [1, 2]
123-- s2 = fromFoldable [3, 4]
124-- s3 = fromFoldable [5, 6]
125-- @
126-- @
127-- [(1,3,5),(2,4,6)]
128-- @
129--
130-- The 'Semigroup' instance of this type works the same way as that of
131-- 'SerialT'.
132--
133-- @since 0.2.0
134newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
135 deriving (Semigroup, Monoid)
137-- |
138-- @since 0.1.0
139{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
140type ZipStream = ZipSerialM
142-- | An IO stream whose applicative instance zips streams serially.
143--
144-- @since 0.2.0
145type ZipSerial = ZipSerialM IO
147-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
148--
149-- @since 0.2.0
150zipSerially :: IsStream t => ZipSerialM m a -> t m a
151zipSerially = K.adapt
153-- | Same as 'zipSerially'.
154--
155-- @since 0.1.0
156{-# DEPRECATED zipping "Please use zipSerially instead." #-}
157zipping :: IsStream t => ZipSerialM m a -> t m a
158zipping = zipSerially
160consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
161consMZip m ms = fromStream $ K.consMStream m (toStream ms)
163instance IsStream ZipSerialM where
164 toStream = getZipSerialM
165 fromStream = ZipSerialM
167 {-# INLINE consM #-}
168 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
169 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
170 consM = consMZip
172 {-# INLINE (|:) #-}
173 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
174 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
175 (|:) = consMZip
177LIST_INSTANCES(ZipSerialM)
178NFDATA1_INSTANCE(ZipSerialM)
180instance Monad m => Functor (ZipSerialM m) where
181 {-# INLINE fmap #-}
182 fmap f (ZipSerialM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
184instance Monad m => Applicative (ZipSerialM m) where
185 pure = ZipSerialM . K.repeat
186 {-# INLINE (<*>) #-}
187 (<*>) = zipWith id
189FOLDABLE_INSTANCE(ZipSerialM)
190TRAVERSABLE_INSTANCE(ZipSerialM)
192------------------------------------------------------------------------------
193-- Parallely Zipping Streams
194------------------------------------------------------------------------------
195--
196-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
197-- be zipped concurrently.
198--
199-- @
200-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
201-- where s1 = fromFoldable [1, 2]
202-- s2 = fromFoldable [3, 4]
203-- s3 = fromFoldable [5, 6]
204-- @
205-- @
206-- [(1,3,5),(2,4,6)]
207-- @
208--
209-- The 'Semigroup' instance of this type works the same way as that of
210-- 'SerialT'.
211--
212-- @since 0.2.0
213newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
214 deriving (Semigroup, Monoid)
216-- | An IO stream whose applicative instance zips streams wAsyncly.
217--
218-- @since 0.2.0
219type ZipAsync = ZipAsyncM IO
221-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
222--
223-- @since 0.2.0
224zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
225zipAsyncly = K.adapt
227-- | Same as 'zipAsyncly'.
228--
229-- @since 0.1.0
230{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
231zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
232zippingAsync = zipAsyncly
234consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
235consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
237instance IsStream ZipAsyncM where
238 toStream = getZipAsyncM
239 fromStream = ZipAsyncM
241 {-# INLINE consM #-}
242 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
243 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
244 consM = consMZipAsync
246 {-# INLINE (|:) #-}
247 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
248 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
249 (|:) = consMZipAsync
251instance Monad m => Functor (ZipAsyncM m) where
252 {-# INLINE fmap #-}
253 fmap f (ZipAsyncM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
255instance MonadAsync m => Applicative (ZipAsyncM m) where
256 pure = ZipAsyncM . K.repeat
257 {-# INLINE (<*>) #-}
258 m1 <*> m2 = zipAsyncWith id m1 m2