git.iko.soy/streamly
on 04c2ae0 help sign in
streamly/src/Streamly/Streams/Zip.hs
236 lines 7.1 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.Streams.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.Streams.Zip
22 (
23 K.zipWith
24 , K.zipWithM
25 , zipAsyncWith
26 , zipAsyncWithM
28 , ZipSerialM
29 , ZipSerial
30 , ZipStream -- deprecated
31 , zipSerially
32 , zipping -- deprecated
34 , ZipAsyncM
35 , ZipAsync
36 , zipAsyncly
37 , zippingAsync -- deprecated
38 )
39where
41import Control.Applicative (liftA2)
42import Control.DeepSeq (NFData(..))
43#if MIN_VERSION_deepseq(1,4,3)
44import Control.DeepSeq (NFData1(..))
45#endif
46import Data.Functor.Identity (Identity, runIdentity)
47import Data.Foldable (fold)
48#if __GLASGOW_HASKELL__ < 808
49import Data.Semigroup (Semigroup(..))
50#endif
51import GHC.Exts (IsList(..), IsString(..))
52import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
53 readListPrecDefault)
54import Prelude hiding (map, repeat, zipWith)
56import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, foldStream)
57import Streamly.Streams.Async (mkAsync')
58import Streamly.Streams.Serial (map)
59import Streamly.Internal.Data.SVar (MonadAsync, adaptState)
61import qualified Streamly.Streams.Prelude as P
62import qualified Streamly.Streams.StreamK as K
64#include "Instances.hs"
66------------------------------------------------------------------------------
67-- Serially Zipping Streams
68------------------------------------------------------------------------------
70-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
71-- i.e. it produces one element from each stream serially and then zips all
72-- those elements.
73--
74-- @
75-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
76-- where s1 = fromFoldable [1, 2]
77-- s2 = fromFoldable [3, 4]
78-- s3 = fromFoldable [5, 6]
79-- @
80-- @
81-- [(1,3,5),(2,4,6)]
82-- @
83--
84-- The 'Semigroup' instance of this type works the same way as that of
85-- 'SerialT'.
86--
87-- @since 0.2.0
88newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
89 deriving (Semigroup, Monoid)
91-- |
92-- @since 0.1.0
93{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
94type ZipStream = ZipSerialM
96-- | An IO stream whose applicative instance zips streams serially.
97--
98-- @since 0.2.0
99type ZipSerial = ZipSerialM IO
101-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
102--
103-- @since 0.2.0
104zipSerially :: IsStream t => ZipSerialM m a -> t m a
105zipSerially = K.adapt
107-- | Same as 'zipSerially'.
108--
109-- @since 0.1.0
110{-# DEPRECATED zipping "Please use zipSerially instead." #-}
111zipping :: IsStream t => ZipSerialM m a -> t m a
112zipping = zipSerially
114consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
115consMZip m ms = fromStream $ K.consMStream m (toStream ms)
117instance IsStream ZipSerialM where
118 toStream = getZipSerialM
119 fromStream = ZipSerialM
121 {-# INLINE consM #-}
122 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
123 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
124 consM = consMZip
126 {-# INLINE (|:) #-}
127 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
128 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
129 (|:) = consMZip
131LIST_INSTANCES(ZipSerialM)
132NFDATA1_INSTANCE(ZipSerialM)
134instance Monad m => Functor (ZipSerialM m) where
135 fmap = map
137instance Monad m => Applicative (ZipSerialM m) where
138 pure = ZipSerialM . K.repeat
139 (<*>) = K.zipWith id
141FOLDABLE_INSTANCE(ZipSerialM)
142TRAVERSABLE_INSTANCE(ZipSerialM)
144------------------------------------------------------------------------------
145-- Parallel Zipping
146------------------------------------------------------------------------------
148-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
149-- are generated concurrently.
150--
151-- @since 0.1.0
152{-# INLINABLE zipAsyncWith #-}
153zipAsyncWith :: (IsStream t, MonadAsync m)
154 => (a -> b -> c) -> t m a -> t m b -> t m c
155zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
156 ma <- mkAsync' (adaptState st) m1
157 mb <- mkAsync' (adaptState st) m2
158 foldStream st stp sng yld (K.zipWith f ma mb)
160-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
161-- are generated concurrently.
162--
163-- @since 0.4.0
164{-# INLINABLE zipAsyncWithM #-}
165zipAsyncWithM :: (IsStream t, MonadAsync m)
166 => (a -> b -> m c) -> t m a -> t m b -> t m c
167zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do
168 ma <- mkAsync' (adaptState st) m1
169 mb <- mkAsync' (adaptState st) m2
170 foldStream st stp sng yld (K.zipWithM f ma mb)
172------------------------------------------------------------------------------
173-- Parallely Zipping Streams
174------------------------------------------------------------------------------
175--
176-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
177-- be zipped concurrently.
178--
179-- @
180-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
181-- where s1 = fromFoldable [1, 2]
182-- s2 = fromFoldable [3, 4]
183-- s3 = fromFoldable [5, 6]
184-- @
185-- @
186-- [(1,3,5),(2,4,6)]
187-- @
188--
189-- The 'Semigroup' instance of this type works the same way as that of
190-- 'SerialT'.
191--
192-- @since 0.2.0
193newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
194 deriving (Semigroup, Monoid)
196-- | An IO stream whose applicative instance zips streams wAsyncly.
197--
198-- @since 0.2.0
199type ZipAsync = ZipAsyncM IO
201-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
202--
203-- @since 0.2.0
204zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
205zipAsyncly = K.adapt
207-- | Same as 'zipAsyncly'.
208--
209-- @since 0.1.0
210{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
211zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
212zippingAsync = zipAsyncly
214consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
215consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
217instance IsStream ZipAsyncM where
218 toStream = getZipAsyncM
219 fromStream = ZipAsyncM
221 {-# INLINE consM #-}
222 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
223 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
224 consM = consMZipAsync
226 {-# INLINE (|:) #-}
227 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
228 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
229 (|:) = consMZipAsync
231instance Monad m => Functor (ZipAsyncM m) where
232 fmap = map
234instance MonadAsync m => Applicative (ZipAsyncM m) where
235 pure = ZipAsyncM . K.repeat
236 m1 <*> m2 = zipAsyncWith id m1 m2