@@ -45,7 +45,8 @@ module Streamly.Internal.Data.Stream.IsStream.Expand
4545 , append
4646
4747 -- ** 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
4950 -- used with 'concatMapWith' as well, however, the interleaving behavior of
5051 -- @n@ streams would be asymmetric giving exponentially more weightage to
5152 -- streams that come earlier in the composition.
@@ -77,6 +78,7 @@ module Streamly.Internal.Data.Stream.IsStream.Expand
7778 -- , merge
7879 , mergeBy
7980 , mergeByM
81+ , mergeByMFused
8082 , mergeAsyncBy
8183 , mergeAsyncByM
8284
@@ -167,20 +169,15 @@ import Streamly.Internal.Data.Stream.IsStream.Common
167169 ( concatM, concatMapM, concatMap, smapM, fromPure, fromEffect, parallelFst
168170 , zipWith, zipWithM)
169171import Streamly.Internal.Data.Stream.IsStream.Type
170− (IsStream(..), fromStreamS, toStreamS, fromStreamD, toStreamD)
172+ (IsStream(..), fromStreamD, toStreamD)
171173import Streamly.Internal.Data.Unfold.Type (Unfold)
172174
173175import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
174176import qualified Streamly.Internal.Data.Stream.Parallel as Par
175177import qualified Streamly.Internal.Data.Stream.Serial as Serial
176178import 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)
178180import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
179−#ifdef USE_STREAMK_ONLY
180−import qualified Streamly.Internal.Data.Stream.StreamK as S
181−#else
182−import qualified Streamly.Internal.Data.Stream.StreamD as S
183−#endif
184181import qualified Streamly.Internal.Data.Stream.Zip as Zip
185182
186183import Prelude hiding (concat, concatMap, zipWith)
@@ -706,6 +703,8 @@ zipAsyncWith f = zipAsyncWithM (\a b -> return (f a b))
706703--
707704-- @
708705--
706+-- See also: 'mergeByMFused'
707+--
709708-- @since 0.6.0
710709{-# INLINE mergeBy #-}
711710mergeBy :: 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)
740739--
741740-- @
742741--
742+-- See also: 'mergeByMFused'
743+--
743744-- @since 0.6.0
744745{-# INLINABLE mergeByM #-}
745746mergeByM
746747 :: (IsStream t, Monad m)
747748 => (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)
750+
751+-- XXX Fused versions should probably go to a separate module using the same
752+-- names for the combinators.
753+--
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'.
758+--
759+-- /Internal/
760+{-# INLINE mergeByMFused #-}
761+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)
749766
750767{-
751768-- | Like 'mergeByM' but stops merging as soon as any of the two streams stops.