{-# LANGUAGE DeriveDataTypeable    #-}
{-# LANGUAGE DeriveGeneric         #-}
{-# LANGUAGE StandaloneDeriving    #-}
{-# LANGUAGE ScopedTypeVariables   #-}
{-# LANGUAGE TemplateHaskell       #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE FlexibleInstances     #-}

-----------------------------------------------------------------------------
-- |
-- Module      :  Control.Distributed.Process.Extras.Internal.Primitives
-- Copyright   :  (c) Tim Watson 2013 - 2017, Parallel Scientific (Jeff Epstein) 2012
-- License     :  BSD3 (see the file LICENSE)
--
-- Maintainers :  Jeff Epstein, Tim Watson
-- Stability   :  experimental
-- Portability :  non-portable (requires concurrency)
--
-- This module provides a set of additional primitives that add functionality
-- to the basic Cloud Haskell APIs.
-----------------------------------------------------------------------------

module Control.Distributed.Process.Extras.Internal.Primitives
  ( -- * General Purpose Process Addressing
    Addressable
  , Routable(..)
  , Resolvable(..)
  , Linkable(..)
  , Killable(..)
  , Monitored(..)

    -- * Spawning and Linking
  , spawnSignalled
  , spawnLinkLocal
  , spawnMonitorLocal
  , linkOnFailure

    -- * Registered Processes
  , whereisRemote
  , whereisOrStart
  , whereisOrStartRemote

    -- * Selective Receive/Matching
  , matchCond
  , awaitResponse

    -- * General Utilities
  , times
  , monitor
  , awaitExit
  , isProcessAlive
  , forever'
  , deliver

    -- * Remote Table
  , __remoteTable
  ) where

import Control.Concurrent (myThreadId, throwTo)
import Control.Distributed.Process hiding (monitor, finally, catch)
import qualified Control.Distributed.Process as P (monitor, unmonitor)
import Control.Distributed.Process.Closure (seqCP, remotable, mkClosure)
import Control.Distributed.Process.Serializable (Serializable)
import Control.Distributed.Process.Extras.Internal.Types
  ( Addressable
  , Linkable(..)
  , Killable(..)
  , Resolvable(..)
  , Routable(..)
  , Monitored(..)
  , RegisterSelf(..)
  , ExitReason(ExitOther)
  , whereisRemote
  )
import Control.Monad (void, (>=>), replicateM_)
import Control.Monad.Catch (finally, catchIf)
import Data.Maybe (isJust, fromJust)
import Data.Foldable (traverse_)

-- utility

-- | Monitor any @Resolvable@ object.
monitor :: Resolvable a => a -> Process (Maybe MonitorRef)
monitor :: forall a. Resolvable a => a -> Process (Maybe MonitorRef)
monitor = a -> Process (Maybe ProcessId)
forall a. Resolvable a => a -> Process (Maybe ProcessId)
resolve (a -> Process (Maybe ProcessId))
-> (Maybe ProcessId -> Process (Maybe MonitorRef))
-> a
-> Process (Maybe MonitorRef)
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> (ProcessId -> Process MonitorRef)
-> Maybe ProcessId -> Process (Maybe MonitorRef)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> Maybe a -> f (Maybe b)
traverse ProcessId -> Process MonitorRef
P.monitor

-- | Wait until @Resolvable@ object will exit. Return immediately
-- if object can't be resolved.
awaitExit :: Resolvable a => a -> Process ()
awaitExit :: forall a. Resolvable a => a -> Process ()
awaitExit = a -> Process (Maybe ProcessId)
forall a. Resolvable a => a -> Process (Maybe ProcessId)
resolve (a -> Process (Maybe ProcessId))
-> (Maybe ProcessId -> Process ()) -> a -> Process ()
forall (m :: * -> *) a b c.
Monad m =>
(a -> m b) -> (b -> m c) -> a -> m c
>=> (ProcessId -> Process ()) -> Maybe ProcessId -> Process ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ ProcessId -> Process ()
await where
  await :: ProcessId -> Process ()
await ProcessId
pid = ProcessId -> (MonitorRef -> Process ()) -> Process ()
forall {c}. ProcessId -> (MonitorRef -> Process c) -> Process c
withMonitorRef ProcessId
pid ((MonitorRef -> Process ()) -> Process ())
-> (MonitorRef -> Process ()) -> Process ()
forall a b. (a -> b) -> a -> b
$ \MonitorRef
ref -> [Match ()] -> Process ()
forall b. [Match b] -> Process b
receiveWait
      [ (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process ()) -> Match ()
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
r ProcessId
_ DiedReason
_) -> MonitorRef
r MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
ref)
                (\ProcessMonitorNotification
_ -> () -> Process ()
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return ())
      ]
  withMonitorRef :: ProcessId -> (MonitorRef -> Process c) -> Process c
withMonitorRef ProcessId
pid = Process MonitorRef
-> (MonitorRef -> Process ())
-> (MonitorRef -> Process c)
-> Process c
forall a b c.
Process a -> (a -> Process b) -> (a -> Process c) -> Process c
bracket (ProcessId -> Process MonitorRef
P.monitor ProcessId
pid) MonitorRef -> Process ()
P.unmonitor

-- | Send message to @Addressable@ object.
deliver :: (Addressable a, Serializable m) => m -> a -> Process ()
deliver :: forall a m. (Addressable a, Serializable m) => m -> a -> Process ()
deliver = (a -> m -> Process ()) -> m -> a -> Process ()
forall a b c. (a -> b -> c) -> b -> a -> c
flip a -> m -> Process ()
forall m. (Serializable m, Resolvable a) => a -> m -> Process ()
forall a m.
(Routable a, Serializable m, Resolvable a) =>
a -> m -> Process ()
sendTo

-- | Check if specified process is alive. Information may be outdated.
isProcessAlive :: ProcessId -> Process Bool
isProcessAlive :: ProcessId -> Process Bool
isProcessAlive ProcessId
pid = Maybe ProcessInfo -> Bool
forall a. Maybe a -> Bool
isJust (Maybe ProcessInfo -> Bool)
-> Process (Maybe ProcessInfo) -> Process Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ProcessId -> Process (Maybe ProcessInfo)
getProcessInfo ProcessId
pid

-- | Apply the supplied expression /n/ times
times :: Int -> Process () -> Process ()
times :: Int -> Process () -> Process ()
times = Int -> Process () -> Process ()
forall (m :: * -> *) a. Applicative m => Int -> m a -> m ()
replicateM_
{-# DEPRECATED times "use replicateM_ instead" #-}

-- | Like 'Control.Monad.forever' but sans space leak
forever' :: Monad m => m a -> m b
forever' :: forall (m :: * -> *) a b. Monad m => m a -> m b
forever' m a
a = let a' :: m b
a' = m a
a m a -> m b -> m b
forall a b. m a -> m b -> m b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> m b
a' in m b
forall {b}. m b
a'
{-# INLINE forever' #-}

-- spawning, linking and generic server startup

-- | Spawn a new (local) process. This variant takes an initialisation
-- action and a secondary expression from the result of the initialisation
-- to @Process ()@. The spawn operation synchronises on the completion of the
-- @before@ action, such that the calling process is guaranteed to only see
-- the newly spawned @ProcessId@ once the initialisation has successfully
-- completed.
spawnSignalled :: Process a -> (a -> Process ()) -> Process ProcessId
spawnSignalled :: forall a. Process a -> (a -> Process ()) -> Process ProcessId
spawnSignalled Process a
before a -> Process ()
after = do
  (SendPort ()
sigStart, ReceivePort ()
recvStart) <- Process (SendPort (), ReceivePort ())
forall a. Serializable a => Process (SendPort a, ReceivePort a)
newChan
  (ProcessId
pid, MonitorRef
mRef) <- Process () -> Process (ProcessId, MonitorRef)
spawnMonitorLocal (Process () -> Process (ProcessId, MonitorRef))
-> Process () -> Process (ProcessId, MonitorRef)
forall a b. (a -> b) -> a -> b
$ do
    a
initProc <- Process a
before
    SendPort () -> () -> Process ()
forall a. Serializable a => SendPort a -> a -> Process ()
sendChan SendPort ()
sigStart ()
    a -> Process ()
after a
initProc
  [Match ProcessId] -> Process ProcessId
forall b. [Match b] -> Process b
receiveWait [
      (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process ProcessId)
-> Match ProcessId
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
ref ProcessId
_ DiedReason
_) -> MonitorRef
ref MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
mRef)
              (\(ProcessMonitorNotification MonitorRef
_ ProcessId
_ DiedReason
dr) -> ExitReason -> Process ProcessId
forall a b. Serializable a => a -> Process b
die (ExitReason -> Process ProcessId)
-> ExitReason -> Process ProcessId
forall a b. (a -> b) -> a -> b
$ String -> ExitReason
ExitOther (DiedReason -> String
forall a. Show a => a -> String
show DiedReason
dr))
    , ReceivePort () -> (() -> Process ProcessId) -> Match ProcessId
forall a b. ReceivePort a -> (a -> Process b) -> Match b
matchChan ReceivePort ()
recvStart (\() -> ProcessId -> Process ProcessId
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return ProcessId
pid)
    ] Process ProcessId -> Process () -> Process ProcessId
forall (m :: * -> *) a b.
(HasCallStack, MonadMask m) =>
m a -> m b -> m a
`finally` (MonitorRef -> Process ()
unmonitor MonitorRef
mRef)

-- | Node local version of 'Control.Distributed.Process.spawnLink'.
-- Note that this is just the sequential composition of 'spawn' and 'link'.
-- (The "Unified" semantics that underlies Cloud Haskell does not even support
-- a synchronous link operation)
spawnLinkLocal :: Process () -> Process ProcessId
spawnLinkLocal :: Process () -> Process ProcessId
spawnLinkLocal Process ()
p = do
  ProcessId
pid <- Process () -> Process ProcessId
spawnLocal Process ()
p
  ProcessId -> Process ()
link ProcessId
pid
  ProcessId -> Process ProcessId
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return ProcessId
pid

-- | Like 'spawnLinkLocal', but monitors the spawned process.
--
spawnMonitorLocal :: Process () -> Process (ProcessId, MonitorRef)
spawnMonitorLocal :: Process () -> Process (ProcessId, MonitorRef)
spawnMonitorLocal Process ()
p = do
  ProcessId
pid <- Process () -> Process ProcessId
spawnLocal Process ()
p
  MonitorRef
ref <- ProcessId -> Process MonitorRef
P.monitor ProcessId
pid
  (ProcessId, MonitorRef) -> Process (ProcessId, MonitorRef)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (ProcessId
pid, MonitorRef
ref)

-- | CH's 'link' primitive, unlike Erlang's, will trigger when the target
-- process dies for any reason. This function has semantics like Erlang's:
-- it will trigger 'ProcessLinkException' only when the target dies abnormally.
--
linkOnFailure :: ProcessId -> Process ()
linkOnFailure :: ProcessId -> Process ()
linkOnFailure ProcessId
them = do
  ProcessId
us <- Process ProcessId
getSelfPid
  ThreadId
tid <- IO ThreadId -> Process ThreadId
forall a. IO a -> Process a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO ThreadId -> Process ThreadId)
-> IO ThreadId -> Process ThreadId
forall a b. (a -> b) -> a -> b
$ IO ThreadId
myThreadId
  Process ProcessId -> Process ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (Process ProcessId -> Process ())
-> Process ProcessId -> Process ()
forall a b. (a -> b) -> a -> b
$ Process () -> Process ProcessId
spawnLocal (Process () -> Process ProcessId)
-> Process () -> Process ProcessId
forall a b. (a -> b) -> a -> b
$ do
    MonitorRef
callerRef <- ProcessId -> Process MonitorRef
P.monitor ProcessId
us
    MonitorRef
calleeRef <- ProcessId -> Process MonitorRef
P.monitor ProcessId
them
    DiedReason
reason <- [Match DiedReason] -> Process DiedReason
forall b. [Match b] -> Process b
receiveWait [
             (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process DiedReason)
-> Match DiedReason
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
mRef ProcessId
_ DiedReason
_) ->
                       MonitorRef
mRef MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
callerRef) -- nothing left to do
                     (\ProcessMonitorNotification
_ -> DiedReason -> Process DiedReason
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return DiedReason
DiedNormal)
           , (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process DiedReason)
-> Match DiedReason
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
mRef' ProcessId
_ DiedReason
_) ->
                       MonitorRef
mRef' MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
calleeRef)
                     (\(ProcessMonitorNotification MonitorRef
_ ProcessId
_ DiedReason
r') -> DiedReason -> Process DiedReason
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return DiedReason
r')
         ]
    case DiedReason
reason of
      DiedReason
DiedNormal -> () -> Process ()
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
      DiedReason
_ -> IO () -> Process ()
forall a. IO a -> Process a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> Process ()) -> IO () -> Process ()
forall a b. (a -> b) -> a -> b
$ ThreadId -> ProcessLinkException -> IO ()
forall e. Exception e => ThreadId -> e -> IO ()
throwTo ThreadId
tid (ProcessId -> DiedReason -> ProcessLinkException
ProcessLinkException ProcessId
us DiedReason
reason)

-- | Returns the pid of the process that has been registered
-- under the given name. This refers to a local, per-node registration,
-- not @global@ registration. If that name is unregistered, a process
-- is started. This is a handy way to start per-node named servers.
--
whereisOrStart :: String -> Process () -> Process ProcessId
whereisOrStart :: String -> Process () -> Process ProcessId
whereisOrStart String
name Process ()
proc = do
  (SendPort ProcessId
sigStart, ReceivePort ProcessId
recvStart) <- Process (SendPort ProcessId, ReceivePort ProcessId)
forall a. Serializable a => Process (SendPort a, ReceivePort a)
newChan
  (ProcessId
_, MonitorRef
mRef) <- Process () -> Process (ProcessId, MonitorRef)
spawnMonitorLocal (Process () -> Process (ProcessId, MonitorRef))
-> Process () -> Process (ProcessId, MonitorRef)
forall a b. (a -> b) -> a -> b
$ do
    ProcessId
us <- Process ProcessId
getSelfPid
    (ProcessRegistrationException -> Bool)
-> Process ()
-> (ProcessRegistrationException -> Process ())
-> Process ()
forall (m :: * -> *) e a.
(HasCallStack, MonadCatch m, Exception e) =>
(e -> Bool) -> m a -> (e -> m a) -> m a
catchIf (\(ProcessRegistrationException String
_ Maybe ProcessId
r) -> Maybe ProcessId -> Bool
forall a. Maybe a -> Bool
isJust Maybe ProcessId
r)
            (String -> ProcessId -> Process ()
register String
name ProcessId
us Process () -> Process () -> Process ()
forall a b. Process a -> Process b -> Process b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> SendPort ProcessId -> ProcessId -> Process ()
forall a. Serializable a => SendPort a -> a -> Process ()
sendChan SendPort ProcessId
sigStart ProcessId
us)
            (\(ProcessRegistrationException String
_ Maybe ProcessId
rPid) ->
                SendPort ProcessId -> ProcessId -> Process ()
forall a. Serializable a => SendPort a -> a -> Process ()
sendChan SendPort ProcessId
sigStart (ProcessId -> Process ()) -> ProcessId -> Process ()
forall a b. (a -> b) -> a -> b
$ Maybe ProcessId -> ProcessId
forall a. HasCallStack => Maybe a -> a
fromJust Maybe ProcessId
rPid)
    Process ()
proc
  [Match ProcessId] -> Process ProcessId
forall b. [Match b] -> Process b
receiveWait [
      (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process ProcessId)
-> Match ProcessId
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
ref ProcessId
_ DiedReason
_) -> MonitorRef
ref MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
mRef)
              (\(ProcessMonitorNotification MonitorRef
_ ProcessId
_ DiedReason
dr) -> ExitReason -> Process ProcessId
forall a b. Serializable a => a -> Process b
die (ExitReason -> Process ProcessId)
-> ExitReason -> Process ProcessId
forall a b. (a -> b) -> a -> b
$ String -> ExitReason
ExitOther (DiedReason -> String
forall a. Show a => a -> String
show DiedReason
dr))
    , ReceivePort ProcessId
-> (ProcessId -> Process ProcessId) -> Match ProcessId
forall a b. ReceivePort a -> (a -> Process b) -> Match b
matchChan ReceivePort ProcessId
recvStart ProcessId -> Process ProcessId
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return
    ] Process ProcessId -> Process () -> Process ProcessId
forall (m :: * -> *) a b.
(HasCallStack, MonadMask m) =>
m a -> m b -> m a
`finally` (MonitorRef -> Process ()
unmonitor MonitorRef
mRef)

-- | Helper function will register itself under a given name and send
-- result to given @Process@.
registerSelf :: (String, ProcessId) -> Process ()
registerSelf :: (String, ProcessId) -> Process ()
registerSelf (String
name,ProcessId
target) =
  do ProcessId
self <- Process ProcessId
getSelfPid
     String -> ProcessId -> Process ()
register String
name ProcessId
self
     ProcessId -> (RegisterSelf, ProcessId) -> Process ()
forall a. Serializable a => ProcessId -> a -> Process ()
send ProcessId
target (RegisterSelf
RegisterSelf, ProcessId
self)
     () <- Process ()
forall a. Serializable a => Process a
expect
     () -> Process ()
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

$(remotable ['registerSelf])

-- | A remote equivalent of 'whereisOrStart'. It deals with the
-- node registry on the given node, and the process, if it needs to be started,
-- will run on that node. If the node is inaccessible, Nothing will be returned.
--
whereisOrStartRemote :: NodeId -> String -> Closure (Process ()) -> Process (Maybe ProcessId)
whereisOrStartRemote :: NodeId
-> String -> Closure (Process ()) -> Process (Maybe ProcessId)
whereisOrStartRemote NodeId
nid String
name Closure (Process ())
proc =
     do MonitorRef
mRef <- NodeId -> Process MonitorRef
monitorNode NodeId
nid
        NodeId -> String -> Process ()
whereisRemoteAsync NodeId
nid String
name
        Maybe (Maybe ProcessId)
res <- [Match (Maybe (Maybe ProcessId))]
-> Process (Maybe (Maybe ProcessId))
forall b. [Match b] -> Process b
receiveWait
          [ (WhereIsReply -> Bool)
-> (WhereIsReply -> Process (Maybe (Maybe ProcessId)))
-> Match (Maybe (Maybe ProcessId))
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(WhereIsReply String
label Maybe ProcessId
_) -> String
label String -> String -> Bool
forall a. Eq a => a -> a -> Bool
== String
name)
                    (\(WhereIsReply String
_ Maybe ProcessId
mPid) -> Maybe (Maybe ProcessId) -> Process (Maybe (Maybe ProcessId))
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe ProcessId -> Maybe (Maybe ProcessId)
forall a. a -> Maybe a
Just Maybe ProcessId
mPid)),
            (NodeMonitorNotification -> Bool)
-> (NodeMonitorNotification -> Process (Maybe (Maybe ProcessId)))
-> Match (Maybe (Maybe ProcessId))
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(NodeMonitorNotification MonitorRef
aref NodeId
_ DiedReason
_) -> MonitorRef
aref MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
mRef)
                    (\(NodeMonitorNotification MonitorRef
_ NodeId
_ DiedReason
_) -> Maybe (Maybe ProcessId) -> Process (Maybe (Maybe ProcessId))
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe (Maybe ProcessId)
forall a. Maybe a
Nothing)
          ]
        case Maybe (Maybe ProcessId)
res of
           Maybe (Maybe ProcessId)
Nothing -> Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe ProcessId
forall a. Maybe a
Nothing
           Just (Just ProcessId
pid) -> MonitorRef -> Process ()
unmonitor MonitorRef
mRef Process ()
-> Process (Maybe ProcessId) -> Process (Maybe ProcessId)
forall a b. Process a -> Process b -> Process b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (ProcessId -> Maybe ProcessId
forall a. a -> Maybe a
Just ProcessId
pid)
           Just Maybe ProcessId
Nothing ->
              do ProcessId
self <- Process ProcessId
getSelfPid
                 SpawnRef
sRef <- NodeId -> Closure (Process ()) -> Process SpawnRef
spawnAsync NodeId
nid ($(mkClosure 'registerSelf) (String
name,ProcessId
self) Closure (Process ())
-> Closure (Process ()) -> Closure (Process ())
forall a b.
(Typeable a, Typeable b) =>
Closure (Process a) -> Closure (Process b) -> Closure (Process b)
`seqCP` Closure (Process ())
proc)
                 Maybe ProcessId
ret <- [Match (Maybe ProcessId)] -> Process (Maybe ProcessId)
forall b. [Match b] -> Process b
receiveWait [
                      (NodeMonitorNotification -> Bool)
-> (NodeMonitorNotification -> Process (Maybe ProcessId))
-> Match (Maybe ProcessId)
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(NodeMonitorNotification MonitorRef
ref NodeId
_ DiedReason
_) -> MonitorRef
ref MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
mRef)
                              (\(NodeMonitorNotification MonitorRef
_ NodeId
_ DiedReason
_) -> Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe ProcessId
forall a. Maybe a
Nothing),
                      (DidSpawn -> Bool)
-> (DidSpawn -> Process (Maybe ProcessId))
-> Match (Maybe ProcessId)
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(DidSpawn SpawnRef
ref ProcessId
_) -> SpawnRef
refSpawnRef -> SpawnRef -> Bool
forall a. Eq a => a -> a -> Bool
==SpawnRef
sRef )
                              (\(DidSpawn SpawnRef
_ ProcessId
pid) ->
                                  do MonitorRef
pRef <- ProcessId -> Process MonitorRef
P.monitor ProcessId
pid
                                     [Match (Maybe ProcessId)] -> Process (Maybe ProcessId)
forall b. [Match b] -> Process b
receiveWait
                                       [ ((RegisterSelf, ProcessId) -> Bool)
-> ((RegisterSelf, ProcessId) -> Process (Maybe ProcessId))
-> Match (Maybe ProcessId)
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(RegisterSelf
RegisterSelf, ProcessId
apid) -> ProcessId
apid ProcessId -> ProcessId -> Bool
forall a. Eq a => a -> a -> Bool
== ProcessId
pid)
                                                 (\(RegisterSelf
RegisterSelf, ProcessId
_) -> do MonitorRef -> Process ()
unmonitor MonitorRef
pRef
                                                                           ProcessId -> () -> Process ()
forall a. Serializable a => ProcessId -> a -> Process ()
send ProcessId
pid ()
                                                                           Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe ProcessId -> Process (Maybe ProcessId))
-> Maybe ProcessId -> Process (Maybe ProcessId)
forall a b. (a -> b) -> a -> b
$ ProcessId -> Maybe ProcessId
forall a. a -> Maybe a
Just ProcessId
pid),
                                         (NodeMonitorNotification -> Bool)
-> (NodeMonitorNotification -> Process (Maybe ProcessId))
-> Match (Maybe ProcessId)
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(NodeMonitorNotification MonitorRef
aref NodeId
_ DiedReason
_) -> MonitorRef
aref MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
mRef)
                                                 (\(NodeMonitorNotification MonitorRef
_aref NodeId
_ DiedReason
_) -> Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe ProcessId
forall a. Maybe a
Nothing),
                                         (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process (Maybe ProcessId))
-> Match (Maybe ProcessId)
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
ref ProcessId
_ DiedReason
_) -> MonitorRef
refMonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
==MonitorRef
pRef)
                                                 (\(ProcessMonitorNotification MonitorRef
_ ProcessId
_ DiedReason
_) -> Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe ProcessId
forall a. Maybe a
Nothing)
                                       ] )
                      ]
                 MonitorRef -> Process ()
unmonitor MonitorRef
mRef
                 case Maybe ProcessId
ret of
                   Maybe ProcessId
Nothing -> NodeId
-> String -> Closure (Process ()) -> Process (Maybe ProcessId)
whereisOrStartRemote NodeId
nid String
name Closure (Process ())
proc
                   Just ProcessId
pid -> Maybe ProcessId -> Process (Maybe ProcessId)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe ProcessId -> Process (Maybe ProcessId))
-> Maybe ProcessId -> Process (Maybe ProcessId)
forall a b. (a -> b) -> a -> b
$ ProcessId -> Maybe ProcessId
forall a. a -> Maybe a
Just ProcessId
pid

-- advanced messaging/matching

-- | An alternative to 'matchIf' that allows both predicate and action
-- to be expressed in one parameter.
matchCond :: (Serializable a) => (a -> Maybe (Process b)) -> Match b
matchCond :: forall a b. Serializable a => (a -> Maybe (Process b)) -> Match b
matchCond a -> Maybe (Process b)
cond =
   let v :: Maybe b -> (Bool, b)
v Maybe b
n = (Maybe b -> Bool
forall a. Maybe a -> Bool
isJust Maybe b
n, Maybe b -> b
forall a. HasCallStack => Maybe a -> a
fromJust Maybe b
n)
       res :: a -> (Bool, Process b)
res = Maybe (Process b) -> (Bool, Process b)
forall {b}. Maybe b -> (Bool, b)
v (Maybe (Process b) -> (Bool, Process b))
-> (a -> Maybe (Process b)) -> a -> (Bool, Process b)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Maybe (Process b)
cond
    in (a -> Bool) -> (a -> Process b) -> Match b
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf ((Bool, Process b) -> Bool
forall a b. (a, b) -> a
fst ((Bool, Process b) -> Bool)
-> (a -> (Bool, Process b)) -> a -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> (Bool, Process b)
res) ((Bool, Process b) -> Process b
forall a b. (a, b) -> b
snd ((Bool, Process b) -> Process b)
-> (a -> (Bool, Process b)) -> a -> Process b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> (Bool, Process b)
res)

-- | Safe (i.e., monitored) waiting on an expected response/message.
awaitResponse :: Addressable a
              => a
              -> [Match (Either ExitReason b)]
              -> Process (Either ExitReason b)
awaitResponse :: forall a b.
Addressable a =>
a -> [Match (Either ExitReason b)] -> Process (Either ExitReason b)
awaitResponse a
addr [Match (Either ExitReason b)]
matches = do
  Maybe ProcessId
mPid <- a -> Process (Maybe ProcessId)
forall a. Resolvable a => a -> Process (Maybe ProcessId)
resolve a
addr
  case Maybe ProcessId
mPid of
    Maybe ProcessId
Nothing -> Either ExitReason b -> Process (Either ExitReason b)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (Either ExitReason b -> Process (Either ExitReason b))
-> Either ExitReason b -> Process (Either ExitReason b)
forall a b. (a -> b) -> a -> b
$ ExitReason -> Either ExitReason b
forall a b. a -> Either a b
Left (ExitReason -> Either ExitReason b)
-> ExitReason -> Either ExitReason b
forall a b. (a -> b) -> a -> b
$ String -> ExitReason
ExitOther String
"UnresolvedAddress"
    Just ProcessId
p  ->
      Process MonitorRef
-> (MonitorRef -> Process ())
-> (MonitorRef -> Process (Either ExitReason b))
-> Process (Either ExitReason b)
forall a b c.
Process a -> (a -> Process b) -> (a -> Process c) -> Process c
bracket (ProcessId -> Process MonitorRef
P.monitor ProcessId
p)
              MonitorRef -> Process ()
P.unmonitor
              ((MonitorRef -> Process (Either ExitReason b))
 -> Process (Either ExitReason b))
-> (MonitorRef -> Process (Either ExitReason b))
-> Process (Either ExitReason b)
forall a b. (a -> b) -> a -> b
$ \MonitorRef
mRef -> [Match (Either ExitReason b)] -> Process (Either ExitReason b)
forall b. [Match b] -> Process b
receiveWait ((MonitorRef -> Match (Either ExitReason b)
forall b. MonitorRef -> Match (Either ExitReason b)
matchRef MonitorRef
mRef)Match (Either ExitReason b)
-> [Match (Either ExitReason b)] -> [Match (Either ExitReason b)]
forall a. a -> [a] -> [a]
:[Match (Either ExitReason b)]
matches)
  where
    matchRef :: MonitorRef -> Match (Either ExitReason b)
    matchRef :: forall b. MonitorRef -> Match (Either ExitReason b)
matchRef MonitorRef
r = (ProcessMonitorNotification -> Bool)
-> (ProcessMonitorNotification -> Process (Either ExitReason b))
-> Match (Either ExitReason b)
forall a b.
Serializable a =>
(a -> Bool) -> (a -> Process b) -> Match b
matchIf (\(ProcessMonitorNotification MonitorRef
r' ProcessId
_ DiedReason
_) -> MonitorRef
r MonitorRef -> MonitorRef -> Bool
forall a. Eq a => a -> a -> Bool
== MonitorRef
r')
                         (\(ProcessMonitorNotification MonitorRef
_ ProcessId
_ DiedReason
d) -> do
                             Either ExitReason b -> Process (Either ExitReason b)
forall a. a -> Process a
forall (m :: * -> *) a. Monad m => a -> m a
return (ExitReason -> Either ExitReason b
forall a b. a -> Either a b
Left (String -> ExitReason
ExitOther (DiedReason -> String
forall a. Show a => a -> String
show DiedReason
d))))