54662d8 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add a predicate to pollCounts

54662d8Harendra Kumar7 years ago

benchmark/Streamly/Benchmark/Prelude.hs+1−1

compare with master

unifiedsplit

@@ -418,7 +418,7 @@ tapRate n str = do

418{-# INLINE pollCounts #-}

419pollCounts :: Int -> Stream IO Int -> IO ()

420pollCounts n str = do

421− composeN n (Internal.pollCounts f FL.drain) str

421+ composeN n (Internal.pollCounts (P.const P.True) f FL.drain) str

422 where f = Internal.rollingMap (P.-) . Internal.delayPost 1

423

424{-# INLINE tapAsyncS #-}

src/Streamly/Internal/Data/Stream/StreamD.hs+4−3

@@ -3457,11 +3457,12 @@ tapOffsetEvery offset n (Fold fstep initial extract) (Stream step state) =

3457{-# INLINE_NORMAL pollCounts #-}

3458pollCounts

3459 :: MonadAsync m

3460+ => (a -> Bool)

3460− => (Stream m Int -> Stream m Int)

3461+ -> (Stream m Int -> Stream m Int)

3461 -> Fold m Int b

3462 -> Fold m Int b

3462 -> Stream m a

3463 -> Stream m a

3464 -> Stream m a

3464−pollCounts transf fld (Stream step state) = Stream step' Nothing

3465+pollCounts predicate transf fld (Stream step state) = Stream step' Nothing

3465 where

3466 where

3466

3467

3467 {-# INLINE_LATE step' #-}

3468 {-# INLINE_LATE step' #-}

@@ -3479,7 +3480,7 @@ pollCounts transf fld (Stream step state) = Stream step' Nothing

3479 r <- step gst st

3480 r <- step gst st

3480 case r of

3481 case r of

3481 Yield x s -> do

3482 Yield x s -> do

3482− liftIO $ modifyVar' countVar (+ 1)

3483+ when (predicate x) $ liftIO $ modifyVar' countVar (+ 1)

3483 return $ Yield x (Just (countVar, tid, s))

3484 return $ Yield x (Just (countVar, tid, s))

3484 Skip s -> return $ Skip (Just (countVar, tid, s))

3485 Skip s -> return $ Skip (Just (countVar, tid, s))

3485 Stop -> do

3486 Stop -> do

src/Streamly/Internal/Prelude.hs+10−8

@@ -3778,15 +3778,16 @@ tapOffsetEvery offset n f xs =

3778tapAsync :: (IsStream t, MonadAsync m) => FL.Fold m a b -> t m a -> t m a

3779tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)

3780

3781−-- | Maintain the count of elements flowing in the stream and poll the count

3781+-- | @pollCounts predicate transform fold stream@ counts those elements in the

3782−-- asynchronously from another thread. The count stream is transformed using

3782+-- stream that pass the @predicate@. The resulting count stream is sent to

3783−-- the supplied transform and then folded using the supplied fold. The thread

3783+-- another thread which transforms it using @transform@ and then folds it using

3784−-- is automatically cleaned up if the stream stops or aborts due to exception.

3784+-- @fold@. The thread is automatically cleaned up if the stream stops or

3785+-- aborts due to exception.

3785--

3786--

3786-- For example, to print the count of elements processed every second:

3787-- For example, to print the count of elements processed every second:

3787--

3788--

3788-- @

3789-- @

3789−-- > S.drain $ S.pollCounts (rollingMap (-) . delayPost 1) (FL.drainBy print)

3790+-- > S.drain $ S.pollCounts (const True) (S.rollingMap (-) . S.delayPost 1) (FL.drainBy print)

3790-- $ S.enumerateFrom 0

3791-- $ S.enumerateFrom 0

3791-- @

3792-- @

3792--

3793--

@@ -3797,13 +3798,14 @@ tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)

3797{-# INLINE pollCounts #-}

3798{-# INLINE pollCounts #-}

3798pollCounts ::

3799pollCounts ::

3799 (IsStream t, MonadAsync m)

3800 (IsStream t, MonadAsync m)

3801+ => (a -> Bool)

3800− => (t m Int -> t m Int)

3802+ -> (t m Int -> t m Int)

3801 -> Fold m Int b

3803 -> Fold m Int b

3802 -> t m a

3804 -> t m a

3803 -> t m a

3805 -> t m a

3804−pollCounts transf f xs =

3806+pollCounts predicate transf f xs =

3805 D.fromStreamD

3807 D.fromStreamD

3806− $ D.pollCounts (D.toStreamD . transf . D.fromStreamD) f

3808+ $ D.pollCounts predicate (D.toStreamD . transf . D.fromStreamD) f

3807 $ (D.toStreamD xs)

3809 $ (D.toStreamD xs)

3808

3810

3809-- | Calls the supplied function with the number of elements consumed

3811-- | Calls the supplied function with the number of elements consumed