54662d8 · streamly · git.iko.soy

git.iko.soy/streamly

Add a predicate to pollCounts

The predicate provides more power to the combinator by allowing you to count only certain elements in the stream and filter out others. This is efficient compared to an async tap followed by a filter and fold as here filtering is done on the source.

commit54662d80c7b7f09685efcda95f0fee7c3796645f authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 10 Feb 2020, 02:36 (7 years ago) parent99a9b61

3 files changed+15−12

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Prelude.hs+1−1 src/Streamly/Internal/Data/Stream/StreamD.hs+4−3 src/Streamly/Internal/Prelude.hs+10−8

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

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

418418{-# INLINE pollCounts #-}

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

420420pollCounts 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

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

423423

424424{-# 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) =

34573457{-# INLINE_NORMAL pollCounts #-}

34583458pollCounts

34593459 :: MonadAsync m

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

3460+ => (a -> Bool)

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

34613462 -> Fold m Int b

34623463 -> Stream m a

34633464 -> Stream m a

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

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

34653466 where

34663467

34673468 {-# INLINE_LATE step' #-}

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

34793480 r <- step gst st

34803481 case r of

34813482 Yield x s -> do

3482− liftIO $ modifyVar' countVar (+ 1)

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

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

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

34853486 Stop -> do

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

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

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

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

37803780

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

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

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

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

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

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

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

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

3785+-- aborts due to exception.

37853786--

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

37873788--

37883789-- @

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)

37903791-- $ S.enumerateFrom 0

37913792-- @

37923793--

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

37973798{-# INLINE pollCounts #-}

37983799pollCounts ::

37993800 (IsStream t, MonadAsync m)

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

3801+ => (a -> Bool)

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

38013803 -> Fold m Int b

38023804 -> t m a

38033805 -> t m a

3804−pollCounts transf f xs =

3806+pollCounts predicate transf f xs =

38053807 D.fromStreamD

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

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

38073809 $ (D.toStreamD xs)

38083810

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