@@ -102,6 +102,9 @@ module Streamly.Internal.Data.Stream.Concurrent
102 , concatMapInterleave
102 , concatMapInterleave
103 , parConcatMap
103 , parConcatMap
106+ , parIterateConcatMap
105 -- ** Reactive
108 -- ** Reactive
106 , fromCallback
109 , fromCallback
107 , pollCountsD
110 , pollCountsD
@@ -701,6 +704,31 @@ parZipWith :: MonadAsync m
701 -> Stream m c
704 -> Stream m c
702parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
705parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
707+-------------------------------------------------------------------------------
709+-------------------------------------------------------------------------------
711+-- | Same as 'iterateConcatMap' but concurrent.
713+{-# INLINE parIterateConcatMap #-}
714+parIterateConcatMap :: MonadAsync m =>
716+ -> (a -> Stream m a)
719+parIterateConcatMap modifier f input =
721+ $ withChannelK modifier (Stream.toStreamK input) iterateStream
725+ iterateStream channel stream =
726+ parConcatMapChanKGeneric modifier channel (generate channel) stream
728+ generate channel x =
729+ -- XXX The channel q should be FIFO for DFS, otherwise it is BFS
730+ x `K.cons` iterateStream channel (Stream.toStreamK $ f x)
704-------------------------------------------------------------------------------
732-------------------------------------------------------------------------------
705-- Generate
733-- Generate
706-------------------------------------------------------------------------------
734-------------------------------------------------------------------------------