src/Streamly/Internal/Data/Stream/Zip.hs · streamly

git.iko.soy/streamly

on 796bb22 help sign in

streamly/src/Streamly/Internal/Data/Stream/Zip.hs

283 lines 8.5 kB history raw

1{-# LANGUAGE UndecidableInstances #-}

3-- |

4-- Module : Streamly.Internal.Data.Stream.Zip

6--

7-- License : BSD3

8-- Maintainer : streamly@composewell.com

9-- Stability : experimental

10-- Portability : GHC

11--

12-- To run examples in this module:

13--

14-- >>> import qualified Streamly.Prelude as Stream

15--

16module Streamly.Internal.Data.Stream.Zip

17 (

18 ZipSerialM

19 , ZipSerial

20 , zipSerially

22 , ZipAsyncM

23 , ZipAsync

24 , zipAsyncly

26 , zipWith

27 , zipWithM

28 , zipAsyncWith

29 , zipAsyncWithM

31 -- * Deprecated

32 , ZipStream

33 , zipping

34 , zippingAsync

35 )

36where

38import Control.Applicative (liftA2)

39import Control.DeepSeq (NFData(..))

40#if MIN_VERSION_deepseq(1,4,3)

41import Control.DeepSeq (NFData1(..))

42#endif

43import Data.Foldable (Foldable(foldl'), fold)

44import Data.Functor.Identity (Identity(..), runIdentity)

45import Data.Maybe (fromMaybe)

46import Data.Semigroup (Endo(..))

47#if __GLASGOW_HASKELL__ < 808

48import Data.Semigroup (Semigroup(..))

49#endif

50import GHC.Exts (IsList(..), IsString(..))

51import Text.Read

52 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec

53 , readListPrecDefault)

54import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)

55import Streamly.Internal.Data.Stream.StreamK (IsStream(..), Stream)

56import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

57import Streamly.Internal.Data.SVar (MonadAsync)

59import qualified Streamly.Internal.Data.Stream.Parallel as Par

60import qualified Streamly.Internal.Data.Stream.Prelude as P

61 (cmpBy, eqBy, foldl', foldr, fromList, toList, fromStreamS, toStreamS)

62import qualified Streamly.Internal.Data.Stream.StreamK as K (repeat)

63import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

64import qualified Streamly.Internal.Data.Stream.StreamD as D (zipWithM)

65import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

66#ifdef USE_STREAMK_ONLY

67import qualified Streamly.Internal.Data.Stream.StreamK as S (zipWith, zipWithM)

68#else

69import qualified Streamly.Internal.Data.Stream.StreamD as S (zipWith, zipWithM)

70#endif

72import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)

74#include "Instances.hs"

76-- $setup

77-- >>> import Control.Concurrent (threadDelay)

78-- >>> :{

79-- delay n = do

80-- threadDelay (n * 1000000) -- sleep for n seconds

81-- putStrLn (show n ++ " sec") -- print "n sec"

82-- return n -- IO Int

83-- :}

85-- | Like 'zipWith' but using a monadic zipping function.

86--

87-- @since 0.4.0

88{-# INLINABLE zipWithM #-}

89zipWithM :: (IsStream t, Monad m) => (a -> b -> m c) -> t m a -> t m b -> t m c

90zipWithM f m1 m2 = P.fromStreamS $ S.zipWithM f (P.toStreamS m1) (P.toStreamS m2)

92-- | Zip two streams serially using a pure zipping function.

93--

94-- @

95-- > S.toList $ S.zipWith (+) (S.fromList [1,2,3]) (S.fromList [4,5,6])

96-- [5,7,9]

97-- @

98--

99-- @since 0.1.0

100{-# INLINABLE zipWith #-}

101zipWith :: (IsStream t, Monad m) => (a -> b -> c) -> t m a -> t m b -> t m c

102zipWith f m1 m2 = P.fromStreamS $ S.zipWith f (P.toStreamS m1) (P.toStreamS m2)

103

104------------------------------------------------------------------------------

105-- Parallel Zipping

106------------------------------------------------------------------------------

107

108-- | Like 'zipWithM' but zips concurrently i.e. both the streams being zipped

109-- are generated concurrently.

110--

111-- @since 0.4.0

112{-# INLINE zipAsyncWithM #-}

113zipAsyncWithM :: (IsStream t, MonadAsync m)

114 => (a -> b -> m c) -> t m a -> t m b -> t m c

115zipAsyncWithM f m1 m2 = D.fromStreamD $

116 D.zipWithM f (Par.mkParallelD $ D.toStreamD m1)

117 (Par.mkParallelD $ D.toStreamD m2)

118

119-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped

120-- are generated concurrently.

121--

122-- @since 0.1.0

123{-# INLINE zipAsyncWith #-}

124zipAsyncWith :: (IsStream t, MonadAsync m)

125 => (a -> b -> c) -> t m a -> t m b -> t m c

126zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))

127

128------------------------------------------------------------------------------

129-- Serially Zipping Streams

130------------------------------------------------------------------------------

131

132-- | For 'ZipSerialM' streams:

133--

134-- @

135-- (<>) = 'Streamly.Prelude.serial'

136-- (<*>) = 'Streamly.Prelude.serial.zipWith' id

137-- @

138--

139-- Applicative evaluates the streams being zipped serially:

140--

141-- >>> s1 = Stream.fromFoldable [1, 2]

142-- >>> s2 = Stream.fromFoldable [3, 4]

143-- >>> s3 = Stream.fromFoldable [5, 6]

144-- >>> Stream.toList $ Stream.zipSerially $ (,,) <$> s1 <*> s2 <*> s3

145-- [(1,3,5),(2,4,6)]

146--

147-- /Since: 0.2.0 ("Streamly")/

148--

149-- @since 0.8.0

150newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: Stream m a}

151 deriving (Semigroup, Monoid)

152

153-- |

154-- @since 0.1.0

155{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}

156type ZipStream = ZipSerialM

157

158-- | An IO stream whose applicative instance zips streams serially.

159--

160-- /Since: 0.2.0 ("Streamly")/

161--

162-- @since 0.8.0

163type ZipSerial = ZipSerialM IO

164

165-- | Fix the type of a polymorphic stream as 'ZipSerialM'.

166--

167-- /Since: 0.2.0 ("Streamly")/

168--

169-- @since 0.8.0

170zipSerially :: IsStream t => ZipSerialM m a -> t m a

171zipSerially = K.adapt

172

173-- | Same as 'zipSerially'.

174--

175-- @since 0.1.0

176{-# DEPRECATED zipping "Please use zipSerially instead." #-}

177zipping :: IsStream t => ZipSerialM m a -> t m a

178zipping = zipSerially

179

180consMZip :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

181consMZip m ms = fromStream $ K.consMStream m (toStream ms)

182

183instance IsStream ZipSerialM where

184 toStream = getZipSerialM

185 fromStream = ZipSerialM

186

187 {-# INLINE consM #-}

188 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

189 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

190 consM = consMZip

191

192 {-# INLINE (|:) #-}

193 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}

194 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a

195 (|:) = consMZip

196

197LIST_INSTANCES(ZipSerialM)

198NFDATA1_INSTANCE(ZipSerialM)

199

200instance Monad m => Functor (ZipSerialM m) where

201 {-# INLINE fmap #-}

202 fmap f (ZipSerialM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m

203

204instance Monad m => Applicative (ZipSerialM m) where

205 pure = ZipSerialM . K.repeat

206 {-# INLINE (<*>) #-}

207 (<*>) = zipWith id

208

209FOLDABLE_INSTANCE(ZipSerialM)

210TRAVERSABLE_INSTANCE(ZipSerialM)

211

212------------------------------------------------------------------------------

213-- Parallely Zipping Streams

214------------------------------------------------------------------------------

215--

216-- | For 'ZipAsyncM' streams:

217--

218-- @

219-- (<>) = 'Streamly.Prelude.serial'

220-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id

221-- @

222--

223-- Applicative evaluates the streams being zipped concurrently, the following

224-- would take half the time that it would take in serial zipping:

225--

226-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]

227-- >>> Stream.toList $ Stream.zipAsyncly $ (,) <$> s <*> s

228-- ...

229-- [(1,1),(1,1),(1,1)]

230--

231-- /Since: 0.2.0 ("Streamly")/

232--

233-- @since 0.8.0

234newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}

235 deriving (Semigroup, Monoid)

236

237-- | An IO stream whose applicative instance zips streams wAsyncly.

238--

239-- /Since: 0.2.0 ("Streamly")/

240--

241-- @since 0.8.0

242type ZipAsync = ZipAsyncM IO

243

244-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.

245--

246-- /Since: 0.2.0 ("Streamly")/

247--

248-- @since 0.8.0

249zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a

250zipAsyncly = K.adapt

251

252-- | Same as 'zipAsyncly'.

253--

254-- @since 0.1.0

255{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}

256zippingAsync :: IsStream t => ZipAsyncM m a -> t m a

257zippingAsync = zipAsyncly

258

259consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

260consMZipAsync m ms = fromStream $ K.consMStream m (toStream ms)

261

262instance IsStream ZipAsyncM where

263 toStream = getZipAsyncM

264 fromStream = ZipAsyncM

265

266 {-# INLINE consM #-}

267 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

268 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

269 consM = consMZipAsync

270

271 {-# INLINE (|:) #-}

272 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}

273 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a

274 (|:) = consMZipAsync

275

276instance Monad m => Functor (ZipAsyncM m) where

277 {-# INLINE fmap #-}

278 fmap f (ZipAsyncM m) = D.fromStreamD $ D.mapM (return . f) $ D.toStreamD m

279

280instance MonadAsync m => Applicative (ZipAsyncM m) where

281 pure = ZipAsyncM . K.repeat

282 {-# INLINE (<*>) #-}

283 m1 <*> m2 = zipAsyncWith id m1 m2