@@ -1,130 +0,0 @@
1−-- | Add message sending capability to a monad
3−-- NOTE: Not thread-safe (you should not do concurrent sends within the same
5−module Control.Distributed.Process.Internal.MessageT
16−import Data.Binary (Binary, encode)
17−import qualified Data.ByteString as BSS (ByteString)
18−import qualified Data.ByteString.Lazy as BSL (toChunks)
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
31− , NodeId(nodeAddress)
32− , LocalNode(localCtrlChan, localEndPoint)
33− , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
34− , DiedReason(DiedDisconnect)
35− , ProcessSignal(Died)
42−import Control.Distributed.Process.Serializable (Serializable)
43−import qualified Network.Transport as NT
48− , Reliability(ReliableOrdered)
49− , defaultConnectHints
52−--------------------------------------------------------------------------------
54−--------------------------------------------------------------------------------
56−runMessageT :: Monad m => LocalNode -> MessageT m a -> m a
57−runMessageT localNode m =
58− evalStateT (unMessageT m) $ initMessageState localNode
60−getLocalNode :: Monad m => MessageT m LocalNode
61−getLocalNode = gets messageLocalNode
63−sendPayload :: MonadIO m => Identifier -> [BSS.ByteString] -> MessageT m ()
64−sendPayload them payload = do
66− didSend <- case mConn of
68− didSend <- liftIO $ NT.send conn payload
70− Left _ -> return False
71− Right _ -> return True
72− Nothing -> return False
75− liftIO . writeChan (localCtrlChan node) $ NCMsg
76− { ctrlMsgSender = them
77− , ctrlMsgSignal = Died them DiedDisconnect
80−sendBinary :: (MonadIO m, Binary a) => Identifier -> a -> MessageT m ()
81−sendBinary them = sendPayload them . BSL.toChunks . encode
83−sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()
84−sendMessage them = sendPayload them . messageToPayload . createMessage
86−--------------------------------------------------------------------------------
88−--------------------------------------------------------------------------------
90−initMessageState :: LocalNode -> MessageState
91−initMessageState localNode = MessageState {
92− messageLocalNode = localNode
93− , _messageConnections = Map.empty
96−setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
98− endPoint <- localEndPoint `liftM` getLocalNode
99− mConn <- liftIO $ NT.connect endPoint
100− (nodeAddress . nodeOf $ them)
102− NT.defaultConnectHints
105− didSend <- liftIO $ NT.send conn (BSL.toChunks . encode $ them)
110− modify $ messageConnectionTo them ^= Just conn
115−connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
117− mConn <- gets (^. messageConnectionTo them)
119− Just conn -> return $ Just conn
120− Nothing -> setupConnTo them
122−--------------------------------------------------------------------------------
124−--------------------------------------------------------------------------------
126−messageConnections :: Accessor MessageState (Map Identifier NT.Connection)
127−messageConnections = accessor _messageConnections (\conns st -> st { _messageConnections = conns })
129−messageConnectionTo :: Identifier -> Accessor MessageState (Maybe NT.Connection)
130−messageConnectionTo them = messageConnections >>> DAC.mapMaybe them