@@ -102,6 +102,9 @@ module Streamly.Internal.Data.Stream.Concurrent
102102 , concatMapInterleave
103103 , parConcatMap
104104
105+ -- *** Iterate
106+ , parIterateConcatMap
107+
105108 -- ** Reactive
106109 , fromCallback
107110 , pollCountsD
@@ -701,6 +704,31 @@ parZipWith :: MonadAsync m
701704 -> Stream m c
702705parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)
703706
707+-------------------------------------------------------------------------------
708+-- iterateConcatMap
709+-------------------------------------------------------------------------------
710+
711+-- | Same as 'iterateConcatMap' but concurrent.
712+--
713+{-# INLINE parIterateConcatMap #-}
714+parIterateConcatMap :: MonadAsync m =>
715+ (Config -> Config)
716+ -> (a -> Stream m a)
717+ -> Stream m a
718+ -> Stream m a
719+parIterateConcatMap modifier f input =
720+ Stream.fromStreamK
721+ $ withChannelK modifier (Stream.toStreamK input) iterateStream
722+
723+ where
724+
725+ iterateStream channel stream =
726+ parConcatMapChanKGeneric modifier channel (generate channel) stream
727+
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)
731+
704732-------------------------------------------------------------------------------
705733-- Generate
706734-------------------------------------------------------------------------------