{- Copyright 2008-2010 Mario Blazevic This file is part of the Streaming Component Combinators (SCC) project. The SCC project is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. SCC is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with SCC. If not, see <http://www.gnu.org/licenses/>. -} {-# LANGUAGE ScopedTypeVariables, Rank2Types, KindSignatures, EmptyDataDecls, MultiParamTypeClasses, FlexibleContexts, FlexibleInstances, FunctionalDependencies, TypeFamilies #-} -- | The "Combinators" module defines combinators applicable to values of the 'Transducer' and 'Splitter' types defined -- in the "Control.Concurrent.SCC.Types" module. module Control.Concurrent.SCC.Combinators (-- * Consumer, producer, and transducer combinators splitterToMarker, consumeBy, prepend, append, substitute, PipeableComponentPair (compose), JoinableComponentPair (join, sequence), -- * Pseudo-logic splitter combinators -- | Combinators 'sAnd' and 'sOr' are only /pseudo/-logic. While the laws of double negation and De Morgan's laws -- hold, 'sAnd' and 'sOr' are in general not commutative, associative, nor idempotent. In the special case when all -- argument splitters are stateless, such as those produced by 'Control.Concurrent.SCC.Types.statelessSplitter', -- these combinators do satisfy all laws of Boolean algebra. sNot, sAnd, sOr, -- ** Zipping logic combinators -- | The 'pAnd' and 'pOr' combinators run the argument splitters in parallel and combine their logical outputs using -- the corresponding logical operation on each output pair, in a manner similar to 'Data.List.zipWith'. They fully -- satisfy the laws of Boolean algebra. pAnd, pOr, -- * Flow-control combinators -- | The following combinators resemble the common flow-control programming language constructs. Combinators -- 'wherever', 'unless', and 'select' are just the special cases of the combinator 'ifs'. -- -- * /transducer/ ``wherever`` /splitter/ = 'ifs' /splitter/ /transducer/ 'Control.Category.id' -- -- * /transducer/ ``unless`` /splitter/ = 'ifs' /splitter/ 'Control.Category.id' /transducer/ -- -- * 'select' /splitter/ = 'ifs' /splitter/ 'Control.Category.id' -- 'Control.Concurrent.SCC.Primitives.suppress' -- ifs, wherever, unless, select, -- ** Recursive while, nestedIn, -- * Section-based combinators -- | All combinators in this section use their 'Control.Concurrent.SCC.Splitter' argument to determine the structure -- of the input. Every contiguous portion of the input that gets passed to one or the other sink of the splitter is -- treated as one section in the logical structure of the input stream. What is done with the section depends on the -- combinator, but the sections, and therefore the logical structure of the input stream, are determined by the -- argument splitter alone. foreach, having, havingOnly, followedBy, even, -- ** first and its variants first, uptoFirst, prefix, -- ** last and its variants last, lastAndAfter, suffix, -- ** positional splitters startOf, endOf, -- ** input ranges between, -- * parser support parseRegions, parseNestedRegions, -- * helper functions groupMarks, findsTrueIn, findsFalseIn, teeConsumers) where import Control.Monad.Coroutine import Control.Monad.Parallel (MonadParallel(..)) import Control.Concurrent.SCC.Streams import Control.Concurrent.SCC.Types import Prelude hiding (even, last, sequence) import Control.Category ((>>>)) import Control.Monad (liftM, when) import qualified Control.Monad as Monad import Control.Monad.Trans (lift) import Data.Maybe (isJust, isNothing, fromJust) import qualified Data.Foldable as Foldable import qualified Data.Sequence as Seq import Data.Sequence (Seq, (|>), (><), ViewL (EmptyL, (:<))) import qualified Control.Category import qualified Data.List -- | Converts a 'Consumer' into a 'Transducer' with no output. consumeBy :: forall m x y r. (Monad m) => Consumer m x r -> Transducer m x y consumeBy c = Transducer $ \ source _sink -> consume c source >> return () -- | Class 'PipeableComponentPair' applies to any two components that can be combined into a third component with the -- following properties: -- -- * The input of the result, if any, becomes the input of the first component. -- -- * The output produced by the first child component is consumed by the second child component. -- -- * The result output, if any, is the output of the second component. class PipeableComponentPair (m :: * -> *) w c1 c2 c3 | c1 c2 -> c3, c1 c3 -> c2, c2 c3 -> c2, c1 -> m w, c2 -> m w, c3 -> m where compose :: Bool -> c1 -> c2 -> c3 instance forall m x. (MonadParallel m) => PipeableComponentPair m x (Producer m x ()) (Consumer m x ()) (Performer m ()) where compose parallel p c = let performPipe :: Coroutine Naught m ((), ()) performPipe = pipePS parallel (produce p) (consume c) in Performer (runCoroutine performPipe >> return ()) instance (MonadParallel m) => PipeableComponentPair m y (Transducer m x y) (Consumer m y r) (Consumer m x r) where compose parallel t c = isolateConsumer $ \source-> liftM snd $ pipePS parallel (transduce t source) (consume c) instance (MonadParallel m) => PipeableComponentPair m x (Producer m x r) (Transducer m x y) (Producer m y r) where compose parallel p t = isolateProducer $ \sink-> liftM fst $ pipePS parallel (produce p) (\source-> transduce t source sink) instance MonadParallel m => PipeableComponentPair m y (Transducer m x y) (Transducer m y z) (Transducer m x z) where compose parallel t1 t2 = if parallel then t1 >|> t2 else t1 >>> t2 class CompatibleSignature c cons (m :: * -> *) input output | c -> cons m class AnyListOrUnit c instance AnyListOrUnit [x] instance AnyListOrUnit () instance (AnyListOrUnit x, AnyListOrUnit y) => CompatibleSignature (Performer m r) (PerformerType r) m x y instance AnyListOrUnit y => CompatibleSignature (Consumer m x r) (ConsumerType r) m [x] y instance AnyListOrUnit y => CompatibleSignature (Producer m x r) (ProducerType r) m y [x] instance CompatibleSignature (Transducer m x y) TransducerType m [x] [y] data PerformerType r data ConsumerType r data ProducerType r data TransducerType -- | Class 'JoinableComponentPair' applies to any two components that can be combined into a third component with the -- following properties: -- -- * if both argument components consume input, the input of the combined component gets distributed to both -- components in parallel, -- -- * if both argument components produce output, the output of the combined component is a concatenation of the -- complete output from the first component followed by the complete output of the second component, and -- -- * the 'join' method may apply the components in any order, the 'sequence' method makes sure its first argument -- has completed before using the second one. class (Monad m, CompatibleSignature c1 t1 m x y, CompatibleSignature c2 t2 m x y, CompatibleSignature c3 t3 m x y) => JoinableComponentPair t1 t2 t3 m x y c1 c2 c3 | c1 c2 -> c3, c1 -> t1 m, c2 -> t2 m, c3 -> t3 m x y, t1 m x y -> c1, t2 m x y -> c2, t3 m x y -> c3 where join :: Bool -> c1 -> c2 -> c3 sequence :: c1 -> c2 -> c3 join = const sequence instance forall m x r1 r2. Monad m => JoinableComponentPair (ProducerType r1) (ProducerType r2) (ProducerType r2) m () [x] (Producer m x r1) (Producer m x r2) (Producer m x r2) where sequence p1 p2 = Producer $ \sink-> produce p1 sink >> produce p2 sink instance forall m x. MonadParallel m => JoinableComponentPair (ConsumerType ()) (ConsumerType ()) (ConsumerType ()) m [x] () (Consumer m x ()) (Consumer m x ()) (Consumer m x ()) where join parallel c1 c2 = Consumer (liftM (const ()) . teeConsumers parallel (consume c1) (consume c2)) sequence c1 c2 = Consumer $ \source-> teeConsumers False (consume c1) getList source >>= \((), list)-> pipe (putList list) (consume c2) >> return () instance forall m x y. (MonadParallel m) => JoinableComponentPair TransducerType TransducerType TransducerType m [x] [y] (Transducer m x y) (Transducer m x y) (Transducer m x y) where join parallel t1 t2 = isolateTransducer $ \source sink-> pipe (\buffer-> teeConsumers parallel (\source-> transduce t1 source sink) (\source-> transduce t2 source buffer) source) getList >>= \(_, list)-> putList list sink sequence t1 t2 = isolateTransducer $ \source sink-> teeConsumers False (flip (transduce t1) sink) getList source >>= \(_, list)-> pipe (putList list) (\source-> transduce t2 source sink) >> return () instance forall m r1 r2. MonadParallel m => JoinableComponentPair (PerformerType r1) (PerformerType r2) (PerformerType r2) m () () (Performer m r1) (Performer m r2) (Performer m r2) where join parallel p1 p2 = Performer $ if parallel then bindM2 (const return) (perform p1) (perform p2) else perform p1 >> perform p2 sequence p1 p2 = Performer $ perform p1 >> perform p2 instance forall m x r1 r2. (MonadParallel m) => JoinableComponentPair (PerformerType r1) (ProducerType r2) (ProducerType r2) m () [x] (Performer m r1) (Producer m x r2) (Producer m x r2) where join parallel pe pr = Producer $ \sink-> if parallel then bindM2 (const return) (lift (perform pe)) (produce pr sink) else lift (perform pe) >> produce pr sink sequence pe pr = Producer $ \sink-> lift (perform pe) >> produce pr sink instance forall m x r1 r2. (MonadParallel m) => JoinableComponentPair (ProducerType r1) (PerformerType r2) (ProducerType r2) m () [x] (Producer m x r1) (Performer m r2) (Producer m x r2) where join parallel pr pe = Producer $ \sink-> if parallel then bindM2 (const return) (produce pr sink) (lift (perform pe)) else produce pr sink >> lift (perform pe) sequence pr pe = Producer $ \sink-> produce pr sink >> lift (perform pe) instance forall m x r1 r2. (MonadParallel m) => JoinableComponentPair (PerformerType r1) (ConsumerType r2) (ConsumerType r2) m [x] () (Performer m r1) (Consumer m x r2) (Consumer m x r2) where join parallel p c = Consumer $ \source-> if parallel then bindM2 (const return) (lift (perform p)) (consume c source) else lift (perform p) >> consume c source sequence p c = Consumer $ \source-> lift (perform p) >> consume c source instance forall m x r1 r2. (MonadParallel m) => JoinableComponentPair (ConsumerType r1) (PerformerType r2) (ConsumerType r2) m [x] () (Consumer m x r1) (Performer m r2) (Consumer m x r2) where join parallel c p = Consumer $ \source-> if parallel then bindM2 (const return) (consume c source) (lift (perform p)) else consume c source >> lift (perform p) sequence c p = Consumer $ \source-> consume c source >> lift (perform p) instance forall m x y r. (MonadParallel m) => JoinableComponentPair (PerformerType r) TransducerType TransducerType m [x] [y] (Performer m r) (Transducer m x y) (Transducer m x y) where join parallel p t = Transducer $ \ source sink -> if parallel then bindM2 (const return) (lift (perform p)) (transduce t source sink) else lift (perform p) >> transduce t source sink sequence p t = Transducer $ \ source sink -> lift (perform p) >> transduce t source sink instance forall m x y r. (MonadParallel m) => JoinableComponentPair TransducerType (PerformerType r) TransducerType m [x] [y] (Transducer m x y) (Performer m r) (Transducer m x y) where join parallel t p = Transducer $ \ source sink -> if parallel then bindM2 (const . return) (transduce t source sink) (lift (perform p)) else do result <- transduce t source sink lift (perform p) return result sequence t p = Transducer $ \ source sink -> do result <- transduce t source sink lift (perform p) return result instance forall m x y. (MonadParallel m) => JoinableComponentPair (ProducerType ()) TransducerType TransducerType m [x] [y] (Producer m y ()) (Transducer m x y) (Transducer m x y) where join parallel p t = if parallel then isolateTransducer $ \source sink-> do (rest, out) <- pipe (\buffer-> bindM2 (const return) (produce p sink) (transduce t source buffer)) getList putList out sink return rest else sequence p t sequence p t = Transducer $ \ source sink -> produce p sink >> transduce t source sink instance forall m x y. (MonadParallel m) => JoinableComponentPair TransducerType (ProducerType ()) TransducerType m [x] [y] (Transducer m x y) (Producer m y ()) (Transducer m x y) where join parallel t p = if parallel then isolateTransducer $ \source sink-> do (rest, out) <- pipe (\buffer-> bindM2 (const . return) (transduce t source sink) (produce p buffer)) getList putList out sink return rest else sequence t p sequence t p = Transducer $ \ source sink -> do result <- transduce t source sink produce p sink return result instance forall m x y. (MonadParallel m) => JoinableComponentPair (ConsumerType ()) TransducerType TransducerType m [x] [y] (Consumer m x ()) (Transducer m x y) (Transducer m x y) where join parallel c t = isolateTransducer $ \source sink-> teeConsumers parallel (consume c) (\source-> transduce t source sink) source >> return () sequence c t = isolateTransducer $ \source sink-> teeConsumers False (consume c) getList source >>= \(_, list)-> pipe (putList list) (\source-> transduce t source sink) >> return () instance forall m x y. MonadParallel m => JoinableComponentPair TransducerType (ConsumerType ()) TransducerType m [x] [y] (Transducer m x y) (Consumer m x ()) (Transducer m x y) where join parallel t c = join parallel c t sequence t c = isolateTransducer $ \source sink-> teeConsumers False (\source-> transduce t source sink) getList source >>= \(_, list)-> pipe (putList list) (consume c) >> return () instance forall m x y. (MonadParallel m) => JoinableComponentPair (ProducerType ()) (ConsumerType ()) TransducerType m [x] [y] (Producer m y ()) (Consumer m x ()) (Transducer m x y) where join parallel p c = Transducer $ \ source sink -> if parallel then bindM2 (\ _ _ -> return ()) (produce p sink) (consume c source) else produce p sink >> consume c source sequence p c = Transducer $ \ source sink -> produce p sink >> consume c source instance forall m x y. (MonadParallel m) => JoinableComponentPair (ConsumerType ()) (ProducerType ()) TransducerType m [x] [y] (Consumer m x ()) (Producer m y ()) (Transducer m x y) where join parallel c p = join parallel p c sequence c p = Transducer $ \ source sink -> consume c source >> produce p sink -- | Combinator 'prepend' converts the given producer to a 'Control.Concurrent.SCC.Types.Transducer' that passes all its -- input through unmodified, except for prepending the output of the argument producer to it. The following law holds: @ -- 'prepend' /prefix/ = 'join' ('substitute' /prefix/) 'Control.Category.id' @ prepend :: forall m x r. (Monad m) => Producer m x r -> Transducer m x x prepend prefix = Transducer $ \ source sink -> produce prefix sink >> pour source sink -- | Combinator 'append' converts the given producer to a 'Control.Concurrent.SCC.Types.Transducer' that passes all its -- input through unmodified, finally appending the output of the argument producer to it. The following law holds: @ -- 'append' /suffix/ = 'join' 'Control.Category.id' ('substitute' /suffix/) @ append :: forall m x r. (Monad m) => Producer m x r -> Transducer m x x append suffix = Transducer $ \ source sink -> pour source sink >> produce suffix sink >> return () -- | The 'substitute' combinator converts its argument producer to a 'Control.Concurrent.SCC.Types.Transducer' that -- produces the same output, while consuming its entire input and ignoring it. substitute :: forall m x y r. (Monad m) => Producer m y r -> Transducer m x y substitute feed = Transducer $ \ source sink -> mapMStream_ (const $ return ()) source >> produce feed sink >> return () -- | The 'sNot' (streaming not) combinator simply reverses the outputs of the argument splitter. In other words, data -- that the argument splitter sends to its /true/ sink goes to the /false/ sink of the result, and vice versa. sNot :: forall m x b. Monad m => Splitter m x b -> Splitter m x b sNot splitter = isolateSplitter $ \ source true false edge -> suppressProducer (split splitter source false true) -- | The 'sAnd' combinator sends the /true/ sink output of its left operand to the input of its right operand for -- further splitting. Both operands' /false/ sinks are connected to the /false/ sink of the combined splitter, but any -- input value to reach the /true/ sink of the combined component data must be deemed true by both splitters. sAnd :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x (b1, b2) sAnd parallel s1 s2 = isolateSplitter $ \ source true false edge -> liftM (fst . fst) $ pipe (\edges-> pipePS parallel (\true-> split s1 source true false (mapSink Left edges)) (\source-> split s2 source true false (mapSink Right edges))) (flip intersectRegions edge) intersectRegions source sink = next Nothing Nothing where next lastLeft lastRight = getWith (either (flip pair lastRight . Just) (pair lastLeft . Just)) source pair l@(Just x) r@(Just y) = put sink (x, y) >> next Nothing Nothing pair l r = next l r -- | A 'sOr' combinator's input value can reach its /false/ sink only by going through both argument splitters' /false/ -- sinks. sOr :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x (Either b1 b2) sOr parallel s1 s2 = isolateSplitter $ \ source true false edge -> liftM fst $ pipePS parallel (\false-> split s1 source true false (mapSink Left edge)) (\source-> split s2 source true false (mapSink Right edge)) -- | Combinator 'pAnd' is a pairwise logical conjunction of two splitters run in parallel on the same input. pAnd :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x (b1, b2) pAnd parallel s1 s2 = isolateSplitter $ \ source true false edge -> pipePS parallel (transduce (splittersToPairMarker parallel s1 s2) source) (\source-> let split l r = getWith (test l r) source test l r (Left (x, t1, t2)) = (if t1 && t2 then put true x else put false x) >> split (if t1 then l else Nothing) (if t2 then r else Nothing) test _ Nothing (Right (Left l)) = split (Just l) Nothing test _ (Just r) (Right (Left l)) = put edge (l, r) >> split (Just l) (Just r) test Nothing _ (Right (Right r)) = split Nothing (Just r) test (Just l) _ (Right (Right r)) = put edge (l, r) >> split (Just l) (Just r) in split Nothing Nothing) >> return () -- | Combinator 'pOr' is a pairwise logical disjunction of two splitters run in parallel on the same input. pOr :: forall c m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x (Either b1 b2) pOr = zipSplittersWith (||) pour ifs :: forall c m x b. (MonadParallel m, Branching c m x ()) => Bool -> Splitter m x b -> c -> c -> c ifs parallel s c1 c2 = combineBranches if' parallel c1 c2 where if' :: forall d. Bool -> (forall a d'. AncestorFunctor d d' => OpenConsumer m a d' x ()) -> (forall a d'. AncestorFunctor d d' => OpenConsumer m a d' x ()) -> forall a. OpenConsumer m a d x () if' parallel c1 c2 source = splitInputToConsumers parallel s source c1 c2 wherever :: forall m x b. MonadParallel m => Bool -> Transducer m x x -> Splitter m x b -> Transducer m x x wherever parallel t s = isolateTransducer wherever' where wherever' :: forall d. Functor d => Source m d x -> Sink m d x -> Coroutine d m () wherever' source sink = pipePS parallel (\true-> split s source true sink (nullSink :: Sink m d b)) (flip (transduce t) sink) >> return () unless :: forall m x b. MonadParallel m => Bool -> Transducer m x x -> Splitter m x b -> Transducer m x x unless parallel t s = wherever parallel t (sNot s) select :: forall m x b. Monad m => Splitter m x b -> Transducer m x x select s = isolateTransducer $ \source sink-> suppressProducer (suppressProducer . split s source sink) -- | Converts a splitter into a parser. parseRegions :: forall m x b. Monad m => Splitter m x b -> Parser m x b parseRegions s = isolateTransducer $ \source sink-> pipe (transduce (splitterToMarker s) source) (\source-> wrapRegions source sink) >> return () where wrapRegions source sink = let wrap Nothing (Left (x, _)) = put sink (Content x) >> return Nothing wrap (Just p) (Left (x, False)) = flush p >> put sink (Content x) >> return Nothing wrap (Just (b, t)) (Left (x, True)) = do Monad.unless t (put sink (Markup (Start b))) put sink (Content x) return (Just (b, True)) wrap (Just p) (Right b') = flush p >> return (Just (b', False)) wrap Nothing (Right b) = return (Just (b, False)) flush (b, t) = put sink $ Markup $ (if t then End else Point) b in foldMStream wrap Nothing source >>= maybe (return ()) flush -- | Converts a boundary-marking splitter into a parser. parseNestedRegions :: forall m x b. Monad m => Splitter m x (Boundary b) -> Parser m x b parseNestedRegions s = isolateTransducer $ \source sink-> split s source (mapSink Content sink) (mapSink Content sink) (mapSink Markup sink) -- | The recursive combinator 'while' feeds the true sink of the argument splitter back to itself, modified by the -- argument transducer. Data fed to the splitter's false sink is passed on unmodified. while :: forall m x b. MonadParallel m => [(Bool, (Transducer m x x, Splitter m x b))] -> Transducer m x x while ((parallel, (t, s)) : rest) = isolateTransducer while' where while' :: forall d. Functor d => Source m d x -> Sink m d x -> Coroutine d m () while' source sink = pipePS parallel (\true-> split s source true sink (nullSink :: Sink m d b)) (\source-> getWith (\x-> liftM fst $ pipe (\sink-> put sink x >> pour source sink) (\source-> transduce while'' source sink)) source) >> return () while'' = compose parallel t (while rest) -- | The recursive combinator 'nestedIn' combines two splitters into a mutually recursive loop acting as a single -- splitter. The true sink of one of the argument splitters and false sink of the other become the true and false sinks -- of the loop. The other two sinks are bound to the other splitter's source. The use of 'nestedIn' makes sense only -- on hierarchically structured streams. If we gave it some input containing a flat sequence of values, and assuming -- both component splitters are deterministic and stateless, an input value would either not loop at all or it would -- loop forever. nestedIn :: forall m x b. MonadParallel m => [(Bool, (Splitter m x b, Splitter m x b))] -> Splitter m x b nestedIn ((parallel, (s1, s2)) : rest) = isolateSplitter $ \ source true false edge -> liftM fst $ pipePS parallel (\false-> split s1 source true false edge) (\source-> pipe (\true-> split s2 source true false (filterMSink (const $ return False) edge)) (\source-> get source >>= maybe (return ((), ())) (\x-> pipe (\sink-> put sink x >> pour source sink) (\source-> split (nestedIn rest) source true false edge)))) -- | The 'foreach' combinator is similar to the combinator 'ifs' in that it combines a splitter and two transducers into -- another transducer. However, in this case the transducers are re-instantiated for each consecutive portion of the -- input as the splitter chunks it up. Each contiguous portion of the input that the splitter sends to one of its two -- sinks gets transducered through the appropriate argument transducer as that transducer's whole input. As soon as the -- contiguous portion is finished, the transducer gets terminated. foreach :: forall m x b c. (MonadParallel m, Branching c m x ()) => Bool -> Splitter m x b -> c -> c -> c foreach parallel s c1 c2 = combineBranches foreach' parallel c1 c2 where foreach' :: forall d. Bool -> (forall a d'. AncestorFunctor d d' => OpenConsumer m a d' x ()) -> (forall a d'. AncestorFunctor d d' => OpenConsumer m a d' x ()) -> forall a. OpenConsumer m a d x () foreach' parallel c1 c2 source = liftM fst $ pipePS parallel (transduce (splitterToMarker s) (liftSource source :: Source m d x)) (\source-> groupMarks source (maybe c2 (const c1))) -- | The 'having' combinator combines two pure splitters into a pure splitter. One splitter is used to chunk the input -- into contiguous portions. Its /false/ sink is routed directly to the /false/ sink of the combined splitter. The -- second splitter is instantiated and run on each portion of the input that goes to first splitter's /true/ sink. If -- the second splitter sends any output at all to its /true/ sink, the whole input portion is passed on to the /true/ -- sink of the combined splitter, otherwise it goes to its /false/ sink. having :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x b1 having parallel s1 s2 = isolateSplitter s where s source true false edge = pipePS parallel (transduce (splitterToMarker s1) source) (flip groupMarks test) >> return () where test Nothing chunk = pour chunk false test (Just mb) chunk = teeConsumers False getList (findsTrueIn s2) chunk >>= \(chunk, maybeFound)-> if isJust maybeFound then maybe (return ()) (put edge) mb >> putList chunk true else putList chunk false -- | The 'havingOnly' combinator is analogous to the 'having' combinator, but it succeeds and passes each chunk of the -- input to its /true/ sink only if the second splitter sends no part of it to its /false/ sink. havingOnly :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x b1 havingOnly parallel s1 s2 = isolateSplitter s where s source true false edge = pipePS parallel (transduce (splitterToMarker s1) source) (flip groupMarks test) >> return () where test Nothing chunk = pour chunk false test (Just mb) chunk = teeConsumers False getList (findsFalseIn s2) chunk >>= \(chunk, anyFalse)-> if anyFalse then putList chunk false else maybe (return ()) (put edge) mb >> putList chunk true -- | The result of combinator 'first' behaves the same as the argument splitter up to and including the first portion of -- the input which goes into the argument's /true/ sink. All input following the first true portion goes into the -- /false/ sink. first :: forall m x b. Monad m => Splitter m x b -> Splitter m x b first splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split 1 (Left (x, False)) = put false x >> return 1 split 1 (Left (x, True)) = put true x >> return 2 split 1 (Right b) = put edge b >> return 2 split 2 b@Right{} = return 3 split 2 (Left (x, True)) = put true x >> return 2 split 2 (Left (x, False)) = put false x >> return 3 split 3 (Left (x, _)) = put false x >> return 3 split 3 (Right _) = return 3 in foldMStream_ split 1 source -- | The result of combinator 'uptoFirst' takes all input up to and including the first portion of the input which goes -- into the argument's /true/ sink and feeds it to the result splitter's /true/ sink. All the rest of the input goes -- into the /false/ sink. The only difference between 'first' and 'uptoFirst' combinators is in where they direct the -- /false/ portion of the input preceding the first /true/ part. uptoFirst :: forall m x b. Monad m => Splitter m x b -> Splitter m x b uptoFirst splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split (Left q) (Left (x, False)) = return (Left (q |> x)) split (Left q) (Left (x, True)) = putQueue q true >> put true x >> return (Right True) split (Left q) (Right b) = putQueue q true >> put edge b >> return (Right True) split (Right True) Right{} = return (Right False) split (Right True) (Left (x, True)) = put true x >> return (Right True) split (Right True) (Left (x, False)) = put false x >> return (Right False) split (Right False) (Left (x, _)) = put false x >> return (Right False) split (Right False) (Right _) = return (Right False) in foldMStream split (Left Seq.empty) source >>= either (flip putQueue false) (const $ return ()) -- | The result of the combinator 'last' is a splitter which directs all input to its /false/ sink, up to the last -- portion of the input which goes to its argument's /true/ sink. That portion of the input is the only one that goes to -- the resulting component's /true/ sink. The splitter returned by the combinator 'last' has to buffer the previous two -- portions of its input, because it cannot know if a true portion of the input is the last one until it sees the end of -- the input or another portion succeeding the previous one. last :: forall m x b. Monad m => Splitter m x b -> Splitter m x b last splitter = wrapMarkedSplitter splitter $ \source true false edge-> let get1 (Left (x, False)) = put false x >> getWith get1 source get1 p@(Left (x, True)) = get2 Nothing Seq.empty p get1 (Right b) = getWith (get2 (Just b) Seq.empty) source get2 mb q (Left (x, True)) = let q' = q |> x in get source >>= maybe (flush mb q') (get2 mb q') get2 mb q p = get3 mb q Seq.empty p get3 mb qt qf (Left (x, False)) = let qf' = qf |> x in get source >>= maybe (flush mb qt >> putQueue qf' false) (get3 mb qt qf') get3 mb qt qf p = do putQueue qt false putQueue qf false get1 p flush mb q = maybe (return ()) (put edge) mb >> putQueue q true in getWith get1 source -- | The result of the combinator 'lastAndAfter' is a splitter which directs all input to its /false/ sink, up to the -- last portion of the input which goes to its argument's /true/ sink. That portion and the remainder of the input is -- fed to the resulting component's /true/ sink. The difference between 'last' and 'lastAndAfter' combinators is where -- they feed the /false/ portion of the input, if any, remaining after the last /true/ part. lastAndAfter :: forall m x b. Monad m => Splitter m x b -> Splitter m x b lastAndAfter splitter = wrapMarkedSplitter splitter $ \source true false edge-> let get1 (Left (x, False)) = put false x >> getWith get1 source get1 p@(Left (x, True)) = get2 Nothing Seq.empty p get1 (Right b) = getWith (get2 (Just b) Seq.empty) source get2 mb q (Left (x, True)) = let q' = q |> x in get source >>= maybe (flush mb q') (get2 mb q') get2 mb q p = get3 mb q p get3 mb q (Left (x, False)) = let q' = q |> x in get source >>= maybe (flush mb q') (get3 mb q') get3 _ q p@(Left (x, True)) = putQueue q false >> get1 p get3 _ q b'@Right{} = putQueue q false >> get1 b' flush mb q = maybe (return ()) (put edge) mb >> putQueue q true in getWith get1 source -- | The 'prefix' combinator feeds its /true/ sink only the prefix of the input that its argument feeds to its /true/ -- sink. All the rest of the input is dumped into the /false/ sink of the result. prefix :: forall m x b. Monad m => Splitter m x b -> Splitter m x b prefix splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split 0 p@Left{} = split 1 p split 0 (Right b) = put edge b >> return 1 split 1 (Left (x, False)) = put false x >> return 2 split 1 (Left (x, True)) = put true x >> return 1 split 1 (Right b) = return 2 split 2 (Left (x, _)) = put false x >> return 2 split 2 Right{} = return 2 in foldMStream_ split 0 source -- | The 'suffix' combinator feeds its /true/ sink only the suffix of the input that its argument feeds to its /true/ -- sink. All the rest of the input is dumped into the /false/ sink of the result. suffix :: forall m x b. Monad m => Splitter m x b -> Splitter m x b suffix splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split Nothing (Left (x, False)) = put false x >> return Nothing split Nothing (Left (x, True)) = return (Just (Nothing, Seq.singleton x)) split Nothing (Right b) = return (Just (Just b, Seq.empty)) split (Just (mb, q)) (Left (x, True)) = return (Just (mb, q |> x)) split (Just (mb, q)) (Left (x, False)) = putQueue q false >> put false x >> return Nothing split (Just (mb, q)) (Right b) = putQueue q false >> return (Just (Just b, Seq.empty)) in foldMStream split Nothing source >>= \r-> case r of Nothing -> return () Just (Nothing, q) -> putQueue q true Just (Just b, q) -> put edge b >> putQueue q true -- | The 'even' combinator takes every input section that its argument /splitter/ deems /true/, and feeds even ones into -- its /true/ sink. The odd sections and parts of input that are /false/ according to its argument splitter are fed to -- 'even' splitter's /false/ sink. even :: forall m x b. Monad m => Splitter m x b -> Splitter m x b even splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split 1 (Left (x, False)) = put false x >> return 1 split 1 p@(Left (x, True)) = split 2 p split 1 (Right b) = return 2 split 2 (Left (x, True)) = put false x >> return 2 split 2 p@(Left (x, False)) = split 3 p split 2 (Right b) = put edge b >> return 4 split 3 (Left (x, False)) = put false x >> return 3 split 3 p@(Left (x, True)) = split 4 p split 3 (Right b) = put edge b >> return 4 split 4 (Left (x, True)) = put true x >> return 4 split 4 p@(Left (x, False)) = split 1 p split 4 (Right b) = return 2 in foldMStream_ split 1 source -- | Splitter 'startOf' issues an empty /true/ section at the beginning of every section considered /true/ by its -- argument splitter, otherwise the entire input goes into its /false/ sink. startOf :: forall m x b. Monad m => Splitter m x b -> Splitter m x (Maybe b) startOf splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split 1 (Left (x, False)) = put false x >> return 1 split 1 p@(Left (x, True)) = put edge Nothing >> split 2 p split 1 (Right b) = put edge (Just b) >> return 2 split 2 (Left (x, True)) = put false x >> return 2 split 2 p = split 1 p in foldMStream_ split 1 source -- | Splitter 'endOf' issues an empty /true/ section at the end of every section considered /true/ by its argument -- splitter, otherwise the entire input goes into its /false/ sink. endOf :: forall m x b. Monad m => Splitter m x b -> Splitter m x (Maybe b) endOf splitter = wrapMarkedSplitter splitter $ \source true false edge-> let split Nothing (Left (x, False)) = put false x >> return Nothing split Nothing p@(Left (x, True)) = split (Just Nothing) p split Nothing (Right b) = return (Just (Just b)) split (Just mb) (Left (x, True)) = put false x >> return (Just mb) split (Just mb) p@(Left (x, False)) = put edge mb >> split Nothing p split (Just mb) (Right b) = put edge mb >> return (Just $ Just b) in foldMStream split Nothing source >>= maybe (return ()) (put edge) -- | Combinator 'followedBy' treats its argument 'Splitter's as patterns components and returns a 'Splitter' that -- matches their concatenation. A section of input is considered /true/ by the result iff its prefix is considered -- /true/ by argument /s1/ and the rest of the section is considered /true/ by /s2/. The splitter /s2/ is started anew -- after every section split to /true/ sink by /s1/. followedBy :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x (b1, b2) followedBy parallel s1 s2 = isolateSplitter $ \ source true false edge -> pipePS parallel (transduce (splitterToMarker s1) source) (\source-> let get0 q = case Seq.viewl q of Seq.EmptyL -> getWith get1 source (Left (x, False)) :< rest -> put false x >> get0 rest (Left (x, True)) :< rest -> get2 Nothing Seq.empty q (Right b) :< rest -> get2 (Just b) Seq.empty rest get1 (Left (x, False)) = put false x >> getWith get1 source get1 p@(Left (x, True)) = get2 Nothing Seq.empty (Seq.singleton p) get1 (Right b) = get2 (Just b) Seq.empty Seq.empty get2 mb q q' = case Seq.viewl q' of Seq.EmptyL -> get source >>= maybe (testEnd mb q) (get2 mb q . Seq.singleton) (Left (x, True)) :< rest -> get2 mb (q |> x) rest (Left (x, False)) :< rest -> get3 mb q q' Right{} :< rest -> get3 mb q q' get3 mb q q' = do ((q1, q2), n) <- pipe (get7 Seq.empty q') (test mb q) case n of Nothing -> putQueue q false >> get0 (q1 >< q2) Just 0 -> get0 (q1 >< q2) Just n -> get8 (Just mb) n (q1 >< q2) get7 q1 q2 sink = case Seq.viewl q2 of Seq.EmptyL -> get source >>= maybe (return (q1, q2)) (\p-> either (put sink . fst) (const $ return ()) p >> get7 (q1 |> p) q2 sink) p :< rest -> either (put sink . fst) (const $ return ()) p >> get7 (q1 |> p) rest sink testEnd mb q = do ((), n) <- pipe (const $ return ()) (test mb q) case n of Nothing -> putQueue q false _ -> return () test mb q source = liftM snd $ pipe (transduce (splitterToMarker s2) source) (\source-> let get4 (Left (_, False)) = return Nothing get4 p@(Left (_, True)) = putQueue q true >> get5 0 p get4 p@(Right b) = maybe (return ()) (\b1-> put edge (b1, b)) mb >> putQueue q true >> get6 0 get5 n (Left (x, True)) = put true x >> get6 (succ n) get5 n _ = return (Just n) get6 n = get source >>= maybe (return $ Just n) (get5 n) in get source >>= maybe (return Nothing) get4) get8 Nothing 0 q = get0 q get8 (Just mb) 0 q = get2 mb Seq.empty q get8 mmb n q = case Seq.viewl q of Left (x, False) :< rest -> get8 Nothing (pred n) rest Left (x, True) :< rest -> get8 (maybe (Just Nothing) Just mmb) (pred n) rest Right b :< rest -> get8 (Just (Just b)) n rest in get0 Seq.empty) >> return () -- | Combinator '...' tracks the running balance of difference between the number of preceding starts of sections -- considered /true/ according to its first argument and the ones according to its second argument. The combinator -- passes to /true/ all input values for which the difference balance is positive. This combinator is typically used -- with 'startOf' and 'endOf' in order to count entire input sections and ignore their lengths. between :: forall m x b1 b2. MonadParallel m => Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x b1 between parallel s1 s2 = isolateSplitter $ \ source true false edge -> pipePS parallel (transduce (splittersToPairMarker parallel s1 s2) source) (let pass n x = (if n > 0 then put true x else put false x) >> return n pass' n x = (if n >= 0 then put true x else put false x) >> return n state n (Left (x, True, False)) = pass (succ n) x state n (Left (x, False, True)) = pass' (pred n) x state n (Left (x, True, True)) = pass' n x state n (Left (x, False, False)) = pass n x state 0 (Right (Left b)) = put edge b >> return 1 state n (Right (Left _)) = return (succ n) state n (Right (Right _)) = return (pred n) in foldMStream_ state 0) >> return () -- Helper functions wrapMarkedSplitter :: forall m x b1 b2. Monad m => Splitter m x b1 -> (forall a1 a2 a3 a4 d. (AncestorFunctor a1 d, AncestorFunctor a2 d, AncestorFunctor a3 d, AncestorFunctor a4 d) => Source m a1 (Either (x, Bool) b1) -> Sink m a2 x -> Sink m a3 x -> Sink m a4 b2 -> Coroutine d m ()) -> Splitter m x b2 wrapMarkedSplitter splitter splitMarked = isolateSplitter $ \ source true false edge -> pipe (transduce (splitterToMarker splitter) source) (\source-> splitMarked source true false edge) >> return () splitterToMarker :: forall m x b. Monad m => Splitter m x b -> Transducer m x (Either (x, Bool) b) splitterToMarker s = isolateTransducer $ \source sink-> split s source (mapSink (\x-> Left (x, True)) sink) (mapSink (\x-> Left (x, False)) sink) (mapSink Right sink) splittersToPairMarker :: forall m x b1 b2. (MonadParallel m) => Bool -> Splitter m x b1 -> Splitter m x b2 -> Transducer m x (Either (x, Bool, Bool) (Either b1 b2)) splittersToPairMarker parallel s1 s2 = let t source sink = pipe (\sync-> teeConsumers parallel (\source1-> split s1 source1 (mapSink (\x-> Left ((x, True), True)) sync) (mapSink (\x-> Left ((x, False), True)) sync) (mapSink (Right. Left) sync)) (\source2-> split s2 source2 (mapSink (\x-> Left ((x, True), False)) sync) (mapSink (\x-> Left ((x, False), False)) sync) (mapSink (Right . Right) sync)) source) (synchronizeMarks sink) >> return () synchronizeMarks :: forall m a1 a2 d. (Monad m, AncestorFunctor a1 d, AncestorFunctor a2 d) => Sink m a1 (Either (x, Bool, Bool) (Either b1 b2)) -> Source m a2 (Either ((x, Bool), Bool) (Either b1 b2)) -> Coroutine d m () synchronizeMarks sink source = foldMStream handleMark Nothing source >>= \Nothing-> return () where handleMark Nothing (Right b) = put sink (Right b) >> return Nothing handleMark Nothing (Left (p, first)) = return (Just (Seq.singleton (Left p), first)) handleMark state@(Just (q, first)) (Left (p, first')) | first == first' = return (Just (q |> Left p, first)) handleMark state@(Just (q, True)) (Right b@Left{}) = return (Just (q |> Right b, True)) handleMark state@(Just (q, False)) (Right b@Right{}) = return (Just (q |> Right b, False)) handleMark state (Right b) = put sink (Right b) >> return state handleMark state@(Just (q, pos')) mark@(Left ((x, t), pos)) = case Seq.viewl q of Seq.EmptyL -> return (Just (Seq.singleton (Left (x, t)), pos)) Right b :< rest -> put sink (Right b) >> handleMark (if Seq.null rest then Nothing else Just (rest, pos')) mark Left (y, t') :< rest -> put sink (Left $ if pos then (y, t, t') else (y, t', t)) >> return (if Seq.null rest then Nothing else Just (rest, pos')) returnQueuedList q = return $ concatMap (either ((:[]) . fst) (const [])) $ Foldable.toList $ Seq.viewl q in isolateTransducer t zipSplittersWith :: forall m x b1 b2 b. MonadParallel m => (Bool -> Bool -> Bool) -> (forall a1 a2 d. (AncestorFunctor a1 d, AncestorFunctor a2 d) => Source m a1 (Either b1 b2) -> Sink m a2 b -> Coroutine d m ()) -> Bool -> Splitter m x b1 -> Splitter m x b2 -> Splitter m x b zipSplittersWith f boundaries parallel s1 s2 = isolateSplitter $ \ source true false edge -> pipe (\edge-> pipePS parallel (transduce (splittersToPairMarker parallel s1 s2) source) (mapMStream_ (either (\(x, t1, t2)-> if f t1 t2 then put true x else put false x) (put edge)))) (flip boundaries edge) >> return () -- | Runs the second argument on every contiguous region of input source (typically produced by 'splitterToMarker') -- whose all values either match @Left (_, True)@ or @Left (_, False)@. groupMarks :: (Monad m, AncestorFunctor a d, AncestorFunctor a (SinkFunctor d x)) => Source m a (Either (x, Bool) b) -> (Maybe (Maybe b) -> Source m (SourceFunctor d x) x -> Coroutine (SourceFunctor d x) m r) -> Coroutine d m () groupMarks source getConsumer = start where start = getWith (either startContent startRegion) source startContent (x, False) = pipe (\sink-> pass False sink x) (getConsumer Nothing) >>= maybe (return ()) (either startContent startRegion) . fst startContent (x, True) = pipe (\sink-> pass True sink x) (getConsumer $ Just Nothing) >>= maybe (return ()) (either startContent startRegion) . fst startRegion b = pipe (next True) (getConsumer (Just $ Just b)) >>= maybe (return ()) (either startContent startRegion) . fst pass t sink x = put sink x >> next t sink next t sink = get source >>= maybe (return Nothing) (continue t sink) continue t sink (Left (x, t')) | t == t' = pass t sink x continue t sink p = return (Just p) -- | 'suppressProducer' runs the /producer/ argument with a new sink, suppressing everything 'put' in the sink. suppressProducer :: forall m a x r. (Functor a, Monad m) => (Sink m a x -> Coroutine a m r) -> Coroutine a m r suppressProducer producer = producer (nullSink :: Sink m a x) findsTrueIn :: forall m a d x b. (Monad m, AncestorFunctor a d) => Splitter m x b -> Source m a x -> Coroutine d m (Maybe (Maybe b)) findsTrueIn splitter source = pipe (\testTrue-> pipe (split splitter (liftSource source :: Source m d x) testTrue (nullSink :: Sink m d x)) get) get >>= \(((), maybeEdge), maybeTrue)-> return $ case maybeEdge of Nothing -> fmap (const Nothing) maybeTrue _ -> Just maybeEdge findsFalseIn :: forall m a d x b. (Monad m, AncestorFunctor a d) => Splitter m x b -> Source m a x -> Coroutine d m Bool findsFalseIn splitter source = pipe (\testFalse-> split splitter (liftSource source :: Source m d x) (nullSink :: Sink m d x) testFalse (nullSink :: Sink m d b)) get >>= \((), maybeFalse)-> return (isJust maybeFalse) teeConsumers :: forall m a d x r1 r2. MonadParallel m => Bool -> (forall a. OpenConsumer m a (SinkFunctor d x) x r1) -> (forall a. OpenConsumer m a (SourceFunctor d x) x r2) -> OpenConsumer m a d x (r1, r2) teeConsumers parallel c1 c2 source = pipePS parallel consume1 c2 where consume1 sink = c1 (teeSource sink source' :: Source m (SinkFunctor d x) x) source' :: Source m d x source' = liftSource source