b70fead · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add concatMap variants with feedback/looping

b70feadHarendra Kumar7 years ago

src/Streamly/Internal/Data/Stream/StreamK/Type.hs+3−38

compare with master

unifiedsplit

@@ -999,44 +999,9 @@ instance Monad m => Monad (Stream m) where

999 (>>=) = flip concatMap

1000

1001{-

1002−-- ConcatMap recursively on itself using a merge strategy.

1002+-- Like concatMap but generates stream using an unfold function. Similar to

1003−concatLoopBy :: IsStream t

1003+-- concatUnfold but for StreamK.

1004− => (forall c. t m c -> t m c -> t m c)

1005− -> (a -> t m a) -> t m a

1006−concatLoopBy = undefined

1007−

1008−-- This is mfix. Put another way, concatMap recursively on the output of a

1009−-- stream. Compare with iterate.

1010−concatLoop :: IsStream t => (a -> t m a) -> t m a

1011−concatLoop = concatLoopBy serial

1012−

1013−instance MonadFix (Stream m) where

1014− mfix = concatLoop

1015−

1016−-- The SVar implementation is something similar to concatFeedBack, so maybe

1017−-- there is an opportunity to share the implementation here. In an SVar we run

1018−-- a part of an action (a stream), it yields an output and the rest of the

1019−-- stream, we yield the output and queue back the rest of the stream for

1020−-- further evaluation. Also see unfoldrA.

1021−--

1022−-- There could be multiple variants of this combinator, for example use a

1023−-- specific way of concating i.e. concatLoopBy. Which also includes combinators

1024−-- with different stop behaviors. For example if the Left values are errors we

1025−-- can stop the whole composition on errors or on specific errors.

1026−--

1027−-- We can also flip the serial/ahead append behavior e.g. instead of processing

1028−-- the Left output after the original stream we can reverse the order.

1029−--

1030−-- | Concat map on the 'Left' output of a stream and merge it back into the

1031−-- stream. The right output is yielded in the output stream.

1032−concatFeedBack :: IsStream t

1033− => (b -> t m (Either a b)) -> t m (Either a b) -> t m a

1034−concatFeedBack = undefined

1035−

1036−-- Compare this with unfoldr. Start with a seed stream and generate a stream

1037−-- with a value and new seeds. The new seeds are fed back to generate a seed

1038−-- stream and so on. This is a stream level unfoldr.

1039concatUnfoldr :: IsStream t

1004concatUnfoldr :: IsStream t

1040− => (b -> t m (Maybe (a, b))) -> t m (Maybe (a, b)) -> t m a

1005+ => (b -> t m (Maybe (a, b))) -> t m b -> t m a

1041concatUnfoldr = undefined

1006concatUnfoldr = undefined

1042-}

1007-}

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

@@ -271,7 +271,10 @@ module Streamly.Internal.Prelude

271 , intercalateSuffix

272 , interpose

273 , interposeSuffix

274+ , concatMapIterateWith

274 , concatMapTreeWith

275 , concatMapTreeWith

276+ , concatMapLoopWith

277+ , concatMapTreeYieldLeavesWith

275

278

276 -- -- ** Breaking

279 -- -- ** Breaking

277

280

@@ -2581,6 +2584,23 @@ interposeSuffix x unf str =

2581-- Flattening Trees

2584-- Flattening Trees

2582------------------------------------------------------------------------------

2585------------------------------------------------------------------------------

2583

2586

2587+-- | Like 'iterateM' but using a stream generator function.

2588+--

2589+-- /Internal/

2590+--

2591+{-# INLINE concatMapIterateWith #-}

2592+concatMapIterateWith

2593+ :: IsStream t

2594+ => (forall c. t m c -> t m c -> t m c)

2595+ -> (a -> t m a)

2596+ -> t m a

2597+ -> t m a

2598+concatMapIterateWith combine f xs = concatMapWith combine go xs

2599+ where

2600+ go x = yield x `combine` concatMapWith combine go (f x)

2601+

2602+-- concatMapIterateLeftsWith

2603+--

2584-- | Traverse a forest with recursive tree structures whose non-leaf nodes are

2604-- | Traverse a forest with recursive tree structures whose non-leaf nodes are

2585-- of type @a@ and leaf nodes are of type @b@, flattening all the trees into

2605-- of type @a@ and leaf nodes are of type @b@, flattening all the trees into

2586-- streams and combining the streams into a single stream consisting of both

2606-- streams and combining the streams into a single stream consisting of both

@@ -2594,6 +2614,13 @@ interposeSuffix x unf str =

2594-- Traversing a directory tree recursively is a canonical use case of

2614-- Traversing a directory tree recursively is a canonical use case of

2595-- 'concatMapTreeWith'.

2615-- 'concatMapTreeWith'.

2596--

2616--

2617+-- @

2618+-- concatMapTreeWith combine f xs = concatMapIterateWith combine g xs

2619+-- where

2620+-- g (Left tree) = f tree

2621+-- g (Right leaf) = nil

2622+-- @

2623+--

2597-- /Internal/

2624-- /Internal/

2598--

2625--

2599{-# INLINE concatMapTreeWith #-}

2626{-# INLINE concatMapTreeWith #-}

@@ -2601,7 +2628,7 @@ concatMapTreeWith

2601 :: IsStream t

2628 :: IsStream t

2602 => (forall c. t m c -> t m c -> t m c)

2629 => (forall c. t m c -> t m c -> t m c)

2603 -> (a -> t m (Either a b))

2630 -> (a -> t m (Either a b))

2604− -> t m (Either a b)

2631+ -> t m (Either a b) -- Should be t m a?

2605 -> t m (Either a b)

2632 -> t m (Either a b)

2606concatMapTreeWith combine f xs = concatMapWith combine go xs

2633concatMapTreeWith combine f xs = concatMapWith combine go xs

2607 where

2634 where

@@ -2610,11 +2637,15 @@ concatMapTreeWith combine f xs = concatMapWith combine go xs

2610

2637

2611{-

2638{-

2612-- | Like concatMapTreeWith but produces only stream of leaf elements.

2639-- | Like concatMapTreeWith but produces only stream of leaf elements.

2640+-- On an either stream, iterate lefts but yield only rights.

2613--

2641--

2614−{-# INLINE concatMapLeavesWith #-}

2642+-- concatMapEitherYieldRightsWith combine f xs =

2643+-- catRights $ concatMapTreeWith combine f xs

2644+--

2645+{-# INLINE concatMapEitherYieldRightsWith #-}

2615−concatMapLeavesWith :: (IsStream t, MonadAsync m)

2646+concatMapEitherYieldRightsWith :: (IsStream t, MonadAsync m)

2616 => _ -> (a -> t m (Either a b)) -> t m (Either a b) -> t m b

2647 => _ -> (a -> t m (Either a b)) -> t m (Either a b) -> t m b

2617−concatMapLeavesWith combine f xs = undefined

2648+concatMapEitherYieldRightsWith combine f xs = undefined

2618-}

2649-}

2619

2650

2620{-

2651{-

@@ -2622,6 +2653,7 @@ concatMapLeavesWith combine f xs = undefined

2622concatUnfoldTree :: (IsStream t, MonadAsync m)

2653concatUnfoldTree :: (IsStream t, MonadAsync m)

2623 => Unfold m a (Either a b) -> t m (Either a b) -> t m (Either a b)

2654 => Unfold m a (Either a b) -> t m (Either a b) -> t m (Either a b)

2624concatUnfoldTree unf xs = undefined

2655concatUnfoldTree unf xs = undefined

2656+-}

2625

2657

2626------------------------------------------------------------------------------

2658------------------------------------------------------------------------------

2627-- Feedback loop

2659-- Feedback loop

@@ -2637,11 +2669,36 @@ concatUnfoldTree unf xs = undefined

2637-- on is the work queue. When evaluated it results in either a leaf element to

2669-- on is the work queue. When evaluated it results in either a leaf element to

2638-- yield or a tail stream to queue back to the work queue.

2670-- yield or a tail stream to queue back to the work queue.

2639--

2671--

2672+-- | Feedback a component of the output back to the input stream.

2673+--

2674+-- /Internal/

2675+--

2640{-# INLINE concatMapLoopWith #-}

2676{-# INLINE concatMapLoopWith #-}

2677+concatMapLoopWith

2641−concatMapLoopWith :: (IsStream t, MonadAsync m)

2678+ :: (IsStream t, MonadAsync m)

2679+ => (forall x. t m x -> t m x -> t m x)

2642− => _ -> (a -> t m (Either b c)) -> (b -> t m a) -> t m a -> t m c

2680+ -> (a -> t m (Either b c))

2681+ -> (b -> t m a) -- feedback function to feed b back into input

2682+ -> t m a

2683+ -> t m c

2643−concatMapLoopWith combine f xs = undefined

2684+concatMapLoopWith combine f fb xs =

2644−-}

2685+ concatMapWith combine go $ concatMapWith combine f xs

2686+ where

2687+ go (Left b) = concatMapLoopWith combine f fb $ fb b

2688+ go (Right c) = yield c

2689+

2690+-- Concat a stream of trees, generating only leaves.

2691+--

2692+-- /Internal/

2693+--

2694+{-# INLINE concatMapTreeYieldLeavesWith #-}

2695+concatMapTreeYieldLeavesWith

2696+ :: (IsStream t, MonadAsync m)

2697+ => (forall x. t m x -> t m x -> t m x)

2698+ -> (a -> t m (Either a b))

2699+ -> t m a

2700+ -> t m b

2701+concatMapTreeYieldLeavesWith combine f = concatMapLoopWith combine f yield

2645

2702

2646------------------------------------------------------------------------------

2703------------------------------------------------------------------------------

2647-- Grouping/Splitting

2704-- Grouping/Splitting