git.iko.soy/streamly
on master help sign in
add intervalsOf combinator
6084321Harendra Kumar
src/Streamly/Fold.hs+29−1
compare with master
unifiedsplit
@@ -207,6 +207,7 @@ module Streamly.Fold
207
207
208 -- Element unaware grouping
208 -- Element unaware grouping
209 , groupsOf
209 , groupsOf
210+ , intervalsOf
210
211
211 -- Element aware grouping
212 -- Element aware grouping
212 , groups
213 , groups
@@ -296,10 +297,13 @@ module Streamly.Fold
296 )
297 )
297where
298where
298
299
300+import Control.Concurrent (threadDelay)
299import Control.Monad (void)
301import Control.Monad (void)
300import Control.Monad.IO.Class (MonadIO(..))
302import Control.Monad.IO.Class (MonadIO(..))
301import Data.Functor.Identity (Identity)
303import Data.Functor.Identity (Identity)
302import Data.Map.Strict (Map)
304import Data.Map.Strict (Map)
305+import Data.Maybe (fromJust, isJust)
306+
303import Foreign.Storable (Storable(..))
307import Foreign.Storable (Storable(..))
304import Prelude
308import Prelude
305 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
309 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
@@ -311,6 +315,7 @@ import Prelude
311
315
312import qualified Data.Map.Strict as Map
316import qualified Data.Map.Strict as Map
313
317
318+import Streamly (MonadAsync, parallel)
314import Streamly.Array (Array)
319import Streamly.Array (Array)
315import Streamly.Fold.Types (Fold(..))
320import Streamly.Fold.Types (Fold(..))
316import Streamly.Streams.Serial (SerialT)
321import Streamly.Streams.Serial (SerialT)
@@ -318,7 +323,6 @@ import Streamly.Streams.StreamK (IsStream())
318
323
319import Streamly.Strict
324import Streamly.Strict
320
325
321−import Streamly (MonadAsync)
322import qualified Streamly.Array as A
326import qualified Streamly.Array as A
323import qualified Streamly.Prelude as S
327import qualified Streamly.Prelude as S
324import qualified Streamly.Streams.StreamD as D
328import qualified Streamly.Streams.StreamD as D
@@ -973,6 +977,30 @@ groupsOf
973 => Int -> Fold m a b -> t m a -> t m b
977 => Int -> Fold m a b -> t m a -> t m b
974groupsOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
978groupsOf n f m = D.fromStreamD $ D.groupsOf n f (D.toStreamD m)
975
979
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+
976------------------------------------------------------------------------------
1004------------------------------------------------------------------------------
977-- Element Aware APIs
1005-- Element Aware APIs
978------------------------------------------------------------------------------
1006------------------------------------------------------------------------------