courier 0.1.0.5 → 0.1.0.6
raw patch · 9 files changed
+652/−577 lines, 9 filesdep −QuickCheckdep −test-framework-quickcheck2
Dependencies removed: QuickCheck, test-framework-quickcheck2
Files
- courier.cabal +13/−7
- src/Network/Transport/Internal.hs +46/−0
- src/Network/Transport/Memory.hs +3/−3
- src/Network/Transport/Sockets.hs +325/−0
- src/Network/Transport/TCP.hs +76/−258
- src/Network/Transport/UDP.hs +181/−0
- tests/TestMemory.hs +0/−76
- tests/TestTCP.hs +0/−223
- tests/Tests.hs +8/−10
courier.cabal view
@@ -2,13 +2,20 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: courier-version: 0.1.0.5-synopsis: A message-passing library, intended for simplifying network applications+version: 0.1.0.6+synopsis: A message-passing library for simplifying network applications description: Inspired by Erlang's simple message-passing facilities, courier provides roughly similar capabilities. Applications simply create one or more endpoints, bind each to a transport using a given name, then can freely send / receive messages to other endpoints just by referencing the name each endpoint bound to its transport.+ .+ Note that while the simplicity is inspired by Erlang, the actual semantics of a receive + operation are not: receive just returns the next message delivered to an endpoint by a + transport. There is no attempt to perform pattern-matching on a range of alternatives, + and thus enabling out-of-order receipt. Consequently, all messages delivered to an endpoint + will always be received in the order delivered. In this sense, endpoints are more akin to + channels in Go but without the strict typing. homepage: http://github.com/hargettp/courier license: MIT@@ -33,9 +40,12 @@ Network.Transport Network.Transport.Memory Network.Transport.TCP+ Network.Transport.UDP ghc-options: -Wall- -- other-modules: + other-modules:+ Network.Transport.Sockets+ Network.Transport.Internal build-depends: base >=4 && <5, async, bytestring,@@ -51,18 +61,14 @@ type: exitcode-stdio-1.0 hs-source-dirs: tests main-is: Tests.hs- other-modules: TestMemory,- TestTCP build-depends: -- base modules base, -- testing support HUnit, - QuickCheck >= 2.0, test-framework, test-framework-hunit,- test-framework-quickcheck2, -- Cabal, -- 3rd party modules cereal,
+ src/Network/Transport/Internal.hs view
@@ -0,0 +1,46 @@+-----------------------------------------------------------------------------+-- |+-- Module : Network.Transport.Internal+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (uses STM)+--+-- Internal definitions used for implementing transports.+-----------------------------------------------------------------------------++module Network.Transport.Internal (+ catchExceptions+)++where++-- local imports+-- external imports++import Control.Exception++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++{-|+Silently ignore 'BlockedIndefinitelyOnSTM' and 'ThreadKilled' exceptions, as+most cases in which they occur are benign and not worth the noise. Use with +caution, as this function could mask undelrying design issues.+-}+catchExceptions :: IO () -> (SomeException -> IO ()) -> IO ()+catchExceptions blk handler = do+ catch (do+ catch (catch blk ignoreBlocked)+ ignoreThreadKilled)+ handler+ where+ ignoreBlocked :: BlockedIndefinitelyOnSTM -> IO ()+ ignoreBlocked _ = return ()+ ignoreThreadKilled :: AsyncException -> IO ()+ ignoreThreadKilled e =+ case e of+ ThreadKilled -> return ()+ _ -> handler $ SomeException e
src/Network/Transport/Memory.hs view
@@ -30,7 +30,6 @@ import Control.Concurrent.STM import qualified Data.Map as M-import Data.Maybe (fromJust) -------------------------------------------------------------------------------- --------------------------------------------------------------------------------@@ -75,8 +74,9 @@ memorySendTo :: MemoryTransport -> Name -> Message -> IO () memorySendTo transport name msg = do mailboxes <- atomically $ readTVar $ boundMailboxes transport- let mailbox = fromJust $ M.lookup name mailboxes- atomically $ writeTQueue mailbox msg+ case M.lookup name mailboxes of+ Just mailbox -> atomically $ writeTQueue mailbox msg+ Nothing -> return () -- error $ "No mailbox for " ++ name memoryUnbind :: MemoryTransport -> Name -> IO () memoryUnbind transport name = do
+ src/Network/Transport/Sockets.hs view
@@ -0,0 +1,325 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveGeneric #-}++-----------------------------------------------------------------------------+-- |+-- Module : Network.Transport.Sockets+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (uses STM)+--+-- Common facilities for socket-based transports, such as UDP and TCP/IP.+--+-----------------------------------------------------------------------------++module Network.Transport.Sockets (++ Bindings,++ SocketTransport(..),++ Connection(..),++ IdentifyMessage(..),++ Messenger(..),+ newMessenger,+ addMessenger,+ deliver,+ closeMessenger,++ dispatcher,+ sender,+ socketSendTo,+ receiver,+ receiveSocketMessage,+ receiveSocketMessages,+ SocketSend,++ parseSocketAddress,+ lookupAddresses,+ lookupAddress++ ) where++-- local imports++import Network.Transport+import Network.Transport.Internal++-- external imports+import Control.Concurrent.Async+import Control.Concurrent.STM++import Control.Exception++import qualified Data.ByteString as B+import qualified Data.Map as M+import Data.Serialize+import qualified Data.Set as S+import qualified Data.Text as T++import GHC.Generics++import Network.Simple.TCP (recv)+import Network.Socket hiding (recv,socket)++import System.Log.Logger++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++_log :: String+_log = "transport.sockets"++type Bindings = TVar (M.Map Name Mailbox)++data SocketTransport = SocketTransport {+ socketMessengers :: TVar (M.Map Address Messenger),+ socketBindings :: Bindings,+ socketConnection :: Address -> IO Connection,+ socketMessenger :: Connection -> Mailbox -> IO Messenger,+ socketInbound :: Mailbox,+ socketDispatchers :: S.Set (Async ()),+ socketResolver :: Resolver++}++{-|+A connection specializes the use of a transport for a particular+destination.+-}+data Connection = Connection {+ connAddress :: Address,+ connSocket :: TMVar Socket,+ connConnect :: IO Socket,+ connSend :: Socket -> B.ByteString -> IO (),+ connReceive :: Socket -> Int -> IO (Maybe B.ByteString),+ connClose :: IO ()+ }++{-|+A messenger is a facility that actual uses the mechanisms of a transport+(and more specifically, of a connection on a transport) to deliver and receive+messages. The messenger uses 'Mailbox'es internally so that the sending/receiving+happens asynchronously, allowing applications to move on without regard for+when any send / receive action actually completes.+-}+data Messenger = Messenger {+ messengerOut :: Mailbox,+ messengerAddress :: Address,+ messengerSender :: Async (),+ messengerReceiver :: Async (),+ messengerConnection :: Connection+ }++data IdentifyMessage = IdentifyMessage Address deriving (Generic)++instance Serialize IdentifyMessage++{-|+Parse a TCP 'Address' into its respective 'HostName' and 'PortNumber' components, on the+assumption the 'Address' has an identifer in the format @host:port@. If+the port number is missing from the supplied address, it will default to 0. If the+hostname component is missing from the identifier (e.g., just @:port@), then hostname+is assumed to be @localhost@.+-}+parseSocketAddress :: Address -> (HostName,ServiceName)+parseSocketAddress address = + let identifer = T.pack $ address + parts = T.splitOn ":" identifer+ in if (length parts) > 1 then+ (host $ T.unpack $ parts !! 0, port $ T.unpack $ parts !! 1)+ else (host $ T.unpack $ parts !! 0, "0")+ where+ host h = if h == "" then+ "localhost"+ else h+ port p = p++lookupAddresses :: (HostName,ServiceName) -> IO [SockAddr]+lookupAddresses hostAndPort = + let (host,port) = hostAndPort+ hints = defaultHints { addrFlags = [AI_ADDRCONFIG, AI_CANONNAME, AI_NUMERICSERV] }+ in do + addresses <- getAddrInfo (Just hints) (Just host) (Just port)+ return $ map addrAddress addresses++lookupAddress :: (HostName,ServiceName) -> IO SockAddr+lookupAddress hostAndPort = do+ addresses <- lookupAddresses hostAndPort+ return $ addresses !! 0++type SocketSend = Socket -> B.ByteString -> IO ()++instance Show Messenger where+ show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")"++newMessenger :: Connection -> Mailbox -> IO Messenger+newMessenger conn inc = do+ out <- newMailbox+ sndr <- async $ sender conn out+ rcvr <- async $ receiver conn inc+ return Messenger {+ messengerOut = out,+ messengerAddress = connAddress conn,+ messengerSender = sndr,+ messengerReceiver = rcvr,+ messengerConnection = conn+ }++addMessenger :: SocketTransport -> Address -> Messenger -> IO ()+addMessenger transport address msngr = do+ msngrs <- atomically $ do+ modifyTVar (socketMessengers transport) $ \msngrs -> M.insert address msngr msngrs+ msngrs <- readTVar (socketMessengers transport)+ return msngrs+ infoM _log $ "Added messenger to " ++ (show address) ++ "; messengers are " ++ (show msngrs)++deliver :: Messenger -> Message -> IO ()+deliver msngr message = atomically $ writeTQueue (messengerOut msngr) message+ +dispatcher :: TVar (M.Map Name Mailbox) -> Mailbox -> IO ()+dispatcher bindings mbox = dispatchMessages+ where+ dispatchMessages = catchExceptions (do+ infoM _log $ "Dispatching messages"+ env <- atomically $ readTQueue mbox+ dispatchMessage env+ dispatchMessages)+ (\e -> do+ warningM _log $ "Dispatch error: " ++ (show (e :: SomeException)))+ dispatchMessage env = do+ infoM _log $ "Dispatching message"+ let envelopeOrErr = decode env+ case envelopeOrErr of+ Left err -> do+ errorM _log $ "Error decoding message for dispatch: " ++ err+ return ()+ Right (Envelope destination msg) -> do + atomically $ do + dests <- readTVar bindings+ let maybeDest = M.lookup destination dests+ case maybeDest of+ Nothing -> return ()+ Just dest -> do + writeTQueue dest msg+ return ()++sender :: Connection -> Mailbox -> IO ()+sender conn mailbox = sendMessages+ where+ sendMessages = do+ reconnect+ catchExceptions (do+ infoM _log $ "Waiting to send to " ++ (show $ connAddress conn)+ msg <- atomically $ readTQueue mailbox+ infoM _log $ "Sending message to " ++ (show $ connAddress conn)+ connected <- atomically $ tryReadTMVar $ connSocket conn+ case connected of+ Just socket -> do+ (connSend conn) socket msg+ Nothing -> return ()+ )+ (\e -> do+ warningM _log $ "Send error: " ++ (show (e :: SomeException))+ disconnect)+ sendMessages+ reconnect = do+ -- TODO need a timeout here, in case connecting always fails+ connected <- atomically $ tryReadTMVar $ connSocket conn+ case connected of+ Just _ -> return ()+ Nothing -> do+ let (host,port) = parseSocketAddress $ connAddress conn+ infoM _log $ "Connecting to " ++ (show host) ++ ":" ++ (show port) -- (show address)+ socket <- connConnect conn+ infoM _log $ "Connected to " ++ (show $ connAddress conn)+ atomically $ putTMVar (connSocket conn) socket+ disconnect = do+ connected <- atomically $ tryTakeTMVar $ connSocket conn+ case connected of+ Just socket -> sClose socket+ Nothing -> return ()+++socketSendTo :: SocketTransport -> Name -> Message -> IO ()+socketSendTo transport name msg = do+ isLocal <- local+ if isLocal+ then return ()+ else remote+ where+ local = do+ found <- atomically $ do+ bindings <- readTVar $ socketBindings transport+ return $ M.lookup name bindings+ case found of+ Nothing -> return False+ Just mbox -> do+ atomically $ writeTQueue mbox msg+ return True+ remote = do+ Just address <- resolve (socketResolver transport) name+ let env = encode $ Envelope {+ envelopeDestination = name,+ envelopeContents = msg+ }+ amsngr <- atomically $ do+ msngrs <- readTVar $ socketMessengers transport+ return $ M.lookup address msngrs+ case amsngr of+ Nothing -> do+ msngrs <- atomically $ readTVar $ socketMessengers transport+ infoM _log $ "No messenger for " ++ (show address) ++ " in " ++ (show msngrs)+ socketVar <- atomically $ newEmptyTMVar+ newConn <- (socketConnection transport) address+ let conn = newConn {connSocket = socketVar}+ msngr <- (socketMessenger transport) conn (socketInbound transport)+ addMessenger transport address msngr+ deliver msngr env+ return ()+ Just msngr -> deliver msngr env++receiver :: Connection -> Mailbox -> IO ()+receiver conn mailbox = do + socket <- atomically $ readTMVar $ connSocket conn+ receiveSocketMessages socket (connAddress conn) mailbox++receiveSocketMessages :: Socket -> Address -> Mailbox -> IO ()+receiveSocketMessages sock addr mailbox = catchExceptions (do+ infoM _log $ "Waiting to receive on " ++ (show addr)+ maybeMsg <- receiveSocketMessage sock+ infoM _log $ "Received message on " ++ (show addr)+ case maybeMsg of+ Nothing -> do+ sClose sock+ return ()+ Just msg -> do+ atomically $ writeTQueue mailbox msg+ receiveSocketMessages sock addr mailbox) (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException)))++receiveSocketMessage :: Socket -> IO (Maybe Message)+receiveSocketMessage socket = do+ maybeLen <- recv socket 8 -- TODO must figure out what defines length of an integer in bytes + case maybeLen of+ Nothing -> do+ infoM _log $ "No length received"+ return Nothing+ Just len -> do + maybeMsg <- recv socket $ msgLength (decode len)+ infoM _log $ "Received message"+ return maybeMsg+ where+ msgLength (Right size) = size+ msgLength (Left err) = error err++closeMessenger :: Messenger -> IO ()+closeMessenger msngr = do+ cancel $ messengerSender msngr+ cancel $ messengerReceiver msngr+ connClose $ messengerConnection msngr
src/Network/Transport/TCP.hs view
@@ -25,6 +25,8 @@ -- local imports import Network.Transport+import Network.Transport.Internal+import Network.Transport.Sockets -- external imports @@ -36,10 +38,7 @@ import qualified Data.Map as M import Data.Serialize import qualified Data.Set as S-import qualified Data.Text as T -import GHC.Generics- import Network.Socket (sClose,accept) import Network.Simple.TCP hiding (accept) @@ -51,18 +50,8 @@ _log :: String _log = "transport.tcp" -data TCPTransport = TCPTransport {- tcpListeners :: TVar (M.Map ServiceName Socket),- tcpMessengers :: TVar (M.Map Address Messenger), - tcpBindings :: TVar (M.Map Name Mailbox),- tcpInbound :: Mailbox,- tcpDispatchers :: S.Set (Async ()),- tcpResolver :: Resolver- }- -data IdentifyMessage = IdentifyMessage Address deriving (Generic)--instance Serialize IdentifyMessage+tcpScheme :: Scheme+tcpScheme = "tcp" {-| Create a new 'Transport' suitable for sending messages over TCP/IP. There can@@ -71,67 +60,42 @@ correct TCP/IP addresses (or hostnames) for communication. -} newTCPTransport :: Resolver -> IO Transport-newTCPTransport resolver = do - listeners <- atomically $ newTVar M.empty+newTCPTransport resolver = do messengers <- atomically $ newTVar M.empty bindings <- atomically $ newTVar M.empty inbound <- newMailbox dispatch <- async $ dispatcher bindings inbound- let transport = TCPTransport {- tcpListeners = listeners,- tcpMessengers = messengers,- tcpBindings = bindings,- tcpInbound = inbound,- tcpDispatchers = S.fromList [dispatch],- tcpResolver = resolver+ let transport = SocketTransport {+ socketMessengers = messengers,+ socketBindings = bindings,+ socketInbound = inbound,+ socketConnection = newTCPConnection,+ socketMessenger = newTCPMessenger bindings resolver,+ socketDispatchers = S.fromList [dispatch],+ socketResolver = resolver } return Transport { scheme = tcpScheme, handles = tcpHandles transport, bind = tcpBind transport,- sendTo = tcpSendTo transport,+ sendTo = socketSendTo transport, shutdown = tcpShutdown transport } ---------------------------------------------------------------------------------- -{-|-Parse a TCP 'Address' into its respective 'HostName' and 'PortNumber' components, on the-assumption the 'Address' has an identifer in the format @host:port@. If-the port number is missing from the supplied address, it will default to 0. If the-hostname component is missing from the identifier (e.g., just @:port@), then hostname-is assumed to be @localhost@.--}-parseTCPAddress :: Address -> (HostName,ServiceName)-parseTCPAddress address = - let identifer = T.pack $ address - parts = T.splitOn ":" identifer- in if (length parts) > 1 then- (host $ T.unpack $ parts !! 0, port $ T.unpack $ parts !! 1)- else (host $ T.unpack $ parts !! 0, "0")- where- host h = if h == "" then- "localhost"- else h- port p = p--tcpScheme :: Scheme-tcpScheme = "tcp"--tcpHandles :: TCPTransport -> Name -> IO Bool+tcpHandles :: SocketTransport -> Name -> IO Bool tcpHandles transport name = do - resolved <- resolve (tcpResolver transport) name+ resolved <- resolve (socketResolver transport) name return $ isJust resolved where isJust (Just _) = True isJust _ = False -tcpBind :: TCPTransport -> Mailbox -> Name -> IO (Either String Binding)-tcpBind transport inc name = do - atomically $ modifyTVar (tcpBindings transport) $ \bindings ->+tcpBind :: SocketTransport -> Mailbox -> Name -> IO (Either String Binding)+tcpBind transport inc name = do+ atomically $ modifyTVar (socketBindings transport) $ \bindings -> M.insert name inc bindings- Just address <- resolve (tcpResolver transport) name- let (_,port) = parseTCPAddress address+ Just address <- resolve (socketResolver transport) name+ let (_,port) = parseSocketAddress address listener <- async $ do infoM _log $ "Binding to address " ++ (show address) tcpListen address port@@ -142,9 +106,9 @@ where tcpListen address port = listen HostAny port $ \(socket,_) -> - catch (do + catchExceptions (do tcpAccept address socket)- (\e -> do + (\e -> do warningM _log $ "Listen error: " ++ (show (e :: SomeException))) tcpAccept address socket = do infoM _log $ "Listening for connections on " ++ (show address) ++ ": " ++ (show socket)@@ -159,9 +123,11 @@ Just (IdentifyMessage clientAddress) -> do infoM _log $ "Identified " ++ (show clientAddress) clientSocket <- atomically $ newTMVar client- msngr <- newMessenger clientSocket clientAddress (tcpInbound transport)- found <- atomically $ do - msngrs <- readTVar $ tcpMessengers transport+ newConn <- newTCPConnection clientAddress+ let conn = newConn {connSocket = clientSocket}+ msngr <- newMessenger conn (socketInbound transport)+ found <- atomically $ do+ msngrs <- readTVar $ socketMessengers transport return $ M.lookup clientAddress msngrs case found of Just _ -> do@@ -171,7 +137,7 @@ addMessenger transport clientAddress msngr tcpIdentify client clientAddress = do infoM _log $ "Awaiting identity from " ++ (show clientAddress)- maybeMsg <- receiveMessage client+ maybeMsg <- receiveSocketMessage client case maybeMsg of Nothing -> return Nothing Just bytes -> do@@ -183,203 +149,55 @@ infoM _log $ "Unbinding from port " ++ (show address) cancel listener -tcpSendTo :: TCPTransport -> Name -> Message -> IO ()-tcpSendTo transport name msg = do- isLocal <- local- if isLocal- then return ()- else remote- where- local = do- found <- atomically $ do- bindings <- readTVar $ tcpBindings transport- return $ M.lookup name bindings- case found of- Nothing -> return False- Just mbox -> do- atomically $ writeTQueue mbox msg- return True- remote = do - Just address <- resolve (tcpResolver transport) name- let env = encode $ Envelope {- envelopeDestination = name,- envelopeContents = msg- }- amsngr <- atomically $ do- msngrs <- readTVar $ tcpMessengers transport- return $ M.lookup address msngrs- case amsngr of- Nothing -> do- msngrs <- atomically $ readTVar $ tcpMessengers transport- infoM _log $ "No messenger for " ++ (show address) ++ " in " ++ (show msngrs)- socketVar <- atomically $ newEmptyTMVar- msngr <- newMessenger socketVar address (tcpInbound transport)- addMessenger transport address msngr- identifyAll msngr- deliver msngr env- return ()- Just msngr -> deliver msngr env- deliver msngr message = atomically $ writeTQueue (messengerOut msngr) message- identifyAll msngr = do- bindings <- atomically $ readTVar $ tcpBindings transport- boundAddresses <- mapM (resolve $ tcpResolver transport) (M.keys bindings)- let uniqueAddresses = S.toList $ S.fromList boundAddresses- mapM_ (identify msngr) uniqueAddresses- identify msngr maybeUniqueAddress= do- case maybeUniqueAddress of- Nothing -> return()- Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress--tcpShutdown :: TCPTransport -> IO ()-tcpShutdown transport = do- infoM _log $ "Closing messengers"- msngrs <- atomically $ readTVar $ tcpMessengers transport- mapM_ closeMessenger $ M.elems msngrs- infoM _log $ "Closing listeners"- listeners <- atomically $ readTVar $ tcpListeners transport- mapM_ sClose $ M.elems listeners- infoM _log $ "Closing dispatcher"- mapM_ cancel $ S.toList $ tcpDispatchers transport- mapM_ wait $ S.toList $ tcpDispatchers transport--data Messenger = Messenger {- messengerOut :: Mailbox,- messengerAddress :: Address,- messengerSender :: Async (),- messengerReceiver :: Async (),- messengerSocket :: TMVar Socket- }- -instance Show Messenger where- show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")"- -newMessenger :: TMVar Socket -> Address -> Mailbox -> IO Messenger -newMessenger socket address inc = do- out <- newMailbox- sndr <- async $ sender socket address out- rcvr <- async $ receiver socket address inc- return Messenger {- messengerOut = out,- messengerAddress = address,- messengerSender = sndr,- messengerReceiver = rcvr,- messengerSocket = socket+newTCPConnection :: Address -> IO Connection+newTCPConnection address = do+ sock <- atomically $ newEmptyTMVar+ let (host,port) = parseSocketAddress address+ return Connection {+ connAddress = address,+ connSocket = sock,+ connConnect = do+ (s,_) <- connectSock host port+ return s,+ -- connSend = send,+ connSend = tcpSend address,+ connReceive = recv,+ connClose = do+ open <- atomically $ tryTakeTMVar sock+ case open of+ Just socket -> sClose socket+ Nothing -> return () }- -addMessenger :: TCPTransport -> Address -> Messenger -> IO ()-addMessenger transport address msngr = do- msngrs <- atomically $ do- modifyTVar (tcpMessengers transport) $ \msngrs -> M.insert address msngr msngrs- msngrs <- readTVar (tcpMessengers transport)- return msngrs- infoM _log $ "Added messenger to " ++ (show address) ++ "; messengers are " ++ (show msngrs) -closeMessenger :: Messenger -> IO () -closeMessenger msngr = do- cancel $ messengerSender msngr- cancel $ messengerReceiver msngr- open <- atomically $ tryTakeTMVar $ messengerSocket msngr- case open of- Just socket -> sClose socket- Nothing -> return ()- +newTCPMessenger :: Bindings -> Resolver -> Connection -> Mailbox -> IO Messenger+newTCPMessenger bindings resolver conn mailbox = do+ msngr <- newMessenger conn mailbox+ identifyAll msngr+ return msngr+ where+ identifyAll msngr = do+ bs <- atomically $ readTVar bindings+ boundAddresses <- mapM (resolve resolver) (M.keys bs)+ let uniqueAddresses = S.toList $ S.fromList boundAddresses+ mapM_ (identify msngr) uniqueAddresses+ identify msngr maybeUniqueAddress= do+ case maybeUniqueAddress of+ Nothing -> return()+ Just uniqueAddress -> deliver msngr $ encode $ IdentifyMessage uniqueAddress -sender :: TMVar Socket -> Address -> Mailbox -> IO ()-sender socketVar address mailbox = sendMessages- where- sendMessages = do- reconnect- catch (do- infoM _log $ "Waiting to send to " ++ (show address)- msg <- atomically $ readTQueue mailbox- infoM _log $ "Sending message to " ++ (show address)- connected <- atomically $ tryReadTMVar socketVar- case connected of- Just socket -> do - send socket $ encode (B.length msg)- infoM _log $ "Length sent"- send socket msg- infoM _log $ "Message sent to" ++ (show address)- Nothing -> return ()- ) (\e -> do- warningM _log $ "Send error: " ++ (show (e :: SomeException))- disconnect)- sendMessages- reconnect = do- -- TODO need a timeout here, in case connecting always fails- connected <- atomically $ tryReadTMVar socketVar- case connected of- Just _ -> return ()- Nothing -> do- let (host,port) = parseTCPAddress address- infoM _log $ "Connecting to " ++ (show host) ++ ":" ++ (show port) -- (show address)- (socket,sockAddr) <- connectSock host port- infoM _log $ "Connected to " ++ (show address) ++ ": " ++ (show sockAddr)- atomically $ putTMVar socketVar socket- disconnect = do- connected <- atomically $ tryTakeTMVar socketVar- case connected of- Just socket -> sClose socket- Nothing -> return ()+tcpSend :: Address -> Socket -> B.ByteString -> IO ()+tcpSend addr sock bs = do+ send sock $ encode (B.length bs)+ infoM _log $ "Length sent"+ send sock bs+ infoM _log $ "Message sent to" ++ (show addr) -dispatcher :: TVar (M.Map Name Mailbox) -> Mailbox -> IO ()-dispatcher bindings mbox = dispatchMessages- where- dispatchMessages = catch (do- infoM _log $ "Dispatching messages"- env <- atomically $ readTQueue mbox- dispatchMessage env- dispatchMessages) - (\e -> do - warningM _log $ "Dispatch error: " ++ (show (e :: SomeException)))- dispatchMessage env = do- infoM _log $ "Dispatching message"- let envelopeOrErr = decode env- case envelopeOrErr of- Left err -> do- errorM _log $ "Error decoding message for dispatch: " ++ err- return ()- Right (Envelope destination msg) -> do - atomically $ do - dests <- readTVar bindings- let maybeDest = M.lookup destination dests- case maybeDest of- Nothing -> return ()- Just dest -> do - writeTQueue dest msg- return ()--receiver :: TMVar Socket -> Address -> Mailbox -> IO ()-receiver socketVar address mailbox = receiveMessages- where- receiveMessages = catch (do- infoM _log $ "Waiting to receive from " ++ (show address)- socket <- atomically $ readTMVar socketVar- maybeMsg <- receiveMessage socket- infoM _log $ "Received message from " ++ (show address)- case maybeMsg of- Nothing -> do- sClose socket- return ()- Just msg -> do- atomically $ writeTQueue mailbox msg- receiveMessages) (\e -> do - warningM _log $ "Receive error: " ++ (show (e :: SomeException)))- -receiveMessage :: Socket -> IO (Maybe Message) -receiveMessage socket = catch (do- maybeLen <- recv socket 8 -- TODO must figure out what defines length of an integer in bytes - case maybeLen of- Nothing -> do- errorM _log $ "No length received"- return Nothing- Just len -> do - maybeMsg <- recv socket $ msgLength (decode len)- infoM _log $ "Received message"- return maybeMsg) (\e -> do - warningM _log $ "Receive error: " ++ (show (e :: SomeException))- throw e)- where- msgLength (Right size) = size- msgLength (Left err) = error err+tcpShutdown :: SocketTransport -> IO ()+tcpShutdown transport = do+ infoM _log $ "Closing messengers"+ msngrs <- atomically $ readTVar $ socketMessengers transport+ mapM_ closeMessenger $ M.elems msngrs+ infoM _log $ "Closing dispatcher"+ mapM_ cancel $ S.toList $ socketDispatchers transport+ mapM_ wait $ S.toList $ socketDispatchers transport
+ src/Network/Transport/UDP.hs view
@@ -0,0 +1,181 @@+{-# LANGUAGE DeriveGeneric #-}++-----------------------------------------------------------------------------+-- |+-- Module : Network.Transport.UDP+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (uses STM)+--+-- UDP transports deliver messages to other 'Network.Endpoints.Endpoint's using UDP/IP.+--+-- Each UDP transport manages socket bindings on behalf of+-- 'Endpoint's, dynamically opening / closing new sockets as needed to deliver+-- messages to other 'Endpoint's using UDP transports.+--+-- There is no reuse of sockets on the sending side, so while messages will be +-- received on a known bound port, the remote sending port will vary arbitrarily.+-- +-- This transport only reads at most 512 bytes from incoming packets: constraining+-- the packet size avoids fragmentation. Applications using this transport should take+-- responsibility for fragmentation, reassembly, retransmission of lost packets,+-- and congestion control.+-----------------------------------------------------------------------------++module Network.Transport.UDP (+ newUDPTransport+ ) where++-- local imports++import Network.Transport+import Network.Transport.Internal+import Network.Transport.Sockets++-- external imports++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception++import qualified Data.ByteString as B+import qualified Data.Map as M+import qualified Data.Set as S++import qualified Network.Socket as N+import Network.Socket.ByteString(sendAllTo,recvFrom)++import System.Log.Logger+-- import qualified System.Random as R++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++_log :: String+_log = "transport.udp"++udpScheme :: Scheme+udpScheme = "udp"++newUDPTransport :: Resolver -> IO Transport+newUDPTransport resolver = do+ messengers <- atomically $ newTVar M.empty+ bindings <- atomically $ newTVar M.empty+ inbound <- newMailbox+ dispatch <- async $ dispatcher bindings inbound+ let transport = SocketTransport {+ socketMessengers = messengers,+ socketBindings = bindings,+ socketInbound = inbound,+ socketConnection = newUDPConnection,+ socketMessenger = newUDPMessenger,+ socketDispatchers = S.fromList [dispatch],+ socketResolver = resolver+ }+ return Transport {+ scheme = udpScheme,+ handles = udpHandles transport,+ bind = udpBind transport,+ sendTo = socketSendTo transport,+ shutdown = udpShutdown transport+ }++udpHandles :: SocketTransport -> Name -> IO Bool+udpHandles transport name = do + resolved <- resolve (socketResolver transport) name+ return $ isJust resolved+ where+ isJust (Just _) = True+ isJust _ = False++udpBind :: SocketTransport -> Mailbox -> Name -> IO (Either String Binding)+udpBind transport inc name = do+ atomically $ modifyTVar (socketBindings transport) $ \bindings ->+ M.insert name inc bindings+ Just address <- resolve (socketResolver transport) name+ let (_,port) = parseSocketAddress address+ addrinfos <- N.getAddrInfo+ (Just (N.defaultHints {N.addrFlags = [N.AI_PASSIVE]}))+ Nothing (Just port)+ let serveraddr = head addrinfos+ sock <- N.socket (N.addrFamily serveraddr) N.Datagram N.defaultProtocol+ -- have to set this option in case we frequently rebind sockets+ infoM _log $ "Binding to " ++ (show port) ++ " over UDP"+ N.setSocketOption sock N.ReuseAddr 1+ N.bindSocket sock $ N.addrAddress serveraddr+ infoM _log $ "Bound to " ++ (show port) ++ " over UDP"+ rcvr <- async $ udpReceiveSocketMessages sock address (socketInbound transport)+ return $ Right Binding {+ bindingName = name,+ unbind = do+ cancel rcvr+ infoM _log $ "Closing UDP socket bound on " ++ (show port)+ N.sClose sock+ }++newUDPConnection :: Address -> IO Connection+newUDPConnection address = do+ sock <- atomically newEmptyTMVar+ return Connection {+ connAddress = address,+ connSocket = sock,+ connConnect = N.socket N.AF_INET N.Datagram N.defaultProtocol,+ connSend = (\s bs -> do+ addr <- lookupAddress $ parseSocketAddress address+ infoM _log $ "Sending via UDP to " ++ (show addr)+ sendAllTo s bs addr+ infoM _log $ "Sent via UDP to " ++ (show addr)),+ connReceive = udpRecvFrom,+ connClose = do+ maybeSocket <- atomically $ tryTakeTMVar sock+ case maybeSocket of+ Just s -> N.sClose s+ Nothing -> return ()+ return ()+ }++newUDPMessenger :: Connection -> Mailbox -> IO Messenger+newUDPMessenger conn mailbox = do+ msngr <- newMessenger conn mailbox+ return msngr++udpReceiveSocketMessages :: N.Socket -> Address -> Mailbox -> IO ()+udpReceiveSocketMessages sock addr mailbox = catchExceptions (do+ infoM _log $ "Waiting to receive via UDP on " ++ (show addr)+ maybeMsg <- udpReceiveSocketMessage+ infoM _log $ "Received message via UDP on " ++ (show addr)+ case maybeMsg of+ Nothing -> do+ N.sClose sock+ return ()+ Just msg -> do+ atomically $ writeTQueue mailbox msg+ udpReceiveSocketMessages sock addr mailbox) (\e -> do + warningM _log $ "Receive error: " ++ (show (e :: SomeException)))+ where+ udpReceiveSocketMessage = do+ maybeMsg <- udpRecvFrom sock 512+ infoM _log $ "Received message"+ return maybeMsg++udpRecvFrom :: N.Socket -> Int -> IO (Maybe B.ByteString)+udpRecvFrom sock count = do+ (bs,addr) <- recvFrom sock count+ infoM _log $ "Received UDP message from " ++ (show addr) ++ ": " ++ (show bs)+ if B.null bs+ then return Nothing+ else return $ Just bs++udpShutdown :: SocketTransport -> IO ()+udpShutdown transport = do+ infoM _log $ "Unbinding transport"+ infoM _log $ "Closing messengers"+ msngrs <- atomically $ readTVar $ socketMessengers transport+ mapM_ closeMessenger $ M.elems msngrs+ infoM _log $ "Closing dispatcher"+ mapM_ cancel $ S.toList $ socketDispatchers transport+ mapM_ wait $ S.toList $ socketDispatchers transport+
− tests/TestMemory.hs
@@ -1,76 +0,0 @@-module TestMemory (tests) where---- local imports--import Network.Endpoints-import Network.Transport.Memory---- external imports--import Data.Serialize--import Test.Framework-import Test.HUnit-import Test.Framework.Providers.HUnit--- import Test.Framework.Providers.QuickCheck2---------------------------------------------------------------------------------------------------------------------------------------------------------------tests :: [Test.Framework.Test]-tests = - [- testCase "mem-endpoints+transport" testEndpointTransport,- testCase "mem-bind" testEndpointBind,- testCase "mem-unbind" testEndpointBindUnbind,- testCase "mem-sendReceive" testEndpointSendReceive,- testCase "mem-transport" testMemoryTransport- ] - -testEndpointTransport :: Assertion -testEndpointTransport = do - transport <- newMemoryTransport- _ <- newEndpoint [transport]- return ()- -testEndpointBind :: Assertion-testEndpointBind = do - let name1 = "endpoint1"- transport <- newMemoryTransport- endpoint <- newEndpoint [transport]- Right () <- bindEndpoint endpoint name1- return ()--testEndpointBindUnbind :: Assertion-testEndpointBindUnbind = do - let name1 = "endpoint1"- transport <- newMemoryTransport- endpoint <- newEndpoint [transport]- Right () <- bindEndpoint endpoint name1- unbound <- unbindEndpoint endpoint name1- case unbound of- Left err -> assertFailure $ "Unbind failed: " ++ err- Right () -> assertBool "Unbind succeeded" True- return ()- -testEndpointSendReceive :: Assertion -testEndpointSendReceive = do- let name1 = "endpoint1"- name2 = "endpoint2"- transport <- newMemoryTransport- endpoint1 <- newEndpoint [transport]- endpoint2 <- newEndpoint [transport]- Right () <- bindEndpoint endpoint1 name1- Right () <- bindEndpoint endpoint2 name2- _ <- sendMessage endpoint1 name2 $ encode "hello!"- msg <- receiveMessage endpoint2 - assertEqual "Received message not same as sent" (Right "hello!") (decode msg)- return ()- --- Memory tests- -testMemoryTransport :: Assertion-testMemoryTransport = do- _ <- newMemoryTransport- return ()-
− tests/TestTCP.hs
@@ -1,223 +0,0 @@-module TestTCP (tests) where---- local imports--import Network.Endpoints-import Network.Transport.TCP---- external imports--import Control.Concurrent-import Control.Exception--import Data.Serialize--import System.Log.Logger--import Test.Framework-import Test.HUnit-import Test.Framework.Providers.HUnit--- import Test.Framework.Providers.QuickCheck2---------------------------------------------------------------------------------------------------------------------------------------------------------------_log :: String-_log = "test.transport.tcp"--testDelay :: Int-testDelay = 1 * 1000000--address1 :: Address-address1 = "localhost:2001"--address2 :: Address-address2 = "localhost:2002"--tests :: [Test.Framework.Test]-tests = - [- testCase "tcp-endpoints+transport" testEndpointTransport,- testCase "tcp-bind-unbind" testEndpointBindUnbind,- testCase "tcp-send-receive" testEndpointSendReceive,- testCase "tcp-double-send-receive" testEndpointDoubleSendReceive,- testCase "tcp-send-receive-reply" testEndpointSendReceiveReply,- testCase "tcp-multiple-send-receive-reply" testEndpointMultipleSendReceiveReply,- testCase "tcp-local-send-receive-reply" testEndpointLocalSendReceiveReply- ] - -testEndpointTransport :: Assertion -testEndpointTransport = do - let name1 = "endpoint1"- name2 = "endpoint2"- let resolver = resolverFromList [(name1,address1),- (name2,address2)]- bracket (newTCPTransport resolver)- shutdown- (\transport -> do - _ <- newEndpoint [transport]- return ())- -testEndpointBindUnbind :: Assertion-testEndpointBindUnbind = do - let name1 = "endpoint1"- name2 = "endpoint2"- let resolver = resolverFromList [(name1,address1),- (name2,address2)]- bracket (newTCPTransport resolver)- shutdown- (\transport -> do - endpoint <- newEndpoint [transport]- Right () <- bindEndpoint endpoint name1- unbound <- unbindEndpoint endpoint name1- case unbound of- Left err -> assertFailure $ "Unbind failed: " ++ err- Right () -> assertBool "Unbind succeeded" True- return ())- -testEndpointSendReceive :: Assertion-testEndpointSendReceive = do- let name1 = "endpoint1"- name2 = "endpoint2"- let resolver = resolverFromList [(name1,address1),- (name2,address2)]- bracket (newTCPTransport resolver)- shutdown- (\transport1 -> - bracket (newTCPTransport resolver)- shutdown- (\transport2 -> do - endpoint1 <- newEndpoint [transport1]- endpoint2 <- newEndpoint [transport2]- Right () <- bindEndpoint endpoint1 name1- Right () <- bindEndpoint endpoint2 name2- threadDelay testDelay-- infoM _log "Sending message from 1 to 2"- sendMessage_ endpoint1 name2 $ encode "hello!"- Just msg <- receiveMessageTimeout endpoint2 testDelay- assertEqual "Received message not same as sent" (Right "hello!") (decode msg)- Right () <- unbindEndpoint endpoint1 name1- Right () <- unbindEndpoint endpoint2 name2- return ()))- -testEndpointDoubleSendReceive :: Assertion-testEndpointDoubleSendReceive = do- let name1 = "endpoint1"- name2 = "endpoint2"- name3 = "endpoint3"- let resolver = resolverFromList [(name1,address1),- (name2,address2),- (name3,address1)]- bracket (newTCPTransport resolver)- shutdown- (\transport1 -> do - bracket (newTCPTransport resolver)- shutdown- (\transport2 -> do - endpoint1 <- newEndpoint [transport1]- endpoint2 <- newEndpoint [transport2]- endpoint3 <- newEndpoint [transport1]- Right () <- bindEndpoint endpoint1 name1- Right () <- bindEndpoint endpoint2 name2- Right () <- bindEndpoint endpoint3 name3- threadDelay testDelay-- verifiedSend endpoint1 endpoint2 name1 name2 "hello"- verifiedSend endpoint3 endpoint2 name3 name2 "ciao!"- - Right () <- unbindEndpoint endpoint1 name1- - verifiedSend endpoint3 endpoint2 name3 name2 "hi!"- - Right () <- unbindEndpoint endpoint2 name2- Right () <- unbindEndpoint endpoint3 name3- return ()))--testEndpointSendReceiveReply :: Assertion-testEndpointSendReceiveReply = do- let name1 = "endpoint1"- name2 = "endpoint2"- let resolver = resolverFromList [(name1,address1),- (name2,address2)]- bracket (newTCPTransport resolver)- shutdown- (\transport1 -> - bracket (newTCPTransport resolver)- shutdown- (\transport2 -> do - endpoint1 <- newEndpoint [transport1]- endpoint2 <- newEndpoint [transport2]- Right () <- bindEndpoint endpoint1 name1- Right () <- bindEndpoint endpoint2 name2- threadDelay testDelay- - verifiedSend endpoint1 endpoint2 name1 name2 "hello"- verifiedSend endpoint2 endpoint1 name2 name1 "hi!"- - Right () <- unbindEndpoint endpoint1 name1- Right () <- unbindEndpoint endpoint2 name2- - return ()))- -testEndpointLocalSendReceiveReply :: Assertion-testEndpointLocalSendReceiveReply = do- let name1 = "endpoint1"- name2 = "endpoint2"- let resolver = resolverFromList [(name1,address1),- (name2,address1)]- bracket (newTCPTransport resolver)- shutdown- (\transport1 -> do- endpoint1 <- newEndpoint [transport1]- endpoint2 <- newEndpoint [transport1]- Right () <- bindEndpoint endpoint1 name1- Right () <- bindEndpoint endpoint2 name2- threadDelay testDelay- - verifiedSend endpoint1 endpoint2 name1 name2 "hello"- verifiedSend endpoint2 endpoint1 name2 name1 "hi!"- - Right () <- unbindEndpoint endpoint1 name1- Right () <- unbindEndpoint endpoint2 name2- - return ())- -testEndpointMultipleSendReceiveReply :: Assertion-testEndpointMultipleSendReceiveReply = do- let name1 = "endpoint1"- name2 = "endpoint2"- let resolver = resolverFromList [(name1,address1),- (name2,address2)]- bracket (newTCPTransport resolver)- shutdown- (\transport1 -> - bracket (newTCPTransport resolver)- shutdown- (\transport2 -> do - endpoint1 <- newEndpoint [transport1]- endpoint2 <- newEndpoint [transport2]- Right () <- bindEndpoint endpoint1 name1- Right () <- bindEndpoint endpoint2 name2- threadDelay testDelay-- roundtrip endpoint1 endpoint2 name1 name2- roundtrip endpoint2 endpoint1 name2 name1- roundtrip endpoint1 endpoint2 name1 name2- roundtrip endpoint2 endpoint1 name2 name1- - Right () <- unbindEndpoint endpoint1 name1- Right () <- unbindEndpoint endpoint2 name2- - return ()))- where- roundtrip endpoint1 endpoint2 name1 name2 = do- verifiedSend endpoint1 endpoint2 name1 name2 "hello" - verifiedSend endpoint2 endpoint1 name2 name1 "hi!"--verifiedSend :: Endpoint -> Endpoint -> Name -> Name -> String -> Assertion-verifiedSend endpoint1 endpoint2 name1 name2 msg = do- infoM _log $ "Sending message from " ++ name1 ++ " to " ++ name2- sendMessage_ endpoint1 name2 $ encode msg- Just msg1 <- receiveMessageTimeout endpoint2 testDelay- assertEqual "Received message not same as sent" (Right msg) (decode msg1)
tests/Tests.hs view
@@ -13,11 +13,11 @@ import Test.Framework import Test.HUnit import Test.Framework.Providers.HUnit-import Test.Framework.Providers.QuickCheck2 -- Test modules import qualified TestMemory as M import qualified TestTCP as T+import qualified TestUDP as U ----------------------------------------------------------------------------- -----------------------------------------------------------------------------@@ -26,29 +26,27 @@ main = do initLogging defaultMain tests- -initLogging :: IO () ++initLogging :: IO () initLogging = do let logFile = "tests.log" exists <- doesFileExist logFile- if exists - then removeFile logFile + if exists+ then removeFile logFile else return () s <- fileHandler logFile INFO updateGlobalLogger rootLoggerName (setLevel WARNING) updateGlobalLogger rootLoggerName (addHandler s) tests :: [Test.Framework.Test]-tests = +tests = [ testCase "hunit" (assertBool "HUnit assertion of truth is false" True),- testProperty "quickcheck" True, testCase "endpoints" testEndpoint- ] + ] ++ M.tests ++ T.tests- -- ++ D.tests- -- ++ S.tests+ ++ U.tests -- Endpoint tests