9922101 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move common zip functions to Core.hs

9922101Harendra Kumar8 years ago

src/Streamly/Core.hs+32−1

compare with master

unifiedsplit

@@ -41,6 +41,10 @@ module Streamly.Core

41 -- * Alternative

42 , alt

44+ -- * zip

45+ , zipWith

46+ , zipParallelWith

47+

44 -- * Transformers

48 -- * Transformers

45 , withLocal

49 , withLocal

46 , withCatchError

50 , withCatchError

@@ -83,7 +87,7 @@ import Data.Maybe (isNothing)

83import Data.Semigroup (Semigroup(..))

87import Data.Semigroup (Semigroup(..))

84import Data.Set (Set)

88import Data.Set (Set)

85import qualified Data.Set as S

89import qualified Data.Set as S

86−import Prelude hiding (repeat)

90+import Prelude hiding (repeat, zipWith)

88------------------------------------------------------------------------------

92------------------------------------------------------------------------------

89-- Parent child thread communication type

93-- Parent child thread communication type

@@ -602,6 +606,33 @@ alt m1 m2 = Stream $ \_ stp sng yld ->

602 let stop = runStream m2 Nothing stp sng yld

606 let stop = runStream m2 Nothing stp sng yld

603 in runStream m1 Nothing stop sng yld

607 in runStream m1 Nothing stop sng yld

604

608

609+------------------------------------------------------------------------------

610+-- Zipping

611+------------------------------------------------------------------------------

612+

613+zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

614+zipWith f m1 m2 = go m1 m2

615+ where

616+ go mx my = Stream $ \_ stp sng yld -> do

617+ let merge a ra =

618+ let single2 b = sng (f a b)

619+ yield2 b rb = yld (f a b) (go ra rb)

620+ in (runStream my) Nothing stp single2 yield2

621+ let single1 a = merge a nil

622+ yield1 a ra = merge a ra

623+ (runStream mx) Nothing stp single1 yield1

624+

625+async :: MonadParallel m => Stream m a -> m (Stream m a)

626+async m = newStreamVar1 (SVarStyle Disjunction LIFO) m

627+ >>= return . fromStreamVar

628+

629+zipParallelWith :: MonadParallel m

630+ => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

631+zipParallelWith f m1 m2 = Stream $ \_ stp sng yld -> do

632+ ma <- async m1

633+ mb <- async m2

634+ (runStream (zipWith f ma mb)) Nothing stp sng yld

635+

605-------------------------------------------------------------------------------

636-------------------------------------------------------------------------------

606-- Transformers

637-- Transformers

607-------------------------------------------------------------------------------

638-------------------------------------------------------------------------------

src/Streamly/Streams.hs+13−24

@@ -577,8 +577,8 @@ instance IsStream CoparallelT where

577-- Semigroup

578------------------------------------------------------------------------------

579

580−-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT', but

580+-- | Polymorphic version of the 'Semigroup' operation '<>' of 'CoparallelT'.

581−-- polymorphic. Merges two streams possibly concurrently, preferring the

581+-- Merges two streams possibly concurrently, preferring the

582-- elements from the left one when available.

583{-# INLINE coparallel #-}

584coparallel :: (IsStream t, MonadParallel m) => t m a -> t m a -> t m a

@@ -723,16 +723,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

723

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

725zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c

726−zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

726+zipWith f m1 m2 = fromStream $ S.zipWith f (toStream m1) (toStream m2)

727− where

728− go mx my = S.Stream $ \_ stp sng yld -> do

729− let merge a ra =

730− let single2 b = sng (f a b)

731− yield2 b rb = yld (f a b) (go ra rb)

732− in (S.runStream my) Nothing stp single2 yield2

733− let single1 a = merge a S.nil

734− yield1 a ra = merge a ra

735− (S.runStream mx) Nothing stp single1 yield1

736

727

737-- | The applicative instance of 'ZipStreamM' zips a number of streams serially

728-- | The applicative instance of 'ZipStreamM' zips a number of streams serially

738-- i.e. it produces one element from each stream serially and then zips all

729-- i.e. it produces one element from each stream serially and then zips all

@@ -754,14 +745,14 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

754newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}

745newtype ZipStreamM m a = ZipStreamM {getZipStreamM :: S.Stream m a}

755 deriving (Functor, Semigroup, Monoid)

746 deriving (Functor, Semigroup, Monoid)

756

747

757−instance Monad m => Applicative (ZipStreamM m) where

758− pure = ZipStreamM . S.repeat

759− (<*>) = zipWith id

760−

761instance IsStream ZipStreamM where

748instance IsStream ZipStreamM where

762 toStream = getZipStreamM

749 toStream = getZipStreamM

763 fromStream = ZipStreamM

750 fromStream = ZipStreamM

764

751

752+instance Monad m => Applicative (ZipStreamM m) where

753+ pure = ZipStreamM . S.repeat

754+ m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)

755+

765------------------------------------------------------------------------------

756------------------------------------------------------------------------------

766-- Parallely Zipping Streams

757-- Parallely Zipping Streams

767------------------------------------------------------------------------------

758------------------------------------------------------------------------------

@@ -770,10 +761,8 @@ instance IsStream ZipStreamM where

770-- generated concurrently) using a pure zipping function.

761-- generated concurrently) using a pure zipping function.

771zipParallelWith :: (IsStream t, MonadParallel m)

762zipParallelWith :: (IsStream t, MonadParallel m)

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

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

773−zipParallelWith f m1 m2 = fromStream $ S.Stream $ \_ stp sng yld -> do

764+zipParallelWith f m1 m2 =

774− ma <- async m1

765+ fromStream $ S.zipParallelWith f (toStream m1) (toStream m2)

775− mb <- async m2

776− (S.runStream (toStream (zipWith f ma mb))) Nothing stp sng yld

777

766

778{-# DEPRECATED zipAsyncWith "Please use zipParallelWith instead." #-}

767{-# DEPRECATED zipAsyncWith "Please use zipParallelWith instead." #-}

779zipAsyncWith :: (IsStream t, MonadParallel m)

768zipAsyncWith :: (IsStream t, MonadParallel m)

@@ -802,14 +791,14 @@ newtype ZipParallelM m a = ZipParallelM {getZipParallelM :: S.Stream m a}

802{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}

791{-# DEPRECATED ZipAsync "Please use ZipParallelM instead." #-}

803type ZipAsync = ZipParallelM

792type ZipAsync = ZipParallelM

804

793

805−instance MonadParallel m => Applicative (ZipParallelM m) where

806− pure = ZipParallelM . S.repeat

807− (<*>) = zipParallelWith id

808−

809instance IsStream ZipParallelM where

794instance IsStream ZipParallelM where

810 toStream = getZipParallelM

795 toStream = getZipParallelM

811 fromStream = ZipParallelM

796 fromStream = ZipParallelM

812

797

798+instance MonadParallel m => Applicative (ZipParallelM m) where

799+ pure = ZipParallelM . S.repeat

800+ m1 <*> m2 = fromStream $ S.zipParallelWith id (toStream m1) (toStream m2)

801+

813-------------------------------------------------------------------------------

802-------------------------------------------------------------------------------

814-- Type adapting combinators

803-- Type adapting combinators

815-------------------------------------------------------------------------------

804-------------------------------------------------------------------------------