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