git.iko.soy/streamly
on 4df1a93 help sign in
streamly/src/Streamly/Streams/Zip.hs
207 lines 6.4 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 UndecidableInstances #-} -- XXX
10-- |
11-- Module : Streamly.Streams.Zip
12-- Copyright : (c) 2017 Harendra Kumar
13--
14-- License : BSD3
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
17-- Portability : GHC
18--
19--
20module Streamly.Streams.Zip
21 (
22 K.zipWith
23 , K.zipWithM
24 , zipAsyncWith
25 , zipAsyncWithM
27 , ZipSerialM
28 , ZipSerial
29 , ZipStream -- deprecated
30 , zipSerially
31 , zipping -- deprecated
33 , ZipAsyncM
34 , ZipAsync
35 , zipAsyncly
36 , zippingAsync -- deprecated
37 )
38where
40import Data.Semigroup (Semigroup(..))
41import Prelude hiding (map, repeat, zipWith)
43import Streamly.Streams.StreamK (IsStream(..), Stream(..))
44import Streamly.Streams.Async (mkAsync')
45import Streamly.Streams.Serial (map)
46import Streamly.SVar (MonadAsync, rstState)
48import qualified Streamly.Streams.StreamK as K
50#include "Instances.hs"
52------------------------------------------------------------------------------
53-- Serially Zipping Streams
54------------------------------------------------------------------------------
56-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
57-- i.e. it produces one element from each stream serially and then zips all
58-- those elements.
59--
60-- @
61-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
62-- where s1 = fromFoldable [1, 2]
63-- s2 = fromFoldable [3, 4]
64-- s3 = fromFoldable [5, 6]
65-- @
66-- @
67-- [(1,3,5),(2,4,6)]
68-- @
69--
70-- The 'Semigroup' instance of this type works the same way as that of
71-- 'SerialT'.
72--
73-- @since 0.2.0
74newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
75 deriving (Semigroup, Monoid)
77-- |
78-- @since 0.1.0
79{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
80type ZipStream = ZipSerialM
82-- | An IO stream whose applicative instance zips streams serially.
83--
84-- @since 0.2.0
85type ZipSerial = ZipSerialM IO
87-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
88--
89-- @since 0.2.0
90zipSerially :: IsStream t => ZipSerialM m a -> t m a
91zipSerially = K.adapt
93-- | Same as 'zipSerially'.
94--
95-- @since 0.1.0
96{-# DEPRECATED zipping "Please use zipSerially instead." #-}
97zipping :: IsStream t => ZipSerialM m a -> t m a
98zipping = zipSerially
100instance IsStream ZipSerialM where
101 toStream = getZipSerialM
102 fromStream = ZipSerialM
104 {-# INLINE consM #-}
105 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
106 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
107 consM m r = fromStream $ K.consMSerial m (toStream r)
109 {-# INLINE (|:) #-}
110 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
111 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
112 m |: r = fromStream $ K.consMSerial m (toStream r)
114instance Monad m => Functor (ZipSerialM m) where
115 fmap = map
117instance Monad m => Applicative (ZipSerialM m) where
118 pure = ZipSerialM . K.repeat
119 m1 <*> m2 = fromStream $ K.zipWith id (toStream m1) (toStream m2)
121------------------------------------------------------------------------------
122-- Parallel Zipping
123------------------------------------------------------------------------------
125-- | Zip two streams concurrently (i.e. both the elements being zipped are
126-- generated concurrently) using a pure zipping function.
127--
128-- @since 0.1.0
129zipAsyncWith :: (IsStream t, MonadAsync m)
130 => (a -> b -> c) -> t m a -> t m b -> t m c
131zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
132 ma <- mkAsync' (rstState st) m1
133 mb <- mkAsync' (rstState st) m2
134 unStream (toStream (K.zipWith f ma mb)) (rstState st) stp sng yld
136-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
137-- concurrently) using a monadic zipping function.
138--
139-- @since 0.4.0
140zipAsyncWithM :: (IsStream t, MonadAsync m)
141 => (a -> b -> m c) -> t m a -> t m b -> t m c
142zipAsyncWithM f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
143 ma <- mkAsync' (rstState st) m1
144 mb <- mkAsync' (rstState st) m2
145 unStream (toStream (K.zipWithM f ma mb)) (rstState st) stp sng yld
147------------------------------------------------------------------------------
148-- Parallely Zipping Streams
149------------------------------------------------------------------------------
150--
151-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
152-- be zipped concurrently.
153--
154-- @
155-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
156-- where s1 = fromFoldable [1, 2]
157-- s2 = fromFoldable [3, 4]
158-- s3 = fromFoldable [5, 6]
159-- @
160-- @
161-- [(1,3,5),(2,4,6)]
162-- @
163--
164-- The 'Semigroup' instance of this type works the same way as that of
165-- 'SerialT'.
166--
167-- @since 0.2.0
168newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
169 deriving (Semigroup, Monoid)
171-- | An IO stream whose applicative instance zips streams wAsyncly.
172--
173-- @since 0.2.0
174type ZipAsync = ZipAsyncM IO
176-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
177--
178-- @since 0.2.0
179zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
180zipAsyncly = K.adapt
182-- | Same as 'zipAsyncly'.
183--
184-- @since 0.1.0
185{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
186zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
187zippingAsync = zipAsyncly
188instance IsStream ZipAsyncM where
189 toStream = getZipAsyncM
190 fromStream = ZipAsyncM
192 {-# INLINE consM #-}
193 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
194 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
195 consM m r = fromStream $ K.consMSerial m (toStream r)
197 {-# INLINE (|:) #-}
198 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
199 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
200 m |: r = fromStream $ K.consMSerial m (toStream r)
202instance Monad m => Functor (ZipAsyncM m) where
203 fmap = map
205instance MonadAsync m => Applicative (ZipAsyncM m) where
206 pure = ZipAsyncM . K.repeat
207 m1 <*> m2 = zipAsyncWith id m1 m2