9922101 · streamly · git.iko.soy

git.iko.soy/streamly

Move common zip functions to Core.hs

commit9922101bc0b46156d7c147d89f11c9189acf6db7 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 3 May 2018, 03:12 (8 years ago) parenta90320a

2 files changed+45−25

compare with master

unifiedsplit

src/Streamly/Core.hs+32−1 src/Streamly/Streams.hs+13−24

src/Streamly/Core.hs+32−1

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

4141 -- * Alternative

4242 , alt

4343

44+ -- * zip

45+ , zipWith

46+ , zipParallelWith

47+

4448 -- * Transformers

4549 , withLocal

4650 , withCatchError

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

8387import Data.Semigroup (Semigroup(..))

8488import Data.Set (Set)

8589import qualified Data.Set as S

86−import Prelude hiding (repeat)

90+import Prelude hiding (repeat, zipWith)

8791

8892------------------------------------------------------------------------------

8993-- Parent child thread communication type

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

602606 let stop = runStream m2 Nothing stp sng yld

603607 in runStream m1 Nothing stop sng yld

604608

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+

605636-------------------------------------------------------------------------------

606637-- Transformers

607638-------------------------------------------------------------------------------

src/Streamly/Streams.hs+13−24

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

577577-- Semigroup

578578------------------------------------------------------------------------------

579579

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

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

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

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

582582-- elements from the left one when available.

583583{-# INLINE coparallel #-}

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

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

723723

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

725725zipWith :: 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)

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

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

736727

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

738729-- 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)

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

755746 deriving (Functor, Semigroup, Monoid)

756747

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

758− pure = ZipStreamM . S.repeat

759− (<*>) = zipWith id

760−

761748instance IsStream ZipStreamM where

762749 toStream = getZipStreamM

763750 fromStream = ZipStreamM

764751

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+

765756------------------------------------------------------------------------------

766757-- Parallely Zipping Streams

767758------------------------------------------------------------------------------

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

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

771762zipParallelWith :: (IsStream t, MonadParallel m)

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

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

774− ma <- async m1

775− mb <- async m2

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

764+zipParallelWith f m1 m2 =

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

777766

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

779768zipAsyncWith :: (IsStream t, MonadParallel m)

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

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

803792type ZipAsync = ZipParallelM

804793

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

806− pure = ZipParallelM . S.repeat

807− (<*>) = zipParallelWith id

808−

809794instance IsStream ZipParallelM where

810795 toStream = getZipParallelM

811796 fromStream = ZipParallelM

812797

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+

813802-------------------------------------------------------------------------------

814803-- Type adapting combinators

815804-------------------------------------------------------------------------------