git.iko.soy/streamly
on ede29df help sign in
streamly/src/Streamly/Streams/Zip.hs
228 lines 6.9 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, foldStream)
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
109consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
110consMZip m ms = fromStream $ K.consMStream m (toStream ms)
112instance IsStream ZipSerialM where
113 toStream = getZipSerialM
114 fromStream = ZipSerialM
116 {-# INLINE consM #-}
117 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
118 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
119 consM = consMZip
121 {-# INLINE (|:) #-}
122 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
123 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
124 (|:) = consMZip
126LIST_INSTANCES(ZipSerialM)
128instance Monad m => Functor (ZipSerialM m) where
129 fmap = map
131instance Monad m => Applicative (ZipSerialM m) where
132 pure = ZipSerialM . K.repeat
133 (<*>) = K.zipWith id
135FOLDABLE_INSTANCE(ZipSerialM)
136TRAVERSABLE_INSTANCE(ZipSerialM)
138------------------------------------------------------------------------------
139-- Parallel Zipping
140------------------------------------------------------------------------------
142-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
143-- are generated concurrently.
144--
145-- @since 0.1.0
146zipAsyncWith :: (IsStream t, MonadAsync m)
147 => (a -> b -> c) -> t m a -> t m b -> t m c
148zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
149 ma <- mkAsync' (adaptState st) m1
150 mb <- mkAsync' (adaptState st) m2
151 foldStream st stp sng yld (K.zipWith f ma mb)
153-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
154-- are generated concurrently.
155--
156-- @since 0.4.0
157zipAsyncWithM :: (IsStream t, MonadAsync m)
158 => (a -> b -> m c) -> t m a -> t m b -> t m c
159zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do
160 ma <- mkAsync' (adaptState st) m1
161 mb <- mkAsync' (adaptState st) m2
162 foldStream st stp sng yld (K.zipWithM f ma mb)
164------------------------------------------------------------------------------
165-- Parallely Zipping Streams
166------------------------------------------------------------------------------
167--
168-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
169-- be zipped concurrently.
170--
171-- @
172-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
173-- where s1 = fromFoldable [1, 2]
174-- s2 = fromFoldable [3, 4]
175-- s3 = fromFoldable [5, 6]
176-- @
177-- @
178-- [(1,3,5),(2,4,6)]
179-- @
180--
181-- The 'Semigroup' instance of this type works the same way as that of
182-- 'SerialT'.
183--
184-- @since 0.2.0
185newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
186 deriving (Semigroup, Monoid)
188-- | An IO stream whose applicative instance zips streams wAsyncly.
189--
190-- @since 0.2.0
191type ZipAsync = ZipAsyncM IO
193-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
194--
195-- @since 0.2.0
196zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
197zipAsyncly = K.adapt
199-- | Same as 'zipAsyncly'.
200--
201-- @since 0.1.0
202{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
203zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
204zippingAsync = zipAsyncly
206consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
207consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
209instance IsStream ZipAsyncM where
210 toStream = getZipAsyncM
211 fromStream = ZipAsyncM
213 {-# INLINE consM #-}
214 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
215 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
216 consM = consMZipAsync
218 {-# INLINE (|:) #-}
219 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
220 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
221 (|:) = consMZipAsync
223instance Monad m => Functor (ZipAsyncM m) where
224 fmap = map
226instance MonadAsync m => Applicative (ZipAsyncM m) where
227 pure = ZipAsyncM . K.repeat
228 m1 <*> m2 = zipAsyncWith id m1 m2