git.iko.soy/streamly
on a296fac help sign in
streamly/src/Streamly/Internal/Data/Stream/Zip.hs
299 lines 9.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.Control.Concurrent (MonadAsync)
55import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace, oneShot)
56import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)
57import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
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-- | Stream @a@ is evaluated first, followed by stream @b@, the resulting
94-- elements @a@ and @b@ are then zipped using the supplied zip function and the
95-- result @c@ is yielded to the consumer.
96--
97-- If stream @a@ or stream @b@ ends, the zipped stream ends. If stream @b@ ends
98-- first, the element @a@ from previous evaluation of stream @a@ is discarded.
99--
100-- @
101-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])
102-- [5,7,9]
103-- @
104--
105-- @since 0.1.0
106{-# INLINABLE zipWith #-}
107zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c
108zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)
110------------------------------------------------------------------------------
111-- Parallel Zipping
112------------------------------------------------------------------------------
114-- | Like 'zipAsyncWith' but with a monadic zipping function.
115--
116-- @since 0.4.0
117{-# INLINE zipAsyncWithM #-}
118zipAsyncWithM :: (IsStream t, MonadAsync m)
119 => (a -> b -> m c) -> t m a -> t m b -> t m c
120zipAsyncWithM f m1 m2 = D.fromStreamD $
121 D.zipWithM f (Par.mkParallelD $ D.toStreamD m1)
122 (Par.mkParallelD $ D.toStreamD m2)
124-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
125-- along with the change of behvaior to end the stream concurrently.
126--
127-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
128-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
129-- style. The maximum number of elements of each stream evaluated in advance
130-- can be controlled by 'maxBuffer'.
131--
132-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
133-- ends while we are still evaluating stream @a@ and waiting for a result then
134-- stream will not end until after the evaluation of stream @a@ finishes. This
135-- behavior can potentially be changed in future to end the stream immediately
136-- as soon as any of the stream end is detected.
137--
138-- @since 0.1.0
139{-# INLINE zipAsyncWith #-}
140zipAsyncWith :: (IsStream t, MonadAsync m)
141 => (a -> b -> c) -> t m a -> t m b -> t m c
142zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
144------------------------------------------------------------------------------
145-- Serially Zipping Streams
146------------------------------------------------------------------------------
148-- | For 'ZipSerialM' streams:
149--
150-- @
151-- (<>) = 'Streamly.Prelude.serial'
152-- (<*>) = 'Streamly.Prelude.serial.zipWith' id
153-- @
154--
155-- Applicative evaluates the streams being zipped serially:
156--
157-- >>> s1 = Stream.fromFoldable [1, 2]
158-- >>> s2 = Stream.fromFoldable [3, 4]
159-- >>> s3 = Stream.fromFoldable [5, 6]
160-- >>> Stream.toList $ Stream.fromZipSerial $ (,,) <$> s1 <*> s2 <*> s3
161-- [(1,3,5),(2,4,6)]
162--
163-- /Since: 0.2.0 ("Streamly")/
164--
165-- @since 0.8.0
166newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
167 deriving (Semigroup, Monoid)
169-- |
170-- @since 0.1.0
171{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
172type ZipStream = ZipSerialM
174-- | An IO stream whose applicative instance zips streams serially.
175--
176-- /Since: 0.2.0 ("Streamly")/
177--
178-- @since 0.8.0
179type ZipSerial = ZipSerialM IO
181-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
182--
183-- /Since: 0.2.0 ("Streamly")/
184--
185-- @since 0.8.0
186fromZipSerial :: IsStream t => ZipSerialM m a -> t m a
187fromZipSerial = K.adapt
189-- | Same as 'fromZipSerial'.
190--
191-- @since 0.1.0
192{-# DEPRECATED zipping "Please use fromZipSerial instead." #-}
193zipping :: IsStream t => ZipSerialM m a -> t m a
194zipping = fromZipSerial
196consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
197consMZip m ms = fromStream $ K.consMStream m (toStream ms)
199instance IsStream ZipSerialM where
200 toStream = getZipSerialM
201 fromStream = ZipSerialM
203 {-# INLINE consM #-}
204 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
205 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
206 consM = consMZip
208 {-# INLINE (|:) #-}
209 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
210 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
211 (|:) = consMZip
213LIST_INSTANCES(ZipSerialM)
214NFDATA1_INSTANCE(ZipSerialM)
216instance Monad m => Functor (ZipSerialM m) where
217 {-# INLINE fmap #-}
218 fmap f (ZipSerialM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
220instance Monad m => Applicative (ZipSerialM m) where
221 pure = ZipSerialM . K.repeat
222 {-# INLINE (<*>) #-}
223 (<*>) = zipWith id
225FOLDABLE_INSTANCE(ZipSerialM)
226TRAVERSABLE_INSTANCE(ZipSerialM)
228------------------------------------------------------------------------------
229-- Parallely Zipping Streams
230------------------------------------------------------------------------------
231--
232-- | For 'ZipAsyncM' streams:
233--
234-- @
235-- (<>) = 'Streamly.Prelude.serial'
236-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id
237-- @
238--
239-- Applicative evaluates the streams being zipped concurrently, the following
240-- would take half the time that it would take in serial zipping:
241--
242-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]
243-- >>> Stream.toList $ Stream.fromZipAsync $ (,) <$> s <*> s
244-- ...
245-- [(1,1),(1,1),(1,1)]
246--
247-- /Since: 0.2.0 ("Streamly")/
248--
249-- @since 0.8.0
250newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
251 deriving (Semigroup, Monoid)
253-- | An IO stream whose applicative instance zips streams wAsyncly.
254--
255-- /Since: 0.2.0 ("Streamly")/
256--
257-- @since 0.8.0
258type ZipAsync = ZipAsyncM IO
260-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
261--
262-- /Since: 0.2.0 ("Streamly")/
263--
264-- @since 0.8.0
265fromZipAsync :: IsStream t => ZipAsyncM m a -> t m a
266fromZipAsync = K.adapt
268-- | Same as 'fromZipAsync'.
269--
270-- @since 0.1.0
271{-# DEPRECATED zippingAsync "Please use fromZipAsync instead." #-}
272zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
273zippingAsync = fromZipAsync
275consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
276consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
278instance IsStream ZipAsyncM where
279 toStream = getZipAsyncM
280 fromStream = ZipAsyncM
282 {-# INLINE consM #-}
283 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
284 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
285 consM = consMZipAsync
287 {-# INLINE (|:) #-}
288 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
289 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
290 (|:) = consMZipAsync
292instance Monad m => Functor (ZipAsyncM m) where
293 {-# INLINE fmap #-}
294 fmap f (ZipAsyncM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m
296instance MonadAsync m => Applicative (ZipAsyncM m) where
297 pure = ZipAsyncM . K.repeat
298 {-# INLINE (<*>) #-}
299 m1 <*> m2 = zipAsyncWith id m1 m2