@@ -8,21 +8,25 @@
8-- semantics are based on /A unified semantics for future Erlang/ by Hans
8-- semantics are based on /A unified semantics for future Erlang/ by Hans
9-- Svensson, Lars-Åke Fredlund and Clara Benac Earle.
9-- Svensson, Lars-Åke Fredlund and Clara Benac Earle.
10module Control.Distributed.Process
10module Control.Distributed.Process
11− ( -- * Basic cloud Haskell API
11+ ( -- * Basic types
15− , expect
15+ -- * Basic messaging
17− , getSelfPid
17+ , expect
18− , spawn
18+ -- * Advanced messaging
19− -- * Matching messages
23 , matchUnknown
24 , matchUnknown
24− , receiveWait
25+ -- * Process management
25− , receiveTimeout
26+ , spawn
26 -- * Monitoring and linking
30 -- * Monitoring and linking
@@ -153,57 +157,42 @@ import Control.Distributed.Process.Internal.MessageT
153-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
157-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
155--------------------------------------------------------------------------------
159--------------------------------------------------------------------------------
156−-- Basic Cloud Haskell API --
160+-- Basic messaging --
157--------------------------------------------------------------------------------
161--------------------------------------------------------------------------------
159−-- | Wait for a message of a specific type
160−expect :: forall a. Serializable a => Process a
161−expect = receiveWait [match return]
163-- | Send a message
163-- | Send a message
164send :: Serializable a => ProcessId -> a -> Process ()
164send :: Serializable a => ProcessId -> a -> Process ()
165-- This requires a lookup on every send. If we want to avoid that we need to
165-- This requires a lookup on every send. If we want to avoid that we need to
166-- modify serializable to allow for stateful (IO) deserialization
166-- modify serializable to allow for stateful (IO) deserialization
167send them msg = lift $ sendMessage (Left them) msg
167send them msg = lift $ sendMessage (Left them) msg
169−-- | Our own process ID
169+-- | Wait for a message of a specific type
170−getSelfPid :: Process ProcessId
170+expect :: forall a. Serializable a => Process a
171−getSelfPid = processId <$> ask
171+expect = receiveWait [match return]
173−-- | Monitor another process
174−monitor :: ProcessId -> Process MonitorRef
176− monitorRef <- getMonitorRefFor them
177− sendCtrlMsg Nothing $ Monitor them monitorRef
180−-- | Link to a remote process
181−link :: ProcessId -> Process ()
182−link = sendCtrlMsg Nothing . Link
184−-- | Remove a monitor
185−unmonitor :: MonitorRef -> Process ()
186−unmonitor = sendCtrlMsg Nothing . Unmonitor
189−unlink :: ProcessId -> Process ()
190−unlink = sendCtrlMsg Nothing . Unlink
193−spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
195− ref <- spawnAsync nid proc
196− receiveWait [ matchIf (\(DidSpawn ref' _) -> ref == ref')
197− (\(DidSpawn _ pid) -> return pid)
200--------------------------------------------------------------------------------
173--------------------------------------------------------------------------------
201−-- Matching --
174+-- Advanced messaging --
202--------------------------------------------------------------------------------
175--------------------------------------------------------------------------------
204-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
177-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
205newtype Match b = Match { unMatch :: Message -> Maybe (Process b) }
178newtype Match b = Match { unMatch :: Message -> Maybe (Process b) }
180+-- | Test the matches in order against each message in the queue
181+receiveWait :: [Match b] -> Process b
183+ queue <- processQueue <$> ask
184+ Just proc <- liftIO $ dequeue queue Nothing (map unMatch ms)
187+-- | Like 'receiveWait' but with a timeout
188+receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
189+receiveTimeout t ms = do
190+ queue <- processQueue <$> ask
191+ mProc <- liftIO $ dequeue queue (Just t) (map unMatch ms)
193+ Nothing -> return Nothing
194+ Just proc -> Just <$> proc
207-- | Match against any message of the right type
196-- | Match against any message of the right type
208match :: forall a b. Serializable a => (a -> Process b) -> Match b
197match :: forall a b. Serializable a => (a -> Process b) -> Match b
209match = matchIf (const True)
198match = matchIf (const True)
@@ -214,28 +203,55 @@ matchIf c p = Match $ \msg ->
214 let decoded :: a
203 let decoded :: a
215 decoded = decode . messageEncoding $ msg in
204 decoded = decode . messageEncoding $ msg in
216 if messageFingerprint msg == fingerprint (undefined :: a) && c decoded
205 if messageFingerprint msg == fingerprint (undefined :: a) && c decoded
217− then Just . p . decode . messageEncoding $ msg
206+ then Just $ p decoded
218 else Nothing
207 else Nothing
220-- | Remove any message from the queue
209-- | Remove any message from the queue
221matchUnknown :: Process b -> Match b
210matchUnknown :: Process b -> Match b
222matchUnknown = Match . const . Just
211matchUnknown = Match . const . Just
224−-- | Test the matches in order against each message in the queue
213+--------------------------------------------------------------------------------
225−receiveWait :: [Match b] -> Process b
214+-- Process management --
226−receiveWait ms = do
215+--------------------------------------------------------------------------------
227− queue <- processQueue <$> ask
228− Just proc <- liftIO $ dequeue queue Nothing (map unMatch ms)
231−-- | Like 'receiveWait' but with a timeout
217+-- | Spawn a process
232−receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
218+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
233−receiveTimeout t ms = do
219+spawn nid proc = do
234− queue <- processQueue <$> ask
220+ ref <- spawnAsync nid proc
235− mProc <- liftIO $ dequeue queue (Just t) (map unMatch ms)
221+ receiveWait [ matchIf (\(DidSpawn ref' _) -> ref == ref')
236− case mProc of
222+ (\(DidSpawn _ pid) -> return pid)
237− Nothing -> return Nothing
223+ ]
238− Just proc -> Just <$> proc
224+
225+-- | Our own process ID
226+getSelfPid :: Process ProcessId
227+getSelfPid = processId <$> ask
229+-- | Get the node ID of our local node
230+getSelfNode :: Process NodeId
231+getSelfNode = localNodeId <$> lift getLocalNode
233+--------------------------------------------------------------------------------
234+-- Monitoring and linking --
235+--------------------------------------------------------------------------------
237+-- | Link to a remote process
238+link :: ProcessId -> Process ()
239+link = sendCtrlMsg Nothing . Link
242+unlink :: ProcessId -> Process ()
243+unlink = sendCtrlMsg Nothing . Unlink
245+-- | Monitor another process
246+monitor :: ProcessId -> Process MonitorRef
248+ monitorRef <- getMonitorRefFor them
249+ sendCtrlMsg Nothing $ Monitor them monitorRef
252+-- | Remove a monitor
253+unmonitor :: MonitorRef -> Process ()
254+unmonitor = sendCtrlMsg Nothing . Unmonitor
240--------------------------------------------------------------------------------
256--------------------------------------------------------------------------------
241-- Closures --
257-- Closures --