git.iko.soy/streamly
on bc4dba9 help sign in
streamly/src/Streamly/Internal/Data/Stream/Zip.hs
284 lines 8.6 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-- To run examples in this module:
13--
14-- >>> import qualified Streamly.Prelude as Stream
15--
16module Streamly.Internal.Data.Stream.Zip
17 (
18 ZipSerialM
19 , ZipSerial
20 , fromZipSerial
22 , ZipAsyncM
23 , ZipAsync
24 , fromZipAsync
26 , zipWith
27 , zipWithM
28 , zipAsyncWith
29 , zipAsyncWithM
31 -- * Deprecated
32 , ZipStream
33 , zipping
34 , zippingAsync
35 )
36where
38import Control.Applicative (liftA2)
39import Control.DeepSeq (NFData(..))
40#if MIN_VERSION_deepseq(1,4,3)
41import Control.DeepSeq (NFData1(..))
42#endif
43import Data.Foldable (Foldable(foldl'), fold)
44import Data.Functor.Identity (Identity(..), runIdentity)
45import Data.Maybe (fromMaybe)
46import Data.Semigroup (Endo(..))
47#if __GLASGOW_HASKELL__ < 808
48import Data.Semigroup (Semigroup(..))
49#endif
50import GHC.Exts (IsList(..), IsString(..))
51import Text.Read
52 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
53 , readListPrecDefault)
54import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace, oneShot)
55import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)
56import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
57import Streamly.Internal.Data.SVar (MonadAsync)
59import qualified Streamly.Internal.Data.Stream.Parallel as Par
60import qualified Streamly.Internal.Data.Stream.Prelude as P
61 (cmpBy, eqBy, foldl', foldr, fromList, toList, fromStreamS, toStreamS)
62import qualified Streamly.Internal.Data.Stream.StreamK as K (repeat)
63import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
64import qualified Streamly.Internal.Data.Stream.StreamD as D (zipWithM)
65import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
66#ifdef USE_STREAMK_ONLY
67import qualified Streamly.Internal.Data.Stream.StreamK as S (zipWith, zipWithM)
68#else
69import qualified Streamly.Internal.Data.Stream.StreamD as S (zipWith, zipWithM)
70#endif
72import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
74#include "Instances.hs"
76-- $setup
77-- >>> import qualified Streamly.Prelude as Stream
78-- >>> import Control.Concurrent (threadDelay)
79-- >>> :{
80-- delay n = do
81-- threadDelay (n * 1000000) -- sleep for n seconds
82-- putStrLn (show n ++ " sec") -- print "n sec"
83-- return n -- IO Int
84-- :}
86-- | Like 'zipWith' but using a monadic zipping function.
87--
88-- @since 0.4.0
89{-# INLINABLE zipWithM #-}
90zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c
91zipWithM f m1 m2 = P.fromStreamS $ S.zipWithM f (P.toStreamS m1) (P.toStreamS m2)
93-- | Zip two streams serially using a pure zipping function.
94--
95-- @
96-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
97-- [5,7,9]
98-- @
99--
100-- @since 0.1.0
101{-# INLINABLE zipWith #-}
102zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
103zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)
105------------------------------------------------------------------------------
106-- Parallel Zipping
107------------------------------------------------------------------------------
109-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
110-- are generated concurrently.
111--
112-- @since 0.4.0
113{-# INLINE zipAsyncWithM #-}
114zipAsyncWithM :: (IsStream t, MonadAsync m)
115 => (a -> b -> m c) -> t m a -> t m b -> t m c
116zipAsyncWithM f m1 m2 = D.fromStreamD $
117 D.zipWithM f (Par.mkParallelD $ D.toStreamD m1)
118 (Par.mkParallelD $ D.toStreamD m2)
120-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
121-- are generated concurrently.
122--
123-- @since 0.1.0
124{-# INLINE zipAsyncWith #-}
125zipAsyncWith :: (IsStream t, MonadAsync m)
126 => (a -> b -> c) -> t m a -> t m b -> t m c
127zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
129------------------------------------------------------------------------------
130-- Serially Zipping Streams
131------------------------------------------------------------------------------
133-- | For 'ZipSerialM' streams:
134--
135-- @
136-- (<>) = 'Streamly.Prelude.serial'
137-- (<*>) = 'Streamly.Prelude.serial.zipWith' id
138-- @
139--
140-- Applicative evaluates the streams being zipped serially:
141--
142-- >>> s1 = Stream.fromFoldable [1, 2]
143-- >>> s2 = Stream.fromFoldable [3, 4]
144-- >>> s3 = Stream.fromFoldable [5, 6]
145-- >>> Stream.toList $ Stream.fromZipSerial $ (,,) <$> s1 <*> s2 <*> s3
146-- [(1,3,5),(2,4,6)]
147--
148-- /Since: 0.2.0 ("Streamly")/
149--
150-- @since 0.8.0
151newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
152 deriving (Semigroup, Monoid)
154-- |
155-- @since 0.1.0
156{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
157type ZipStream = ZipSerialM
159-- | An IO stream whose applicative instance zips streams serially.
160--
161-- /Since: 0.2.0 ("Streamly")/
162--
163-- @since 0.8.0
164type ZipSerial = ZipSerialM IO
166-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
167--
168-- /Since: 0.2.0 ("Streamly")/
169--
170-- @since 0.8.0
171fromZipSerial :: IsStream t => ZipSerialM m a -> t m a
172fromZipSerial = K.adapt
174-- | Same as 'fromZipSerial'.
175--
176-- @since 0.1.0
177{-# DEPRECATED zipping "Please use fromZipSerial instead." #-}
178zipping :: IsStream t => ZipSerialM m a -> t m a
179zipping = fromZipSerial
181consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
182consMZip m ms = fromStream $ K.consMStream m (toStream ms)
184instance IsStream ZipSerialM where
185 toStream = getZipSerialM
186 fromStream = ZipSerialM
188 {-# INLINE consM #-}
189 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
190 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
191 consM = consMZip
193 {-# INLINE (|:) #-}
194 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
195 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
196 (|:) = consMZip
198LIST_INSTANCES(ZipSerialM)
199NFDATA1_INSTANCE(ZipSerialM)
201instance Monad m => Functor (ZipSerialM m) where
202 {-# INLINE fmap #-}
203 fmap f (ZipSerialM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
205instance Monad m => Applicative (ZipSerialM m) where
206 pure = ZipSerialM . K.repeat
207 {-# INLINE (<*>) #-}
208 (<*>) = zipWith id
210FOLDABLE_INSTANCE(ZipSerialM)
211TRAVERSABLE_INSTANCE(ZipSerialM)
213------------------------------------------------------------------------------
214-- Parallely Zipping Streams
215------------------------------------------------------------------------------
216--
217-- | For 'ZipAsyncM' streams:
218--
219-- @
220-- (<>) = 'Streamly.Prelude.serial'
221-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id
222-- @
223--
224-- Applicative evaluates the streams being zipped concurrently, the following
225-- would take half the time that it would take in serial zipping:
226--
227-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]
228-- >>> Stream.toList $ Stream.fromZipAsync $ (,) <$> s <*> s
229-- ...
230-- [(1,1),(1,1),(1,1)]
231--
232-- /Since: 0.2.0 ("Streamly")/
233--
234-- @since 0.8.0
235newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
236 deriving (Semigroup, Monoid)
238-- | An IO stream whose applicative instance zips streams wAsyncly.
239--
240-- /Since: 0.2.0 ("Streamly")/
241--
242-- @since 0.8.0
243type ZipAsync = ZipAsyncM IO
245-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
246--
247-- /Since: 0.2.0 ("Streamly")/
248--
249-- @since 0.8.0
250fromZipAsync :: IsStream t => ZipAsyncM m a -> t m a
251fromZipAsync = K.adapt
253-- | Same as 'fromZipAsync'.
254--
255-- @since 0.1.0
256{-# DEPRECATED zippingAsync "Please use fromZipAsync instead." #-}
257zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
258zippingAsync = fromZipAsync
260consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
261consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
263instance IsStream ZipAsyncM where
264 toStream = getZipAsyncM
265 fromStream = ZipAsyncM
267 {-# INLINE consM #-}
268 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
269 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
270 consM = consMZipAsync
272 {-# INLINE (|:) #-}
273 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
274 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
275 (|:) = consMZipAsync
277instance Monad m => Functor (ZipAsyncM m) where
278 {-# INLINE fmap #-}
279 fmap f (ZipAsyncM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
281instance MonadAsync m => Applicative (ZipAsyncM m) where
282 pure = ZipAsyncM . K.repeat
283 {-# INLINE (<*>) #-}
284 m1 <*> m2 = zipAsyncWith id m1 m2