@@ -1,28 +1,114 @@
11module Network.Transport.TCP
22 ( mkTransport
3+ , TCPConfig (..)
34 ) where
45
56import Network.Transport
7+
8+import Control.Concurrent (forkIO)
9+import Control.Concurrent.Chan
10+import Control.Concurrent.MVar
11+import Control.Monad (forever, unless)
12+import Data.ByteString.Char8 (ByteString)
13+import Data.IntMap (IntMap)
614import Data.Word
15+import Network.Socket
16+import Safe
17+
18+import qualified Data.ByteString.Char8 as BS
19+import qualified Data.IntMap as IntMap
20+import qualified Network.Socket as N
21+import qualified Network.Socket.ByteString as NBS
722
8−data Address = Address String
9−type Port = Word16
23+type Chans = MVar (Int, IntMap (Chan ByteString))
24+type ChanId = Int
1025
1126-- | This deals with several different configuration properties:
1227-- * Buffer size, specified in Hints
1328-- * LAN/WAN, since we can inspect the addresses
14−data TCPConfig = TCPConfig Hints Port Address Address
29+-- Note that HostName could be an IP address, and ServiceName could be
30+-- a port number
31+data TCPConfig = TCPConfig Hints HostName ServiceName
1532
33+-- | This creates a TCP connection between a server and a number of
34+-- clients. Behind the scenes, the server hostname is passed as the SendAddr
35+-- and when a connection is made, messages sent down the SendEnd go
36+-- via a socket made for the client that connected.
37+-- Messages are all queued using an unbounded Chan.
1638mkTransport :: TCPConfig -> IO Transport
17−mkTransport tcpConfig =
18− -- create buffer
19− -- allocate listening socket
20− -- fork thread on that socket
39+mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
40+ channels <- newMVar (0, IntMap.empty)
41+ serverAddrs <- getAddrInfo
42+ (Just (N.defaultHints { addrFlags = [AI_PASSIVE] }))
43+ Nothing
44+ (Just service)
45+ let serverAddr = case serverAddrs of
46+ [] -> error "mkTransport: getAddrInfo returned []"
47+ as -> head as
48+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
49+ bindSocket sock (addrAddress serverAddr)
50+ listen sock 5
51+ forkIO $ procConnections channels sock
52+
2153 return Transport
22− { newConnectionWith = undefined
23− -- create listening mailbox
24− -- send back mailbox details
54+ { newConnectionWith = \_ -> do
55+ (chanId, chanMap) <- takeMVar channels
56+ chan <- newChan
57+ putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
58+ return (mkSendAddr host service chanId, mkReceiveEnd chan)
2559 , newMulticastWith = undefined
26− , deserialize = undefined
60+ , deserialize = \bs ->
61+ case readMay . BS.unpack $ bs of
62+ Nothing -> error "deserialize: cannot parse"
63+ Just (host, service, chanId) -> Just $ mkSendAddr host service chanId
2764 }
2865
66+ where
67+ mkSendAddr :: HostName -> ServiceName -> ChanId -> SendAddr
68+ mkSendAddr host service chanId = SendAddr
69+ { connectWith = \_ -> mkSendEnd chanId
70+ , serialize = BS.pack . show $ (host, service, chanId)
71+ }
72+
73+ mkSendEnd :: ChanId -> IO SendEnd
74+ mkSendEnd chanId = withSocketsDo $ do
75+ serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
76+ let serverAddr = case serverAddrs of
77+ [] -> error "mkSendEnd: getAddrInfo returned []"
78+ as -> head as
79+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
80+ setSocketOption sock KeepAlive 1
81+ N.connect sock (addrAddress serverAddr)
82+ NBS.sendAll sock $ BS.pack . show $ chanId
83+ return $ SendEnd
84+ { Network.Transport.send = \bss -> NBS.sendMany sock bss
85+ }
86+
87+ mkReceiveEnd :: Chan ByteString -> ReceiveEnd
88+ mkReceiveEnd chan = ReceiveEnd
89+ { -- for now we will implement this as a Chan
90+ receive = do
91+ bs <- readChan chan
92+ return [bs]
93+ }
94+
95+ procConnections :: Chans -> Socket -> IO ()
96+ procConnections chans sock = forever $ do
97+ (clientSock, _clientAddr) <- accept sock
98+ -- decode the first message to find the correct chanId
99+ bs <- NBS.recv clientSock 4096
100+ case BS.readInt bs of
101+ Nothing -> error "procConnections: cannot parse chanId"
102+ Just (chanId, bs') -> do
103+ -- lookup the channel
104+ (_, chanMap) <- readMVar chans
105+ case IntMap.lookup chanId chanMap of
106+ Nothing -> error "procConnections: cannot find chanId"
107+ Just chan -> forkIO $ do
108+ unless (BS.null bs') $ writeChan chan bs'
109+ procMessages chan clientSock
110+
111+ procMessages :: Chan ByteString -> Socket -> IO ()
112+ procMessages chan sock = forever $ do
113+ bs <- NBS.recv sock 4096
114+ writeChan chan bs