git.iko.soy/streamly
on master help sign in
Add a "foldFilter" combinator to serial streams
commitddfe6bd5384c20b2122b2e58d1f64ddc5858969b authorHarendra Kumar <harendra@composewell.com> dateWed 23 Mar 2022, 00:23 (5 years ago) parent8dabc2e
3 files changed+43
compare with master
unifiedsplit
benchmark/Streamly/Benchmark/Prelude/Serial/Transformation.hs+9 core/src/Streamly/Internal/Data/Stream/Serial.hs+20 core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+14
benchmark/Streamly/Benchmark/Prelude/Serial/Transformation.hs+9
@@ -31,6 +31,7 @@ import Test.Inspection
3131#endif
3232
3333import qualified Streamly.Prelude as S
34+import qualified Streamly.Internal.Data.Stream.Serial as Serial
3435import qualified Streamly.Internal.Data.Stream.IsStream as Internal
3536import qualified Streamly.Internal.Data.Fold as FL
3637import qualified Streamly.Internal.Data.Unfold as Unfold
@@ -288,6 +289,10 @@ filterMAllOut value n = composeN n $ S.filterM (\x -> return $ x > (value + 1))
288289filterMAllIn :: MonadIO m => Int -> Int -> SerialT m Int -> m ()
289290filterMAllIn value n = composeN n $ S.filterM (\x -> return $ x <= (value + 1))
290291
292+{-# INLINE foldFilterEven #-}
293+foldFilterEven :: MonadIO m => Int -> SerialT m Int -> m ()
294+foldFilterEven n = composeN n $ Serial.foldFilter (FL.satisfy even)
295+
291296{-# INLINE _takeOne #-}
292297_takeOne :: MonadIO m => Int -> SerialT m Int -> m ()
293298_takeOne n = composeN n $ S.take 1
@@ -396,6 +401,8 @@ o_1_space_filtering value =
396401 , benchIOSink value "filterM-all-out" (filterMAllOut value 1)
397402 , benchIOSink value "filterM-all-in" (filterMAllIn value 1)
398403
404+ , benchIOSink value "foldFilter-even" (foldFilterEven 1)
405+
399406 -- Trimming
400407 , benchIOSink value "take-all" (takeAll value 1)
401408 , benchIOSink
@@ -441,6 +448,8 @@ o_1_space_filteringX4 value =
441448 , benchIOSink value "filterM-all-out" (filterMAllOut value 4)
442449 , benchIOSink value "filterM-all-in" (filterMAllIn value 4)
443450
451+ , benchIOSink value "foldFilter-even" (foldFilterEven 4)
452+
444453 -- trimming
445454 , benchIOSink value "take-all" (takeAll value 4)
446455 , benchIOSink value "takeWhile-true" (takeWhileTrue value 4)
core/src/Streamly/Internal/Data/Stream/Serial.hs+20
@@ -42,6 +42,7 @@ module Streamly.Internal.Data.Stream.Serial
4242 -- * Transformation
4343 , map
4444 , mapM
45+ , foldFilter
4546
4647 -- * Deprecated
4748 , StreamT
@@ -72,12 +73,14 @@ import Text.Read
7273 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
7374 , readListPrecDefault)
7475import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace, oneShot)
76+import Streamly.Internal.Data.Fold.Type (Fold)
7577import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
7678import Streamly.Internal.Data.Stream.StreamK.Type
7779 (Stream, mkStream, foldStream)
7880
7981import qualified Streamly.Internal.Data.Stream.Common as P
8082import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
83+import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D
8184import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
8285import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
8386
@@ -259,6 +262,23 @@ NFDATA1_INSTANCE(SerialT)
259262FOLDABLE_INSTANCE(SerialT)
260263TRAVERSABLE_INSTANCE(SerialT)
261264
265+{-# INLINE toStreamD #-}
266+toStreamD :: Applicative m => SerialT m a -> D.Stream m a
267+toStreamD (SerialT m) = D.fromStreamK m
268+
269+{-# INLINE fromStreamD #-}
270+fromStreamD :: Monad m => D.Stream m a -> SerialT m a
271+fromStreamD m = SerialT $ D.toStreamK m
272+
273+-- | Use a filtering fold on a stream.
274+--
275+-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.fromList [1..10]
276+-- 40
277+--
278+{-# INLINE foldFilter #-}
279+foldFilter :: Monad m => Fold m a (Maybe b) -> SerialT m a -> SerialT m b
280+foldFilter p = fromStreamD . D.foldFilter p . toStreamD
281+
262282------------------------------------------------------------------------------
263283-- WSerialT
264284------------------------------------------------------------------------------
core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+14
@@ -70,6 +70,7 @@ module Streamly.Internal.Data.Stream.StreamD.Transform
7070 -- | Produce a subset of the stream.
7171 , filter
7272 , filterM
73+ , foldFilter
7374 , deleteBy
7475 , uniq
7576
@@ -741,6 +742,19 @@ scanl1' f = scanl1M' (\x y -> return (f x y))
741742-- Filtering
742743-------------------------------------------------------------------------------
743744
745+-- XXX nested foldMany does not fuse, therefore, this may not be very useful as
746+-- a filter unless that is fixed.
747+--
748+-- | Use a filtering fold on a stream.
749+--
750+-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.fromList [1..10]
751+-- 40
752+--
753+-- /Pre-release/
754+{-# INLINE foldFilter #-}
755+foldFilter :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b
756+foldFilter f = catMaybes . foldMany f
757+
744758-- Adapted from the vector package
745759{-# INLINE_NORMAL filterM #-}
746760filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a