git.iko.soy/streamly
on e08ff86 help sign in
streamly/src/Streamly/Streams/Zip.hs
221 lines 6.8 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 : harendra.kumar@gmail.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(..), NFData1(..), rnf1)
43import Data.Functor.Identity (Identity, runIdentity)
44import Data.Foldable (fold)
45import Data.Semigroup (Semigroup(..))
46import GHC.Exts (IsList(..), IsString(..))
47import Text.Read (Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec,
48 readListPrecDefault)
49import Prelude hiding (map, repeat, zipWith)
51import Streamly.Streams.StreamK (IsStream(..), Stream, mkStream, unStream)
52import Streamly.Streams.Async (mkAsync')
53import Streamly.Streams.Serial (map)
54import Streamly.SVar (MonadAsync, adaptState)
56import qualified Streamly.Streams.Prelude as P
57import qualified Streamly.Streams.StreamK as K
59#include "Instances.hs"
61------------------------------------------------------------------------------
62-- Serially Zipping Streams
63------------------------------------------------------------------------------
65-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
66-- i.e. it produces one element from each stream serially and then zips all
67-- those elements.
68--
69-- @
70-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
71-- where s1 = fromFoldable [1, 2]
72-- s2 = fromFoldable [3, 4]
73-- s3 = fromFoldable [5, 6]
74-- @
75-- @
76-- [(1,3,5),(2,4,6)]
77-- @
78--
79-- The 'Semigroup' instance of this type works the same way as that of
80-- 'SerialT'.
81--
82-- @since 0.2.0
83newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
84 deriving (Semigroup, Monoid)
86-- |
87-- @since 0.1.0
88{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
89type ZipStream = ZipSerialM
91-- | An IO stream whose applicative instance zips streams serially.
92--
93-- @since 0.2.0
94type ZipSerial = ZipSerialM IO
96-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
97--
98-- @since 0.2.0
99zipSerially :: IsStream t => ZipSerialM m a -> t m a
100zipSerially = K.adapt
102-- | Same as 'zipSerially'.
103--
104-- @since 0.1.0
105{-# DEPRECATED zipping "Please use zipSerially instead." #-}
106zipping :: IsStream t => ZipSerialM m a -> t m a
107zipping = zipSerially
109instance IsStream ZipSerialM where
110 toStream = getZipSerialM
111 fromStream = ZipSerialM
113 {-# INLINE consM #-}
114 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
115 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
116 consM m r = fromStream $ K.consMSerial m (toStream r)
118 {-# INLINE (|:) #-}
119 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
120 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
121 m |: r = fromStream $ K.consMSerial m (toStream r)
123LIST_INSTANCES(ZipSerialM)
125instance Monad m => Functor (ZipSerialM m) where
126 fmap = map
128instance Monad m => Applicative (ZipSerialM m) where
129 pure = ZipSerialM . K.repeat
130 m1 <*> m2 = fromStream $ K.zipWith id (toStream m1) (toStream m2)
132FOLDABLE_INSTANCE(ZipSerialM)
133TRAVERSABLE_INSTANCE(ZipSerialM)
135------------------------------------------------------------------------------
136-- Parallel Zipping
137------------------------------------------------------------------------------
139-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
140-- are generated concurrently.
141--
142-- @since 0.1.0
143zipAsyncWith :: (IsStream t, MonadAsync m)
144 => (a -> b -> c) -> t m a -> t m b -> t m c
145zipAsyncWith f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do
146 ma <- mkAsync' (adaptState st) m1
147 mb <- mkAsync' (adaptState st) m2
148 unStream (toStream (K.zipWith f ma mb)) st stp sng yld
150-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
151-- are generated concurrently.
152--
153-- @since 0.4.0
154zipAsyncWithM :: (IsStream t, MonadAsync m)
155 => (a -> b -> m c) -> t m a -> t m b -> t m c
156zipAsyncWithM f m1 m2 = fromStream $ mkStream $ \st stp sng yld -> do
157 ma <- mkAsync' (adaptState st) m1
158 mb <- mkAsync' (adaptState st) m2
159 unStream (toStream (K.zipWithM f ma mb)) st stp sng yld
161------------------------------------------------------------------------------
162-- Parallely Zipping Streams
163------------------------------------------------------------------------------
164--
165-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
166-- be zipped concurrently.
167--
168-- @
169-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
170-- where s1 = fromFoldable [1, 2]
171-- s2 = fromFoldable [3, 4]
172-- s3 = fromFoldable [5, 6]
173-- @
174-- @
175-- [(1,3,5),(2,4,6)]
176-- @
177--
178-- The 'Semigroup' instance of this type works the same way as that of
179-- 'SerialT'.
180--
181-- @since 0.2.0
182newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
183 deriving (Semigroup, Monoid)
185-- | An IO stream whose applicative instance zips streams wAsyncly.
186--
187-- @since 0.2.0
188type ZipAsync = ZipAsyncM IO
190-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
191--
192-- @since 0.2.0
193zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
194zipAsyncly = K.adapt
196-- | Same as 'zipAsyncly'.
197--
198-- @since 0.1.0
199{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
200zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
201zippingAsync = zipAsyncly
202instance IsStream ZipAsyncM where
203 toStream = getZipAsyncM
204 fromStream = ZipAsyncM
206 {-# INLINE consM #-}
207 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
208 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
209 consM m r = fromStream $ K.consMSerial m (toStream r)
211 {-# INLINE (|:) #-}
212 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
213 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
214 m |: r = fromStream $ K.consMSerial m (toStream r)
216instance Monad m => Functor (ZipAsyncM m) where
217 fmap = map
219instance MonadAsync m => Applicative (ZipAsyncM m) where
220 pure = ZipAsyncM . K.repeat
221 m1 <*> m2 = zipAsyncWith id m1 m2