machinecell 3.0.1 → 3.1.0

raw patch · 8 files changed

+541/−253 lines, 8 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +18/−0
• machinecell.cabal +2/−2
• src/Control/Arrow/Machine.hs +39/−57
• src/Control/Arrow/Machine/Misc/Discrete.hs +100/−6
• src/Control/Arrow/Machine/Misc/Exception.hs +10/−1
• src/Control/Arrow/Machine/Types.hs +136/−98
• src/Control/Arrow/Machine/Utils.hs +150/−57
• test/spec.hs +86/−32

@@ -1,3 +1,21 @@++3.1.0+-----------+* Add `Discrete` utilities+    * eval+    * refer+    * kSwitch+    * dkSwitch+    * Num instance definition+* Add source utilities+    * blockingSource+    * interleave+    * blocking+* Delete `sample`+* Change a switching behavior. With previous implementation, a switching doesn't occur+  when a runnning transducer emits a trigger event using `now` transducer.++ 3.0.1 ----------- * Fix performance issue of switch, dSwitch, accum, dAccum.

@@ -1,5 +1,5 @@ name:                machinecell-version:             3.0.1+version:             3.1.0 synopsis:            Arrow based stream transducers license:             BSD3 license-file:        LICENSE@@ -53,4 +53,4 @@ source-repository this   type:		git   location:	https://github.com/as-capabl/machinecell.git-  tag:		release-3.0.1+  tag:		release-3.1.0

@@ -15,7 +15,7 @@       (         -- * Quick introduction         -- $introduction-        +         -- * Note         -- $note @@ -72,14 +72,12 @@ -- -- "ProcessA a (Event b) (Event c)" transducers are actually one-directional composable pipes. ----- They can be constructed from `Plan` monads.+-- They can be constructed from the `Plan` monad. -- In `Plan` monad context, `await` and `yield` can be used to get and emit values. -- And actions of base monads can be `lift`ed to the context. ----- Then, resulting processes are composed as `Category` using `(\>\>\>)` operator.--- -- @--- source :: ProcessA (Kleisli IO) (Event ()) (Event String)  +-- source :: ProcessA (Kleisli IO) (Event ()) (Event String) -- source = repeatedlyT kleisli0 $ --   do --     _ \<- await@@ -100,31 +98,41 @@ --     lift $ putStrLn x -- @ ----- >>> runKleisli (run_ $ source >>> pipe >>> sink) (repeat ())+-- Then, resulting processes are composed as `Category` using `(\>\>\>)` operator. ----- The above code reads two lines from stdin, puts a concatenated line to stdout and finishes.+-- > runKleisli (run_ $ source >>> pipe >>> sink) (repeat ()) ----- Unlike other pipe libraries, even a source must call `await`.+-- This reads two lines from stdin, puts a concatenated line to stdout and finishes. --+-- Unlike other pipe libraries, even the source calls `await`. -- The source awaits dummy input, namely "(repeat ())", and discard input values.+-- -- Even the input is an infinite list, this program stops when the "pipe" transducer stops. -- -- == More details on finalizing -- -- Finalizing behavior of transducers obey the following scenario.--- +-- -- 1. Signals of type `Event` can carry /end signs/. -- 2. Most transducers stop when they get an end sign. --    (Some exceptions can be made by `onEnd` or `catchP`) -- 3. If `run` function detects an end sign as an output of a running transducer, --    it stops feeding input values and alternatively feeds end signs. -- 4. Continue iteration until no more events can be occurred.--- +-- -- So "await \`catchP\` some_cleanup" can handle any stop of both upstream and downstream. -- -- On the other hand, a plan never gets end sign without calling await.--- That's why even sources must call await.+-- So it is better that even a source calls await. --+-- A source that calls await periodically is an "interleaved source".+-- Interleaved sources have a number of advantages.+-- They can be controled their output timings by their upstream, or can be stopped any time.+--+-- There is another kind of source that doesn't call await, namely "blocking source".+--+-- see "sources" section of "Control.Arrow.Machine.Utils" documentation.+-- -- = Arrow composition -- -- One of the most attractive feature of machinecell is the /arrow composition/.@@ -162,7 +170,7 @@ -- The transducers we have already seen are all have input and output type wrapped by `Event`. -- We have not taken care of them so far because all of them are cancelled each other. ----- But several built-in transducers provides non-event values like below.+-- But several built-in transducers provide non-event values like below. -- -- @ -- hold :: ArrowApply a =\> b -\> ProcessA a (Event b) b@@ -177,7 +185,7 @@ -- values that appear naked in arrow notations are /behaviour/, -- that means /coutinuous/ time-varying values, -- whereas /event/ values are /discrete/.--- +-- -- Note that all values that can be input, output, or taken effects must be discrete. -- -- To use continuous values anyhow interacting the real world,@@ -209,26 +217,27 @@  -- $note -- = Purity of `ProcessA (-\>)`--- Since `a` of `ProcessA a b c` represents base monad(ArrowApply), `ProcessA (-\>)` is expected to be pure.+-- Since the 1st type parameter of `ProcessA` represents base monad(ArrowApply),+-- "ProcessA (-\>)" is expected to be pure. ----- In other words, the following arrow results the same result for arbitrary `f`.+-- In other words, the following arrow results the same result for arbitrary f. -- -- @ -- proc x -\> --   do---     _ \<- fit arr f -\< x+--     _ \<- `fit` arr f -\< x --     g -\< x -- @--- --- Which is desugared to `f &&& g \>\>\> arr snd`. At least if `Event` constructor is exported,--- the proposition is falsible.--- When `f` is "arr (replicate k) \>\>\> fork" for some integer k and `g` is "arr (const $ Event ())",+--+-- Which is desugared to "fit arr f &&& g \>\>\> arr snd". At least if `Event` constructor is exported,+-- someone can make a counter example.+-- When f is "arr (replicate k) \>\>\> fork" for some integer k and g is "arr (const $ Event ())", -- g yields ()s for k times. That is because, the result value of arrow "f &&& g" is -- nothing but "(Event x, Event ())" and its number of yields is k because "Event x" must--- be yielded k times. +-- be yielded k times. ----- That's because `Event` constructor is hidden.--- Using primitives exported by this module, it works almost correctly.+-- This is the reason why the `Event` constructor is hidden.+-- Using exported primitives, it works almost correctly. -- Event number is conserved by inserting an appropriate number of `NoEvent`s. -- But there is still a loophole. --@@ -245,10 +254,12 @@ -- the problem will be avoided. -- -- = Looping--- +-- -- Although `ProcessA` is an instance of `ArrowLoop`,--- to send values to upstream, there is a little difficulties.--- +-- there is a large limitation.+--+-- The limitation is, Events mustn't be looped back to upstream.+-- -- In example below, result is [0, 0, 0, 0], not [1, 2, 3, 4]. -- -- @@@ -269,35 +280,6 @@ -- almost always `NoEvent`s. -- -- A better way to send events to upstream is, to encode them to behaviours using `dHold`,--- `dAccum`, and so on, then send to upstream in rec statement.------ = Unsafe primitives------ In the code below, `edge` does not fire.------ @--- encloseState False (sta \>\>\> peekState) \>\>\> edge--- @------ where------ @--- sta = constructT (ary0 $ statefully unArrowMonad) (put True \>\> await \>\> put False)--- @------ That is because, when "put True" is executing, the backtracking is going up and never hits `edge`--- until "put False" is executed.------ The same occurs for "proc b -> if b then (now -< ()) else (returnA -< noEvent)" instead of `edge`.------ Even worse, it again breaks the purity of `ProcessA`.--- `await` gets `NoEvent` if some "arr (replicate k) \>\>\> fork" is inserted somewhere in upstream.--- Then `edge` may fire because "put False" execution is delayed.------ This means that, `encloseState`, `peekState`, `edge`, and `ArrowChoice` instance for `ProcessA`--- should never be existed simultaneously.------ Moreover, their primitives `unsafeSteady`, `unsafeExhaust`, `fitEx` are so.+-- `dAccum` and so on, then send to upstream in rec statement. ----- But I hope some of them can be rescued. So for now, this library contains them all.-       +

@@ -5,15 +5,17 @@ {-# LANGUAGE TypeSynonymInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} + module     Control.Arrow.Machine.Misc.Discrete       (-        -- *Discrete-        -- | This module should be imported manually.+        -- * Discrete type+        -- $type+         T(),         updates,         value,-        +         arr,         arr2,         arr3,@@ -24,10 +26,18 @@         hold,         accum,         fromEq,-        +         edge,         asUpdater,-        Alg+        kSwitch,+        dkSwitch,++        -- * Discrete algebra+        -- $alg++        Alg(Alg),+        eval,+        refer       ) where @@ -39,6 +49,24 @@ import qualified Control.Arrow.Machine as P import Data.Monoid (mconcat, mappend) +{-$type+This module should be imported manually. Qualified import is recommended.++This module provides an abstraction that continuous values with+finite number of changing points.++>>> import qualified Control.Arrow.Machine.Misc.Discrete as D+>>> run (D.hold "apple" >>> D.arr reverse >>> D.edge) ["orange", "grape"]+["elppa", "egnaro", "eparg"]++In above example, input data of "reverse" is continuous.+But the "D.edge" transducer extracts changing points without calling string comparison.++This is possible because the intermediate type `T` has the information of changes+together with the value information.+-}++-- |The discrete signal type. data T a = T {     updates :: (P.Event ()),     value :: a@@ -49,6 +77,16 @@     P.ProcessA a (P.Event (), b) (T b) makeT = Arr.arr $ uncurry T +-- TODO this should be implemented by switch+rising ::+    ArrowApply a =>+    P.ProcessA a b (T c) ->+    P.ProcessA a b (T c)+rising sf = proc x ->+  do+    (dy, n) <- sf &&& P.now -< x+    makeT -< (updates dy `mappend` n, value dy)+ arr ::     ArrowApply a =>     (b->c) ->@@ -138,9 +176,53 @@ asUpdater ar = edge >>> P.anytime ar  +kSwitch ::+    ArrowApply a =>+    P.ProcessA a b (T c) ->+    P.ProcessA a (b, T c) (P.Event t) ->+    (P.ProcessA a b (T c) -> t -> P.ProcessA a b (T c)) ->+    P.ProcessA a b (T c)+kSwitch sf test k = P.kSwitch sf test (\sf' x -> rising (k sf' x)) -newtype Alg a i o = Alg { eval :: P.ProcessA a i (T o) }+dkSwitch ::+    ArrowApply a =>+    P.ProcessA a b (T c) ->+    P.ProcessA a (b, T c) (P.Event t) ->+    (P.ProcessA a b (T c) -> t -> P.ProcessA a b (T c)) ->+    P.ProcessA a b (T c)+dkSwitch sf test k = P.dkSwitch sf test (\sf' x -> rising (k sf' x)) ++{-$alg+Calculations between discrete types.++An example is below.++@+holdAdd ::+    (ArrowApply a, Num b) =>+    ProcessA a (Event b, Event b) (Discrete b)+holdAdd = proc (evx, evy) ->+  do+    x <- D.hold 0 -< evx+    y <- D.hold 0 -< evy+    D.eval (refer fst + refer snd) -< (x, y)+@++The last line is equivalent to "arr2 (+) -< (x, y)".+Using Alg, you can construct more complex calculations+between discrete signals.+-}++-- |Discrete algebra type.+newtype Alg a i o =+    Alg { eval :: P.ProcessA a i (T o) }++refer ::+    ArrowApply a =>+    (e -> T b) -> Alg a e b+refer = Alg . Arr.arr+ instance     ArrowApply a => Functor (Alg a i)   where@@ -151,3 +233,15 @@   where     pure = Alg . constant     af <*> aa = Alg $ (eval af &&& eval aa) >>> arr2 ($)++instance+    (ArrowApply a, Num o) =>+    Num (Alg a i o)+  where+    abs = fmap abs+    signum = fmap signum+    fromInteger = pure . fromInteger+    (+) = liftA2 (+)+    (-) = liftA2 (-)+    (*) = liftA2 (*)+

@@ -3,6 +3,9 @@ module     Control.Arrow.Machine.Misc.Exception       (+        -- * Variations of catchP+        -- $variation+         catch,         handle,         bracket,@@ -13,9 +16,15 @@        ) where - import Control.Arrow.Machine.Types ++{-$variation+This module provides variations of catchP.++If you use this module together with "Control.Exception" module of base package,+import this package qualified.+-}  catch :: Monad m =>     PlanT i o m a -> PlanT i o m a -> PlanT i o m a

@@ -30,7 +30,7 @@         filterLeft,         filterRight,         evMap,-        +         -- * Coroutine monad         -- | Procedural coroutine monad that can await or yield values.         --@@ -58,7 +58,7 @@         run,         runOn,         run_,-        +         -- * Running machines (step-by-step)         ExecInfo(..),         stepRun,@@ -66,7 +66,7 @@          -- * Primitive machines - switches         -- | Switches inspired by Yampa library.-        -- Signature is almost same, but collection requirement is  not only 'Functor', +        -- Signature is almost same, but collection requirement is  not only 'Functor',         -- but 'Tv.Traversable'. This is because of side effects.         switch,         dSwitch,@@ -115,8 +115,8 @@ -- Once a value `Feed`ed, the machine is `Sweep`ed until it `Suspend`s. data Phase = Feed | Sweep | Suspend deriving (Eq, Show) -instance -    Monoid Phase +instance+    Monoid Phase   where     mempty = Sweep @@ -150,7 +150,7 @@     step :: ArrowApply a => ProcessA a b c -> a b (f c, ProcessA a b c)     helperToMaybe :: f a -> Maybe a     weakly :: a -> f a-  +     step' :: ArrowApply a => ProcessA a b c -> a (f b) (f c, ProcessA a b c)     step' pa = proc hx ->       do@@ -161,12 +161,32 @@             mx                 -<< () +    testStep' ::+        ArrowApply a =>+        (x -> a b (f c, x)) ->+        (x -> b -> c) ->+        x ->+        ProcessA a (b, c) t ->+        a b (f c, f t, x, ProcessA a (b, c) t)++testStep ::+    (ArrowApply a, ProcessHelper f) =>+    ProcessA a b c ->+    ProcessA a (b, c) t ->+    a b (f c, f t, ProcessA a b c, ProcessA a (b, c) t)+testStep = testStep' step suspend+ instance     ProcessHelper Identity   where     step pa = feed pa >>> first (arr Identity)     helperToMaybe = Just . runIdentity     weakly = Identity+    testStep' stp' _ sf test = proc x ->+      do+        (Identity y, sf') <- stp' sf -< x+        (t, test') <- feed test -< (x, y)+        returnA -< (return y, return t, sf', test')  instance     ProcessHelper Maybe@@ -174,6 +194,26 @@     step = sweep     helperToMaybe = id     weakly _ = Nothing+    testStep' stp' sus' sf0 test0 = proc x ->+      do+        let y = sus' sf0 x+        (mt, test') <- sweep test0 -< (x, y)+        (case mt of+            Just t -> arr $ const (Just y, Just t, sf0, test')+            Nothing -> cont sf0 test')+                -<< x+      where+        cont sf test = proc x ->+          do+            (my, sf') <- stp' sf -< x+            (case my of+                Just y -> cont2 y sf' test+                Nothing -> arr $ const (Nothing, Nothing, sf', test))+                    -<< x+        cont2 y sf test = proc x ->+          do+            (t, test') <- feed test -< (x, y)+            returnA -< (Just y, Just t, sf, test')  makePA ::     Arrow a =>@@ -186,12 +226,12 @@     sweep = h,     suspend = sus   }-            -       ++ -- |Natural transformation fit ::-    (ArrowApply a, ArrowApply a') => -    (forall p q. a p q -> a' p q) -> +    (ArrowApply a, ArrowApply a') =>+    (forall p q. a p q -> a' p q) ->     ProcessA a b c -> ProcessA a' b c fit f pa =     arr Identity >>>@@ -216,7 +256,7 @@     {-# INLINE dimap #-}  dimapProc ::-    ArrowApply a => +    ArrowApply a =>     (b->c)->(d->e)->     ProcType a c d -> ProcType a b e dimapProc f g pa = makePA@@ -247,7 +287,7 @@     {-# INLINE (.) #-}  -instance +instance     ArrowApply a => Arrow (ProcessA a)   where     arr = arrProc@@ -294,7 +334,7 @@ {-# NOINLINE arrProc #-}  -- |Composition is proceeded by the backtracking strategy.-compositeProc :: ArrowApply a => +compositeProc :: ArrowApply a =>               ProcType a b d -> ProcType a d c -> ProcType a b c compositeProc f0 g0 = ProcessA {     feed = proc x ->@@ -338,7 +378,7 @@ "ProcessA: */id"     forall f. compositeProc f idProc = f -"ProcessA: concat/concat" +"ProcessA: concat/concat"     forall f g h. compositeProc (compositeProc f g) h = compositeProc f (compositeProc g h)  "ProcessA: dimap/dimap"@@ -404,7 +444,7 @@     ArrowApply a => ArrowLoop (ProcessA a)   where     loop pa =-        makePA +        makePA             (proc x ->               do                 (hyd, pa') <- step pa -< (x, loopSusD x)@@ -418,8 +458,8 @@ data Event a = Event a | NoEvent | End  -instance -    Functor Event +instance+    Functor Event   where     fmap _ NoEvent = NoEvent     fmap _ End = End@@ -441,7 +481,7 @@  -- | Signals that can be absent(`NoEvent`) or end. -- For composite structure, `collapse` can be defined as monoid sum of all member occasionals.-class +class     Occasional' a   where     collapse :: a -> Event ()@@ -465,12 +505,12 @@     noEvent = (noEvent, noEvent)     end = (end, end) -instance +instance     Occasional' (Event a)   where     collapse = (() <$) -instance +instance     Occasional (Event a)   where     noEvent = NoEvent@@ -502,7 +542,7 @@ -- -- While "ProcessA a (Event b) (Event c)" means a transducer from b to c, -- function b->c can be lifted into a transducer by fhis function.--- +-- -- But in most cases you needn't call this function in proc-do notations, -- because `arr`s are completed automatically while desugaring. --@@ -516,12 +556,12 @@ -- -- @ -- evMap f--- @            +-- @ evMap ::  Arrow a => (b->c) -> a (Event b) (Event c) evMap = arr . fmap  -stopped :: +stopped ::     (ArrowApply a, Occasional c) => ProcessA a b c stopped = arr (const end) @@ -591,7 +631,7 @@ catchP:: Monad m =>     PlanT i o m a -> PlanT i o m a -> PlanT i o m a -catchP (PlanT pl) cont0 = +catchP (PlanT pl) cont0 =     PlanT $ F.FT $ \pr free ->         F.runFT             pl@@ -621,18 +661,18 @@   constructT ::-    (Monad m, ArrowApply a) => +    (Monad m, ArrowApply a) =>     (forall b. m b -> a () b) ->-    PlanT i o m r -> +    PlanT i o m r ->     ProcessA a (Event i) (Event o) constructT = constructT'   constructT' ::     forall a m i o r.-    (Monad m, ArrowApply a) => +    (Monad m, ArrowApply a) =>     (forall b. m b -> a () b) ->-    PlanT i o m r -> +    PlanT i o m r ->     ProcessA a (Event i) (Event o) constructT' fit0 (PlanT pl0) = prependProc $ F.runFT pl0 pr free   where@@ -651,11 +691,11 @@     prependFeed (Event x, pa) = arr $ const (Event x, pa)     prependFeed (NoEvent, pa) = feed pa     prependFeed (End, _) = arr $ const (End, stopped)-  +     prependSweep (Event x, pa) = arr $ const (Just (Event x), pa)     prependSweep (NoEvent, pa) = sweep pa     prependSweep (End, _) = arr $ const (Just End, stopped)-  +     pr _ = return (End, stopped)      free ::@@ -685,9 +725,9 @@     eToMpure e = Just e  -repeatedlyT :: (Monad m, ArrowApply a) => +repeatedlyT :: (Monad m, ArrowApply a) =>               (forall b. m b -> a () b) ->-              PlanT i o m r -> +              PlanT i o m r ->               ProcessA a (Event i) (Event o)  repeatedlyT f pl = constructT f $ forever pl@@ -695,12 +735,12 @@  -- for pure construct :: ArrowApply a =>-             Plan i o t -> +             Plan i o t ->              ProcessA a (Event i) (Event o) construct pl = constructT (ary0 unArrowMonad) pl  repeatedly :: ArrowApply a =>-              Plan i o t -> +              Plan i o t ->               ProcessA a (Event i) (Event o) repeatedly pl = construct $ forever pl @@ -708,9 +748,9 @@ -- -- Switches ---switch :: -    ArrowApply a => -    ProcessA a b (c, Event t) -> +switch ::+    ArrowApply a =>+    ProcessA a b (c, Event t) ->     (t -> ProcessA a b c) ->     ProcessA a b c switch sf k = makePA@@ -726,9 +766,9 @@     (fst . suspend sf)  -dSwitch :: -    ArrowApply a => -    ProcessA a b (c, Event t) -> +dSwitch ::+    ArrowApply a =>+    ProcessA a b (c, Event t) ->     (t -> ProcessA a b c) ->     ProcessA a b c dSwitch sf k = makePA@@ -745,8 +785,8 @@     (fst . suspend sf)  -rSwitch :: -    ArrowApply a => ProcessA a b c -> +rSwitch ::+    ArrowApply a => ProcessA a b c ->     ProcessA a (b, Event (ProcessA a b c)) c rSwitch p = rSwitch' (p *** Cat.id) >>> arr fst   where@@ -755,8 +795,8 @@     test = proc (_, (_, r)) -> returnA -< r  -drSwitch :: -    ArrowApply a => ProcessA a b c -> +drSwitch ::+    ArrowApply a => ProcessA a b c ->     ProcessA a (b, Event (ProcessA a b c)) c  drSwitch p =  drSwitch' (p *** Cat.id)@@ -765,7 +805,7 @@   kSwitch ::-    ArrowApply a => +    ArrowApply a =>     ProcessA a b c ->     ProcessA a (b, c) (Event t) ->     (ProcessA a b c -> t -> ProcessA a b c) ->@@ -773,8 +813,7 @@ kSwitch sf test k = makePA     (proc x ->       do-        (hy, sf') <- step sf -< x-        (hevt, test') <- step' test -< (x,) <$> hy+        (hy, hevt, sf', test') <- testStep sf test -< x         (case (helperToMaybe hevt)           of             Just (Event t) -> step (k sf' t)@@ -783,7 +822,7 @@     (suspend sf)  dkSwitch ::-    ArrowApply a => +    ArrowApply a =>     ProcessA a b c ->     ProcessA a (b, c) (Event t) ->     (ProcessA a b c -> t -> ProcessA a b c) ->@@ -791,16 +830,15 @@ dkSwitch sf test k = makePA     (proc x ->       do-        (hy, sf') <- step sf -< x-        (hevt, test') <- step' test -< (x,) <$> hy+        (hy, hevt, sf', test') <- testStep sf test -< x         (case (helperToMaybe hevt)           of             Just (Event t) -> arr $ const (hy, k sf' t)             _ -> arr $ const (hy, dkSwitch sf' test' k))                 -<< x)     (suspend sf)-  -broadcast :: ++broadcast ::     Functor col =>     b -> col sf -> col (b, sf) broadcast x sfs = fmap (\sf -> (x, sf)) sfs@@ -821,7 +859,7 @@     ProcessA a b (col c) parB = par broadcast -suspendAll :: +suspendAll ::     (ArrowApply a, Tv.Traversable col) =>     (forall sf. (b -> col sf -> col (ext, sf))) ->     col (ProcessA a ext c) ->@@ -829,7 +867,7 @@ suspendAll r sfs = (sus <$>) . (r `flip` sfs)   where     sus (ext, sf) = suspend sf ext-     + traverseResult ::     forall h col c.     (Tv.Traversable col, ProcessHelper h) =>@@ -847,7 +885,7 @@         result = fst <$> hxs       in         if exist then result else join (weakly result)-     + parCore ::     (ArrowApply a, Tv.Traversable col, ProcessHelper h) =>     (forall sf. (b -> col sf -> col (ext, sf))) ->@@ -878,15 +916,15 @@ pSwitch r sfs test k = makePA     (proc x ->       do-        (hzs, sfs') <- parCore r sfs -<< x-        (hevt, test') <- step' test -< (x,) <$> hzs+        (hzs, hevt, sfs', test') <-+            testStep' (parCore r) (suspendAll r) sfs test -< x         (case helperToMaybe hevt           of             Just (Event t) -> (step (k sfs' t))             _ -> arr $ const (hzs, pSwitch r sfs' test' k))                 -<< x)     (suspendAll r sfs)-  + pSwitchB ::     (ArrowApply a, Tv.Traversable col) =>     col (ProcessA a b c) ->@@ -911,8 +949,8 @@         (hzs, sfs') <- parCore r sfsNew -<< x         returnA -< (hzs, rpSwitch r sfs'))     (fst >>> suspendAll r sfs)-     + rpSwitchB ::     (ArrowApply a, Tv.Traversable col) =>     col (ProcessA a b c) ->@@ -927,9 +965,9 @@ -- -- Unsafe primitives ---    + -- | Repeatedly call `p`.---    +-- -- How many times `p` is called is indefinite. -- So `p` must satisfy the equation below; --@@ -974,7 +1012,7 @@         else return ()  -- | Monoid wrapper-data WithEnd r = WithEnd { +data WithEnd r = WithEnd {     getRWE :: r,     getContWE :: !Bool   }@@ -1006,8 +1044,8 @@     ProcessA a (Event i) o ->     RM a (Event i) o m x ->     m x-runRM f pa mx = -    evalStateT mx $ +runRM f pa mx =+    evalStateT mx $         RunInfo {             freezeRI = pa,             getInputRI = NoEvent,@@ -1018,8 +1056,8 @@   -feed_ :: -    Monad m => +feed_ ::+    Monad m =>     i -> i -> RM a i o m Bool feed_ input padding =   do@@ -1037,15 +1075,15 @@         else             return False -feedR :: -    Monad m => +feedR ::+    Monad m =>     i -> RM a (Event i) o m Bool feedR x = feed_ (Event x) NoEvent   {--finalizeE :: -    Monad m => +finalizeE ::+    Monad m =>     RM a (Event i) o m Bool finalizeE = feed_ End End -}@@ -1054,9 +1092,9 @@     Monad m =>     RM a i o m (ProcessA a i o) freeze = gets freezeRI-     -sweepR :: ++sweepR ::     Monad m =>     RM a i o m o sweepR =@@ -1075,7 +1113,7 @@                 getPhaseRI = Sweep               }             return y-      Sweep ->  +      Sweep ->         do             fit0 <- gets getFitRI             x <- gets getPaddingRI@@ -1089,17 +1127,17 @@         do             x <- gets getPaddingRI             return $ suspend pa x-    -    -      -sweepAll :: ++++sweepAll ::     (ArrowApply a, Monoid r, Monad m) =>     (o->r) ->     WriterT (WithEnd r) (RM a i (Event o) m) ()-sweepAll outpre = -        while_ +sweepAll outpre =+        while_             ((not . (== Suspend)) `liftM` lift (gets getPhaseRI)) $           do             evx <- lift sweepR@@ -1121,7 +1159,7 @@     a (f b) r runOn outpre pa0 = unArrowMonad $ \xs ->   do-    wer <- runRM arrowMonad pa0 $ execWriterT $ +    wer <- runRM arrowMonad pa0 $ execWriterT $       do         -- Sweep initial events.         (_, wer) <- listen $ sweepAll outpre@@ -1158,20 +1196,20 @@         Builder $ \c e -> g c (f c e)  -- | Run a machine.-run :: -    ArrowApply a => -    ProcessA a (Event b) (Event c) -> +run ::+    ArrowApply a =>+    ProcessA a (Event b) (Event c) ->     a [b] [c]-run pa = +run pa =     runOn (\x -> Builder $ \c e -> c x e) pa >>>     arr (\b -> build (unBuilder b))  -- | Run a machine discarding all results.-run_ :: -    ArrowApply a => -    ProcessA a (Event b) (Event c) -> +run_ ::+    ArrowApply a =>+    ProcessA a (Event b) (Event c) ->     a [b] ()-run_ pa = +run_ pa =     runOn (const ()) pa  @@ -1194,18 +1232,18 @@     Alternative f => Monoid (ExecInfo (f a))   where     mempty = ExecInfo empty False False-    ExecInfo y1 c1 s1 `mappend` ExecInfo y2 c2 s2 = +    ExecInfo y1 c1 s1 `mappend` ExecInfo y2 c2 s2 =         ExecInfo (y1 <|> y2) (c1 || c2) (s1 || s2)   -- | Execute until an input consumed and the machine suspends.-stepRun :: +stepRun ::     ArrowApply a =>     ProcessA a (Event b) (Event c) ->     a b (ExecInfo [c], ProcessA a (Event b) (Event c)) stepRun pa0 = unArrowMonad $ \x ->   do-    (pa, wer)  <- runRM arrowMonad pa0 $ runWriterT $ +    (pa, wer)  <- runRM arrowMonad pa0 $ runWriterT $       do         sweepAll singleton         _ <- lift $ feedR x@@ -1217,13 +1255,13 @@     singleton x = Endo (x:)      retval WithEnd {..} = ExecInfo {-        yields = appEndo getRWE [], -        hasConsumed = True, +        yields = appEndo getRWE [],+        hasConsumed = True,         hasStopped = not getContWE       }  -- | Execute until an output produced.-stepYield :: +stepYield ::     ArrowApply a =>     ProcessA a (Event b) (Event c) ->     a b (ExecInfo (Maybe c), ProcessA a (Event b) (Event c))@@ -1234,20 +1272,20 @@     pa <- lift freeze     return (r, pa) -  where +  where     go x =       do         csmd <- lift $ feedR x         modify $ \ri -> ri { hasConsumed = csmd }-                             +         evo <- lift sweepR-        +         case evo           of             Event y ->               do                 modify $ \ri -> ri { yields = Just y }-    +             NoEvent ->               do                 csmd' <- gets hasConsumed

@@ -24,7 +24,7 @@          -- * Switches         -- | Switches inspired by Yampa library.-        -- Signature is almost same, but collection requirement is  not only 'Functor', +        -- Signature is almost same, but collection requirement is  not only 'Functor',         -- but 'Tv.Traversable'. This is because of side effects.         switch,         dSwitch,@@ -37,11 +37,17 @@         rpSwitch,         rpSwitchB, +        -- * Sources+        -- $sources++        source,+        blockingSource,+        interleave,+        blocking,+         -- * Other utility arrows         tee,         gather,-        sample,-        source,         fork,         filter,         echo,@@ -50,10 +56,10 @@         parB,         now,         onEnd,-    +         -- * Transformer         readerProc-       )+     ) where  import Prelude hiding (filter)@@ -74,15 +80,15 @@   -hold :: +hold ::     ArrowApply a => b -> ProcessA a (Event b) b-hold old = proc evx -> +hold old = proc evx ->   do     rSwitch (pure old) -< ((), pure <$> evx) -dHold :: +dHold ::     ArrowApply a => b -> ProcessA a (Event b) b-dHold old = proc evx -> +dHold old = proc evx ->   do     drSwitch (pure old) -< ((), pure <$> evx) @@ -97,7 +103,7 @@ dAccum x = dSwitch (pure x &&& arr (($x)<$>)) dAccum  -edge :: +edge ::     (ArrowApply a, Eq b) =>     ProcessA a b (Event b) edge = proc x ->@@ -110,46 +116,151 @@     judge (prv, x) = if prv == Just x then Nothing else Just x  +-- $sources+-- In addition to the main event stream privided by `run`,+-- there are two other ways to provide additional input streams,+-- "interleaved" sources and "blocking" sources.+--+-- Interleaved sources are actually Event -> Event transformers+-- that don't see the values of the input events.+-- They discard input values and emit their values according to input event timing.+--+-- Blocking sources emit their events independent from upstream.+-- Until they exhaust their values, they block upstream transducers.+--+-- Here is a demonstration of two kind of sources.+--+-- @+-- a = proc x ->+--   do+--     y1 <- source [1, 2, 3] -< x+--     y2 <- source [4, 5, 6] -< x+--+--     gather -< [y1, y2]+-- -- run a (repeat ()) => [1, 4, 2, 5, 3, 6]+--+-- b = proc _ ->+--   do+--     y1 <- blockingSource [1, 2, 3] -< ()+--     y2 <- blockingSource [4, 5, 6] -< ()+--+--     gather -< [y1, y2]+-- -- run b [] => [4, 5, 6, 1, 2, 3]+-- @+--+-- In above code, you'll see that output values of `source`+-- (an interleaved source) are actually interelaved,+-- while `blockingSource` blocks another upstream source.+--+-- And they can both implemented using `PlanT`.+-- The only one deference is `await` call to listen upstream event timing.+--+-- An example is below.+--+-- @+-- interleavedStdin = constructT kleisli0 (forever pl)+--   where+--     pl =+--       do+--         _ <- await+--         eof <- isEOF+--         if isEOF then stop else return()+--         getLine >>= yield+--+-- blockingStdin = pure noEvent >>> constructT kleisli0 (forever pl)+--   where+--     pl =+--       do+--         -- No await here+--         eof <- isEOF+--         if isEOF then stop else return()+--         getLine >>= yield+-- @+--+-- They are different in the end behavior.+-- When upstream stops, an interleaved source stops because await call fails.+-- But a blocking source doesn't stop until its own termination. ++-- | Provides a source event stream.+-- A dummy input event stream is needed. --+-- @+--   run af [...]+-- @+--+-- is equivalent to+--+-- @+--   run (source [...] >>> af) (repeat ())+-- @+source ::+    (ArrowApply a, Fd.Foldable f) =>+    f c -> ProcessA a (Event b) (Event c)+source l = construct $ Fd.mapM_ yd l+  where+    yd x = await >> yield x++-- | Provides a blocking event stream.+blockingSource ::+    (ArrowApply a, Fd.Foldable f) =>+    f c -> ProcessA a () (Event c)+blockingSource l = pure noEvent >>> construct (Fd.mapM_ yield l)++-- | Make a blocking source interleaved.+interleave ::+    ArrowApply a =>+    ProcessA a () (Event c) ->+    ProcessA a (Event b) (Event c)+interleave bs0 = sweep1 (pure () >>> bs0)+  where+    waiting bs r =+        dSwitch+            (handler bs r)+            sweep1+    sweep1 bs =+        kSwitch+            bs+            (arr snd)+            waiting+    handler bs r = proc ev ->+      do+        ev' <- splitter bs r -< ev+        returnA -< (filterJust (fst <$> ev'), snd <$> ev')+    splitter bs r =+        construct $+          do+            _ <- await+            yield (Just r, bs)+          `catchP`+            yield (Nothing, bs >>> muted)++-- | Make an interleaved source blocking.+blocking ::+    ArrowApply a =>+    ProcessA a (Event ()) (Event c) ->+    ProcessA a () (Event c)+blocking is = dSwitch (blockingSource (repeat ()) >>> is >>> (Cat.id &&& onEnd)) (const stopped)+++-- -- other utility arrow  -- |Make two event streams into one. -- Actually `gather` is more general and convenient;--- +-- -- @... \<- tee -\< (e1, e2)@--- +-- -- is equivalent to--- +-- -- @... \<- gather -\< [Left \<$\> e1, Right \<$\> e2]@--- +-- tee ::     ArrowApply a => ProcessA a (Event b1, Event b2) (Event (Either b1 b2)) tee = proc (e1, e2) -> gather -< [Left <$> e1, Right <$> e2]   -sample ::-    ArrowApply a =>-    ProcessA a (Event b1, Event b2) [b1]-{--sample = join >>> construct (go id) >>> hold []-  where-    go l = -      do-        (evx, evy) <- await `catch` return (NoEvent, End)-        let l2 = evMaybe l (\x -> l . (x:)) evx-        if isEnd evy-          then-          do-            yield $ l2 []-            stop-          else-            return ()-        evMaybe (go l2) (\_ -> yield (l2 []) >> go id) evy--}-sample = undefined- -- |Make multiple event channels into one. -- If simultaneous events are given, lefter one is emitted earlier. gather ::@@ -159,35 +270,17 @@   where     singleton x = x NonEmpty.:| [] --- | Provides a source event stream.--- A dummy input event stream is needed.---   --- @---   run af [...]--- @---   --- is equivalent to------ @---   run (source [...] >>> af) (repeat ())--- @-source ::-    (ArrowApply a, Fd.Foldable f) =>-    f c -> ProcessA a (Event b) (Event c)-source l = construct $ Fd.mapM_ yd l-  where-    yd x = await >> yield x  -- |Given an array-valued event and emit it's values as inidvidual events. fork ::     (ArrowApply a, Fd.Foldable f) =>     ProcessA a (Event (f b)) (Event b) -fork = repeatedly $ +fork = repeatedly $     await >>= Fd.mapM_ yield  -- |Executes an action once per an input event is provided.-anytime :: +anytime ::     ArrowApply a =>     a b c ->     ProcessA a (Event b) (Event c)@@ -210,7 +303,7 @@     if b then yield x else return ()  -echo :: +echo ::     ArrowApply a =>     ProcessA a (Event b) (Event b) @@ -241,4 +334,4 @@   where     swap :: (a, b) -> (b, a)     swap ~(a, b) = (b, a)-    +

@@ -10,10 +10,10 @@     Main where -import Prelude hiding (filter) import Data.Maybe (fromMaybe)-import Control.Arrow.Machine as P-import Control.Applicative ((<$>), (<*>), (<$))+import qualified Control.Arrow.Machine as P+import Control.Arrow.Machine hiding (filter, source)+import Control.Applicative import qualified Control.Category as Cat import Control.Arrow import Control.Monad.State@@ -30,8 +30,8 @@   -main = hspec $ -  do +main = hspec $+  do     basics     rules     loops@@ -39,6 +39,7 @@     plans     utility     switches+    source     execution     loopUtil @@ -62,7 +63,7 @@          let             -- 入力1度につき同じ値を2回出力する-            doubler = repeatedly $ +            doubler = repeatedly $                       do {x <- await; yield x; yield x}             -- 入力値をStateのリストの先頭にPushする副作用を行い、同じ値を出力する             pusher = repeatedlyT (Kleisli . const) $@@ -96,7 +97,7 @@         it "never spoils any FEED" $           let               counter = construct $ counterDo 1-              counterDo n = +              counterDo n =                 do                   x <- await                   yield $ n * 100 + x@@ -110,7 +111,7 @@               split2' = fmap fst &&& fmap snd               gen = arr (fmap $ \x -> [x, x]) >>> fork >>> arr split2'               r1 = runKI (run (gen >>> arr fst)) (l::[(Int, [Int])])-              r2 = runKI (run (gen >>> second (fork >>> echo) >>> arr fst)) +              r2 = runKI (run (gen >>> second (fork >>> echo) >>> arr fst))                    (l::[(Int, [Int])])             in               r1 == r2@@ -119,7 +120,7 @@ rules =   do     describe "ProcessA as Category" $-      do        +      do         prop "has asocciative composition" $ \fx gx hx cond ->           let               f = mkProc fx@@ -138,7 +139,7 @@               (f >>> g) `equiv` (f >>> Cat.id >>> g)      describe "ProcessA as Arrow" $-      do        +      do         it "can be made from pure function(arr)" $           do             (run . arr . fmap $ (+ 2)) [1, 2, 3]@@ -156,7 +157,7 @@           do             pendingWith "to correct" {--            let +            let                 myProc2 = repeatedlyT (Kleisli . const) $                   do                     x <- await@@ -171,10 +172,10 @@                  (result, state) =                     stateProc (arr de >>> first myProc2 >>> arr en) l-                                  -            (result >>= maybe mzero return . fst) ++            (result >>= maybe mzero return . fst)                 `shouldBe` [1,2,2,3,3,3]-            (result >>= maybe mzero return . snd) +            (result >>= maybe mzero return . snd)                 `shouldBe` [1,2,3]             state `shouldBe` [1,2,3] -}@@ -191,7 +192,7 @@           let               f = first $ mkProc fx               g = second (arr $ fmap (+2))-              +               equiv = mkEquivTest2 cond             in               (f >>> g) `equiv` (g >>> f)@@ -253,7 +254,7 @@                     rec r <- dHold True -< False <$ ev2                         ev2 <- fork -< [(), ()] <$ ev                     returnA -< r <$ ev-            run pa [1, 2, 3] `shouldBe` [True, True, True] +            run pa [1, 2, 3] `shouldBe` [True, True, True]       describe "Rules for ArrowLoop" $@@ -292,8 +293,8 @@                 aj1 = arr Right                 aj2 = arr $ either id id                 l = [1]-                r1 = stateProc -                       (aj1 >>> left af >>> aj2) +                r1 = stateProc+                       (aj1 >>> left af >>> aj2)                        l               in                 r1 `shouldBe` ([1],[])@@ -302,7 +303,7 @@             let                 f = mkProc fx                 g = mkProc gx-                +                 equiv = mkEquivTest cond                     ::(MyTestT (Either (Event Int) (Event Int))                                (Either (Event Int) (Event Int)))@@ -312,7 +313,7 @@  plans = describe "Plan" $   do-    let pl = +    let pl =           do             x <- await             yield x@@ -324,7 +325,7 @@      it "can be constructed into ProcessA" $       do-        let +        let             result = run (construct pl) l         result `shouldBe` [2, 3, 5, 6] @@ -364,7 +365,7 @@       do         it "acts like fold." $           do-            let +            let                 pa = proc evx ->                   do                     val <- accum 0 -< (+1) <$ evx@@ -376,7 +377,7 @@       do         it "fires only once at the end of a stream." $           do-            let +            let                 pa = proc evx ->                   do                     x <- hold 0 -< evx@@ -391,7 +392,7 @@             let                 pa = proc x ->                   do-                    r1 <- filter $ arr (\x -> x `mod` 3 == 0) -< x+                    r1 <- P.filter $ arr (\x -> x `mod` 3 == 0) -< x                     r2 <- stopped -< x::Event Int                     r3 <- returnA -< r2                     fin <- gather -< [r1, r2, r3]@@ -399,16 +400,16 @@                     end <- onEnd -< fin                     returnA -< val <$ end             run pa [1, 2, 3, 4, 5] `shouldBe` ([3]::[Int])-                     + switches =   do     describe "switch" $       do         it "switches once" $           do-            let -                before = proc evx -> +            let+                before = proc evx ->                   do                     ch <- P.filter (arr $ (\x -> x `mod` 2 == 0)) -< evx                     returnA -< (noEvent, ch)@@ -448,12 +449,65 @@              ret `shouldBe` [7, 2, 6, 18, 21]             retD `shouldBe` [7, 3, 6, 12, 21]+    describe "kSwitch" $+      do+        it "switches spontaneously" $+          do+            let+                oneshot x = pure () >>> blockingSource [x]+                theArrow sw = sw (oneshot False) (arr snd) $ \_ _ -> oneshot True+            run (theArrow kSwitch) [] `shouldBe` [True]+            run (theArrow dkSwitch) [] `shouldBe` [False, True] +source =+  do+    describe "source" $+      do+        it "provides interleaved source stream" $+          do+            let+                pa = proc cl ->+                  do+                    s1 <- P.source [1, 2, 3] -< cl+                    s2 <- P.source [4, 5, 6] -< cl+                    P.gather -< [s1, s2]+            P.run pa (repeat ()) `shouldBe` [1, 4, 2, 5, 3, 6]+    describe "blockingSource" $+      do+        it "provides blocking source stream" $+          do+            let+                pa = proc _ ->+                  do+                    s1 <- P.blockingSource [1, 2, 3] -< ()+                    s2 <- P.blockingSource [4, 5, 6] -< ()+                    P.gather -< [s1, s2]+            P.run pa (repeat ()) `shouldBe` [4, 5, 6, 1, 2, 3] +    describe "source and blockingSource" $+      do+        prop "[interleave blockingSource = source]" $ \l cond ->+            let+                _ = l::[Int]+                equiv = mkEquivTest cond+                    ::(MyTestT (Event Int) (Event Int))+              in+                P.source l `equiv` P.interleave (P.blockingSource l)++        prop "[blocking source = blockingSource]" $ \l cond ->+            let+                _ = l::[Int]+                equiv = mkEquivTest cond+                    ::(MyTestT (Event Int) (Event Int))+              in+                (pure () >>> P.blockingSource l)+                    `equiv` (pure () >>> P.blocking (P.source l))++ execution = describe "Execution of ProcessA" $     do       let-          pl = +          pl =             do               x <- await               yield x@@ -481,25 +535,25 @@           yields ret `shouldBe` ([]::[Int])           hasStopped ret `shouldBe` True -      it "supports step execution (2)" $ +      it "supports step execution (2)" $           pendingWith "Correct stop handling" {-       prop "supports step execution (2)" $ \p l ->           let               pa = mkProc p-              all pc (x:xs) ys = +              all pc (x:xs) ys =                 do                   (r, cont) <- runKleisli (stepRun pc) x                   all cont (if hasStopped r then [] else xs) (ys ++ yields r)               all pc [] ys = runKleisli (run pc) [] >>= return . (ys++)             in               runState (all pa (l::[Int]) []) [] == stateProc pa l--}          +-}        it "supports yield-driven step" $         do           let-              init = construct $ +              init = construct $                 do                   yield (-1)                   x <- await