git.iko.soy/streamly
on 6b192fb help sign in
streamly/src/Streamly/Streams/Zip.hs
208 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 StandaloneDeriving #-}
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 Data.Semigroup (Semigroup(..))
42import Prelude hiding (map, repeat, zipWith)
44import Streamly.Streams.StreamK (IsStream(..), Stream(..))
45import Streamly.Streams.Async (mkAsync')
46import Streamly.Streams.Serial (map)
47import Streamly.SVar (MonadAsync, rstState)
49import qualified Streamly.Streams.StreamK as K
51#include "Instances.hs"
53------------------------------------------------------------------------------
54-- Serially Zipping Streams
55------------------------------------------------------------------------------
57-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
58-- i.e. it produces one element from each stream serially and then zips all
59-- those elements.
60--
61-- @
62-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
63-- where s1 = fromFoldable [1, 2]
64-- s2 = fromFoldable [3, 4]
65-- s3 = fromFoldable [5, 6]
66-- @
67-- @
68-- [(1,3,5),(2,4,6)]
69-- @
70--
71-- The 'Semigroup' instance of this type works the same way as that of
72-- 'SerialT'.
73--
74-- @since 0.2.0
75newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}
76 deriving (Semigroup, Monoid)
78-- |
79-- @since 0.1.0
80{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
81type ZipStream = ZipSerialM
83-- | An IO stream whose applicative instance zips streams serially.
84--
85-- @since 0.2.0
86type ZipSerial a = ZipSerialM IO a
88-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
89--
90-- @since 0.2.0
91zipSerially :: IsStream t => ZipSerialM m a -> t m a
92zipSerially = K.adapt
94-- | Same as 'zipSerially'.
95--
96-- @since 0.1.0
97{-# DEPRECATED zipping "Please use zipSerially instead." #-}
98zipping :: IsStream t => ZipSerialM m a -> t m a
99zipping = zipSerially
101instance IsStream ZipSerialM where
102 toStream = getZipSerialM
103 fromStream = ZipSerialM
105 {-# INLINE consM #-}
106 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
107 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
108 consM m r = fromStream $ K.consMSerial m (toStream r)
110 {-# INLINE (|:) #-}
111 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
112 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
113 m |: r = fromStream $ K.consMSerial m (toStream r)
115instance Monad m => Functor (ZipSerialM m) where
116 fmap = map
118instance Monad m => Applicative (ZipSerialM m) where
119 pure = ZipSerialM . K.repeat
120 m1 <*> m2 = fromStream $ K.zipWith id (toStream m1) (toStream m2)
122------------------------------------------------------------------------------
123-- Parallel Zipping
124------------------------------------------------------------------------------
126-- | Zip two streams concurrently (i.e. both the elements being zipped are
127-- generated concurrently) using a pure zipping function.
128--
129-- @since 0.1.0
130zipAsyncWith :: (IsStream t, MonadAsync m)
131 => (a -> b -> c) -> t m a -> t m b -> t m c
132zipAsyncWith f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
133 ma <- mkAsync' (rstState st) m1
134 mb <- mkAsync' (rstState st) m2
135 unStream (toStream (K.zipWith f ma mb)) (rstState st) stp sng yld
137-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
138-- concurrently) using a monadic zipping function.
139--
140-- @since 0.4.0
141zipAsyncWithM :: (IsStream t, MonadAsync m)
142 => (a -> b -> m c) -> t m a -> t m b -> t m c
143zipAsyncWithM f m1 m2 = fromStream $ Stream $ \st stp sng yld -> do
144 ma <- mkAsync' (rstState st) m1
145 mb <- mkAsync' (rstState st) m2
146 unStream (toStream (K.zipWithM f ma mb)) (rstState st) stp sng yld
148------------------------------------------------------------------------------
149-- Parallely Zipping Streams
150------------------------------------------------------------------------------
151--
152-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
153-- be zipped concurrently.
154--
155-- @
156-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
157-- where s1 = fromFoldable [1, 2]
158-- s2 = fromFoldable [3, 4]
159-- s3 = fromFoldable [5, 6]
160-- @
161-- @
162-- [(1,3,5),(2,4,6)]
163-- @
164--
165-- The 'Semigroup' instance of this type works the same way as that of
166-- 'SerialT'.
167--
168-- @since 0.2.0
169newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
170 deriving (Semigroup, Monoid)
172-- | An IO stream whose applicative instance zips streams wAsyncly.
173--
174-- @since 0.2.0
175type ZipAsync a = ZipAsyncM IO a
177-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
178--
179-- @since 0.2.0
180zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
181zipAsyncly = K.adapt
183-- | Same as 'zipAsyncly'.
184--
185-- @since 0.1.0
186{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
187zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
188zippingAsync = zipAsyncly
189instance IsStream ZipAsyncM where
190 toStream = getZipAsyncM
191 fromStream = ZipAsyncM
193 {-# INLINE consM #-}
194 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
195 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
196 consM m r = fromStream $ K.consMSerial m (toStream r)
198 {-# INLINE (|:) #-}
199 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
200 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
201 m |: r = fromStream $ K.consMSerial m (toStream r)
203instance Monad m => Functor (ZipAsyncM m) where
204 fmap = map
206instance MonadAsync m => Applicative (ZipAsyncM m) where
207 pure = ZipAsyncM . K.repeat
208 m1 <*> m2 = zipAsyncWith id m1 m2