eb7da51 · streamly · git.iko.soy

git.iko.soy/streamly

Expose tapCount

Renamed pollCounts to tapCount

commiteb7da51798ecc6a5c42f3c9d7029b2d68baa15fb authorHarendra Kumar <harendra@composewell.com> dateTue 3 Jan 2023, 06:17 (4 years ago) parentc5cb2d8

4 files changed+58−26

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Stream/Transform.hs+36−6 src/Streamly/Data/Stream/Concurrent.hs+3 src/Streamly/Internal/Data/Stream/Concurrent.hs+18−19 src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+1−1

core/src/Streamly/Internal/Data/Stream/Transform.hs+36−6

@@ -168,6 +168,7 @@ import Prelude hiding

168168-- $setup

169169-- >>> :m

170170-- >>> import Control.Concurrent (threadDelay)

171+-- >>> import Control.Monad (void)

171172-- >>> import Control.Monad.IO.Class (MonadIO (liftIO))

172173-- >>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)

173174-- >>> import Data.Function ((&))

@@ -730,6 +731,10 @@ intersperseSuffixBySpan n eff =

730731

731732-- | Insert a side effect before consuming an element of a stream.

732733--

734+-- Definition:

735+--

736+-- >>> interspersePrefix_ m = Stream.mapM (\x -> void m >> return x)

737+--

733738-- >>> input = Stream.fromList "hello"

734739-- >>> Stream.fold Fold.toList $ Stream.trace putChar $ Stream.interspersePrefix_ (putChar '.' >> return ',') input

735740-- .h.e.l.l.o"hello"

@@ -746,24 +751,42 @@ interspersePrefix_ m = mapM (\x -> void m >> return x)

746751-- Inserting Time

747752------------------------------------------------------------------------------

748753

754+-- XXX This should be in Prelude, should we export this as a helper function?

755+

756+-- | Block the current thread for specified number of seconds.

757+{-# INLINE sleep #-}

758+sleep :: MonadIO m => Double -> m ()

759+sleep n = liftIO $ threadDelay $ round $ n * 1000000

760+

749761-- | Introduce a delay of specified seconds between elements of the stream.

750762--

763+-- Definition:

764+--

765+-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000

766+-- >>> delay = Stream.intersperseM_ . sleep

767+--

768+-- Example:

769+--

751770-- >>> input = Stream.enumerateFromTo 1 3

752771-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delay 1 input

753772-- 1

754773-- 2

755774-- 3

756775--

757−-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000

758−-- >>> delay n = Stream.intersperseM_ $ sleep n

759−--

760776{-# INLINE delay #-}

761777delay :: MonadIO m => Double -> Stream m a -> Stream m a

762−delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000

778+delay = intersperseM_ . sleep

763779

764780-- | Introduce a delay of specified seconds after consuming an element of a

765781-- stream.

766782--

783+-- Definition:

784+--

785+-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000

786+-- >>> delayPost = Stream.intersperseSuffix_ . sleep

787+--

788+-- Example:

789+--

767790-- >>> input = Stream.enumerateFromTo 1 3

768791-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPost 1 input

769792-- 1

@@ -774,11 +797,18 @@ delay n = intersperseM_ $ liftIO $ threadDelay $ round $ n * 1000000

774797--

775798{-# INLINE delayPost #-}

776799delayPost :: MonadIO m => Double -> Stream m a -> Stream m a

777−delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000

800+delayPost = intersperseSuffix_ . sleep

778801

779802-- | Introduce a delay of specified seconds before consuming an element of a

780803-- stream.

781804--

805+-- Definition:

806+--

807+-- >>> sleep n = liftIO $ threadDelay $ round $ n * 1000000

808+-- >>> delayPre = Stream.interspersePrefix_ . sleep

809+--

810+-- Example:

811+--

782812-- >>> input = Stream.enumerateFromTo 1 3

783813-- >>> Stream.fold (Fold.drainMapM print) $ Stream.delayPre 1 input

784814-- 1

@@ -789,7 +819,7 @@ delayPost n = intersperseSuffix_ $ liftIO $ threadDelay $ round $ n * 1000000

789819--

790820{-# INLINE delayPre #-}

791821delayPre :: MonadIO m => Double -> Stream m a -> Stream m a

792−delayPre n = trace_ $ liftIO $ threadDelay $ round $ n * 1000000

822+delayPre = interspersePrefix_ . sleep

793823

794824------------------------------------------------------------------------------

795825-- Reorder in sequence

src/Streamly/Data/Stream/Concurrent.hs+3

@@ -172,6 +172,9 @@ module Streamly.Data.Stream.Concurrent

172172

173173 -- **** ConcatIterate

174174 , parConcatIterate

175+

176+ -- *** Observation

177+ , tapCount

175178 )

176179where

177180

src/Streamly/Internal/Data/Stream/Concurrent.hs+18−19

@@ -107,8 +107,8 @@ module Streamly.Internal.Data.Stream.Concurrent

107107

108108 -- ** Reactive

109109 , fromCallback

110− , pollCountsD

111− , pollCounts

110+ , tapCountD

111+ , tapCount

112112 )

113113where

114114

@@ -147,7 +147,7 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel

147147-- >>> import qualified Streamly.Data.Array as Array

148148-- >>> import qualified Streamly.Data.Fold as Fold

149149-- >>> import qualified Streamly.Data.Parser as Parser

150−-- >>> import qualified Streamly.Data.Stream as Stream

150+-- >>> import qualified Streamly.Internal.Data.Stream as Stream hiding (append2)

151151-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream

152152-- >>> import Prelude hiding (concatMap, concat, zipWith)

153153-- >>> :{

@@ -825,14 +825,14 @@ fromCallback setCallback = Stream.concatEffect $ do

825825 setCallback callback

826826 return stream

827827

828−{-# INLINE_NORMAL pollCountsD #-}

829−pollCountsD

828+{-# INLINE_NORMAL tapCountD #-}

829+tapCountD

830830 :: MonadAsync m

831831 => (a -> Bool)

832832 -> (D.Stream m Int -> m b)

833833 -> D.Stream m a

834834 -> D.Stream m a

835−pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing

835+tapCountD predicate fld (D.Stream step state) = D.Stream step' Nothing

836836 where

837837

838838 {-# INLINE_LATE step' #-}

@@ -858,31 +858,30 @@ pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing

858858 liftIO $ killThread tid

859859 return Stop

860860

861−-- | @pollCounts predicate transform fold stream@ counts those elements in the

861+-- | @tapCount predicate fold stream@ taps the count of those elements in the

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

863−-- another thread which transforms it using @transform@ and then folds it using

864−-- @fold@. The thread is automatically cleaned up if the stream stops or

865−-- aborts due to exception.

863+-- another thread which folds it using @fold@.

866864--

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

868866--

869−-- @

870−-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)

871−-- $ Stream.enumerateFrom 0

872−-- @

867+-- >>> rate = Stream.rollingMap2 (flip (-)) . Stream.delayPost 1

868+-- >>> report = Stream.fold (Fold.drainMapM print) . rate

869+-- >>> tap = Stream.tapCount (const True) report

870+-- >>> go = Stream.fold Fold.drain $ tap $ Stream.enumerateFrom 0

873871--

874−-- Note: This may not work correctly on 32-bit machines.

872+-- Note: This may not work correctly on 32-bit machines because of Int

873+-- overflow.

875874--

876875-- /Pre-release/

877876--

878−{-# INLINE pollCounts #-}

879−pollCounts ::

877+{-# INLINE tapCount #-}

878+tapCount ::

880879 (MonadAsync m)

881880 => (a -> Bool)

882881 -> (Stream m Int -> m b)

883882 -> Stream m a

884883 -> Stream m a

885−pollCounts predicate f xs =

884+tapCount predicate f xs =

886885 Stream.fromStreamD

887− $ pollCountsD predicate (f . Stream.fromStreamD)

886+ $ tapCountD predicate (f . Stream.fromStreamD)

888887 $ Stream.toStreamD xs

src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+1−1

@@ -583,7 +583,7 @@ pollCounts ::

583583 -> t m a

584584pollCounts predicate f xs =

585585 fromStreamD

586− $ Concur.pollCountsD predicate (f . fromStreamD)

586+ $ Concur.tapCountD predicate (f . fromStreamD)

587587 $ toStreamD xs

588588

589589-- | Apply a monadic function to each element flowing through the stream and