@@ -65,6 +65,8 @@ module Streamly.Internal.Data.Stream.Concurrent
6565 , parTwo
6666 , parZipWithM
6767 , parZipWith
68+ , parMergeByM
69+ , parMergeBy
6870
6971 -- ** List of streams
7072 -- | Shares a single channel across many streams.
@@ -582,6 +584,36 @@ parZipWith :: MonadAsync m
582584 -> Stream m c
583585parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
584586
587+-- | Like 'mergeByM' but evaluates both the streams concurrently.
588+--
589+-- Definition:
590+--
591+-- >>> parMergeByM cfg f m1 m2 = Stream.mergeByM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)
592+--
593+{-# INLINE parMergeByM #-}
594+parMergeByM :: MonadAsync m
595+ => (Config -> Config)
596+ -> (a -> a -> m Ordering)
597+ -> Stream m a
598+ -> Stream m a
599+ -> Stream m a
600+parMergeByM cfg f m1 m2 = Stream.mergeByM f (parEval cfg m1) (parEval cfg m2)
601+
602+-- | Like 'mergeBy' but evaluates both the streams concurrently.
603+--
604+-- Definition:
605+--
606+-- >>> parMergeBy cfg f = Stream.parMergeByM cfg (\a b -> return $ f a b)
607+--
608+{-# INLINE parMergeBy #-}
609+parMergeBy :: MonadAsync m
610+ => (Config -> Config)
611+ -> (a -> a -> Ordering)
612+ -> Stream m a
613+ -> Stream m a
614+ -> Stream m a
615+parMergeBy cfg f = parMergeByM cfg (\a b -> return $ f a b)
616+
585617-------------------------------------------------------------------------------
586618-- concatIterate
587619-------------------------------------------------------------------------------