8af66f4 · distributed-process

git.iko.soy/distributed-process

on master help sign in

extend AbstractMessage with deferred type checking

8af66f4Tim Watson14 years ago

distributed-process/src/Control/Distributed/Process.hs+2

compare with master

unifiedsplit

@@ -39,6 +39,7 @@ module Control.Distributed.Process

39 , matchUnknown

40 , AbstractMessage(..)

41 , matchAny

42+ , matchAnyIf

42 , matchChan

43 , matchChan

43 -- * Process management

44 -- * Process management

44 , spawn

45 , spawn

@@ -192,6 +193,7 @@ import Control.Distributed.Process.Internal.Primitives

192 , matchUnknown

193 , matchUnknown

193 , AbstractMessage(..)

194 , AbstractMessage(..)

194 , matchAny

195 , matchAny

196+ , matchAnyIf

195 , matchChan

197 , matchChan

196 -- Process management

198 -- Process management

197 , terminate

199 , terminate

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+53−14

@@ -21,6 +21,7 @@ module Control.Distributed.Process.Internal.Primitives

21 , matchUnknown

22 , AbstractMessage(..)

23 , matchAny

24+ , matchAnyIf

24 , matchChan

25 , matchChan

25 -- * Process management

26 -- * Process management

26 , terminate

27 , terminate

@@ -311,24 +312,62 @@ matchIf c p = Match $ MatchMsg $ \msg ->

311 !decoded = decode (messageEncoding msg)

312 !decoded = decode (messageEncoding msg)

312 _ -> Nothing

313 _ -> Nothing

313

314

315+-- | Represents a received message and provides two operations on it.

314data AbstractMessage = AbstractMessage {

316data AbstractMessage = AbstractMessage {

315− forward :: ProcessId -> Process ()

317+ forward :: ProcessId -> Process () -- ^ forward the message to @ProcessId@

318+ , maybeHandleMessage :: forall a b. (Serializable a)

319+ => (a -> Process b) -> Process (Maybe b) -- ^ handle the message if it is of the given type

316 }

320 }

317

321

318−-- | Match against an arbitrary message

322+-- | Match against an arbitrary message. 'matchAny' removes the first available

323+-- message from the process mailbox, and via the 'AbstractMessage' type,

324+-- supports forwarding /or/ handling the message /if/ it is of the correct

325+-- type. If /not/ of the right type, then the 'AbstractMessage'

326+-- @maybeHandleMessage@ function will not evaluate the supplied expression,

327+-- /but/ the message will still have been removed from the process mailbox!

328+--

319matchAny :: forall b. (AbstractMessage -> Process b) -> Match b

329matchAny :: forall b. (AbstractMessage -> Process b) -> Match b

320−matchAny p = Match $ MatchMsg $ Just . p . abstract

330+matchAny p = Match $ MatchMsg $ Just . p . abstract

331+

332+-- | Match against an arbitrary message. 'matchAnyIf' will /only/ remove the

333+-- message from the process mailbox, /if/ the supplied condition matches. The

334+-- success (or failure) of runtime type checks in @maybeHandleMessage@ does not

335+-- count here, i.e., if the condition evaluates to @True@ then the message will

336+-- be removed from the process mailbox and decoded, but that does /not/

337+-- guarantee that an expression passed to @maybeHandleMessage@ will pass the

338+-- runtime type checks and therefore be evaluated. If the types do not match

339+-- up, then @maybeHandleMessage@ returns 'Nothing'.

340+matchAnyIf :: forall a b. (Serializable a)

341+ => (a -> Bool)

342+ -> (AbstractMessage -> Process b)

343+ -> Match b

344+matchAnyIf c p = Match $ MatchMsg $ \msg ->

345+ case messageFingerprint msg == fingerprint (undefined :: a) of

346+ True | c decoded -> Just (p (abstract msg))

321− where

347+ where

348+ decoded :: a

349+ -- Make sure the value is fully decoded so that we don't hang to

350+ -- bytestrings when the calling process doesn't evaluate immediately

351+ !decoded = decode (messageEncoding msg)

352+ _ -> Nothing

353+

322− abstract :: Message -> AbstractMessage

354+abstract :: Message -> AbstractMessage

323− abstract msg = AbstractMessage {

355+abstract msg = AbstractMessage {

324− forward = \them -> do

356+ forward = \them -> do

325− proc <- ask

357+ proc <- ask

326− liftIO $ sendPayload (processNode proc)

358+ liftIO $ sendPayload (processNode proc)

327− (ProcessIdentifier (processId proc))

359+ (ProcessIdentifier (processId proc))

328− (ProcessIdentifier them)

360+ (ProcessIdentifier them)

329− NoImplicitReconnect

361+ NoImplicitReconnect

330− (messageToPayload msg)

362+ (messageToPayload msg)

363+ , maybeHandleMessage = \(proc :: (a -> Process b)) -> do

364+ case messageFingerprint msg == fingerprint (undefined :: a) of

365+ True -> do { r <- proc (decoded :: a); return (Just r) }

366+ where

367+ decoded :: a

368+ !decoded = decode (messageEncoding msg)

369+ _ -> return Nothing

331− }

370+ }

332

371

333-- | Remove any message from the queue

372-- | Remove any message from the queue

334matchUnknown :: Process b -> Match b

373matchUnknown :: Process b -> Match b

distributed-process/tests/TestCH.hs+118

@@ -757,6 +757,121 @@ testMatchAny transport = do

757

758 takeMVar clientDone

759

760+-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

761+-- in between, however we block 'Divide' requests ....

762+testMatchAnyHandle :: NT.Transport -> Assertion

763+testMatchAnyHandle transport = do

764+ proxyAddr <- newEmptyMVar

765+ clientDone <- newEmptyMVar

766+

767+ -- Math server

768+ forkIO $ do

769+ localNode <- newLocalNode transport initRemoteTable

770+ mathServer <- forkProcess localNode math

771+ proxyServer <- forkProcess localNode $ forever $ do

772+ receiveWait [

773+ matchAny (maybeForward mathServer)

774+ ]

775+ putMVar proxyAddr proxyServer

776+

777+ -- Client

778+ forkIO $ do

779+ localNode <- newLocalNode transport initRemoteTable

780+ mathServer <- readMVar proxyAddr

781+

782+ runProcess localNode $ do

783+ pid <- getSelfPid

784+ send mathServer (Add pid 1 2)

785+ 3 <- expect :: Process Double

786+ send mathServer (Divide pid 8 2)

787+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)

788+ liftIO $ putMVar clientDone ()

789+

790+ takeMVar clientDone

791+ where maybeForward :: ProcessId -> AbstractMessage -> Process (Maybe ())

792+ maybeForward s msg =

793+ maybeHandleMessage msg (\m@(Add _ _ _) -> send s m)

794+

795+testMatchAnyNoHandle :: NT.Transport -> Assertion

796+testMatchAnyNoHandle transport = do

797+ addr <- newEmptyMVar

798+ clientDone <- newEmptyMVar

799+ serverDone <- newEmptyMVar

800+

801+ -- Math server

802+ forkIO $ do

803+ localNode <- newLocalNode transport initRemoteTable

804+ server <- forkProcess localNode $ forever $ do

805+ receiveWait [

806+ matchAnyIf

807+ -- the condition has type `Add -> Bool`

808+ (\(Add _ _ _) -> True)

809+ -- the match `AbstractMessage -> Process ()` will succeed!

810+ (\m -> do

811+ -- `String -> Process ()` does *not* match the input types however

812+ r <- (maybeHandleMessage m (\(_ :: String) -> die "NONSENSE" ))

813+ case r of

814+ Nothing -> return ()

815+ Just _ -> die "NONSENSE")

816+ ]

817+ -- we *must* have removed the message from our mailbox though!!!

818+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]

819+ liftIO $ putMVar serverDone ()

820+ putMVar addr server

821+

822+ -- Client

823+ forkIO $ do

824+ localNode <- newLocalNode transport initRemoteTable

825+ server <- readMVar addr

826+

827+ runProcess localNode $ do

828+ pid <- getSelfPid

829+ send server (Add pid 1 2)

830+ -- we only care about the client having sent a message, so we're done

831+ liftIO $ putMVar clientDone ()

832+

833+ takeMVar clientDone

834+ takeMVar serverDone

835+

836+-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests

837+-- unless the text body @/= "bar"@ - this case should time out rather than

838+-- removing the message from the process mailbox.

839+testMatchAnyIf :: NT.Transport -> Assertion

840+testMatchAnyIf transport = do

841+ echoAddr <- newEmptyMVar

842+ clientDone <- newEmptyMVar

843+

844+ -- echo server

845+ forkIO $ do

846+ localNode <- newLocalNode transport initRemoteTable

847+ echoServer <- forkProcess localNode $ forever $ do

848+ receiveWait [

849+ matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")

850+ handleMessage

851+ ]

852+ putMVar echoAddr echoServer

853+

854+ -- Client

855+ forkIO $ do

856+ localNode <- newLocalNode transport initRemoteTable

857+ server <- readMVar echoAddr

858+

859+ runProcess localNode $ do

860+ pid <- getSelfPid

861+ send server (pid, "foo")

862+ "foo" <- expect

863+ send server (pid, "baz")

864+ "baz" <- expect

865+ send server (pid, "bar")

866+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)

867+ liftIO $ putMVar clientDone ()

868+

869+ takeMVar clientDone

870+ where handleMessage :: AbstractMessage -> Process (Maybe ())

871+ handleMessage msg =

872+ maybeHandleMessage msg (\(pid :: ProcessId, (m :: String))

873+ -> do { send pid m; return () })

874+

760-- Test 'receiveChanTimeout'

875-- Test 'receiveChanTimeout'

761testReceiveChanTimeout :: NT.Transport -> Assertion

876testReceiveChanTimeout :: NT.Transport -> Assertion

762testReceiveChanTimeout transport = do

877testReceiveChanTimeout transport = do

@@ -987,6 +1102,9 @@ tests (transport, transportInternals) = [

987 , testCase "RemoteRegistry" (testRemoteRegistry transport)

1102 , testCase "RemoteRegistry" (testRemoteRegistry transport)

988 , testCase "SpawnLocal" (testSpawnLocal transport)

1103 , testCase "SpawnLocal" (testSpawnLocal transport)

989 , testCase "MatchAny" (testMatchAny transport)

1104 , testCase "MatchAny" (testMatchAny transport)

1105+ , testCase "MatchAnyHandle" (testMatchAnyHandle transport)

1106+ , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle transport)

1107+ , testCase "MatchAnyIf" (testMatchAnyIf transport)

990 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

1108 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

991 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

1109 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

992 , testCase "KillLocal" (testKillLocal transport)

1110 , testCase "KillLocal" (testKillLocal transport)