64d211c · distributed-process

6import Control.Applicative ((<$>))

7import Control.Category ((>>>))

8import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)

9−import Control.Monad (forM_)

9+import Control.Monad (forM_, when)

10import Data.Map (Map)

11import qualified Data.Map as Map (empty, insert, size, delete, findWithDefault)

12+import Data.ByteString (ByteString)

12import qualified Data.ByteString.Char8 as BSC (pack)

13import qualified Data.ByteString.Char8 as BSC (pack)

13import Data.Lens.Lazy (Lens, lens, (^.), (^%=), (^=), (^+=), mapLens)

14import Data.Lens.Lazy (Lens, lens, (^.), (^%=), (^=), (^+=), mapLens)

14

15

49 , _nextConnectionId = Map.empty

50 , _nextConnectionId = Map.empty

50 , _multigroups = Map.empty

51 , _multigroups = Map.empty

51 }

52 }

52− return Transport { newEndPoint = chanNewEndPoint state }

53+ return Transport { newEndPoint = apiNewEndPoint state }

53

54

54−-- Create a new end point

55+-- | Create a new end point

55−chanNewEndPoint :: MVar TransportState -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)

56+apiNewEndPoint :: MVar TransportState -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)

56−chanNewEndPoint state = do

57+apiNewEndPoint state = do

57 chan <- newChan

58 chan <- newChan

58 addr <- modifyMVar state $ \st -> do

59 addr <- modifyMVar state $ \st -> do

59 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels

60 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels

60 return ((channelAt addr ^= chan) . (nextConnectionIdAt addr ^= 1) $ st, addr)

61 return ((channelAt addr ^= chan) . (nextConnectionIdAt addr ^= 1) $ st, addr)

61 return . Right $ EndPoint { receive = readChan chan

62 return . Right $ EndPoint { receive = readChan chan

62 , address = addr

63 , address = addr

63− , connect = chanConnect addr state

64+ , connect = apiConnect addr state

64− , newMulticastGroup = chanNewMulticastGroup state chan

65+ , newMulticastGroup = apiNewMulticastGroup state chan

65− , resolveMulticastGroup = chanResolveMulticastGroup state chan

66+ , resolveMulticastGroup = apiResolveMulticastGroup state chan

66 }

67 }

67

68

68−-- Create a new connection

69+-- | Create a new connection

69−chanConnect :: EndPointAddress -> MVar TransportState -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)

70+apiConnect :: EndPointAddress -> MVar TransportState -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)

70−chanConnect myAddress state theirAddress _ = do

71+apiConnect myAddress state theirAddress _ = do

71 (chan, conn) <- modifyMVar state $ \st -> do

72 (chan, conn) <- modifyMVar state $ \st -> do

72 let chan = st ^. channelAt theirAddress

73 let chan = st ^. channelAt theirAddress

73 let conn = st ^. nextConnectionIdAt theirAddress

74 let conn = st ^. nextConnectionIdAt theirAddress

74 return (nextConnectionIdAt theirAddress ^+= 1 $ st, (chan, conn))

75 return (nextConnectionIdAt theirAddress ^+= 1 $ st, (chan, conn))

75 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress

76 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress

77+ connAlive <- newMVar True

76− return . Right $ Connection { send = \msg -> do writeChan chan (Received conn msg)

78+ return . Right $ Connection { send = apiSend chan conn connAlive

77− return (Right ())

79+ , close = apiClose chan conn connAlive

78− , close = writeChan chan $ ConnectionClosed conn

79 }

80 }

80

81

82+-- | Send a message over a connection

83+apiSend :: Chan Event -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())

84+apiSend chan conn connAlive msg =

85+ modifyMVar connAlive $ \alive -> do

86+ if alive

87+ then do writeChan chan (Received conn msg)

88+ return (alive, Right ())

89+ else do return (alive, Left (FailedWith SendConnectionClosed "Connection closed"))

90+

91+-- | Close a connection

92+apiClose :: Chan Event -> ConnectionId -> MVar Bool -> IO ()

93+apiClose chan conn connAlive =

94+ modifyMVar_ connAlive $ \alive -> do

95+ when alive . writeChan chan $ ConnectionClosed conn

96+ return False

97+

81−-- Create a new multicast group

98+-- | Create a new multicast group

82−chanNewMulticastGroup :: MVar TransportState -> Chan Event -> IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)

99+apiNewMulticastGroup :: MVar TransportState -> Chan Event -> IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)

83−chanNewMulticastGroup state endpoint = do

100+apiNewMulticastGroup state endpoint = do

84 group <- newMVar []

101 group <- newMVar []

85 addr <- modifyMVar state $ \st -> do

102 addr <- modifyMVar state $ \st -> do

86 let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups

103 let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups

87 return (multigroupAt addr ^= group $ st, addr)

104 return (multigroupAt addr ^= group $ st, addr)

88− return . Right $ chanMulticastGroup state endpoint addr group

105+ return . Right $ createMulticastGroup state endpoint addr group

89

106

90−-- Construct a multicast group

107+-- | Construct a multicast group

91--

108--

92-- When the group is deleted some endpoints may still receive messages, but

109-- When the group is deleted some endpoints may still receive messages, but

93-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact

110-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact

97-- TODO: subscribing twice means you will receive messages twice, but as soon

114-- TODO: subscribing twice means you will receive messages twice, but as soon

98-- as you unsubscribe you will stop receiving all messages. Not sure if this

115-- as you unsubscribe you will stop receiving all messages. Not sure if this

99-- is the right behaviour.

116-- is the right behaviour.

100−chanMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> MVar [Chan Event] -> MulticastGroup

117+createMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> MVar [Chan Event] -> MulticastGroup

101−chanMulticastGroup state endpoint addr group =

118+createMulticastGroup state endpoint addr group =

102 MulticastGroup { multicastAddress = addr

119 MulticastGroup { multicastAddress = addr

103 , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^%= Map.delete addr)

120 , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^%= Map.delete addr)

104 , maxMsgSize = Nothing

121 , maxMsgSize = Nothing

110 , multicastClose = return ()

127 , multicastClose = return ()

111 }

128 }

112

129

113−-- Resolve a multicast group

130+-- | Resolve a multicast group

114−chanResolveMulticastGroup :: MVar TransportState

131+apiResolveMulticastGroup :: MVar TransportState

115 -> Chan Event

132 -> Chan Event

116 -> MulticastAddress

133 -> MulticastAddress

117 -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)

134 -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)

118−chanResolveMulticastGroup state endpoint addr = do

135+apiResolveMulticastGroup state endpoint addr = do

119 group <- (^. (multigroups >>> mapLens addr)) <$> readMVar state

136 group <- (^. (multigroups >>> mapLens addr)) <$> readMVar state

120 case group of

137 case group of

121 Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show addr ++ " not found")

138 Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show addr ++ " not found")

122− Just mvar -> return . Right $ chanMulticastGroup state endpoint addr mvar

139+ Just mvar -> return . Right $ createMulticastGroup state endpoint addr mvar