git.iko.soy/streamly
on d1146fb help sign in
streamly/src/Streamly/Internal/Data/Stream/Zip.hs
226 lines 7.4 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 , consMZip
21 , zipWithK
22 , zipWithMK
24 , ZipAsyncM(..)
25 , ZipAsync
26 , consMZipAsync
27 , zipAsyncWithK
28 , zipAsyncWithMK
30 -- * Deprecated
31 , ZipStream
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, oneShot)
52import Streamly.Internal.Control.Concurrent (MonadAsync)
53import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
54import Streamly.Internal.Data.Stream.Serial (SerialT(..))
55import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
57import qualified Streamly.Internal.Data.Stream.Prelude as P
58 (cmpBy, eqBy, foldl', foldr, fromList, toList)
59import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
60import qualified Streamly.Internal.Data.Stream.StreamK as K
61import qualified Streamly.Internal.Data.Stream.StreamD as D
62import qualified Streamly.Internal.Data.Stream.Serial as Serial
63import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
64import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
66import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
67import Streamly.Internal.Data.SVar
69#include "Instances.hs"
71-- $setup
72-- >>> import qualified Streamly.Prelude as Stream
73-- >>> import Control.Concurrent (threadDelay)
74-- >>> :{
75-- delay n = do
76-- threadDelay (n * 1000000) -- sleep for n seconds
77-- putStrLn (show n ++ " sec") -- print "n sec"
78-- return n -- IO Int
79-- :}
81{-# INLINE zipWithMK #-}
82zipWithMK :: Monad m =>
83 (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
84zipWithMK f m1 m2 =
85 D.toStreamK $ D.zipWithM f (D.fromStreamK m1) (D.fromStreamK m2)
87{-# INLINE zipWithK #-}
88zipWithK :: Monad m
89 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
90zipWithK f = zipWithMK (\a b -> return (f a b))
92------------------------------------------------------------------------------
93-- Parallel Zipping
94------------------------------------------------------------------------------
96-- | Like 'zipAsyncWith' but with a monadic zipping function.
97--
98-- @since 0.4.0
99{-# INLINE zipAsyncWithMK #-}
100zipAsyncWithMK :: MonadAsync m
101 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
102zipAsyncWithMK f m1 m2 = K.mkStream $ \st yld sng stp -> do
103 sv <- newParallelVar StopNone (adaptState st)
104 SVar.toSVarParallel (adaptState st) sv $ D.fromStreamK m2
105 K.foldStream st yld sng stp $ K.zipWithM f m1 (getSerialT (SVar.fromSVar sv))
107-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
108-- along with the change of behvaior to end the stream concurrently.
109--
110-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
111-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
112-- style. The maximum number of elements of each stream evaluated in advance
113-- can be controlled by 'maxBuffer'.
114--
115-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
116-- ends while we are still evaluating stream @a@ and waiting for a result then
117-- stream will not end until after the evaluation of stream @a@ finishes. This
118-- behavior can potentially be changed in future to end the stream immediately
119-- as soon as any of the stream end is detected.
120--
121-- @since 0.1.0
122{-# INLINE zipAsyncWithK #-}
123zipAsyncWithK :: MonadAsync m
124 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
125zipAsyncWithK f = zipAsyncWithMK (\a b -> return (f a b))
127------------------------------------------------------------------------------
128-- Serially Zipping Streams
129------------------------------------------------------------------------------
131-- | For 'ZipSerialM' streams:
132--
133-- @
134-- (<>) = 'Streamly.Prelude.serial'
135-- (<*>) = 'Streamly.Prelude.serial.zipWith' id
136-- @
137--
138-- Applicative evaluates the streams being zipped serially:
139--
140-- >>> s1 = Stream.fromFoldable [1, 2]
141-- >>> s2 = Stream.fromFoldable [3, 4]
142-- >>> s3 = Stream.fromFoldable [5, 6]
143-- >>> Stream.toList $ Stream.fromZipSerial $ (,,) <$> s1 <*> s2 <*> s3
144-- [(1,3,5),(2,4,6)]
145--
146-- /Since: 0.2.0 ("Streamly")/
147--
148-- @since 0.8.0
149newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
150 deriving (Semigroup, Monoid)
152-- |
153-- @since 0.1.0
154{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
155type ZipStream = ZipSerialM
157-- | An IO stream whose applicative instance zips streams serially.
158--
159-- /Since: 0.2.0 ("Streamly")/
160--
161-- @since 0.8.0
162type ZipSerial = ZipSerialM IO
164consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
165consMZip m (ZipSerialM r) = ZipSerialM $ K.consM m r
167LIST_INSTANCES(ZipSerialM)
168NFDATA1_INSTANCE(ZipSerialM)
170instance Monad m => Functor (ZipSerialM m) where
171 {-# INLINE fmap #-}
172 fmap f (ZipSerialM m) = ZipSerialM $ getSerialT $ fmap f (SerialT m)
174instance Monad m => Applicative (ZipSerialM m) where
175 pure = ZipSerialM . getSerialT . Serial.repeat
177 {-# INLINE (<*>) #-}
178 ZipSerialM m1 <*> ZipSerialM m2 = ZipSerialM $ zipWithK id m1 m2
180FOLDABLE_INSTANCE(ZipSerialM)
181TRAVERSABLE_INSTANCE(ZipSerialM)
183------------------------------------------------------------------------------
184-- Parallely Zipping Streams
185------------------------------------------------------------------------------
186--
187-- | For 'ZipAsyncM' streams:
188--
189-- @
190-- (<>) = 'Streamly.Prelude.serial'
191-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id
192-- @
193--
194-- Applicative evaluates the streams being zipped concurrently, the following
195-- would take half the time that it would take in serial zipping:
196--
197-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]
198-- >>> Stream.toList $ Stream.fromZipAsync $ (,) <$> s <*> s
199-- ...
200-- [(1,1),(1,1),(1,1)]
201--
202-- /Since: 0.2.0 ("Streamly")/
203--
204-- @since 0.8.0
205newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
206 deriving (Semigroup, Monoid)
208-- | An IO stream whose applicative instance zips streams wAsyncly.
209--
210-- /Since: 0.2.0 ("Streamly")/
211--
212-- @since 0.8.0
213type ZipAsync = ZipAsyncM IO
215consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
216consMZipAsync m (ZipAsyncM r) = ZipAsyncM $ K.consM m r
218instance Monad m => Functor (ZipAsyncM m) where
219 {-# INLINE fmap #-}
220 fmap f (ZipAsyncM m) = ZipAsyncM $ getSerialT $ fmap f (SerialT m)
222instance MonadAsync m => Applicative (ZipAsyncM m) where
223 pure = ZipAsyncM . getSerialT . Serial.repeat
225 {-# INLINE (<*>) #-}
226 ZipAsyncM m1 <*> ZipAsyncM m2 = ZipAsyncM $ zipAsyncWithK id m1 m2