@@ -1,12 +1,21 @@
11-- | TCP transport
2−module Network.Transport.TCP where
2+module Network.Transport.TCP ( -- * Main API
3+ createTransport
4+ , -- TCP specific functionality
5+ EndPointIx
6+ , decodeEndPointAddress
7+ ) where
38
49import Network.Transport
10+import Network.Transport.Internal.TCP ( forkServer
11+ , recvExact
12+ , recvWithLength
13+ , sendWithLength
14+ )
515import Network.Transport.Internal ( encodeInt16
616 , decodeInt16
7− , encodeInt32
8− , decodeInt32
917 , maybeToErrorT
18+ , maybeTToErrorT
1019 )
1120import qualified Network.Socket as N ( HostName
1221 , ServiceName
@@ -14,10 +23,7 @@ import qualified Network.Socket as N ( HostName
1423 , SocketType(Stream)
1524 , SocketOption(ReuseAddr)
1625 , getAddrInfo
17− , defaultHints
1826 , socket
19− , bindSocket
20− , listen
2127 , addrFamily
2228 , addrAddress
2329 , defaultProtocol
@@ -26,22 +32,21 @@ import qualified Network.Socket as N ( HostName
2632 , sClose
2733 , accept
2834 )
29−import qualified Network.Socket.ByteString as NBS (sendMany, recv)
30−import Control.Concurrent (forkIO, ThreadId)
35+import qualified Network.Socket.ByteString as NBS (sendMany)
36+import Control.Concurrent (forkIO)
3137import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
3238import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, readMVar)
3339import Control.Category ((>>>))
3440import Control.Applicative ((<*>), (*>), (<$>), pure)
35−import Control.Monad (forever, mzero)
41+import Control.Monad (forever)
3642import Control.Monad.Error (ErrorT, liftIO, runErrorT)
3743import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)
3844import Data.ByteString (ByteString)
39−import qualified Data.ByteString as BS (length, concat, null)
45+import qualified Data.ByteString as BS (concat)
4046import qualified Data.ByteString.Char8 as BSC (pack, unpack)
4147import Data.IntMap (IntMap)
4248import qualified Data.IntMap as IntMap (empty, size)
4349import Data.Lens.Lazy (Lens, lens, intMapLens, (^.), (^=), (^+=))
44−import Data.Int (Int32)
4550import Text.Regex.Applicative (RE, few, anySym, sym, many, match)
4651
4752data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState) }
@@ -55,7 +60,7 @@ endPoints = lens _endPoints (\es st -> st { _endPoints = es })
5560nextConnectionIx :: Lens EndPointState ConnectionIx
5661nextConnectionIx = lens _nextConnectionIx (\cix st -> st { _nextConnectionIx = cix })
5762
58−endPointAt :: Int -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
63+endPointAt :: EndPointIx -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
5964endPointAt ix = endPoints >>> intMapLens ix
6065
6166-- | Create a TCP transport
@@ -74,19 +79,6 @@ createTransport host port = do
7479 forkServer host port (transportServer state)
7580 return Transport { newEndPoint = tcpNewEndPoint state host port }
7681
77−-- | Start a server at the specified address
78−forkServer :: N.HostName -> N.ServiceName -> (N.Socket -> IO ()) -> IO ThreadId
79−forkServer host port server = do
80− -- Resolve the specified address. By specification, getAddrInfo will never
81− -- return an empty list (but will throw an exception instead) and will return
82− -- the "best" address first, whatever that means
83− addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)
84− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
85− N.setSocketOption sock N.ReuseAddr 1
86− N.bindSocket sock (N.addrAddress addr)
87− N.listen sock 5
88− forkIO $ server sock
89−
9082-- | The transport server handles incoming connections for all endpoints
9183--
9284-- TODO: lots of error checking
@@ -129,7 +121,7 @@ endpointServer sock theirAddress chan state = forever $ do
129121 writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)
130122
131123 -- Read a message and output it on the endPoint's channel
132− readMessage :: Int -> MaybeT IO ()
124+ readMessage :: ConnectionIx -> MaybeT IO ()
133125 readMessage connIx = do
134126 payload <- recvWithLength sock
135127 liftIO $ writeChan chan (Received connIx payload)
@@ -145,24 +137,29 @@ tcpNewEndPoint transportState host port = do
145137 return . Right $ EndPoint { receive = readChan chan
146138 , address = addr
147139 , connect = tcpConnect addr
148− , newMulticastGroup = undefined
149− , resolveMulticastGroup = undefined
140+ , newMulticastGroup = return . Left $ newMulticastGroupError
141+ , resolveMulticastGroup = \_ -> return . Left $ resolveMulticastGroupError
150142 }
143+ where
144+ newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
145+ resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
146+
151147
152148-- | Connnect to an endpoint
153149tcpConnect :: EndPointAddress -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)
154150tcpConnect myAddress theirAddress _ = runErrorT $ do
155− sock <- socketForEndPoint myAddress theirAddress
156− connIx <- liftIO $ requestNewConnection sock
157− return $ Connection { send = sendWithLength sock (Just connIx)
158− , close = N.sClose sock
159− }
151+ sock <- socketForEndPoint myAddress theirAddress
152+ -- TODO: test this failure
153+ connIx <- maybeTToErrorT (FailedWith undefined undefined) $ requestNewConnection sock
154+ return $ Connection { send = sendWithLength sock (Just connIx)
155+ , close = N.sClose sock
156+ }
160157
161158-- | Request a new connection
162−requestNewConnection :: N.Socket -> IO ByteString
159+requestNewConnection :: N.Socket -> MaybeT IO ByteString
163160requestNewConnection sock = do
164− pure <$> encodeInt16 0 >>= NBS.sendMany sock
165− NBS.recv sock 2
161+ liftIO $ pure <$> encodeInt16 0 >>= NBS.sendMany sock
162+ BS.concat <$> recvExact sock 2
166163
167164-- | Find a socket to the specified endpoint
168165--
@@ -172,7 +169,7 @@ requestNewConnection sock = do
172169socketForEndPoint :: EndPointAddress -- ^ Our address
173170 -> EndPointAddress -- ^ Their address
174171 -> ErrorT (FailedWith ConnectErrorCode) IO N.Socket
175−socketForEndPoint (EndPointAddress myAddress) (EndPointAddress theirAddress) = do
172+socketForEndPoint (EndPointAddress myAddress) theirAddress = do
176173 (host, port, endPointIx) <- maybeToErrorT invalidAddress (decodeEndPointAddress theirAddress)
177174 liftIO $ do
178175 -- Connect to the destination transport
@@ -188,46 +185,11 @@ socketForEndPoint (EndPointAddress myAddress) (EndPointAddress theirAddress) = d
188185 where
189186 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"
190187
191−-- | Send a bunch of bytestrings prepended with their length
192−sendWithLength :: N.Socket -- ^ Socket to send on
193− -> Maybe ByteString -- ^ Optional header to send before the length
194− -> [ByteString] -- ^ Payload
195− -> IO ()
196−sendWithLength sock header payload = do
197− lengthBs <- encodeInt32 (fromIntegral . sum . map BS.length $ payload)
198− let msg = maybe id (:) header $ lengthBs : payload
199− NBS.sendMany sock msg
200−
201−-- | Read a length and then a payload of that length
202−-- TODO: error handling
203−recvWithLength :: N.Socket -> MaybeT IO [ByteString]
204−recvWithLength sock = do
205− msgLengthBs <- recvExact sock 4
206− decodeInt32 (BS.concat msgLengthBs) >>= recvExact sock
207−
208188-- | Decode end point address
209189-- TODO: This uses regular expression parsing, which is nice, but unnecessary
210−decodeEndPointAddress :: ByteString -> Maybe (N.HostName, N.ServiceName, Int)
211−decodeEndPointAddress bs = match endPointAddressRE (BSC.unpack bs)
190+decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointIx)
191+decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack bs)
212192 where
213− endPointAddressRE :: RE Char (N.HostName, N.ServiceName, Int)
193+ endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointIx)
214194 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))
215195
216−-- | Read an exact number of bytes from a socket
217−--
218−-- Returns 'Nothing' if the socket closes prematurely.
219−recvExact :: N.Socket -- ^ Socket to read from
220− -> Int32 -- ^ Number of bytes to read
221− -> MaybeT IO [ByteString]
222−recvExact sock len = do
223− (socketClosed, input) <- liftIO $ go [] len
224− if socketClosed then mzero else return input
225− where
226− -- Returns input read and whether the socket closed prematurely
227− go :: [ByteString] -> Int32 -> IO (Bool, [ByteString])
228− go acc 0 = return (False, reverse acc)
229− go acc l = do
230− bs <- NBS.recv sock (fromIntegral l `min` 4096)
231− if BS.null bs
232− then return (True, reverse acc)
233− else go (bs : acc) (l - (fromIntegral $ BS.length bs))