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

git.iko.soy/streamly

on 4300a9f help sign in

50 lines 1.5 kB history raw

1{-# LANGUAGE UndecidableInstances #-}

2-- |

3-- Module : Streamly.Internal.Data.Stream.Concurrent

5-- License : BSD-3-Clause

6-- Maintainer : streamly@composewell.com

7-- Stability : experimental

8-- Portability : GHC

9--

10-- To run examples in this module:

12-- >>> import qualified Streamly.Internal.Data.Stream as Stream

13-- >>> import qualified Streamly.Data.Fold as Fold

15module Streamly.Internal.Data.Stream.Zip.Concurrent

16 (

17 ZipConcurrent (..)

18 )

19where

21import Streamly.Internal.Data.Stream.StreamD (Stream)

22import Streamly.Internal.Data.Stream.Concurrent (MonadAsync, parZipWith)

24import qualified Streamly.Internal.Data.Stream.StreamD as Stream (repeat)

25import Prelude hiding (map, repeat, zipWith)

27-- $setup

28-- >>> import qualified Streamly.Internal.Data.Stream as Stream

29-- >>> import qualified Streamly.Data.Fold as Fold

32newtype ZipConcurrent m a = ZipConcurrent {getZipConcurrent :: Stream m a}

33 deriving (Functor)

35-- | An IO stream whose applicative instance zips streams concurrently. Note

36-- that it uses the default concurrency options.

38-- >>> s = ZipConcurrent $ Stream.fromList [1, 2, 3]

39-- >>> x = (,,) <$> s <*> s <*> s

40-- >>> Stream.fold Fold.toList (getZipConcurrent x)

41-- [(1,1,1),(2,2,2),(3,3,3)]

43-- @since 0.9.0

45instance MonadAsync m => Applicative (ZipConcurrent m) where

46 pure = ZipConcurrent . Stream.repeat

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

49 ZipConcurrent m1 <*> ZipConcurrent m2 =

50 ZipConcurrent $ parZipWith id id m1 m2