@@ -207,6 +207,7 @@ module Streamly.Fold
207207
208208 -- Element unaware grouping
209209 , groupsOf
210+ , intervalsOf
210211
211212 -- Element aware grouping
212213 , groups
@@ -296,10 +297,13 @@ module Streamly.Fold
296297 )
297298where
298299
300+import Control.Concurrent (threadDelay)
299301import Control.Monad (void)
300302import Control.Monad.IO.Class (MonadIO(..))
301303import Data.Functor.Identity (Identity)
302304import Data.Map.Strict (Map)
305+import Data.Maybe (fromJust, isJust)
306+
303307import Foreign.Storable (Storable(..))
304308import Prelude
305309 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
@@ -311,6 +315,7 @@ import Prelude
311315
312316import qualified Data.Map.Strict as Map
313317
318+import Streamly (MonadAsync, parallel)
314319import Streamly.Array (Array)
315320import Streamly.Fold.Types (Fold(..))
316321import Streamly.Streams.Serial (SerialT)
@@ -318,7 +323,6 @@ import Streamly.Streams.StreamK (IsStream())
318323
319324import Streamly.Strict
320325
321−import Streamly (MonadAsync)
322326import qualified Streamly.Array as A
323327import qualified Streamly.Prelude as S
324328import qualified Streamly.Streams.StreamD as D
@@ -973,6 +977,30 @@ groupsOf
973977 => Int -> Fold m a b -> t m a -> t m b
974978groupsOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
975979
980+-- | Transform a fold from a pure input to a 'Maybe' input, consuming only
981+-- 'Just' values.
982+lcatMaybes :: Monad m => Fold m a b -> Fold m (Maybe a) b
983+lcatMaybes = lfilter isJust . lmap fromJust
984+
985+-- XXX add this example after fixing the serial stream rate control
986+-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
987+-- > [3,7,11,15,19]
988+--
989+-- | Group the input stream into windows of @n@ second each and then fold each
990+-- group using the provided fold function.
991+--
992+-- @since 0.7.0
993+{-# INLINE intervalsOf #-}
994+intervalsOf
995+ :: (IsStream t, MonadAsync m)
996+ => Double -> Fold m a b -> t m a -> t m b
997+intervalsOf n f m = grouped (lcatMaybes f) s
998+ where
999+ s = S.map (\x -> (Just x, False)) m `parallel` S.repeatM timeout
1000+ timeout = do
1001+ liftIO $ threadDelay (round $ n * 1000000)
1002+ return (Nothing, True)
1003+
9761004------------------------------------------------------------------------------
9771005-- Element Aware APIs
9781006------------------------------------------------------------------------------