git.iko.soy/distributed-process
on master help sign in
Proposal: matchAny
1052f4eEdsko de Vries
distributed-process/src/Control/Distributed/Process.hs+4
compare with master
unifiedsplit
@@ -35,6 +35,8 @@ module Control.Distributed.Process
35 , match
35 , match
36 , matchIf
36 , matchIf
37 , matchUnknown
37 , matchUnknown
38+ , AbstractMessage(..)
39+ , matchAny
38 -- * Process management
40 -- * Process management
39 , spawn
41 , spawn
40 , call
42 , call
@@ -173,6 +175,8 @@ import Control.Distributed.Process.Internal.Primitives
173 , match
175 , match
174 , matchIf
176 , matchIf
175 , matchUnknown
177 , matchUnknown
178+ , AbstractMessage(..)
179+ , matchAny
176 -- Process management
180 -- Process management
177 , terminate
181 , terminate
178 , ProcessTerminationException(..)
182 , ProcessTerminationException(..)
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+23
@@ -19,6 +19,8 @@ module Control.Distributed.Process.Internal.Primitives
19 , match
19 , match
20 , matchIf
20 , matchIf
21 , matchUnknown
21 , matchUnknown
22+ , AbstractMessage(..)
23+ , matchAny
22 -- * Process management
24 -- * Process management
23 , terminate
25 , terminate
24 , ProcessTerminationException(..)
26 , ProcessTerminationException(..)
@@ -127,10 +129,12 @@ import Control.Distributed.Process.Internal.Types
127 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
129 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
128 , LocalProcessState
130 , LocalProcessState
129 , LocalSendPortId
131 , LocalSendPortId
132+ , messageToPayload
130 )
133 )
131import Control.Distributed.Process.Internal.Messaging
134import Control.Distributed.Process.Internal.Messaging
132 ( sendMessage
135 ( sendMessage
133 , sendBinary
136 , sendBinary
137+ , sendPayload
134 , disconnect
138 , disconnect
135 )
139 )
136import Control.Distributed.Process.Internal.WeakTQueue
140import Control.Distributed.Process.Internal.WeakTQueue
@@ -269,6 +273,25 @@ matchIf c p = Match $ \msg ->
269 then Just $ p decoded
273 then Just $ p decoded
270 else Nothing
274 else Nothing
271
275
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+
272-- | Remove any message from the queue
295-- | Remove any message from the queue
273matchUnknown :: Process b -> Match b
296matchUnknown :: Process b -> Match b
274matchUnknown = Match . const . Just
297matchUnknown = Match . const . Just
distributed-process/tests/TestCH.hs+35−1
@@ -15,7 +15,7 @@ import Control.Concurrent.MVar
15 , takeMVar
15 , takeMVar
16 , readMVar
16 , readMVar
17 )
17 )
18−import Control.Monad (replicateM_, replicateM)
18+import Control.Monad (replicateM_, replicateM, forever)
19import Control.Exception (throwIO)
19import Control.Exception (throwIO)
20import Control.Applicative ((<$>), (<*>))
20import Control.Applicative ((<$>), (<*>))
21import qualified Network.Transport as NT (Transport, closeEndPoint)
21import qualified Network.Transport as NT (Transport, closeEndPoint)
@@ -660,6 +660,39 @@ testReconnect transport transportInternals = do
660
660
661 takeMVar registerTestOk
661 takeMVar registerTestOk
662
662
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+
663main :: IO ()
696main :: IO ()
664main = do
697main = do
665 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
698 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
@@ -675,6 +708,7 @@ main = do
675 , ("Registry", testRegistry transport)
708 , ("Registry", testRegistry transport)
676 , ("RemoteRegistry", testRemoteRegistry transport)
709 , ("RemoteRegistry", testRemoteRegistry transport)
677 , ("SpawnLocal", testSpawnLocal transport)
710 , ("SpawnLocal", testSpawnLocal transport)
711+ , ("MatchAny", testMatchAny transport)
678 -- Monitoring processes
712 -- Monitoring processes
679 --
713 --
680 -- The "missing" combinations in the list below don't make much sense, as
714 -- The "missing" combinations in the list below don't make much sense, as