machinecell 2.0.1 → 2.1.0
raw patch · 10 files changed
+679/−567 lines, 10 files
Files
- CHANGELOG.md +12/−0
- README.md +86/−4
- machinecell.cabal +4/−2
- src/Control/Arrow/Machine.hs +8/−15
- src/Control/Arrow/Machine/Misc/Pump.hs +2/−3
- src/Control/Arrow/Machine/Types.hs +10/−5
- src/Control/Arrow/Machine/Utils.hs +15/−7
- test/LoopUtil.hs +17/−0
- test/RandomProc.hs +1/−1
- test/spec.hs +524/−530
CHANGELOG.md view
@@ -1,4 +1,16 @@ +2.1.0+-----------+* Added `dHold`, `dAccum`.+* Deprecated `cycleDelay`.+* Fixed `muted`.+* Slightly changed the ArrowLoop instance declaration.+ * Right tightening rule will be preserved.+ * For IO processes, "Indefinite access to MVar" errors, which used to occur in some+ situations in old versions, will be suppressed.+ * This will not change any existing code unless it loops back+ any Event-type signal.+ 2.0.1 ------------ * Support free-4.12
README.md view
@@ -5,9 +5,91 @@ Description ----------------Coroutine-style stream processing library with support of arrow combinatins.-AFRP-like utilities are also available. -Usage+As other iteratee or pipe libraries, machinecell abstracts general iteration processes.++Here is an example that is a simple iteration over a list.++```+>>> run (evMap (+1)) [1, 2, 3]+[2, 3, 4]+```++In above statement, "`evMap` (+1)" has a type "ProcessA (-\>) (Event Int) (Event Int)",+which denotes "A stream transducer that takes a series of Int as input,+gives a series of Int as output, run on base arrow (-\>)."+++In addition to this simple iteration, machinecell has following features.++* Side effects+* Composite pipelines+* Arrow compositions+* Behaviours and switches++See [Control.Arrow.Machine](https://hackage.haskell.org/package/machinecell/docs/Control-Arrow-Machine.html) documentation.++++Comparison to other libraries. ----------------See example of test/Main.hs++Some part of machinecell is similar to other stream transducer+libraries, namely pipes, conduit, or machines. machinecell can be+seen as a restricted variation of them to one-directional. But+additionally, machinecell supports arrow compositions.+Bidirectional communications can be taken place by ArrowLoop+feature.++Rather, there are several other arrowised stream transducer+libraries. streamproc shares the most concept to machinecell. But+actually it has a problem described later in this post. Machinecell+can be said as "Streamproc done right."++auto is a brand-new arrowised stream transducer library. Compared+to it, machinecell's advantage is await/yield coroutines, while+auto's one is serialization.++++Motivation and background+---------------++"Generalizing monads to arrows," The original paper of arrow calculation+mentions a kind of stream transducer, which later implemented as streamproc.++http://www.cse.chalmers.se/~rjmh/Papers/arrows.pdf+++And other people propose instance declarations of Arrow class for several existing stream processors.++http://stackoverflow.com/questions/19758744/haskell-splitting-pipes-broadcast-without-using-spawn++https://www.fpcomplete.com/school/to-infinity-and-beyond/pick-of-the-week/coroutines-for-streaming/part-4-category-and-arrow+++But actually, there is a problem argued in this post.++https://mail.haskell.org/pipermail/haskell-cafe/2010-January/072193.html+++The core problem is, while arrow uses tuples as parallel data+stream, they cannot represent a composite streams if they carry+different numbers of data in parallel.++To solve this problem, some arrow libraries restrict transducers to+one-to-one data transformation. Yampa and netwire does so, as+mentioned in above post. And auto also takes this approach.++Machinecell's approach is different, but simple too. The key idea+is wrapping all types of data stream into a maybe-like type. Then+even tuples can represent different numbers of data, by inserting+appropreate number of 'Nothing's.++Furthermore, I identified the maybe-like type as the 'Event' type,+which appears in Yampa and netwire. Then I successively implemented+several arrows of Yampa and netwire.++API names come from stream libraries are named after machines',+while ones from FRPs are after Yampa's. Now, machinecell may be+seen as a hybrid of machines and Yampa.
machinecell.cabal view
@@ -1,5 +1,5 @@ name: machinecell-version: 2.0.1+version: 2.1.0 synopsis: Arrow based stream transducers license: BSD3 license-file: LICENSE@@ -19,6 +19,8 @@ . Arrow combinatins are supported and can be used with the arrow notation. AFRP-like utilities are also available.+ .+ A quick introduction is available in the Control.Arrow.Machine documentation. library exposed-modules:@@ -51,4 +53,4 @@ source-repository this type: git location: https://github.com/as-capabl/machinecell.git- tag: release-2.0.1+ tag: release-2.1.0
src/Control/Arrow/Machine.hs view
@@ -99,7 +99,7 @@ -- lift $ putStrLn x -- @ ----- >>> runKleisli (run_ $ source \>\>\> pipe \>\>\> sink) (repeat ())+-- >>> runKleisli (run_ $ source >>> pipe >>> sink) (repeat ()) -- -- The above code reads two lines from stdin, puts a concatenated line to stdout and finishes. --@@ -248,13 +248,13 @@ -- Although `ProcessA` is an instance of `ArrowLoop`, -- to send values to upstream, there is a little difficulties. -- --- In example below, result is [0, 1, 1, 1], not [0, 1, 2, 3].+-- In example below, result is [0, 0, 0, 0], not [1, 2, 3, 4]. -- -- @ -- f = proc x -\> -- do -- rec--- b \<- dHold 0 -\< y+-- b \<- hold 0 -\< y -- y \<- fork -\< (\xx -\> [xx, xx+1, xx+2, xx+3]) \<$\> x -- returnA -\< b \<$ y --@@ -262,20 +262,13 @@ -- @ -- -- >>> run f [1]--- [0, 1, 1, 1]------ This is because of machinecell's execution strategy.--- It's much similar to Prolog's backtracking stategy.--- At the time backtracking reaches `fork` three values are--- found and backtracking go and back three times between fork and returnA,--- but not reaches to dHold until all outputs are done.------ In general, `Event` values should not be refered at upstream.+-- [0, 0, 0, 0] ----- Rather, they should be encoded to behaviours and send to upstream in--- rec statement and delayed by `cycleDelay`.+-- In general, `Event` values refered at upstream in rec statements are+-- almost always `NoEvent`s. ----- Another way to send values to upstream is `encloseState`.+-- 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 --
src/Control/Arrow/Machine/Misc/Pump.hs view
@@ -47,7 +47,7 @@ do cl2 <- oneMore -< clock append <- returnA -< (\x y -> y `mappend` Endo (x:)) <$> ev- e <- P.accum (Endo id) <<< P.gather -< [ (const $ Endo id) <$ cl2, append ]+ e <- P.dAccum (Endo id) <<< P.gather -< [ (const $ Endo id) <$ cl2, append ] returnA -< Duct e outlet ::@@ -56,6 +56,5 @@ outlet = proc (~(Duct dct), clock) -> do cl2 <- oneMore -< clock- dct' <- P.cycleDelay -< dct- P.fork -< appEndo dct' [] <$ cl2+ P.fork -< appEndo dct [] <$ cl2
src/Control/Arrow/Machine/Types.hs view
@@ -333,12 +333,16 @@ instance (ArrowApply a, ArrowLoop a) => ArrowLoop (ProcessA a) where- loop = fitEx (\f -> loop (lp f))+ loop pa = ProcessA $ proc (ph, x) ->+ do+ (_, d) <- loop suspended -< x+ (ph', (y, _), pa') <- step pa -< (ph, (x, d))+ returnA -< (ph', y, loop pa') where- lp f = proc ((p, x), d) ->+ suspended = proc (x, d) -> do- (q, (y, d')) <- f -< (p, (x, d))- returnA -< ((q, y), d')+ (_, (y, d'), _) <- step pa -< (Suspend, (x, d))+ returnA -< ((y, d'), d') instance@@ -477,7 +481,8 @@ (ArrowApply a, Occasional' b, Occasional c) => ProcessA a b c muted = proc x -> do- rSwitch (arr $ const noEvent) -< ((), stopped <$ collapse x)+ ed <- repeatedly $ await `catchP` yield () -< collapse x+ rSwitch (arr $ const noEvent) -< ((), stopped <$ ed)
src/Control/Arrow/Machine/Utils.hs view
@@ -10,7 +10,9 @@ ( -- * AFRP-like utilities hold,+ dHold, accum,+ dAccum, edge, passRecent, withRecent,@@ -73,23 +75,26 @@ hold :: ArrowApply a => b -> ProcessA a (Event b) b-{--hold old = ProcessA $ proc (ph, evx) ->- do- let new = fromEvent old evx- returnA -< (ph `mappend` Suspend, new, hold new)--} hold old = proc evx -> do rSwitch (pure old) -< ((), pure <$> evx) +dHold :: + ArrowApply a => b -> ProcessA a (Event b) b+dHold old = proc evx -> + do+ drSwitch (pure old) -< ((), pure <$> evx)+ accum :: ArrowApply a => b -> ProcessA a (Event (b->b)) b accum x = switch (pure x &&& arr (($x)<$>)) accum' where accum' y = dSwitch (pure y &&& Cat.id) (const (accum y))- +dAccum ::+ ArrowApply a => b -> ProcessA a (Event (b->b)) b+dAccum x = dSwitch (pure x &&& arr (($x)<$>)) dAccum+ edge :: (ArrowApply a, Eq b) => ProcessA a b (Event b)@@ -292,6 +297,8 @@ -- |Observe a previous value of a signal. -- Tipically used with rec statement.++{-# DEPRECATED cycleDelay "Simply use `dHold` or `dAccum`" #-} cycleDelay :: ArrowApply a => ProcessA a b b cycleDelay =@@ -312,4 +319,5 @@ appStore (Just x) = (proc _ -> store -< (Just x, Nothing), ()) appStore _ = (Cat.id, ())+
test/LoopUtil.hs view
@@ -18,6 +18,23 @@ loopUtil = do+ describe "loop" $+ do+ it "is possible that value by `dHold` or `dAccum` can refer at upstream." $+ do+ let + pa :: ProcessA (Kleisli IO) (Event Int) (Event Int)+ pa = proc evx ->+ do+ rec+ anytime (Kleisli print) -< y <$ evx+ anytime (Kleisli putStr) -< "" <$ evx -- side effect+ evx2 <- doubler -< evx+ y <- P.dAccum 0 -< (+) <$> evx2+ returnA -< y <$ evx+ ret <- liftIO $ runKleisli (P.run pa) [1, 2, 3]+ ret `shouldBe` [0, 1+1, 1+1+2+2]+ describe "cycleDelay" $ do it "can refer a recent value at downstream." $
test/RandomProc.hs view
@@ -173,7 +173,7 @@ input = proc evx -> do -- 一個前の値で分岐してみる- b <- cycleDelay <<< hold True -< + b <- dHold True -< (\x -> x `mod` 2 == 0) <$> evx if b
test/spec.hs view
@@ -1,530 +1,524 @@-{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE Arrows #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE TypeSynonymInstances #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE NoMonomorphismRestriction #-} -{-# LANGUAGE FlexibleContexts #-} - -module - Main -where - -import Prelude hiding (filter) -import Data.Maybe (fromMaybe) -import Control.Arrow.Machine as P -import Control.Applicative ((<$>), (<*>), (<$)) -import qualified Control.Category as Cat -import Control.Arrow -import Control.Monad.State -import Control.Monad -import Control.Monad.Trans -import Control.Monad.Identity (Identity, runIdentity) -import Debug.Trace -import Test.Hspec -import Test.Hspec.QuickCheck (prop) -import Test.QuickCheck (Arbitrary, arbitrary, oneof, frequency, sized) -import RandomProc -import LoopUtil -runKI a x = runIdentity (runKleisli a x) - - - -main = hspec $ - do - basics - rules - loops - choice - plans - utility - switches - execution - loopUtil - - -basics = - do - describe "ProcessA" $ - do - it "is stream transducer." $ - do - let - process = repeatedly $ - do - x <- await - yield x - yield (x + 1) - - resultA = run process [1,2,4] - - resultA `shouldBe` [1, 2, 2, 3, 4, 5] - - let - -- 入力1度につき同じ値を2回出力する - doubler = repeatedly $ - do {x <- await; yield x; yield x} - -- 入力値をStateのリストの先頭にPushする副作用を行い、同じ値を出力する - pusher = repeatedlyT (Kleisli . const) $ - do {x <- await; lift $ modify (x:); yield x} - - it "has stop state" $ - let - -- 一度だけ入力をそのまま出力し、すぐに停止する - onlyOnce = construct $ await >>= yield - - x = stateProc (doubler >>> pusher >>> onlyOnce) [3, 3] - in - -- 最後尾のMachineが停止した時点で処理を停止するが、 - -- 既にa2が出力した値の副作用は処理する - x `shouldBe` ([3], [3, 3]) - - it "has side-effect" $ - let - incl = arr $ fmap (+1) - - -- doublerで信号が2つに分岐する。 - -- このとき、副作用は1つ目の信号について末尾まで - -- -> 二つ目の信号について分岐点から末尾まで ... - -- の順で処理される。 - a = pusher >>> doubler >>> incl >>> pusher >>> incl >>> pusher - - x = stateProc a [1000] - in - x `shouldBe` ([1002, 1002], reverse [1000,1001,1002,1001,1002]) - - it "never spoils any FEED" $ - let - counter = construct $ counterDo 1 - counterDo n = - do - x <- await - yield $ n * 100 + x - counterDo (n+1) - x = stateProc (doubler >>> doubler >>> counter) [1,2] - in - fst x `shouldBe` [101, 201, 301, 401, 502, 602, 702, 802] - - prop "each path can have independent number of events" $ \l -> - let - 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)) - (l::[(Int, [Int])]) - in - r1 == r2 - - -rules = - do - describe "ProcessA as Category" $ - do - prop "has asocciative composition" $ \fx gx hx cond -> - let - f = mkProc fx - g = mkProc gx - h = mkProc hx - equiv = mkEquivTest cond - in - ((f >>> g) >>> h) `equiv` (f >>> (g >>> h)) - - prop "has identity" $ \fx gx cond -> - let - f = mkProc fx - g = mkProc gx - equiv = mkEquivTest cond - in - (f >>> g) `equiv` (f >>> Cat.id >>> g) - - describe "ProcessA as Arrow" $ - do - it "can be made from pure function(arr)" $ - do - (run . arr . fmap $ (+ 2)) [1, 2, 3] - `shouldBe` [3, 4, 5] - - prop "arr id is identity" $ \fx gx cond -> - let - f = mkProc fx - g = mkProc gx - equiv = mkEquivTest cond - in - (f >>> g) `equiv` (f >>> arr id >>> g) - - it "can be parallelized" $ - do - pendingWith "to correct" -{- - let - myProc2 = repeatedlyT (Kleisli . const) $ - do - x <- await - lift $ modify (++ [x]) - yield `mapM` (take x $ repeat x) - - toN = evMaybe Nothing Just - en (ex, ey) = Event (toN ex, toN ey) - de evxy = (fst <$> evxy, snd <$> evxy) - - l = map (\x->(x,x)) [1,2,3] - - (result, state) = - stateProc (arr de >>> first myProc2 >>> arr en) l - - (result >>= maybe mzero return . fst) - `shouldBe` [1,2,2,3,3,3] - (result >>= maybe mzero return . snd) - `shouldBe` [1,2,3] - state `shouldBe` [1,2,3] --} - - prop "first and composition." $ \fx gx cond -> - let - f = mkProc fx - g = mkProc gx - equiv = mkEquivTest2 cond - in - (first (f >>> g)) `equiv` (first f >>> first g) - - prop "first-second commutes" $ \fx cond -> - let - f = first $ mkProc fx - g = second (arr $ fmap (+2)) - - equiv = mkEquivTest2 cond - in - (f >>> g) `equiv` (g >>> f) - - prop "first-fst commutes" $ \fx cond -> - let - f = mkProc fx - equiv = mkEquivTest cond - ::(MyTestT (Event Int, Event Int) (Event Int)) - in - (first f >>> arr fst) `equiv` (arr fst >>> f) - - prop "assoc relation" $ \fx cond -> - let - f = mkProc fx - assoc ((a,b),c) = (a,(b,c)) - - equiv = mkEquivTest cond - ::(MyTestT ((Event Int, Event Int), Event Int) - (Event Int, (Event Int, Event Int))) - in - (first (first f) >>> arr assoc) `equiv` (arr assoc >>> first f) - -loops = - do - describe "ProcessA as ArrowLoop" $ - do - it "can be used with rec statement(pure)" $ - let - a = proc ev -> - do - x <- hold 0 -< ev - rec l <- returnA -< x:l - returnA -< l <$ ev - result = fst $ stateProc a [2, 5] - in - take 3 (result!!1) `shouldBe` [5, 5, 5] - -{- - it "can be used with rec statement(macninery)" $ - let - mc = anytime Cat.id - a = proc x -> - do - rec l <- mc -< (:l') <$> x - l' <- returnA -< fromEvent [] l - returnA -< l - result = fst $ stateProc a [2, 5] - in - take 3 (result!!1) `shouldBe` [5, 5, 5] - - it "the last value is valid." $ - do - let - mc = repeatedly $ - do - x <- await - yield x - yield (x*2) - pa = proc x -> - do - rec y <- mc -< (+z) <$> x - z <- hold 0 <<< delay -< y - returnA -< y - run pa [1, 10] `shouldBe` [1, 2, 12, 24] --} - - describe "Rules for ArrowLoop" $ - do - let - fixcore f y = if y `mod` 5 == 0 then y else y + f (y-1) - pure (evx, f) = (f <$> evx, fixcore f) - apure = arr pure - - prop "left tightening" $ \fx cond -> - let - f = mkProc fx - - equiv = mkEquivTest cond - in - (loop (first f >>> apure)) `equiv` (f >>> loop apure) - - it "right tigntening" - pending -{- - prop "right tightening" $ \fx cond -> - let - f = mkProc fx - - equiv = mkEquivTest cond - in - (loop (apure >>> first f)) `equiv` (loop apure >>> f) --} - -choice = - do - describe "ProcessA as ArrowChoice" $ - do - it "temp1" $ - do - let - af = mkProc $ PgStop - ag = mkProc $ PgOdd PgNop - aj1 = arr Right - aj2 = arr $ either id id - l = [1] - r1 = stateProc - (aj1 >>> left af >>> aj2) - l - in - r1 `shouldBe` ([1],[]) - - prop "left (f >>> g) = left f >>> left g" $ \fx gx cond -> - let - f = mkProc fx - g = mkProc gx - - equiv = mkEquivTest cond - ::(MyTestT (Either (Event Int) (Event Int)) - (Either (Event Int) (Event Int))) - in - (left (f >>> g)) `equiv` (left f >>> left g) - - -plans = describe "Plan" $ - do - let pl = - do - x <- await - yield x - yield (x+1) - x <- await - yield x - yield (x+1) - l = [2, 5, 10, 20, 100] - - it "can be constructed into ProcessA" $ - do - let - result = run (construct pl) l - result `shouldBe` [2, 3, 5, 6] - - it "can be repeatedly constructed into ProcessA" $ - do - let - result = run (repeatedly pl) l - result `shouldBe` [2, 3, 5, 6, 10, 11, 20, 21, 100, 101] - - it "can handle the end with catchP." $ - do - let - plCatch = - do - x <- await `catchP` (yield 1 >> stop) - yield x - y <- (yield 2 >> await >> yield 3 >> await) `catchP` (yield 4 >> return 5) - yield y - y <- (await >>= yield >> stop) `catchP` (yield 6 >> return 7) - yield y - run (construct plCatch) [] `shouldBe` [1] - run (construct plCatch) [100] `shouldBe` [100, 2, 4, 5, 6, 7] - run (construct plCatch) [100, 200] `shouldBe` [100, 2, 3, 4, 5, 6, 7] - run (construct plCatch) [100, 200, 300] `shouldBe` [100, 2, 3, 300, 6, 7] - run (construct plCatch) [100, 200, 300, 400] `shouldBe` [100, 2, 3, 300, 400, 6, 7] - -utility = - do - describe "edge" $ - do - it "detects edges of input behaviour" $ - do - run (hold 0 >>> edge) [1, 1, 2, 2, 2, 3] `shouldBe` [0, 1, 2, 3] - run (hold 0 >>> edge) [0, 1, 1, 2, 2, 2, 3] `shouldBe` [0, 1, 2, 3] - - describe "accum" $ - do - it "acts like fold." $ - do - let - pa = proc evx -> - do - val <- accum 0 -< (+1) <$ evx - returnA -< val <$ evx - - run pa (replicate 10 ()) `shouldBe` [1..10] - - describe "onEnd" $ - do - it "fires only once at the end of a stream." $ - do - let - pa = proc evx -> - do - x <- hold 0 -< evx - ed <- onEnd -< evx - returnA -< x <$ ed - run pa [1..4] `shouldBe` [4] - - describe "gather" $ - do - it "correctly handles the end" $ - do - let - pa = proc x -> - do - r1 <- filter $ arr (\x -> x `mod` 3 == 0) -< x - r2 <- stopped -< x::Event Int - r3 <- returnA -< r2 - fin <- gather -< [r1, r2, r3] - val <- hold 0 -< r1 - 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 -> - do - ch <- P.filter (arr $ (\x -> x `mod` 2 == 0)) -< evx - returnA -< (noEvent, ch) - - after t = proc evx -> returnA -< (t*) <$> evx - - l = [1,3,4,1,3,2] - - -- 最初に偶数が与えられるまでは、入力を無視(NoEvent)し、 - -- それ以降は最初に与えられた偶数 * 入力値を返す - ret = run (switch before after) l - - -- dが付くと次回からの切り替えとなる - retD = run (dSwitch before after) l - - ret `shouldBe` [16, 4, 12, 8] - retD `shouldBe` [4, 12, 8] - - describe "rSwitch" $ - do - it "switches any times" $ - do - let - theArrow sw = proc evtp -> - do - evx <- P.fork -< fst <$> evtp - evarr <- P.fork -< snd <$> evtp - sw (arr $ fmap (+2)) -< (evx, evarr) - - l = [(Just 5, Nothing), - (Just 1, Just (arr $ fmap (*2))), - (Just 3, Nothing), - (Just 6, Just (arr $ fmap (*3))), - (Just 7, Nothing)] - ret = run (theArrow rSwitch) l - retD = run (theArrow drSwitch) l - - ret `shouldBe` [7, 2, 6, 18, 21] - retD `shouldBe` [7, 3, 6, 12, 21] - - -execution = describe "Execution of ProcessA" $ - do - let - pl = - do - x <- await - yield x - yield (x+1) - x <- await - yield x - yield (x+1) - yield (x+5) - init = construct pl - - it "supports step execution" $ - do - let - (ret, now) = stepRun init 1 - yields ret `shouldBe` [1, 2] - hasStopped ret `shouldBe` False - - let - (ret, now2) = stepRun now 1 - yields ret `shouldBe` [1, 2, 6] - hasStopped ret `shouldBe` True - - let - (ret, _) = stepRun now2 1 - yields ret `shouldBe` ([]::[Int]) - hasStopped ret `shouldBe` True - - 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 = - 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 $ - do - yield (-1) - x <- await - mapM yield (iterate (+1) x) -- infinite - - (ret, now) = stepYield init 5 - yields ret `shouldBe` Just (-1) - hasConsumed ret `shouldBe` False - hasStopped ret `shouldBe` False - - let - (ret, now2) = stepYield now 10 - yields ret `shouldBe` Just 10 - hasConsumed ret `shouldBe` True - hasStopped ret `shouldBe` False - - let - (ret, now3) = stepYield now2 10 - yields ret `shouldBe` Just 11 - hasConsumed ret `shouldBe` False - hasStopped ret `shouldBe` False - +{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE Arrows #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE FlexibleContexts #-}++module+ Main+where++import Prelude hiding (filter)+import Data.Maybe (fromMaybe)+import Control.Arrow.Machine as P+import Control.Applicative ((<$>), (<*>), (<$))+import qualified Control.Category as Cat+import Control.Arrow+import Control.Monad.State+import Control.Monad+import Control.Monad.Trans+import Control.Monad.Identity (Identity, runIdentity)+import Debug.Trace+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck (Arbitrary, arbitrary, oneof, frequency, sized)+import RandomProc+import LoopUtil+runKI a x = runIdentity (runKleisli a x)++++main = hspec $ + do + basics+ rules+ loops+ choice+ plans+ utility+ switches+ execution+ loopUtil+++basics =+ do+ describe "ProcessA" $+ do+ it "is stream transducer." $+ do+ let+ process = repeatedly $+ do+ x <- await+ yield x+ yield (x + 1)++ resultA = run process [1,2,4]++ resultA `shouldBe` [1, 2, 2, 3, 4, 5]++ let+ -- 入力1度につき同じ値を2回出力する+ doubler = repeatedly $ + do {x <- await; yield x; yield x}+ -- 入力値をStateのリストの先頭にPushする副作用を行い、同じ値を出力する+ pusher = repeatedlyT (Kleisli . const) $+ do {x <- await; lift $ modify (x:); yield x}++ it "has stop state" $+ let+ -- 一度だけ入力をそのまま出力し、すぐに停止する+ onlyOnce = construct $ await >>= yield++ x = stateProc (doubler >>> pusher >>> onlyOnce) [3, 3]+ in+ -- 最後尾のMachineが停止した時点で処理を停止するが、+ -- 既にa2が出力した値の副作用は処理する+ x `shouldBe` ([3], [3, 3])++ it "has side-effect" $+ let+ incl = arr $ fmap (+1)++ -- doublerで信号が2つに分岐する。+ -- このとき、副作用は1つ目の信号について末尾まで+ -- -> 二つ目の信号について分岐点から末尾まで ...+ -- の順で処理される。+ a = pusher >>> doubler >>> incl >>> pusher >>> incl >>> pusher++ x = stateProc a [1000]+ in+ x `shouldBe` ([1002, 1002], reverse [1000,1001,1002,1001,1002])++ it "never spoils any FEED" $+ let+ counter = construct $ counterDo 1+ counterDo n = + do+ x <- await+ yield $ n * 100 + x+ counterDo (n+1)+ x = stateProc (doubler >>> doubler >>> counter) [1,2]+ in+ fst x `shouldBe` [101, 201, 301, 401, 502, 602, 702, 802]++ prop "each path can have independent number of events" $ \l ->+ let+ 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)) + (l::[(Int, [Int])])+ in+ r1 == r2+++rules =+ do+ describe "ProcessA as Category" $+ do + prop "has asocciative composition" $ \fx gx hx cond ->+ let+ f = mkProc fx+ g = mkProc gx+ h = mkProc hx+ equiv = mkEquivTest cond+ in+ ((f >>> g) >>> h) `equiv` (f >>> (g >>> h))++ prop "has identity" $ \fx gx cond ->+ let+ f = mkProc fx+ g = mkProc gx+ equiv = mkEquivTest cond+ in+ (f >>> g) `equiv` (f >>> Cat.id >>> g)++ describe "ProcessA as Arrow" $+ do + it "can be made from pure function(arr)" $+ do+ (run . arr . fmap $ (+ 2)) [1, 2, 3]+ `shouldBe` [3, 4, 5]++ prop "arr id is identity" $ \fx gx cond ->+ let+ f = mkProc fx+ g = mkProc gx+ equiv = mkEquivTest cond+ in+ (f >>> g) `equiv` (f >>> arr id >>> g)++ it "can be parallelized" $+ do+ pendingWith "to correct"+{-+ let + myProc2 = repeatedlyT (Kleisli . const) $+ do+ x <- await+ lift $ modify (++ [x])+ yield `mapM` (take x $ repeat x)++ toN = evMaybe Nothing Just+ en (ex, ey) = Event (toN ex, toN ey)+ de evxy = (fst <$> evxy, snd <$> evxy)++ l = map (\x->(x,x)) [1,2,3]++ (result, state) =+ stateProc (arr de >>> first myProc2 >>> arr en) l+ + (result >>= maybe mzero return . fst) + `shouldBe` [1,2,2,3,3,3]+ (result >>= maybe mzero return . snd) + `shouldBe` [1,2,3]+ state `shouldBe` [1,2,3]+-}++ prop "first and composition." $ \fx gx cond ->+ let+ f = mkProc fx+ g = mkProc gx+ equiv = mkEquivTest2 cond+ in+ (first (f >>> g)) `equiv` (first f >>> first g)++ prop "first-second commutes" $ \fx cond ->+ let+ f = first $ mkProc fx+ g = second (arr $ fmap (+2))+ + equiv = mkEquivTest2 cond+ in+ (f >>> g) `equiv` (g >>> f)++ prop "first-fst commutes" $ \fx cond ->+ let+ f = mkProc fx+ equiv = mkEquivTest cond+ ::(MyTestT (Event Int, Event Int) (Event Int))+ in+ (first f >>> arr fst) `equiv` (arr fst >>> f)++ prop "assoc relation" $ \fx cond ->+ let+ f = mkProc fx+ assoc ((a,b),c) = (a,(b,c))++ equiv = mkEquivTest cond+ ::(MyTestT ((Event Int, Event Int), Event Int)+ (Event Int, (Event Int, Event Int)))+ in+ (first (first f) >>> arr assoc) `equiv` (arr assoc >>> first f)++loops =+ do+ describe "ProcessA as ArrowLoop" $+ do+ it "can be used with rec statement(pure)" $+ let+ a = proc ev ->+ do+ x <- hold 0 -< ev+ rec l <- returnA -< x:l+ returnA -< l <$ ev+ result = fst $ stateProc a [2, 5]+ in+ take 3 (result!!1) `shouldBe` [5, 5, 5]++ it "the last value is valid." $+ do+ let+ mc = repeatedly $+ do+ x <- await+ yield x+ yield (x*2)+ pa = proc x ->+ do+ rec y <- mc -< (+z) <$> x+ z <- dHold 0 -< y+ returnA -< y+ run pa [1, 10] `shouldBe` [1, 2, 12, 24]++ it "carries no events to upstream." $+ do+ let+ pa = proc ev ->+ do+ rec r <- dHold True -< False <$ ev2+ ev2 <- fork -< [(), ()] <$ ev+ returnA -< r <$ ev+ run pa [1, 2, 3] `shouldBe` [True, True, True] +++ describe "Rules for ArrowLoop" $+ do+ let+ fixcore f y = if y `mod` 5 == 0 then y else y + f (y-1)+ pure (evx, f) = (f <$> evx, fixcore f)+ apure = arr pure++ prop "left tightening" $ \fx cond ->+ let+ f = mkProc fx++ equiv = mkEquivTest cond+ in+ (loop (first f >>> apure)) `equiv` (f >>> loop apure)++ prop "right tightening" $ \fx cond ->+ let+ f = mkProc fx++ equiv = mkEquivTest cond+ in+ (loop (apure >>> first f)) `equiv` (loop apure >>> f)+++choice =+ do+ describe "ProcessA as ArrowChoice" $+ do+ it "temp1" $+ do+ let+ af = mkProc $ PgStop+ ag = mkProc $ PgOdd PgNop+ aj1 = arr Right+ aj2 = arr $ either id id+ l = [1]+ r1 = stateProc + (aj1 >>> left af >>> aj2) + l+ in+ r1 `shouldBe` ([1],[])++ prop "left (f >>> g) = left f >>> left g" $ \fx gx cond ->+ let+ f = mkProc fx+ g = mkProc gx+ + equiv = mkEquivTest cond+ ::(MyTestT (Either (Event Int) (Event Int))+ (Either (Event Int) (Event Int)))+ in+ (left (f >>> g)) `equiv` (left f >>> left g)+++plans = describe "Plan" $+ do+ let pl = + do+ x <- await+ yield x+ yield (x+1)+ x <- await+ yield x+ yield (x+1)+ l = [2, 5, 10, 20, 100]++ it "can be constructed into ProcessA" $+ do+ let + result = run (construct pl) l+ result `shouldBe` [2, 3, 5, 6]++ it "can be repeatedly constructed into ProcessA" $+ do+ let+ result = run (repeatedly pl) l+ result `shouldBe` [2, 3, 5, 6, 10, 11, 20, 21, 100, 101]++ it "can handle the end with catchP." $+ do+ let+ plCatch =+ do+ x <- await `catchP` (yield 1 >> stop)+ yield x+ y <- (yield 2 >> await >> yield 3 >> await) `catchP` (yield 4 >> return 5)+ yield y+ y <- (await >>= yield >> stop) `catchP` (yield 6 >> return 7)+ yield y+ run (construct plCatch) [] `shouldBe` [1]+ run (construct plCatch) [100] `shouldBe` [100, 2, 4, 5, 6, 7]+ run (construct plCatch) [100, 200] `shouldBe` [100, 2, 3, 4, 5, 6, 7]+ run (construct plCatch) [100, 200, 300] `shouldBe` [100, 2, 3, 300, 6, 7]+ run (construct plCatch) [100, 200, 300, 400] `shouldBe` [100, 2, 3, 300, 400, 6, 7]++utility =+ do+ describe "edge" $+ do+ it "detects edges of input behaviour" $+ do+ run (hold 0 >>> edge) [1, 1, 2, 2, 2, 3] `shouldBe` [0, 1, 2, 3]+ run (hold 0 >>> edge) [0, 1, 1, 2, 2, 2, 3] `shouldBe` [0, 1, 2, 3]++ describe "accum" $+ do+ it "acts like fold." $+ do+ let + pa = proc evx ->+ do+ val <- accum 0 -< (+1) <$ evx+ returnA -< val <$ evx++ run pa (replicate 10 ()) `shouldBe` [1..10]++ describe "onEnd" $+ do+ it "fires only once at the end of a stream." $+ do+ let + pa = proc evx ->+ do+ x <- hold 0 -< evx+ ed <- onEnd -< evx+ returnA -< x <$ ed+ run pa [1..4] `shouldBe` [4]++ describe "gather" $+ do+ it "correctly handles the end" $+ do+ let+ pa = proc x ->+ do+ r1 <- filter $ arr (\x -> x `mod` 3 == 0) -< x+ r2 <- stopped -< x::Event Int+ r3 <- returnA -< r2+ fin <- gather -< [r1, r2, r3]+ val <- hold 0 -< r1+ 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 -> + do+ ch <- P.filter (arr $ (\x -> x `mod` 2 == 0)) -< evx+ returnA -< (noEvent, ch)++ after t = proc evx -> returnA -< (t*) <$> evx++ l = [1,3,4,1,3,2]++ -- 最初に偶数が与えられるまでは、入力を無視(NoEvent)し、+ -- それ以降は最初に与えられた偶数 * 入力値を返す+ ret = run (switch before after) l++ -- dが付くと次回からの切り替えとなる+ retD = run (dSwitch before after) l++ ret `shouldBe` [16, 4, 12, 8]+ retD `shouldBe` [4, 12, 8]++ describe "rSwitch" $+ do+ it "switches any times" $+ do+ let+ theArrow sw = proc evtp ->+ do+ evx <- P.fork -< fst <$> evtp+ evarr <- P.fork -< snd <$> evtp+ sw (arr $ fmap (+2)) -< (evx, evarr)++ l = [(Just 5, Nothing),+ (Just 1, Just (arr $ fmap (*2))),+ (Just 3, Nothing),+ (Just 6, Just (arr $ fmap (*3))),+ (Just 7, Nothing)]+ ret = run (theArrow rSwitch) l+ retD = run (theArrow drSwitch) l++ ret `shouldBe` [7, 2, 6, 18, 21]+ retD `shouldBe` [7, 3, 6, 12, 21]+++execution = describe "Execution of ProcessA" $+ do+ let+ pl = + do+ x <- await+ yield x+ yield (x+1)+ x <- await+ yield x+ yield (x+1)+ yield (x+5)+ init = construct pl++ it "supports step execution" $+ do+ let+ (ret, now) = stepRun init 1+ yields ret `shouldBe` [1, 2]+ hasStopped ret `shouldBe` False++ let+ (ret, now2) = stepRun now 1+ yields ret `shouldBe` [1, 2, 6]+ hasStopped ret `shouldBe` True++ let+ (ret, _) = stepRun now2 1+ yields ret `shouldBe` ([]::[Int])+ hasStopped ret `shouldBe` True++ 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 = + 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 $ + do+ yield (-1)+ x <- await+ mapM yield (iterate (+1) x) -- infinite++ (ret, now) = stepYield init 5+ yields ret `shouldBe` Just (-1)+ hasConsumed ret `shouldBe` False+ hasStopped ret `shouldBe` False++ let+ (ret, now2) = stepYield now 10+ yields ret `shouldBe` Just 10+ hasConsumed ret `shouldBe` True+ hasStopped ret `shouldBe` False++ let+ (ret, now3) = stepYield now2 10+ yields ret `shouldBe` Just 11+ hasConsumed ret `shouldBe` False+ hasStopped ret `shouldBe` False+