@@ -134,12 +134,12 @@ module Streamly.Fold
134134
135135 -- ** Mapping
136136 --, transform
137− , lmap
137+ -- , lmap
138138 --, lsequence
139− , lmapM
139+ -- , lmapM
140140
141141 -- -- ** Filtering
142− , lfilter
142+ -- , lfilter
143143 -- , lfilterM
144144 -- , ldeleteBy
145145 -- , luniq
@@ -247,13 +247,7 @@ module Streamly.Fold
247247 )
248248where
249249
250−import Control.Concurrent (threadDelay, forkIO, killThread)
251−import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
252−import Control.Exception (SomeException(..), catch, mask)
253250import Control.Monad (void)
254−import Control.Monad.Catch (throwM)
255−import Control.Monad.IO.Class (MonadIO(..))
256−import Control.Monad.Trans.Control (control)
257251import Data.Map.Strict (Map)
258252
259253import Prelude
@@ -267,8 +261,7 @@ import Prelude
267261import qualified Data.Map.Strict as Map
268262import qualified Prelude
269263
270−import Streamly.SVar (MonadAsync)
271−import Streamly.Fold.Types (Fold(..))
264+import Streamly.Fold.Types (Fold(..), lmap)
272265import Streamly.Pipe.Types (Pipe (..), PipeState(..))
273266import Streamly.Strict
274267
@@ -336,27 +329,6 @@ _transform (Pipe pstep1 pstep2 pinitial) (Fold fstep finitial fextract) =
336329
337330 extract (Tuple' _ fs) = fextract fs
338331
339−-- | @(lmap f fold)@ maps the function @f@ on the input of the fold.
340−--
341−-- >>> S.runFold (FL.lmap (\x -> x * x) FL.sum) (S.enumerateFromTo 1 100)
342−-- 338350
343−--
344−-- @since 0.7.0
345−{-# INLINABLE lmap #-}
346−lmap :: (a -> b) -> Fold m b r -> Fold m a r
347−lmap f (Fold step begin done) = Fold step' begin done
348− where
349− step' x a = step x (f a)
350−
351−-- | @(lmapM f fold)@ maps the monadic function @f@ on the input of the fold.
352−--
353−-- @since 0.7.0
354−{-# INLINABLE lmapM #-}
355−lmapM :: Monad m => (a -> m b) -> Fold m b r -> Fold m a r
356−lmapM f (Fold step begin done) = Fold step' begin done
357− where
358− step' x a = f a >>= step x
359−
360332------------------------------------------------------------------------------
361333-- Utilities
362334------------------------------------------------------------------------------
@@ -774,65 +746,6 @@ and = Fold (\x a -> return $ x && a) (return True) return
774746or :: Monad m => Fold m Bool Bool
775747or = Fold (\x a -> return $ x || a) (return False) return
776748
777−------------------------------------------------------------------------------
778−-- Filtering
779−------------------------------------------------------------------------------
780−
781−-- | Include only those elements that pass a predicate.
782−--
783−-- >>> S.runFold (lfilter (> 5) FL.sum) [1..10]
784−-- 40
785−--
786−-- @since 0.7.0
787−{-# INLINABLE lfilter #-}
788−lfilter :: Monad m => (a -> Bool) -> Fold m a r -> Fold m a r
789−lfilter f (Fold step begin done) = Fold step' begin done
790− where
791− step' x a = if f a then step x a else return x
792−
793−-- | Like 'lfilter' but with a monadic predicate.
794−--
795−-- @since 0.7.0
796−{-# INLINABLE _lfilterM #-}
797−_lfilterM :: Monad m => (a -> m Bool) -> Fold m a r -> Fold m a r
798−_lfilterM f (Fold step begin done) = Fold step' begin done
799− where
800− step' x a = do
801− use <- f a
802− if use then step x a else return x
803−
804−-- | Take first 'n' elements from the stream and discard the rest.
805−--
806−-- @since 0.7.0
807−{-# INLINABLE _ltake #-}
808−_ltake :: Monad m => Int -> Fold m a b -> Fold m a b
809−_ltake n (Fold step initial done) = Fold step' initial' done'
810− where
811− initial' = fmap (Tuple' 0) initial
812− step' (Tuple' i r) a = do
813− if i < n
814− then do
815− res <- step r a
816− return $ Tuple' (i + 1) res
817− else return $ Tuple' i r
818− done' (Tuple' _ r) = done r
819−
820−-- | Takes elements from the input as long as the predicate succeeds.
821−--
822−-- @since 0.7.0
823−{-# INLINABLE _ltakeWhile #-}
824−_ltakeWhile :: Monad m => (a -> Bool) -> Fold m a b -> Fold m a b
825−_ltakeWhile predicate (Fold step initial done) = Fold step' initial' done'
826− where
827− initial' = fmap Left' initial
828− step' (Left' r) a = do
829− if predicate a
830− then fmap Left' $ step r a
831− else return (Right' r)
832− step' r _ = return r
833− done' (Left' r) = done r
834− done' (Right' r) = done r
835−
836749------------------------------------------------------------------------------
837750-- Distributing
838751------------------------------------------------------------------------------
@@ -1231,23 +1144,6 @@ unzip = unzipWith id
12311144-- Nesting
12321145------------------------------------------------------------------------------
12331146--
1234−-- | Modify the fold such that when the fold is done, instead of returning the
1235−-- accumulator, it returns a fold. The returned fold starts from where we left
1236−-- i.e. it uses the last accumulator value as the initial value of the
1237−-- accumulator. Thus we can resume the fold later and feed it more input.
1238−--
1239−-- >> do
1240−-- > more <- S.runFold (FL.duplicate FL.sum) (S.enumerateFromTo 1 10)
1241−-- > evenMore <- S.runFold (FL.duplicate more) (S.enumerateFromTo 11 20)
1242−-- > S.runFold evenMore (S.enumerateFromTo 21 30)
1243−-- > 465
1244−--
1245−-- @since 0.7.0
1246−{-# INLINABLE _duplicate #-}
1247−_duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)
1248−_duplicate (Fold step begin done) =
1249− Fold step begin (\x -> pure (Fold step (pure x) done))
1250−
12511147{-
12521148-- All the stream flattening transformations can also be applied to a fold
12531149-- input stream.
@@ -1281,101 +1177,3 @@ lchunksInRange :: Monad m
12811177lchunksInRange low high (Fold step1 initial1 extract1)
12821178 (Fold step2 initial2 extract2) = undefined
12831179-}
1284−
1285−-- | Group the input stream into groups of @n@ elements each and then fold each
1286−-- group using the provided fold function.
1287−--
1288−-- @
1289−--
1290−-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1291−--
1292−-- @
1293−--
1294−{-# INLINE _lchunksOf #-}
1295−_lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c
1296−_lchunksOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1297− Fold step' initial' extract'
1298−
1299− where
1300−
1301− initial' = (Tuple3' 0) <$> initial1 <*> initial2
1302− step' (Tuple3' i r1 r2) a = do
1303− if i < n
1304− then do
1305− res <- step1 r1 a
1306− return $ Tuple3' (i + 1) res r2
1307− else do
1308− res <- extract1 r1
1309− acc2 <- step2 r2 res
1310−
1311− i1 <- initial1
1312− acc1 <- step1 i1 a
1313− return $ Tuple3' 1 acc1 acc2
1314− extract' (Tuple3' _ _ r) = extract2 r
1315−
1316−-- | Group the input stream into windows of n second each and then fold each
1317−-- group using the provided fold function.
1318−--
1319−-- For example, we can copy and distribute a stream to multiple folds where
1320−-- each fold can group the input differently e.g. by one second, one minute and
1321−-- one hour windows respectively and fold each resulting stream of folds.
1322−--
1323−-- @
1324−--
1325−-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c
1326−--
1327−-- @
1328−{-# INLINE _lsessionsOf #-}
1329−_lsessionsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
1330−_lsessionsOf n (Fold step1 initial1 extract1) (Fold step2 initial2 extract2) =
1331− Fold step' initial' extract'
1332−
1333− where
1334−
1335− -- XXX MVar may be expensive we need a cheaper synch mechanism here
1336− initial' = do
1337− i1 <- initial1
1338− i2 <- initial2
1339− mv1 <- liftIO $ newMVar i1
1340− mv2 <- liftIO $ newMVar (Right i2)
1341− t <- control $ \run ->
1342− mask $ \restore -> do
1343− tid <- forkIO $ catch (restore $ void $ run (timerThread mv1 mv2))
1344− (handleChildException mv2)
1345− run (return tid)
1346− return $ Tuple3' t mv1 mv2
1347− step' acc@(Tuple3' _ mv1 _) a = do
1348− r1 <- liftIO $ takeMVar mv1
1349− res <- step1 r1 a
1350− liftIO $ putMVar mv1 res
1351− return acc
1352− extract' (Tuple3' tid _ mv2) = do
1353− r2 <- liftIO $ takeMVar mv2
1354− liftIO $ killThread tid
1355− case r2 of
1356− Left e -> throwM e
1357− Right x -> extract2 x
1358−
1359− timerThread mv1 mv2 = do
1360− liftIO $ threadDelay (round $ n * 1000000)
1361−
1362− r1 <- liftIO $ takeMVar mv1
1363− i1 <- initial1
1364− liftIO $ putMVar mv1 i1
1365−
1366− res1 <- extract1 r1
1367− r2 <- liftIO $ takeMVar mv2
1368− res <- case r2 of
1369− Left _ -> return r2
1370− Right x -> fmap Right $ step2 x res1
1371− liftIO $ putMVar mv2 res
1372− timerThread mv1 mv2
1373−
1374− handleChildException ::
1375− MVar (Either SomeException a) -> SomeException -> IO ()
1376− handleChildException mv2 e = do
1377− r2 <- takeMVar mv2
1378− let r = case r2 of
1379− Left _ -> r2
1380− Right _ -> Left e
1381− putMVar mv2 r