@@ -3,14 +3,41 @@ module Network.Transport.Chan (createTransport) where
33
44import Network.Transport
55import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
6−import Control.Concurrent.MVar (MVar, newMVar, modifyMVar)
6+import Control.Applicative ((<$>))
7+import Control.Category ((>>>))
8+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)
9+import Control.Monad (forM_)
710import Data.Map (Map)
8−import qualified Data.Map as Map (empty, insert, (!))
9−import Data.ByteString (ByteString)
11+import qualified Data.Map as Map (empty, insert, (!), size, delete)
1012import qualified Data.ByteString.Char8 as BSC (pack)
13+import Data.Lens.Lazy (Lens, lens, (^.), (^%=), (^=), (^+=), mapLens)
1114
1215-- Global state: next available "address", mapping from addresses to channels and next available connection
13−type TransportState = (Int, Map ByteString (Chan Event, Int))
16+data TransportState = State { _channels :: Map Address (Chan Event)
17+ , _connections :: Map Address Int
18+ , _multigroups :: Map MulticastAddress (MVar [Chan Event])
19+ }
20+
21+channels :: Lens TransportState (Map Address (Chan Event))
22+channels = lens _channels (\ch st -> st { _channels = ch })
23+
24+connections :: Lens TransportState (Map Address Int)
25+connections = lens _connections (\conn st -> st { _connections = conn })
26+
27+multigroups :: Lens TransportState (Map MulticastAddress (MVar [Chan Event]))
28+multigroups = lens _multigroups (\gs st -> st { _multigroups = gs })
29+
30+at :: Ord k => k -> Lens (Map k v) v
31+at k = lens (Map.! k) (Map.insert k)
32+
33+channelAt :: Address -> Lens TransportState (Chan Event)
34+channelAt addr = channels >>> at addr
35+
36+connectionAt :: Address -> Lens TransportState Int
37+connectionAt addr = connections >>> at addr
38+
39+multigroupAt :: MulticastAddress -> Lens TransportState (MVar [Chan Event])
40+multigroupAt addr = multigroups >>> at addr
1441
1542-- | Create a new Transport.
1643--
@@ -18,30 +45,74 @@ type TransportState = (Int, Map ByteString (Chan Event, Int))
1845-- (threads can, and should, create their own endpoints though).
1946createTransport :: IO Transport
2047createTransport = do
21− endpoints <- newMVar (0, Map.empty)
22− return Transport { newEndPoint = chanNewEndPoint endpoints }
48+ state <- newMVar $ State { _channels = Map.empty
49+ , _connections = Map.empty
50+ , _multigroups = Map.empty
51+ }
52+ return Transport { newEndPoint = chanNewEndPoint state }
2353
2454-- Create a new end point
2555chanNewEndPoint :: MVar TransportState -> IO (Either Error EndPoint)
26−chanNewEndPoint endpoints = do
56+chanNewEndPoint state = do
2757 chan <- newChan
28− addr <- modifyMVar endpoints $ \(next, state) -> do
29− let addr = BSC.pack (show next)
30− return ((next + 1, Map.insert addr (chan, 0) state), addr)
58+ addr <- modifyMVar state $ \st -> do
59+ let addr = Address . BSC.pack . show . Map.size $ st ^. channels
60+ return ((channelAt addr ^= chan) . (connectionAt addr ^= 0) $ st, addr)
3161 return . Right $ EndPoint { receive = readChan chan
32− , address = Address addr
33− , connect = chanConnect (Address addr) endpoints
34− , newMulticastGroup = undefined
35− , resolveMulticastGroup = undefined
62+ , address = addr
63+ , connect = chanConnect addr state
64+ , newMulticastGroup = chanNewMulticastGroup state chan
65+ , resolveMulticastGroup = chanResolveMulticastGroup state chan
3666 }
3767
3868-- Create a new connection
3969chanConnect :: Address -> MVar TransportState -> Address -> Reliability -> IO (Either Error Connection)
40−chanConnect myAddress endpoints (Address theirAddress) _ = do
41− (chan, conn) <- modifyMVar endpoints $ \(next, state) -> do
42− let (chan, conn) = state Map.! theirAddress
43− return ((next, Map.insert theirAddress (chan, conn + 1) state), (chan, conn))
70+chanConnect myAddress state theirAddress _ = do
71+ (chan, conn) <- modifyMVar state $ \st -> do
72+ let chan = st ^. channelAt theirAddress
73+ let conn = st ^. connectionAt theirAddress
74+ return (connectionAt theirAddress ^+= 1 $ st, (chan, conn))
4475 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress
45− return . Right $ Connection { send = writeChan chan . Receive conn
46− , close = writeChan chan $ ConnectionClosed conn
76+ return . Right $ Connection { send = writeChan chan . Received conn
77+ , close = writeChan chan $ ConnectionClosed conn
4778 }
79+
80+-- Create a new multicast group
81+chanNewMulticastGroup :: MVar TransportState -> Chan Event -> IO (Either Error MulticastGroup)
82+chanNewMulticastGroup state endpoint = do
83+ group <- newMVar []
84+ addr <- modifyMVar state $ \st -> do
85+ let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups
86+ return (multigroupAt addr ^= group $ st, addr)
87+ return . Right $ chanMulticastGroup state endpoint addr group
88+
89+-- Construct a multicast group
90+--
91+-- When the group is deleted some endpoints may still receive messages, but
92+-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact
93+-- that some multicast messages may still be in transit when the group is
94+-- deleted.
95+--
96+-- TODO: subscribing twice means you will receive messages twice, but as soon
97+-- as you unsubscribe you will stop receiving all messages. Not sure if this
98+-- is the right behaviour.
99+chanMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> MVar [Chan Event] -> MulticastGroup
100+chanMulticastGroup state endpoint addr group =
101+ MulticastGroup { multicastAddress = addr
102+ , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^%= Map.delete addr)
103+ , maxMsgSize = Nothing
104+ , multicastSend = \payload -> do
105+ cs <- readMVar group
106+ forM_ cs $ \ch -> writeChan ch (ReceivedMulticast addr payload)
107+ , multicastSubscribe = modifyMVar_ group $ return . (endpoint :)
108+ , multicastUnsubscribe = modifyMVar_ group $ return . (filter (/= endpoint))
109+ , multicastClose = return ()
110+ }
111+
112+-- Resolve a multicast group
113+chanResolveMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> IO (Either Error MulticastGroup)
114+chanResolveMulticastGroup state endpoint addr = do
115+ group <- (^. (multigroups >>> mapLens addr)) <$> readMVar state
116+ case group of
117+ Nothing -> return . Left $ Error undefined ("Group " ++ show addr ++ " not found")
118+ Just mvar -> return . Right $ chanMulticastGroup state endpoint addr mvar