git.iko.soy/distributed-process
on master help sign in
Proposal: matchAny
commit1052f4e618c98048e30ff9e0fe640ba5df7cb26b authorEdsko de Vries <edsko@edsko.net> dateWed 19 Sep 2012, 15:41 (14 years ago) parent2ee6440
3 files changed+62−1
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process.hs+4 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+23 distributed-process/tests/TestCH.hs+35−1
distributed-process/src/Control/Distributed/Process.hs+4
@@ -35,6 +35,8 @@ module Control.Distributed.Process
3535 , match
3636 , matchIf
3737 , matchUnknown
38+ , AbstractMessage(..)
39+ , matchAny
3840 -- * Process management
3941 , spawn
4042 , call
@@ -173,6 +175,8 @@ import Control.Distributed.Process.Internal.Primitives
173175 , match
174176 , matchIf
175177 , matchUnknown
178+ , AbstractMessage(..)
179+ , matchAny
176180 -- Process management
177181 , terminate
178182 , ProcessTerminationException(..)
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+23
@@ -19,6 +19,8 @@ module Control.Distributed.Process.Internal.Primitives
1919 , match
2020 , matchIf
2121 , matchUnknown
22+ , AbstractMessage(..)
23+ , matchAny
2224 -- * Process management
2325 , terminate
2426 , ProcessTerminationException(..)
@@ -127,10 +129,12 @@ import Control.Distributed.Process.Internal.Types
127129 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
128130 , LocalProcessState
129131 , LocalSendPortId
132+ , messageToPayload
130133 )
131134import Control.Distributed.Process.Internal.Messaging
132135 ( sendMessage
133136 , sendBinary
137+ , sendPayload
134138 , disconnect
135139 )
136140import Control.Distributed.Process.Internal.WeakTQueue
@@ -269,6 +273,25 @@ matchIf c p = Match $ \msg ->
269273 then Just $ p decoded
270274 else Nothing
271275
276+data AbstractMessage = AbstractMessage {
277+ forward :: ProcessId -> Process ()
278+ }
279+
280+-- | Match against an arbitrary message
281+matchAny :: forall b. (AbstractMessage -> Process b) -> Match b
282+matchAny p = Match $ Just . p . abstract
283+ where
284+ abstract :: Message -> AbstractMessage
285+ abstract msg = AbstractMessage {
286+ forward = \them -> do
287+ proc <- ask
288+ liftIO $ sendPayload (processNode proc)
289+ (ProcessIdentifier (processId proc))
290+ (ProcessIdentifier them)
291+ NoImplicitReconnect
292+ (messageToPayload msg)
293+ }
294+
272295-- | Remove any message from the queue
273296matchUnknown :: Process b -> Match b
274297matchUnknown = Match . const . Just
distributed-process/tests/TestCH.hs+35−1
@@ -15,7 +15,7 @@ import Control.Concurrent.MVar
1515 , takeMVar
1616 , readMVar
1717 )
18−import Control.Monad (replicateM_, replicateM)
18+import Control.Monad (replicateM_, replicateM, forever)
1919import Control.Exception (throwIO)
2020import Control.Applicative ((<$>), (<*>))
2121import qualified Network.Transport as NT (Transport, closeEndPoint)
@@ -660,6 +660,39 @@ testReconnect transport transportInternals = do
660660
661661 takeMVar registerTestOk
662662
663+-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
664+-- in between
665+testMatchAny :: NT.Transport -> IO ()
666+testMatchAny transport = do
667+ proxyAddr <- newEmptyMVar
668+ clientDone <- newEmptyMVar
669+
670+ -- Math server
671+ forkIO $ do
672+ localNode <- newLocalNode transport initRemoteTable
673+ mathServer <- forkProcess localNode math
674+ proxyServer <- forkProcess localNode $ forever $ do
675+ msg <- receiveWait [ matchAny return ]
676+ forward msg mathServer
677+ putMVar proxyAddr proxyServer
678+
679+ -- Client
680+ forkIO $ do
681+ localNode <- newLocalNode transport initRemoteTable
682+ mathServer <- readMVar proxyAddr
683+
684+ runProcess localNode $ do
685+ pid <- getSelfPid
686+ send mathServer (Add pid 1 2)
687+ 3 <- expect :: Process Double
688+ send mathServer (Divide pid 8 2)
689+ 4 <- expect :: Process Double
690+ send mathServer (Divide pid 8 0)
691+ DivByZero <- expect
692+ liftIO $ putMVar clientDone ()
693+
694+ takeMVar clientDone
695+
663696main :: IO ()
664697main = do
665698 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
@@ -675,6 +708,7 @@ main = do
675708 , ("Registry", testRegistry transport)
676709 , ("RemoteRegistry", testRemoteRegistry transport)
677710 , ("SpawnLocal", testSpawnLocal transport)
711+ , ("MatchAny", testMatchAny transport)
678712 -- Monitoring processes
679713 --
680714 -- The "missing" combinations in the list below don't make much sense, as