eb7da51 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Expose tapCount

eb7da51Harendra Kumar4 years ago

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

compare with master

unifiedsplit

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

168-- $setup

169-- >>> :m

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

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

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

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

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

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

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

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

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

730

731

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

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

732--

733--

734+-- Definition:

735+--

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

737+--

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

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

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

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

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

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

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

746-- Inserting Time

751-- Inserting Time

747------------------------------------------------------------------------------

752------------------------------------------------------------------------------

748

753

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+

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

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

750--

762--

763+-- Definition:

764+--

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

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

767+--

768+-- Example:

769+--

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

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

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

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

753-- 1

772-- 1

754-- 2

773-- 2

755-- 3

774-- 3

756--

775--

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

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

759−--

760{-# INLINE delay #-}

776{-# INLINE delay #-}

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

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

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

778+delay = intersperseM_ . sleep

763

779

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

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

765-- stream.

781-- stream.

766--

782--

783+-- Definition:

784+--

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

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

787+--

788+-- Example:

789+--

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

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

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

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

769-- 1

792-- 1

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

774--

797--

775{-# INLINE delayPost #-}

798{-# INLINE delayPost #-}

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

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

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

800+delayPost = intersperseSuffix_ . sleep

778

801

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

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

780-- stream.

803-- stream.

781--

804--

805+-- Definition:

806+--

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

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

809+--

810+-- Example:

811+--

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

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

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

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

784-- 1

814-- 1

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

789--

819--

790{-# INLINE delayPre #-}

820{-# INLINE delayPre #-}

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

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

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

822+delayPre = interspersePrefix_ . sleep

793

823

794------------------------------------------------------------------------------

824------------------------------------------------------------------------------

795-- Reorder in sequence

825-- Reorder in sequence

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

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

172

173 -- **** ConcatIterate

174 , parConcatIterate

175+

176+ -- *** Observation

177+ , tapCount

175 )

178 )

176where

179where

177

180

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

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

107

108 -- ** Reactive

109 , fromCallback

110− , pollCountsD

110+ , tapCountD

111− , pollCounts

111+ , tapCount

112 )

113where

114

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

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

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

149-- >>> 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)

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

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

153-- >>> :{

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

825 setCallback callback

826 return stream

827

828−{-# INLINE_NORMAL pollCountsD #-}

828+{-# INLINE_NORMAL tapCountD #-}

829−pollCountsD

829+tapCountD

830 :: MonadAsync m

831 => (a -> Bool)

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

833 -> D.Stream m a

834 -> 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

836 where

837

838 {-# INLINE_LATE step' #-}

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

858 liftIO $ killThread tid

859 return Stop

860

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

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

862-- 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

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

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

865−-- aborts due to exception.

866--

864--

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

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

868--

866--

869−-- @

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

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

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

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

871−-- $ Stream.enumerateFrom 0

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

872−-- @

873--

871--

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.

875--

874--

876-- /Pre-release/

875-- /Pre-release/

877--

876--

878−{-# INLINE pollCounts #-}

877+{-# INLINE tapCount #-}

879−pollCounts ::

878+tapCount ::

880 (MonadAsync m)

879 (MonadAsync m)

881 => (a -> Bool)

880 => (a -> Bool)

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

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

883 -> Stream m a

882 -> Stream m a

884 -> Stream m a

883 -> Stream m a

885−pollCounts predicate f xs =

884+tapCount predicate f xs =

886 Stream.fromStreamD

885 Stream.fromStreamD

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

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

888 $ Stream.toStreamD xs

887 $ Stream.toStreamD xs

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

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

583 -> t m a

584pollCounts predicate f xs =

585 fromStreamD

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

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

587 $ toStreamD xs

588

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