git.iko.soy/streamly
on ee691b6 help sign in
streamly/src/Streamly/Internal/Data/Stream/Zip.hs
274 lines 8.3 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.Zip
5-- Copyright : (c) 2017 Composewell Technologies
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.Parallel as Par
57import qualified Streamly.Internal.Data.Stream.Prelude as P
58 (cmpBy, eqBy, foldl', foldr, fromList, toList, fromStreamS, toStreamS)
59import qualified Streamly.Internal.Data.Stream.StreamK as K (repeat)
60import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
61import qualified Streamly.Internal.Data.Stream.StreamD as D (zipWithM)
62import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
63#ifdef USE_STREAMK_ONLY
64import qualified Streamly.Internal.Data.Stream.StreamK as S (zipWith, zipWithM)
65#else
66import qualified Streamly.Internal.Data.Stream.StreamD as S (zipWith, zipWithM)
67#endif
69import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
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 (Par.mkParallelD $ D.toStreamD m1)
105 (Par.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 ("Streamly")/
138--
139-- @since 0.8.0
140newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
141 deriving (Semigroup, Monoid)
143-- |
144-- @since 0.1.0
145{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
146type ZipStream = ZipSerialM
148-- | An IO stream whose applicative instance zips streams serially.
149--
150-- /Since: 0.2.0 ("Streamly")/
151--
152-- @since 0.8.0
153type ZipSerial = ZipSerialM IO
155-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
156--
157-- /Since: 0.2.0 ("Streamly")/
158--
159-- @since 0.8.0
160zipSerially :: IsStream t => ZipSerialM m a -> t m a
161zipSerially = K.adapt
163-- | Same as 'zipSerially'.
164--
165-- @since 0.1.0
166{-# DEPRECATED zipping "Please use zipSerially instead." #-}
167zipping :: IsStream t => ZipSerialM m a -> t m a
168zipping = zipSerially
170consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
171consMZip m ms = fromStream $ K.consMStream m (toStream ms)
173instance IsStream ZipSerialM where
174 toStream = getZipSerialM
175 fromStream = ZipSerialM
177 {-# INLINE consM #-}
178 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
179 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
180 consM = consMZip
182 {-# INLINE (|:) #-}
183 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
184 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
185 (|:) = consMZip
187LIST_INSTANCES(ZipSerialM)
188NFDATA1_INSTANCE(ZipSerialM)
190instance Monad m => Functor (ZipSerialM m) where
191 {-# INLINE fmap #-}
192 fmap f (ZipSerialM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
194instance Monad m => Applicative (ZipSerialM m) where
195 pure = ZipSerialM . K.repeat
196 {-# INLINE (<*>) #-}
197 (<*>) = zipWith id
199FOLDABLE_INSTANCE(ZipSerialM)
200TRAVERSABLE_INSTANCE(ZipSerialM)
202------------------------------------------------------------------------------
203-- Parallely Zipping Streams
204------------------------------------------------------------------------------
205--
206-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
207-- be zipped concurrently.
208--
209-- @
210-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
211-- where s1 = fromFoldable [1, 2]
212-- s2 = fromFoldable [3, 4]
213-- s3 = fromFoldable [5, 6]
214-- @
215-- @
216-- [(1,3,5),(2,4,6)]
217-- @
218--
219-- The 'Semigroup' instance of this type works the same way as that of
220-- 'SerialT'.
221--
222-- /Since: 0.2.0 ("Streamly")/
223--
224-- @since 0.8.0
225newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
226 deriving (Semigroup, Monoid)
228-- | An IO stream whose applicative instance zips streams wAsyncly.
229--
230-- /Since: 0.2.0 ("Streamly")/
231--
232-- @since 0.8.0
233type ZipAsync = ZipAsyncM IO
235-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
236--
237-- /Since: 0.2.0 ("Streamly")/
238--
239-- @since 0.8.0
240zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
241zipAsyncly = K.adapt
243-- | Same as 'zipAsyncly'.
244--
245-- @since 0.1.0
246{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
247zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
248zippingAsync = zipAsyncly
250consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
251consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
253instance IsStream ZipAsyncM where
254 toStream = getZipAsyncM
255 fromStream = ZipAsyncM
257 {-# INLINE consM #-}
258 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
259 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
260 consM = consMZipAsync
262 {-# INLINE (|:) #-}
263 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
264 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
265 (|:) = consMZipAsync
267instance Monad m => Functor (ZipAsyncM m) where
268 {-# INLINE fmap #-}
269 fmap f (ZipAsyncM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
271instance MonadAsync m => Applicative (ZipAsyncM m) where
272 pure = ZipAsyncM . K.repeat
273 {-# INLINE (<*>) #-}
274 m1 <*> m2 = zipAsyncWith id m1 m2