@@ -45,7 +45,8 @@ module Streamly.Internal.Data.Stream.IsStream.Expand
47 -- ** wSerial
47 -- ** wSerial
48− -- | 'wSerial' is a CPS based stream interleaving functions. It can be
48+ -- | 'wSerial' is a CPS based stream interleaving functions. Use
49+ -- 'concatPairsWith wSerial' to interleave @n@ streams uniformly. It can be
49 -- used with 'concatMapWith' as well, however, the interleaving behavior of
50 -- used with 'concatMapWith' as well, however, the interleaving behavior of
50 -- @n@ streams would be asymmetric giving exponentially more weightage to
51 -- @n@ streams would be asymmetric giving exponentially more weightage to
51 -- streams that come earlier in the composition.
52 -- streams that come earlier in the composition.
@@ -77,6 +78,7 @@ module Streamly.Internal.Data.Stream.IsStream.Expand
77 -- , merge
78 -- , merge
79 , mergeByM
80 , mergeByM
80 , mergeAsyncBy
82 , mergeAsyncBy
81 , mergeAsyncByM
83 , mergeAsyncByM
@@ -167,20 +169,15 @@ import Streamly.Internal.Data.Stream.IsStream.Common
167 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect, parallelFst
169 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect, parallelFst
168 , zipWith, zipWithM)
170 , zipWith, zipWithM)
169import Streamly.Internal.Data.Stream.IsStream.Type
171import Streamly.Internal.Data.Stream.IsStream.Type
170− (IsStream(..), fromStreamS, toStreamS, fromStreamD, toStreamD)
172+ (IsStream(..), fromStreamD, toStreamD)
171import Streamly.Internal.Data.Unfold.Type (Unfold)
173import Streamly.Internal.Data.Unfold.Type (Unfold)
173import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
175import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
174import qualified Streamly.Internal.Data.Stream.Parallel as Par
176import qualified Streamly.Internal.Data.Stream.Parallel as Par
175import qualified Streamly.Internal.Data.Stream.Serial as Serial
177import qualified Streamly.Internal.Data.Stream.Serial as Serial
176import qualified Streamly.Internal.Data.Stream.StreamD as D
178import qualified Streamly.Internal.Data.Stream.StreamD as D
177−import qualified Streamly.Internal.Data.Stream.StreamK as K (mergeBy)
179+import qualified Streamly.Internal.Data.Stream.StreamK as K (mergeBy, mergeByM)
178import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
180import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
179−#ifdef USE_STREAMK_ONLY
180−import qualified Streamly.Internal.Data.Stream.StreamK as S
182−import qualified Streamly.Internal.Data.Stream.StreamD as S
184import qualified Streamly.Internal.Data.Stream.Zip as Zip
181import qualified Streamly.Internal.Data.Stream.Zip as Zip
186import Prelude hiding (concat, concatMap, zipWith)
183import Prelude hiding (concat, concatMap, zipWith)
@@ -706,6 +703,8 @@ zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
706+-- See also: 'mergeByMFused'
709-- @since 0.6.0
708-- @since 0.6.0
710{-# INLINE mergeBy #-}
709{-# INLINE mergeBy #-}
711mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a
710mergeBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a -> t m a
@@ -740,12 +739,30 @@ mergeBy f m1 m2 = fromStream $ K.mergeBy f (toStream m1) (toStream m2)
742+-- See also: 'mergeByMFused'
743-- @since 0.6.0
744-- @since 0.6.0
744{-# INLINABLE mergeByM #-}
745{-# INLINABLE mergeByM #-}
746 :: (IsStream t, Monad m)
747 :: (IsStream t, Monad m)
747 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
748 => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
748−mergeByM f m1 m2 = fromStreamS $ S.mergeByM f (toStreamS m1) (toStreamS m2)
749+mergeByM f m1 m2 = fromStream $ K.mergeByM f (toStream m1) (toStream m2)
751+-- XXX Fused versions should probably go to a separate module using the same
752+-- names for the combinators.
754+-- | Like 'mergeByM' but much faster, works best when merging statically known
755+-- number of streams. When merging more than two streams try to merge pairs and
756+-- pair pf pairs in a tree like structure.'mergeByM' works better with variable
757+-- number of streams being merged using 'concatPairsWith'.
760+{-# INLINE mergeByMFused #-}
762+ :: (IsStream t, Monad m)
763+ => (a -> a -> m Ordering) -> t m a -> t m a -> t m a
764+mergeByMFused f m1 m2 =
765+ fromStreamD $ D.mergeByM f (toStreamD m1) (toStreamD m2)
751-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.
768-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.