@@ -1,11 +1,7 @@
1+{-# LANGUAGE DeriveGeneric #-}
12module Main where
23
34import Control.Concurrent (threadDelay)
4−import Control.Concurrent.MVar
5− ( newEmptyMVar
6− , tryPutMVar
7− , takeMVar
8− )
95import Control.Distributed.Process
106import Control.Distributed.Process.Management
117 ( MxEvent(..)
@@ -13,18 +9,27 @@ import Control.Distributed.Process.Management
139 , mxAgent
1410 , mxSink
1511 , mxReady
16− , mxLift
12+ , liftMX
1713 , mxGetLocal
1814 , mxSetLocal
1915 , mxNotify
16+ , mxBroadcast
17+ , mxGetId
18+ , mxGet
19+ , mxSet
20+ , mxClear
21+ , mxPurgeTable
22+ , mxDropTable
2023 )
2124import Control.Distributed.Process.Node
2225 ( LocalNode
2326 )
27+import Data.Binary
2428import Data.List (find)
2529import Data.Maybe (isJust)
30+import Data.Typeable
31+import GHC.Generics
2632import qualified Network.Transport as NT
27−
2833#if ! MIN_VERSION_base(4,6,0)
2934import Prelude hiding (catch, log)
3035#endif
@@ -36,41 +41,71 @@ import Test.Framework
3641import Test.Framework.Providers.HUnit (testCase)
3742import TestUtils
3843
44+data Publish = Publish
45+ deriving (Typeable, Generic, Eq)
46+
47+instance Binary Publish where
48+
49+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
50+testAgentBroadcast result = do
51+ (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
52+
53+ publisher <- mxAgent (MxAgentId "publisher-agent") () [
54+ mxSink $ \() -> mxBroadcast Publish >> mxReady
55+ ]
56+
57+ consumer <- mxAgent (MxAgentId "consumer-agent") () [
58+ mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady
59+ ]
60+
61+ mxNotify ()
62+ -- once the publisher has seen our message, it will broadcast the Publish
63+ stash result =<< receiveChanTimeout 1000000 resultRP
64+
65+ kill publisher "finished"
66+ kill consumer "finished"
67+
3968testAgentEventHandling :: TestResult Bool -> Process ()
4069testAgentEventHandling result = do
4170 let initState = [] :: [MxEvent]
42− agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
71+ agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
4372 (mxSink $ \ev -> do
4473 st <- mxGetLocal
45− let act = case ev of
46− (MxSpawned _) -> mxSetLocal (ev:st)
47− (MxProcessDied _ _) -> mxSetLocal (ev:st)
48− _ -> return ()
74+ let act =
75+ case ev of
76+ (MxSpawned _) -> mxSetLocal (ev:st)
77+ (MxProcessDied _ _) -> mxSetLocal (ev:st)
78+ _ -> return ()
4979 act >> mxReady),
5080 (mxSink $ \(ev, sp :: SendPort Bool) -> do
5181 st <- mxGetLocal
52− let found = case ev of
53− MxSpawned p ->
54− isJust $ find (\ev' ->
55− case ev' of
56− (MxSpawned p') -> p' == p
57− _ -> False) st
58− MxProcessDied p r ->
59− isJust $ find (\ev' ->
60− case ev' of
61− (MxProcessDied p' r') -> p' == p && r == r'
62− _ -> False) st
63− _ -> False
64− mxLift $ sendChan sp found
65− liftIO $ putStrLn $ "replied to " ++ (show sp)
82+ let found =
83+ case ev of
84+ MxSpawned p ->
85+ isJust $ find (\ev' ->
86+ case ev' of
87+ (MxSpawned p') -> p' == p
88+ _ -> False) st
89+ MxProcessDied p r ->
90+ isJust $ find (\ev' ->
91+ case ev' of
92+ (MxProcessDied p' r') -> p' == p && r == r'
93+ _ -> False) st
94+ _ -> False
95+ liftMX $ sendChan sp found
6696 mxReady)
6797 ]
6898
69− mref <- monitor agentPid
99+ _ <- monitor agentPid
70100 (sp, rp) <- newChan
71101 pid <- spawnLocal $ sendChan sp ()
72102 () <- receiveChan rp
73103
104+ -- By waiting for a monitor notification, we have a
105+ -- higher probably that the agent has seen
106+ monitor pid
107+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
108+
74109 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
75110 mxNotify (MxSpawned pid, replyTo)
76111 mxNotify (MxProcessDied pid DiedNormal, replyTo)
@@ -80,21 +115,92 @@ testAgentEventHandling result = do
80115
81116 stash result $ seenAlive && seenDead
82117
118+testAgentPublication :: TestResult (Maybe Publish) -> Process ()
119+testAgentPublication result = do
120+ (syncChan, rp) <- newChan
121+ let ourId = MxAgentId "publication-agent"
122+ agentPid <- mxAgent ourId () [
123+ mxSink $ \() -> do
124+ selfId <- mxGetId
125+ liftMX $ mxSet selfId "publish" Publish >> sendChan syncChan ()
126+ mxReady
127+ ]
128+
129+ mxNotify ()
130+ () <- receiveChan rp
131+
132+ stash result =<< mxGet ourId "publish"
133+
134+ mref <- monitor agentPid
83135 kill agentPid "finished"
84136 receiveWait [
85137 matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)
86138 ((\_ -> return ()))
87139 ]
88140
141+testAgentTableClear :: Int -> TestResult (Maybe Int, Maybe Int) -> Process ()
142+testAgentTableClear val result =
143+ let tId = (MxAgentId "agent-1")
144+ tKey = "key-1" in do
145+ mxSet tId tKey val
146+ get1 <- mxGet tId tKey
147+ mxClear tId tKey
148+ get2 <- mxGet tId tKey
149+ stash result (get1, get2)
150+
151+testAgentTablePurge :: TestResult (Maybe Int) -> Process ()
152+testAgentTablePurge result =
153+ let tId = MxAgentId "agent-2"
154+ tKey = "key-2" in do
155+ mxSet tId tKey (12345 :: Int)
156+ mxPurgeTable tId
157+ stash result =<< mxGet tId tKey
158+
159+testAgentTableDelete :: Int
160+ -> TestResult (Maybe Int, Maybe Int, Maybe Int)
161+ -> Process ()
162+testAgentTableDelete val result =
163+ let tId = (MxAgentId "agent-3")
164+ tKey = "key-3" in do
165+ mxSet tId tKey val
166+ get1 <- mxGet tId tKey
167+ mxDropTable tId
168+ get2 <- mxGet tId tKey
169+ mxSet tId tKey val
170+ get3 <- mxGet tId tKey
171+ stash result (get1, get2, get3)
172+
89173tests :: LocalNode -> IO [Test]
90174tests node1 = do
91175 return [
92− testGroup "Basic Agent Handling" [
93− testCase "testAgentEventHandling"
176+ testGroup "Mx Agents" [
177+ testCase "Event Handling"
94178 (delayedAssertion
95− "expected "
179+ "expected True, but events where not as expected"
96180 node1 True testAgentEventHandling)
97− ] ]
181+ , testCase "Inter-Agent Broadcast"
182+ (delayedAssertion
183+ "expected (Just ()), but no broadcast was received"
184+ node1 (Just ()) testAgentBroadcast)
185+ ],
186+ testGroup "Mx Global Properties" [
187+ testCase "Global Property Publication"
188+ (delayedAssertion
189+ "expected (Just Publish), but no table entry was found"
190+ node1 (Just Publish) testAgentPublication)
191+ , testCase "Clearing Global Properties"
192+ (delayedAssertion
193+ "expected (Just 1024, Nothing): invalid table entry found!"
194+ node1 (Just 1024, Nothing) (testAgentTableClear 1024))
195+ , testCase "Purging Global Tables"
196+ (delayedAssertion
197+ "expected Nothing, but a table entry was found"
198+ node1 Nothing testAgentTablePurge)
199+ , testCase "Deleting and (Re)Creating Global Tables"
200+ (delayedAssertion
201+ "expected (Just 15, Nothing, Just 15): invalid table entry found!"
202+ node1 (Just 15, Nothing, Just 15) (testAgentTableDelete 15))
203+ ]]
98204
99205statsTests :: NT.Transport -> IO [Test]
100206statsTests _ = do