@@ -9,7 +9,7 @@ module Control.Distributed.Process.Node
99 ) where
1010
1111import Prelude hiding (catch)
12−import System.IO (fixIO)
12+import System.IO (fixIO, hPutStrLn, stderr)
1313import qualified Data.ByteString.Lazy as BSL (fromChunks)
1414import Data.Binary (decode)
1515import Data.Map (Map)
@@ -120,6 +120,7 @@ import Control.Distributed.Process.Internal.Closure
120120 , resolveClosure
121121 )
122122import Control.Distributed.Process.Internal.Node (runLocalProcess)
123+import Control.Distributed.Process.Internal.Primitives (expect, register)
123124
124125--------------------------------------------------------------------------------
125126-- Initialization --
@@ -131,26 +132,42 @@ import Control.Distributed.Process.Internal.Node (runLocalProcess)
131132-- to do.
132133newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
133134newLocalNode transport rtable = do
134− mEndPoint <- NT.newEndPoint transport
135− case mEndPoint of
136− Left ex -> throwIO ex
137− Right endPoint -> do
138− unq <- randomIO
139− state <- newMVar LocalNodeState
140− { _localProcesses = Map.empty
141− , _localPidCounter = 0
142− , _localPidUnique = unq
143− }
144− ctrlChan <- newChan
145− let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
146− , localEndPoint = endPoint
147− , localState = state
148− , localCtrlChan = ctrlChan
149− , remoteTable = rtable
150− }
151− void . forkIO $ runNodeController node
152− void . forkIO $ handleIncomingMessages node
153− return node
135+ mEndPoint <- NT.newEndPoint transport
136+ case mEndPoint of
137+ Left ex -> throwIO ex
138+ Right endPoint -> do
139+ localNode <- createBareLocalNode endPoint rtable
140+ startServiceProcesses localNode
141+ return localNode
142+
143+-- | Create a new local node (without any service processes running)
144+createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
145+createBareLocalNode endPoint rtable = do
146+ unq <- randomIO
147+ state <- newMVar LocalNodeState
148+ { _localProcesses = Map.empty
149+ , _localPidCounter = 0
150+ , _localPidUnique = unq
151+ }
152+ ctrlChan <- newChan
153+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
154+ , localEndPoint = endPoint
155+ , localState = state
156+ , localCtrlChan = ctrlChan
157+ , remoteTable = rtable
158+ }
159+ void . forkIO $ runNodeController node
160+ void . forkIO $ handleIncomingMessages node
161+ return node
162+
163+-- | Start and register the service processes on a node
164+-- (for now, this is only the logger)
165+startServiceProcesses :: LocalNode -> IO ()
166+startServiceProcesses node = do
167+ logger <- forkProcess node . forever $ do
168+ (time, pid, string) <- expect :: Process (String, ProcessId, String)
169+ liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
170+ runProcess node $ register "logger" logger
154171
155172-- | Force-close a local node
156173--