@@ -7,7 +7,7 @@
77module Network.Transport.TCP ( -- * Main API
88 createTransport
99 , -- TCP specific functionality
10− EndPointIx
10+ EndPointId
1111 , decodeEndPointAddress
1212 ) where
1313
@@ -15,12 +15,13 @@ import Network.Transport
1515import Network.Transport.Internal.TCP ( forkServer
1616 , connectTo
1717 , recvExact
18− , sendMany
1918 , recvWithLength
2019 , sendWithLength
20+ , recvInt16
21+ , recvInt32
22+ , sendInt32
2123 )
2224import Network.Transport.Internal ( encodeInt16
23− , decodeInt16
2425 , maybeToErrorT
2526 , maybeTToErrorT
2627 )
@@ -34,7 +35,7 @@ import Control.Concurrent (forkIO)
3435import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
3536import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, readMVar)
3637import Control.Category ((>>>))
37−import Control.Applicative ((<*>), (*>), (<$>), pure)
38+import Control.Applicative ((<*>), (*>), (<$>))
3839import Control.Monad (forever, forM_)
3940import Control.Monad.Error (ErrorT, liftIO, runErrorT)
4041import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)
@@ -43,24 +44,24 @@ import Control.Monad.Writer.Class (tell)
4344import Data.ByteString (ByteString)
4445import qualified Data.ByteString as BS (concat)
4546import qualified Data.ByteString.Char8 as BSC (pack, unpack)
47+import Data.Int (Int16)
4648import Data.IntMap (IntMap)
4749import qualified Data.IntMap as IntMap (empty, size)
4850import Data.Lens.Lazy (Lens, lens, intMapLens, (^.), (^=), (^+=))
4951import Text.Regex.Applicative (RE, few, anySym, sym, many, match)
5052
5153data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState) }
52−data EndPointState = EndPointState { _nextConnectionIx :: ConnectionIx }
53−type EndPointIx = Int
54−type ConnectionIx = Int
54+data EndPointState = EndPointState { _nextConnectionId :: ConnectionId }
55+type EndPointId = Int16
5556
5657endPoints :: Lens TransportState (IntMap (Chan Event, MVar EndPointState))
5758endPoints = lens _endPoints (\es st -> st { _endPoints = es })
5859
59−nextConnectionIx :: Lens EndPointState ConnectionIx
60−nextConnectionIx = lens _nextConnectionIx (\cix st -> st { _nextConnectionIx = cix })
60+nextConnectionId :: Lens EndPointState ConnectionId
61+nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
6162
62−endPointAt :: EndPointIx -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
63−endPointAt ix = endPoints >>> intMapLens ix
63+endPointAt :: EndPointId -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
64+endPointAt ix = endPoints >>> intMapLens (fromIntegral ix)
6465
6566-- | Create a TCP transport
6667--
@@ -92,18 +93,14 @@ transportServer state sock = forever $ do
9293handleClient :: MVar TransportState -> N.Socket -> IO ()
9394handleClient state sock = do
9495 mendpoint <- runMaybeT $ do
95− -- Our endpoint the other side is interested in
96− endPointBS <- recvExact sock 2
97− endPointIx <- decodeInt16 (BS.concat endPointBS)
98− -- Full endpoint address of the other side
99− theirAddress <- BS.concat <$> recvWithLength sock
100− -- Find our endpoint
101− (endPointCh, endPointSt) <- MaybeT $ (^. endPointAt (fromIntegral endPointIx)) <$> readMVar state
102− return (endPointCh, endPointSt, theirAddress)
96+ ourEndPointId <- recvInt16 sock
97+ theirAddress <- BS.concat <$> recvWithLength sock
98+ ourEndPoint <- MaybeT $ (^. endPointAt (fromIntegral ourEndPointId)) <$> readMVar state
99+ return (ourEndPoint, EndPointAddress theirAddress)
103100 case mendpoint of
104− Just (endPointCh, endPointSt, theirAddress) -> do
105− openConnections <- endpointServer sock (EndPointAddress theirAddress) endPointCh endPointSt
106− forM_ openConnections $ \cix -> writeChan endPointCh (ConnectionClosed cix)
101+ Just ((endPointCh, endPointSt), theirAddress) -> do
102+ unclosedConnections <- endpointServer sock theirAddress endPointCh endPointSt
103+ forM_ unclosedConnections $ \cix -> writeChan endPointCh (ConnectionClosed cix)
107104 Nothing ->
108105 -- Invalid request
109106 return ()
@@ -111,41 +108,39 @@ handleClient state sock = do
111108-- | Handle requests for one endpoint. Returns only if an error occurs,
112109-- in which case it returns the connections that are still open (normally,
113110-- this should be an empty list).
114−endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> IO [ConnectionIx]
111+endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> IO [ConnectionId]
115112endpointServer sock theirAddress chan state = execWriterT . runMaybeT . forever $ do
116− connBS <- recvExact sock 2
117− connIx <- fromIntegral <$> decodeInt16 (BS.concat connBS)
118− if connIx == 0
119− then createNewConnection
120− else readMessage connIx
113+ connId <- recvInt32 sock
114+ case connId of
115+ 0 -> createNewConnection
116+ _ -> readMessage connId
121117 where
122118 -- A brief note on the types:
123119 -- These functions may fail (connection error): hence MaybeT
124120 -- We also need to keep track of which connections are still open: hence WriterT
125121
126122 -- Create a new connection
127− createNewConnection :: MaybeT (WriterT [ConnectionIx] IO) ()
123+ createNewConnection :: MaybeT (WriterT [ConnectionId] IO) ()
128124 createNewConnection = do
129− newIx <- liftIO $ modifyMVar state $ \st -> return (nextConnectionIx ^+= 1 $ st, st ^. nextConnectionIx)
130− newBs <- encodeInt16 (fromIntegral newIx)
131− sendMany sock [newBs]
132− liftIO $ writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)
133− tell [newIx]
125+ newId <- liftIO $ modifyMVar state $ \st -> return (nextConnectionId ^+= 1 $ st, st ^. nextConnectionId)
126+ sendInt32 sock newId
127+ liftIO $ writeChan chan (ConnectionOpened newId ReliableOrdered theirAddress)
128+ -- We add the new connection ID to the list of open connections only once the
129+ -- endpoint has been notified of the new connection (sendInt32 may fail)
130+ tell [newId]
134131
135132 -- Read a message and output it on the endPoint's channel
136− readMessage :: ConnectionIx -> MaybeT (WriterT [ConnectionIx] IO) ()
137− readMessage connIx = do
138− payload <- recvWithLength sock
139− liftIO $ writeChan chan (Received connIx payload)
133+ readMessage :: ConnectionId -> MaybeT (WriterT [ConnectionId] IO) ()
134+ readMessage connId = recvWithLength sock >>= liftIO . writeChan chan . Received connId
140135
141136-- | Create a new endpoint
142137tcpNewEndPoint :: MVar TransportState -> N.HostName -> N.ServiceName -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
143138tcpNewEndPoint transportState host port = do
144139 chan <- newChan
145− endPointState <- newMVar EndPointState { _nextConnectionIx = 1 } -- Connection id 0 is reserved for requesting new connections
146− endPointIx <- modifyMVar transportState $ \st -> do let ix = IntMap.size (st ^. endPoints)
140+ endPointState <- newMVar EndPointState { _nextConnectionId = firstNonReservedConnectionId }
141+ endPointId <- modifyMVar transportState $ \st -> do let ix = fromIntegral $ IntMap.size (st ^. endPoints)
147142 return ((endPointAt ix ^= Just (chan, endPointState)) $ st, ix)
148− let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointIx
143+ let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointId
149144 return . Right $ EndPoint { receive = readChan chan
150145 , address = addr
151146 , connect = tcpConnect addr
@@ -155,24 +150,26 @@ tcpNewEndPoint transportState host port = do
155150 where
156151 newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
157152 resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
158−
153+
154+-- | We reserve a bunch of connection IDs for control messages
155+firstNonReservedConnectionId :: ConnectionId
156+firstNonReservedConnectionId = 1024
159157
160158-- | Connnect to an endpoint
161159tcpConnect :: EndPointAddress -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)
162160tcpConnect myAddress theirAddress _ = runErrorT $ do
163161 sock <- socketForEndPoint myAddress theirAddress
164− connIx <- maybeTToErrorT undefined $ requestNewConnection sock
162+ connId <- maybeTToErrorT undefined $ requestNewConnection sock
165163 return Connection { send = runErrorT .
166164 maybeTToErrorT (FailedWith SendFailed "Send failed") .
167− sendWithLength sock (Just connIx)
165+ sendWithLength sock (Just connId)
168166 , close = N.sClose sock
169167 }
170168
171169-- | Request a new connection
170+-- Returns the new connection ID in serialized form
172171requestNewConnection :: N.Socket -> MaybeT IO ByteString
173−requestNewConnection sock = do
174− liftIO (pure <$> encodeInt16 0) >>= sendMany sock
175− BS.concat <$> recvExact sock 2
172+requestNewConnection sock = sendInt32 sock 0 >> BS.concat <$> recvExact sock 4
176173
177174-- | Find a socket to the specified endpoint
178175--
@@ -183,9 +180,9 @@ socketForEndPoint :: EndPointAddress -- ^ Our address
183180 -> EndPointAddress -- ^ Their address
184181 -> ErrorT (FailedWith ConnectErrorCode) IO N.Socket
185182socketForEndPoint (EndPointAddress myAddress) theirAddress = do
186− (host, port, endPointIx) <- maybeToErrorT invalidAddress (decodeEndPointAddress theirAddress)
183+ (host, port, endPointId) <- maybeToErrorT invalidAddress (decodeEndPointAddress theirAddress)
187184 sock <- maybeTToErrorT undefined (connectTo host port)
188− endPointBS <- liftIO $ encodeInt16 (fromIntegral endPointIx)
185+ endPointBS <- encodeInt16 (fromIntegral endPointId)
189186 maybeTToErrorT undefined $ sendWithLength sock (Just endPointBS) [myAddress]
190187 return sock
191188 where
@@ -193,8 +190,8 @@ socketForEndPoint (EndPointAddress myAddress) theirAddress = do
193190
194191-- | Decode end point address
195192-- TODO: This uses regular expression parsing, which is nice, but unnecessary
196−decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointIx)
193+decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointId)
197194decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack bs)
198195 where
199− endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointIx)
196+ endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointId)
200197 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))