@@ -6,9 +6,12 @@ import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
6import Control.Applicative ((<$>))
6import Control.Applicative ((<$>))
7import Control.Category ((>>>))
7import Control.Category ((>>>))
8import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)
8import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)
9+import Control.Exception (throw)
9import Control.Monad (forM_, when)
10import Control.Monad (forM_, when)
10import Data.Map (Map)
11import Data.Map (Map)
11import qualified Data.Map as Map (empty, insert, size, delete, findWithDefault)
12import qualified Data.Map as Map (empty, insert, size, delete, findWithDefault)
14+import qualified Data.Set as Set (empty, elems, insert, delete)
12import Data.ByteString (ByteString)
15import Data.ByteString (ByteString)
13import qualified Data.ByteString.Char8 as BSC (pack)
16import qualified Data.ByteString.Char8 as BSC (pack)
14import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
17import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
@@ -17,30 +20,9 @@ import qualified Data.Accessor.Container as DAC (mapMaybe)
17-- Global state: next available "address", mapping from addresses to channels and next available connection
20-- Global state: next available "address", mapping from addresses to channels and next available connection
18data TransportState = State { _channels :: Map EndPointAddress (Chan Event)
21data TransportState = State { _channels :: Map EndPointAddress (Chan Event)
19 , _nextConnectionId :: Map EndPointAddress ConnectionId
22 , _nextConnectionId :: Map EndPointAddress ConnectionId
20− , _multigroups :: Map MulticastAddress (MVar [Chan Event])
23+ , _multigroups :: Map MulticastAddress (MVar (Set EndPointAddress))
23−channels :: Accessor TransportState (Map EndPointAddress (Chan Event))
24−channels = accessor _channels (\ch st -> st { _channels = ch })
26−nextConnectionId :: Accessor TransportState (Map EndPointAddress ConnectionId)
27−nextConnectionId = accessor _nextConnectionId (\cid st -> st { _nextConnectionId = cid })
29−multigroups :: Accessor TransportState (Map MulticastAddress (MVar [Chan Event]))
30−multigroups = accessor _multigroups (\gs st -> st { _multigroups = gs })
32−at :: Ord k => k -> String -> Accessor (Map k v) v
33−at k err = accessor (Map.findWithDefault (error err) k) (Map.insert k)
35−channelAt :: EndPointAddress -> Accessor TransportState (Chan Event)
36−channelAt addr = channels >>> at addr "Invalid channel"
38−nextConnectionIdAt :: EndPointAddress -> Accessor TransportState ConnectionId
39−nextConnectionIdAt addr = nextConnectionId >>> at addr "Invalid connection ID"
41−multigroupAt :: MulticastAddress -> Accessor TransportState (MVar [Chan Event])
42−multigroupAt addr = multigroups >>> at addr "Invalid multigroup"
44-- | Create a new Transport.
26-- | Create a new Transport.
46-- Only a single transport should be created per Haskell process
28-- Only a single transport should be created per Haskell process
@@ -51,7 +33,9 @@ createTransport = do
51 , _nextConnectionId = Map.empty
33 , _nextConnectionId = Map.empty
52 , _multigroups = Map.empty
34 , _multigroups = Map.empty
54− return Transport { newEndPoint = apiNewEndPoint state }
36+ return Transport { newEndPoint = apiNewEndPoint state
37+ , closeTransport = throw (userError "closeEndPoint not implemented")
56-- | Create a new end point
40-- | Create a new end point
57apiNewEndPoint :: MVar TransportState -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
41apiNewEndPoint :: MVar TransportState -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
@@ -60,11 +44,12 @@ apiNewEndPoint state = do
60 addr <- modifyMVar state $ \st -> do
44 addr <- modifyMVar state $ \st -> do
61 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels
45 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels
62 return ((channelAt addr ^= chan) . (nextConnectionIdAt addr ^= 1) $ st, addr)
46 return ((channelAt addr ^= chan) . (nextConnectionIdAt addr ^= 1) $ st, addr)
63− return . Right $ EndPoint { receive = readChan chan
47+ return . Right $ EndPoint { receive = readChan chan
64− , address = addr
48+ , address = addr
65− , connect = apiConnect addr state
49+ , connect = apiConnect addr state
50+ , closeEndPoint = throw (userError "closeEndPoint not implemented")
66− , newMulticastGroup = apiNewMulticastGroup state chan
51+ , newMulticastGroup = apiNewMulticastGroup state addr
67− , resolveMulticastGroup = apiResolveMulticastGroup state chan
52+ , resolveMulticastGroup = apiResolveMulticastGroup state addr
70-- | Create a new connection
55-- | Create a new connection
@@ -97,13 +82,13 @@ apiClose chan conn connAlive =
97 return False
82 return False
99-- | Create a new multicast group
84-- | Create a new multicast group
100−apiNewMulticastGroup :: MVar TransportState -> Chan Event -> IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)
85+apiNewMulticastGroup :: MVar TransportState -> EndPointAddress -> IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)
101−apiNewMulticastGroup state endpoint = do
86+apiNewMulticastGroup state ourAddress = do
102− group <- newMVar []
87+ group <- newMVar Set.empty
103− addr <- modifyMVar state $ \st -> do
88+ groupAddr <- modifyMVar state $ \st -> do
104 let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups
89 let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups
105 return (multigroupAt addr ^= group $ st, addr)
90 return (multigroupAt addr ^= group $ st, addr)
106− return . Right $ createMulticastGroup state endpoint addr group
91+ return . Right $ createMulticastGroup state ourAddress groupAddr group
108-- | Construct a multicast group
93-- | Construct a multicast group
@@ -111,30 +96,55 @@ apiNewMulticastGroup state endpoint = do
111-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact
96-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact
112-- that some multicast messages may still be in transit when the group is
97-- that some multicast messages may still be in transit when the group is
113-- deleted.
98-- deleted.
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
99+createMulticastGroup :: MVar TransportState -> EndPointAddress -> MulticastAddress -> MVar (Set EndPointAddress) -> MulticastGroup
119−createMulticastGroup state endpoint addr group =
100+createMulticastGroup state ourAddress groupAddress group =
120− MulticastGroup { multicastAddress = addr
101+ MulticastGroup { multicastAddress = groupAddress
121− , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^: Map.delete addr)
102+ , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^: Map.delete groupAddress)
122 , maxMsgSize = Nothing
103 , maxMsgSize = Nothing
123 , multicastSend = \payload -> do
104 , multicastSend = \payload -> do
105+ cs <- (^. channels) <$> readMVar state
124− cs <- readMVar group
106+ es <- readMVar group
107+ forM_ (Set.elems es) $ \ep -> do
108+ let ch = (cs ^. at ep "Invalid endpoint")
125− forM_ cs $ \ch -> writeChan ch (ReceivedMulticast addr payload)
109+ writeChan ch (ReceivedMulticast groupAddress payload)
126− , multicastSubscribe = modifyMVar_ group $ return . (endpoint :)
110+ , multicastSubscribe = modifyMVar_ group $ return . (Set.insert ourAddress)
127− , multicastUnsubscribe = modifyMVar_ group $ return . filter (/= endpoint)
111+ , multicastUnsubscribe = modifyMVar_ group $ return . (Set.delete ourAddress)
128 , multicastClose = return ()
112 , multicastClose = return ()
131-- | Resolve a multicast group
115-- | Resolve a multicast group
132apiResolveMulticastGroup :: MVar TransportState
116apiResolveMulticastGroup :: MVar TransportState
133− -> Chan Event
117+ -> EndPointAddress
134− -> MulticastAddress
118+ -> MulticastAddress
135− -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)
119+ -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)
136−apiResolveMulticastGroup state endpoint addr = do
120+apiResolveMulticastGroup state ourAddress groupAddress = do
137− group <- (^. (multigroups >>> DAC.mapMaybe addr)) <$> readMVar state
121+ group <- (^. (multigroups >>> DAC.mapMaybe groupAddress)) <$> readMVar state
138 case group of
122 case group of
139− Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show addr ++ " not found")
123+ Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show groupAddress ++ " not found")
140− Just mvar -> return . Right $ createMulticastGroup state endpoint addr mvar
124+ Just mvar -> return . Right $ createMulticastGroup state ourAddress groupAddress mvar
126+--------------------------------------------------------------------------------
127+-- Lens definitions --
128+--------------------------------------------------------------------------------
130+channels :: Accessor TransportState (Map EndPointAddress (Chan Event))
131+channels = accessor _channels (\ch st -> st { _channels = ch })
133+nextConnectionId :: Accessor TransportState (Map EndPointAddress ConnectionId)
134+nextConnectionId = accessor _nextConnectionId (\cid st -> st { _nextConnectionId = cid })
136+multigroups :: Accessor TransportState (Map MulticastAddress (MVar (Set EndPointAddress)))
137+multigroups = accessor _multigroups (\gs st -> st { _multigroups = gs })
139+at :: Ord k => k -> String -> Accessor (Map k v) v
140+at k err = accessor (Map.findWithDefault (error err) k) (Map.insert k)
142+channelAt :: EndPointAddress -> Accessor TransportState (Chan Event)
143+channelAt addr = channels >>> at addr "Invalid channel"
145+nextConnectionIdAt :: EndPointAddress -> Accessor TransportState ConnectionId
146+nextConnectionIdAt addr = nextConnectionId >>> at addr "Invalid connection ID"
148+multigroupAt :: MulticastAddress -> Accessor TransportState (MVar (Set EndPointAddress))
149+multigroupAt addr = multigroups >>> at addr "Invalid multigroup"