@@ -1,130 +0,0 @@
1−-- | Add message sending capability to a monad
2−--
3−-- NOTE: Not thread-safe (you should not do concurrent sends within the same
4−-- monad).
5−module Control.Distributed.Process.Internal.MessageT
6− ( MessageT
7− , runMessageT
8− , getLocalNode
9− , sendPayload
10− , sendBinary
11− , sendMessage
12− , payloadToMessage
13− , createMessage
14− ) where
15−
16−import Data.Binary (Binary, encode)
17−import qualified Data.ByteString as BSS (ByteString)
18−import qualified Data.ByteString.Lazy as BSL (toChunks)
19−import Data.Map (Map)
20−import qualified Data.Map as Map (empty)
21−import Data.Accessor (Accessor, accessor, (^=), (^.))
22−import qualified Data.Accessor.Container as DAC (mapMaybe)
23−import Control.Category ((>>>))
24−import Control.Monad (unless, liftM)
25−import Control.Monad.State (gets, modify, evalStateT)
26−import Control.Monad.IO.Class (MonadIO, liftIO)
27−import Control.Concurrent.Chan (writeChan)
28−import Control.Distributed.Process.Internal.Types
29− ( Identifier(..)
30− , nodeOf
31− , NodeId(nodeAddress)
32− , LocalNode(localCtrlChan, localEndPoint)
33− , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
34− , DiedReason(DiedDisconnect)
35− , ProcessSignal(Died)
36− , MessageT(..)
37− , MessageState(..)
38− , createMessage
39− , messageToPayload
40− , payloadToMessage
41− )
42−import Control.Distributed.Process.Serializable (Serializable)
43−import qualified Network.Transport as NT
44− ( EndPoint
45− , Connection
46− , connect
47− , send
48− , Reliability(ReliableOrdered)
49− , defaultConnectHints
50− )
51−
52−--------------------------------------------------------------------------------
53−-- API --
54−--------------------------------------------------------------------------------
55−
56−runMessageT :: Monad m => LocalNode -> MessageT m a -> m a
57−runMessageT localNode m =
58− evalStateT (unMessageT m) $ initMessageState localNode
59−
60−getLocalNode :: Monad m => MessageT m LocalNode
61−getLocalNode = gets messageLocalNode
62−
63−sendPayload :: MonadIO m => Identifier -> [BSS.ByteString] -> MessageT m ()
64−sendPayload them payload = do
65− mConn <- connTo them
66− didSend <- case mConn of
67− Just conn -> do
68− didSend <- liftIO $ NT.send conn payload
69− case didSend of
70− Left _ -> return False
71− Right _ -> return True
72− Nothing -> return False
73− unless didSend $ do
74− node <- getLocalNode
75− liftIO . writeChan (localCtrlChan node) $ NCMsg
76− { ctrlMsgSender = them
77− , ctrlMsgSignal = Died them DiedDisconnect
78− }
79−
80−sendBinary :: (MonadIO m, Binary a) => Identifier -> a -> MessageT m ()
81−sendBinary them = sendPayload them . BSL.toChunks . encode
82−
83−sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()
84−sendMessage them = sendPayload them . messageToPayload . createMessage
85−
86−--------------------------------------------------------------------------------
87−-- Internal --
88−--------------------------------------------------------------------------------
89−
90−initMessageState :: LocalNode -> MessageState
91−initMessageState localNode = MessageState {
92− messageLocalNode = localNode
93− , _messageConnections = Map.empty
94− }
95−
96−setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
97−setupConnTo them = do
98− endPoint <- localEndPoint `liftM` getLocalNode
99− mConn <- liftIO $ NT.connect endPoint
100− (nodeAddress . nodeOf $ them)
101− NT.ReliableOrdered
102− NT.defaultConnectHints
103− case mConn of
104− Right conn -> do
105− didSend <- liftIO $ NT.send conn (BSL.toChunks . encode $ them)
106− case didSend of
107− Left _ ->
108− return Nothing
109− Right () -> do
110− modify $ messageConnectionTo them ^= Just conn
111− return $ Just conn
112− Left _ ->
113− return Nothing
114−
115−connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
116−connTo them = do
117− mConn <- gets (^. messageConnectionTo them)
118− case mConn of
119− Just conn -> return $ Just conn
120− Nothing -> setupConnTo them
121−
122−--------------------------------------------------------------------------------
123−-- Accessors --
124−--------------------------------------------------------------------------------
125−
126−messageConnections :: Accessor MessageState (Map Identifier NT.Connection)
127−messageConnections = accessor _messageConnections (\conns st -> st { _messageConnections = conns })
128−
129−messageConnectionTo :: Identifier -> Accessor MessageState (Maybe NT.Connection)
130−messageConnectionTo them = messageConnections >>> DAC.mapMaybe them