@@ -6,9 +6,12 @@ import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
66import Control.Applicative ((<$>))
77import Control.Category ((>>>))
88import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)
9+import Control.Exception (throw)
910import Control.Monad (forM_, when)
1011import Data.Map (Map)
1112import qualified Data.Map as Map (empty, insert, size, delete, findWithDefault)
13+import Data.Set (Set)
14+import qualified Data.Set as Set (empty, elems, insert, delete)
1215import Data.ByteString (ByteString)
1316import qualified Data.ByteString.Char8 as BSC (pack)
1417import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
@@ -17,30 +20,9 @@ import qualified Data.Accessor.Container as DAC (mapMaybe)
1720-- Global state: next available "address", mapping from addresses to channels and next available connection
1821data TransportState = State { _channels :: Map EndPointAddress (Chan Event)
1922 , _nextConnectionId :: Map EndPointAddress ConnectionId
20− , _multigroups :: Map MulticastAddress (MVar [Chan Event])
23+ , _multigroups :: Map MulticastAddress (MVar (Set EndPointAddress))
2124 }
2225
23−channels :: Accessor TransportState (Map EndPointAddress (Chan Event))
24−channels = accessor _channels (\ch st -> st { _channels = ch })
25−
26−nextConnectionId :: Accessor TransportState (Map EndPointAddress ConnectionId)
27−nextConnectionId = accessor _nextConnectionId (\cid st -> st { _nextConnectionId = cid })
28−
29−multigroups :: Accessor TransportState (Map MulticastAddress (MVar [Chan Event]))
30−multigroups = accessor _multigroups (\gs st -> st { _multigroups = gs })
31−
32−at :: Ord k => k -> String -> Accessor (Map k v) v
33−at k err = accessor (Map.findWithDefault (error err) k) (Map.insert k)
34−
35−channelAt :: EndPointAddress -> Accessor TransportState (Chan Event)
36−channelAt addr = channels >>> at addr "Invalid channel"
37−
38−nextConnectionIdAt :: EndPointAddress -> Accessor TransportState ConnectionId
39−nextConnectionIdAt addr = nextConnectionId >>> at addr "Invalid connection ID"
40−
41−multigroupAt :: MulticastAddress -> Accessor TransportState (MVar [Chan Event])
42−multigroupAt addr = multigroups >>> at addr "Invalid multigroup"
43−
4426-- | Create a new Transport.
4527--
4628-- Only a single transport should be created per Haskell process
@@ -51,7 +33,9 @@ createTransport = do
5133 , _nextConnectionId = Map.empty
5234 , _multigroups = Map.empty
5335 }
54− return Transport { newEndPoint = apiNewEndPoint state }
36+ return Transport { newEndPoint = apiNewEndPoint state
37+ , closeTransport = throw (userError "closeEndPoint not implemented")
38+ }
5539
5640-- | Create a new end point
5741apiNewEndPoint :: MVar TransportState -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
@@ -60,11 +44,12 @@ apiNewEndPoint state = do
6044 addr <- modifyMVar state $ \st -> do
6145 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels
6246 return ((channelAt addr ^= chan) . (nextConnectionIdAt addr ^= 1) $ st, addr)
63− return . Right $ EndPoint { receive = readChan chan
64− , address = addr
65− , connect = apiConnect addr state
66− , newMulticastGroup = apiNewMulticastGroup state chan
67− , resolveMulticastGroup = apiResolveMulticastGroup state chan
47+ return . Right $ EndPoint { receive = readChan chan
48+ , address = addr
49+ , connect = apiConnect addr state
50+ , closeEndPoint = throw (userError "closeEndPoint not implemented")
51+ , newMulticastGroup = apiNewMulticastGroup state addr
52+ , resolveMulticastGroup = apiResolveMulticastGroup state addr
6853 }
6954
7055-- | Create a new connection
@@ -97,13 +82,13 @@ apiClose chan conn connAlive =
9782 return False
9883
9984-- | Create a new multicast group
100−apiNewMulticastGroup :: MVar TransportState -> Chan Event -> IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)
101−apiNewMulticastGroup state endpoint = do
102− group <- newMVar []
103− addr <- modifyMVar state $ \st -> do
85+apiNewMulticastGroup :: MVar TransportState -> EndPointAddress -> IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)
86+apiNewMulticastGroup state ourAddress = do
87+ group <- newMVar Set.empty
88+ groupAddr <- modifyMVar state $ \st -> do
10489 let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups
10590 return (multigroupAt addr ^= group $ st, addr)
106− return . Right $ createMulticastGroup state endpoint addr group
91+ return . Right $ createMulticastGroup state ourAddress groupAddr group
10792
10893-- | Construct a multicast group
10994--
@@ -111,30 +96,55 @@ apiNewMulticastGroup state endpoint = do
11196-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact
11297-- that some multicast messages may still be in transit when the group is
11398-- deleted.
114−--
115−-- TODO: subscribing twice means you will receive messages twice, but as soon
116−-- as you unsubscribe you will stop receiving all messages. Not sure if this
117−-- is the right behaviour.
118−createMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> MVar [Chan Event] -> MulticastGroup
119−createMulticastGroup state endpoint addr group =
120− MulticastGroup { multicastAddress = addr
121− , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^: Map.delete addr)
99+createMulticastGroup :: MVar TransportState -> EndPointAddress -> MulticastAddress -> MVar (Set EndPointAddress) -> MulticastGroup
100+createMulticastGroup state ourAddress groupAddress group =
101+ MulticastGroup { multicastAddress = groupAddress
102+ , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^: Map.delete groupAddress)
122103 , maxMsgSize = Nothing
123104 , multicastSend = \payload -> do
124− cs <- readMVar group
125− forM_ cs $ \ch -> writeChan ch (ReceivedMulticast addr payload)
126− , multicastSubscribe = modifyMVar_ group $ return . (endpoint :)
127− , multicastUnsubscribe = modifyMVar_ group $ return . filter (/= endpoint)
105+ cs <- (^. channels) <$> readMVar state
106+ es <- readMVar group
107+ forM_ (Set.elems es) $ \ep -> do
108+ let ch = (cs ^. at ep "Invalid endpoint")
109+ writeChan ch (ReceivedMulticast groupAddress payload)
110+ , multicastSubscribe = modifyMVar_ group $ return . (Set.insert ourAddress)
111+ , multicastUnsubscribe = modifyMVar_ group $ return . (Set.delete ourAddress)
128112 , multicastClose = return ()
129113 }
130114
131115-- | Resolve a multicast group
132116apiResolveMulticastGroup :: MVar TransportState
133− -> Chan Event
134− -> MulticastAddress
135− -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)
136−apiResolveMulticastGroup state endpoint addr = do
137− group <- (^. (multigroups >>> DAC.mapMaybe addr)) <$> readMVar state
117+ -> EndPointAddress
118+ -> MulticastAddress
119+ -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)
120+apiResolveMulticastGroup state ourAddress groupAddress = do
121+ group <- (^. (multigroups >>> DAC.mapMaybe groupAddress)) <$> readMVar state
138122 case group of
139− Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show addr ++ " not found")
140− Just mvar -> return . Right $ createMulticastGroup state endpoint addr mvar
123+ Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show groupAddress ++ " not found")
124+ Just mvar -> return . Right $ createMulticastGroup state ourAddress groupAddress mvar
125+
126+--------------------------------------------------------------------------------
127+-- Lens definitions --
128+--------------------------------------------------------------------------------
129+
130+channels :: Accessor TransportState (Map EndPointAddress (Chan Event))
131+channels = accessor _channels (\ch st -> st { _channels = ch })
132+
133+nextConnectionId :: Accessor TransportState (Map EndPointAddress ConnectionId)
134+nextConnectionId = accessor _nextConnectionId (\cid st -> st { _nextConnectionId = cid })
135+
136+multigroups :: Accessor TransportState (Map MulticastAddress (MVar (Set EndPointAddress)))
137+multigroups = accessor _multigroups (\gs st -> st { _multigroups = gs })
138+
139+at :: Ord k => k -> String -> Accessor (Map k v) v
140+at k err = accessor (Map.findWithDefault (error err) k) (Map.insert k)
141+
142+channelAt :: EndPointAddress -> Accessor TransportState (Chan Event)
143+channelAt addr = channels >>> at addr "Invalid channel"
144+
145+nextConnectionIdAt :: EndPointAddress -> Accessor TransportState ConnectionId
146+nextConnectionIdAt addr = nextConnectionId >>> at addr "Invalid connection ID"
147+
148+multigroupAt :: MulticastAddress -> Accessor TransportState (MVar (Set EndPointAddress))
149+multigroupAt addr = multigroups >>> at addr "Invalid multigroup"
150+