git.iko.soy/streamly
on b604937 help sign in
streamly/src/Streamly/Streams/Zip.hs
230 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
146{-# INLINABLE zipAsyncWith #-}
147zipAsyncWith :: (IsStream t, MonadAsync m)
148 => (a -> b -> c) -> t m a -> t m b -> t m c
149zipAsyncWith f m1 m2 = mkStream $ \st stp sng yld -> do
150 ma <- mkAsync' (adaptState st) m1
151 mb <- mkAsync' (adaptState st) m2
152 foldStream st stp sng yld (K.zipWith f ma mb)
154-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped
155-- are generated concurrently.
156--
157-- @since 0.4.0
158{-# INLINABLE zipAsyncWithM #-}
159zipAsyncWithM :: (IsStream t, MonadAsync m)
160 => (a -> b -> m c) -> t m a -> t m b -> t m c
161zipAsyncWithM f m1 m2 = mkStream $ \st stp sng yld -> do
162 ma <- mkAsync' (adaptState st) m1
163 mb <- mkAsync' (adaptState st) m2
164 foldStream st stp sng yld (K.zipWithM f ma mb)
166------------------------------------------------------------------------------
167-- Parallely Zipping Streams
168------------------------------------------------------------------------------
169--
170-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
171-- be zipped concurrently.
172--
173-- @
174-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
175-- where s1 = fromFoldable [1, 2]
176-- s2 = fromFoldable [3, 4]
177-- s3 = fromFoldable [5, 6]
178-- @
179-- @
180-- [(1,3,5),(2,4,6)]
181-- @
182--
183-- The 'Semigroup' instance of this type works the same way as that of
184-- 'SerialT'.
185--
186-- @since 0.2.0
187newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
188 deriving (Semigroup, Monoid)
190-- | An IO stream whose applicative instance zips streams wAsyncly.
191--
192-- @since 0.2.0
193type ZipAsync = ZipAsyncM IO
195-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
196--
197-- @since 0.2.0
198zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
199zipAsyncly = K.adapt
201-- | Same as 'zipAsyncly'.
202--
203-- @since 0.1.0
204{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
205zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
206zippingAsync = zipAsyncly
208consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
209consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)
211instance IsStream ZipAsyncM where
212 toStream = getZipAsyncM
213 fromStream = ZipAsyncM
215 {-# INLINE consM #-}
216 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
217 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
218 consM = consMZipAsync
220 {-# INLINE (|:) #-}
221 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
222 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
223 (|:) = consMZipAsync
225instance Monad m => Functor (ZipAsyncM m) where
226 fmap = map
228instance MonadAsync m => Applicative (ZipAsyncM m) where
229 pure = ZipAsyncM . K.repeat
230 m1 <*> m2 = zipAsyncWith id m1 m2