courier (empty) → 0.1.0.0
raw patch · 10 files changed
+1062/−0 lines, 10 filesdep +HUnitdep +QuickCheckdep +asyncsetup-changed
Dependencies added: HUnit, QuickCheck, async, base, bytestring, cereal, containers, courier, directory, hslogger, network, network-simple, stm, test-framework, test-framework-hunit, test-framework-quickcheck2, text
Files
- LICENSE +19/−0
- Setup.hs +2/−0
- courier.cabal +74/−0
- src/Network/Endpoints.hs +151/−0
- src/Network/Transport.hs +136/−0
- src/Network/Transport/Memory.hs +84/−0
- src/Network/Transport/TCP.hs +328/−0
- tests/TestMemory.hs +76/−0
- tests/TestTCP.hs +133/−0
- tests/Tests.hs +59/−0
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright (c) 2013 Phil Hargett++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ courier.cabal view
@@ -0,0 +1,74 @@+-- Initial courier.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/++name: courier+version: 0.1.0.0+synopsis: A message-passing library, intended for simplifying network applications+description: Inspired 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.+homepage: http://github.com/hargettp/courier+license: MIT+license-file: LICENSE+author: Phil Hargett+maintainer: phil@haphazardhouse.net+copyright: Copyright (c) 2013 Phil Hargett++category: Network+build-type: Simple+cabal-version: >=1.8++source-repository head+ type: git+ location: https://github.com/hargettp/courier.git++library+ hs-source-dirs: src++ exposed-modules:+ Network.Endpoints+ Network.Transport+ Network.Transport.Memory+ Network.Transport.TCP++ ghc-options: -Wall+ -- other-modules: + build-depends: base >=4 && <5,+ async,+ bytestring,+ cereal,+ containers,+ hslogger,+ network,+ network-simple,+ stm,+ text++Test-Suite test-courier+ 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+ async,+ cereal,+ directory,+ hslogger,+ stm,+ -- this project's modules+ courier++ ghc-options: -Wall -threaded
+ src/Network/Endpoints.hs view
@@ -0,0 +1,151 @@+{-# LANGUAGE ExistentialQuantification #-}++-----------------------------------------------------------------------------+-- |+-- Module : Network.Endpoints+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- 'Endpoint's are a generalized abstraction for communication between parts of a program,+-- whether on the same physical host or distributed over a network. 'Endpoint's are intended+-- to simplify the development of network-centric applications by providing a small transport-independent+-- message-passing interface, and application writers can independently alter their implementation+-- by enabling their 'Endpoint's with different 'Transport's without modifying the logic of their+-- application that sends / receives 'Message's.+--+-----------------------------------------------------------------------------++module Network.Endpoints (+ + -- * Primary API+ Endpoint,+ newEndpoint,+ + bindEndpoint,+ unbindEndpoint,+ + sendMessage,+ receiveMessage,+ + -- * Transports+ {-|+ Transports define specific implementations of message-passing techniques (e.g.,+ memory-based, TCP, UDP, HTTP, etc.). Typical use of the 'Endpoint's does not+ require direct use of 'Transport's, beyond creating specific 'Transport's (such as+ found in "Network.Transport.Memory" and "Network.Transport.TCP") and adding+ them to an 'Endpoint'.+ -}+ module Network.Transport+ + ) where++-- local imports++import Network.Transport++-- external imports++import Control.Concurrent.STM++import qualified Data.Map as M++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++{-|+Endpoints are a locus of communication, used for sending and receive messages.+-}+data Endpoint = Endpoint {+ endpointTransports :: TVar [Transport],+ endpointBindings :: TVar (M.Map Name Binding),+ endpointMailbox :: Mailbox+ }++{-|+Create a new 'Endpoint' using the provided transports.+-}+newEndpoint :: [Transport] -> IO Endpoint+newEndpoint trans = do+ transports <- atomically $ newTVar trans+ bindings <- atomically $ newTVar M.empty+ mailbox <- atomically $ newTQueue+ return Endpoint {+ endpointTransports = transports,+ endpointBindings = bindings,+ endpointMailbox = mailbox+ }++{-|+Binding an 'Endpoint' to a 'Name' prepares the 'Endpoint' to receive+messages sent to the bound name. Upon success, the result will be @Right ()@, but+if failed, @Left text-of-error-message@.+-}+bindEndpoint :: Endpoint -> Name -> IO (Either String ())+bindEndpoint endpoint name = do + maybeTransport <- findTransport endpoint name+ case maybeTransport of+ Nothing -> return $ Left $ "No transport to handle name: " ++ (show name)+ Just transport -> do+ eitherBinding <- bind transport (endpointMailbox endpoint) name+ case eitherBinding of+ Left err -> return $ Left err+ Right binding -> do + atomically $ modifyTVar (endpointBindings endpoint)+ (\bindings -> M.insert name binding bindings)+ return $ Right ()++{-|+Unbind an 'Endpoint' from a 'Name', after which the 'Endpoint' will eventually not +receive messages sent to that 'Name'. Note that there is no guarantee that after 'Unbind'+succeeds that additional messages to that 'Name' will not be delivered: the only guarantee+is that eventually messages will no longer be delivered.+Upon success, the result will be @Right ()@ but+if failed, @Left text-of-error-message@.+-}+unbindEndpoint :: Endpoint -> Name -> IO (Either String ())+unbindEndpoint endpoint name = do+ bindings <- atomically $ readTVar $ endpointBindings endpoint+ let maybeBinding = M.lookup name bindings+ case maybeBinding of+ Nothing -> return $ Left $ "Endpoint not bound to address: " ++ (show name)+ Just binding -> do + unbind binding+ return $ Right ()++{-|+Send a 'Message' to specific 'Name' via the indicated 'Endpoint'. While a successful+response (indicated by returning @Right ()@) indicates that there was no error initiating+transport of the message, success does not guarantee that an 'Endpoint' received the message.+Failure initiating transport is indicated by returning @Left text-of-error-message@.+-}+sendMessage :: Endpoint -> Name -> Message -> IO (Either String ())+sendMessage endpoint name msg = do+ maybeTransport <- findTransport endpoint name+ case maybeTransport of+ Nothing -> return $ Left $ "No transport to handle name: " ++ (show name)+ Just transport -> do + sendTo transport name msg+ return $ Right ()++{-|+Receive the next 'Message' sent to the 'Endpoint'.+-}+receiveMessage :: Endpoint -> IO Message+receiveMessage endpoint = atomically $ readTQueue $ endpointMailbox endpoint++findTransport :: Endpoint -> Name -> IO (Maybe Transport)+findTransport endpoint name = do+ transports <- atomically $ readTVar $ endpointTransports endpoint+ findM canHandle transports+ where+ canHandle transport = (handles transport) name+ findM mf (a:as) = do+ result <- mf a+ if result+ then return $ Just a+ else findM mf as+ findM _ [] = return Nothing
+ src/Network/Transport.hs view
@@ -0,0 +1,136 @@+{-# LANGUAGE DeriveGeneric #-}++-----------------------------------------------------------------------------+-- |+-- Module : Network.Transport+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- A 'Transport' abstracts the details of message delivery, and defines the interfaces+-- that specific 'Transport' implementations should provide in order to deliver messages+-- for 'Endpoint's.+--+-----------------------------------------------------------------------------++module Network.Transport (+ Address,+ Binding(..),+ Envelope(..),+ Mailbox,+ newMailbox,+ Message,+ Name,+ Resolver,+ resolve,+ resolverFromList,+ Scheme,+ Transport(..), + ) where++-- local imports++-- external imports+import Control.Concurrent.STM++import Data.ByteString as B+import Data.Serialize++import GHC.Generics++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++{-|+Messages are containers for arbitrary data that may be sent to other 'Network.Endpoints.Endpoint's.+-}+type Message = B.ByteString++{-|+Name for uniquely identifying an 'Endpoint'; suitable for identifying+the target destination for a 'Message'.++-}+type Name = String++{-|+An 'Envelope' is a container for a 'Message' with the 'Address' of the 'Message''s destination.++-}+data Envelope = Envelope {+ envelopeDestination :: Name,+ envelopeContents :: Message+ } deriving (Eq,Show,Generic)++instance Serialize Envelope++{-|+A 'Mailbox' is a place where transports can put messages for 'Network.Endpoint.Endpoint's+to receive. Typically 'Network.Endpoint.Endpoint's will use the same 'Mailbox' when+binding or connecting with a 'Transport'.+-}+type Mailbox = TQueue Message++{-|+Create a new mailbox.+-}+newMailbox :: IO Mailbox+newMailbox = atomically $ newTQueue++{-|+An address is a logical identifier suitable for establishing a connection to+another 'Endpoint' over a 'Transport'. It's use (if at all) is specific to the 'Transport'+in question.+-}+type Address = String++{-|+A scheme is an identifier for a discrete type of transport.+-}+type Scheme = String++{-|+Bindings are a site for receiving messages on a particular 'Address'+through a 'Transport'.+-}+data Binding = Binding {+ bindingName :: Name,+ unbind :: IO ()+ }++{-|+A 'Transport' defines a specific method for establishing connections+between 'Endpoint's.+-}+data Transport = Transport {+ scheme :: String,+ handles :: Name -> IO Bool,+ bind :: Mailbox -> Name -> IO (Either String Binding),+ sendTo :: Name -> Message -> IO (),+ shutdown :: IO ()+ }++{-|+A 'Resolver' translates a name into an 'Address', if possible. +'Transport's may find resolvers useful for determing+where to reach a specific 'Endpoint', given it''s 'Name'.+-}+newtype Resolver = Resolver (Name -> IO (Maybe Address))++{-|+Ask the 'Resolver' to find one or more 'Address'es for the provided+'Name', if any are available from this resolver.++-}+resolve :: Resolver -> Name -> IO (Maybe Address)+resolve (Resolver resolver) name = resolver name++{-|+A simple 'Resolver' that accepts an association list of 'Name's to 'Address'es+and returns the addresses associated with a given name in the list.+-}+resolverFromList :: [(Name,Address)] -> Resolver+resolverFromList addresses = Resolver (\name -> return $ lookup name addresses)
+ src/Network/Transport/Memory.hs view
@@ -0,0 +1,84 @@+-----------------------------------------------------------------------------+-- |+-- Module : Network.Transport.Memory+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- Memory transports deliver messages to other 'Network.Endpoints.Endpoint's within the same shared+-- address space, or operating system process.++-- Internally memory transports use a set of 'TQueue's to deliver messages to 'Network.Endpoint.Endpoint's. +-- Memory transports are not global in nature: 'Network.Endpoint.Endpoint's can only communicate with+-- one another if each has added the same memory 'Transport' and each invoked 'bind' on that shared+-- transport.+--+-----------------------------------------------------------------------------++module Network.Transport.Memory (+ newMemoryTransport+ ) where++-- local imports++import Network.Transport++-- external imports++import Control.Concurrent.STM+import qualified Data.Map as M+import Data.Maybe (fromJust)++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++memoryScheme :: Scheme+memoryScheme = "mem"++data MemoryTransport = MemoryTransport {+ boundMailboxes :: TVar (M.Map Name Mailbox)+ }+ +{-|+Create a new memory 'Transport' for use by 'Network.Endpoint.Endpoint's.+-}+newMemoryTransport :: IO Transport+newMemoryTransport = do + bindings <- atomically $ newTVar M.empty+ let transport = MemoryTransport {+ boundMailboxes = bindings+ }+ return Transport {+ scheme = memoryScheme,+ handles = memoryHandles transport,+ bind = memoryBind transport,+ sendTo = memorySendTo transport,+ shutdown = return ()+ }++memoryBind :: MemoryTransport -> Mailbox -> Name -> IO (Either String Binding)+memoryBind transport mailbox name = do+ atomically $ modifyTVar (boundMailboxes transport) + (\mailboxes -> M.insert name mailbox mailboxes)+ return $ Right Binding {+ bindingName = name,+ unbind = memoryUnbind transport name+ }+ +memoryHandles :: MemoryTransport -> Name -> IO Bool+-- memoryHandles transport name = True+memoryHandles _ _ = return True++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++memoryUnbind :: MemoryTransport -> Name -> IO ()+memoryUnbind transport name = do+ atomically $ modifyTVar (boundMailboxes transport) deleteBinding+ where deleteBinding m = M.delete name m
+ src/Network/Transport/TCP.hs view
@@ -0,0 +1,328 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveGeneric #-}++-----------------------------------------------------------------------------+-- |+-- Module : Network.Transport.TCP+-- Copyright : (c) Phil Hargett 2013+-- License : MIT (see LICENSE file)+-- +-- Maintainer : phil@haphazardhouse.net+-- Stability : experimental+-- Portability : non-portable (uses STM)+--+-- TCP transports deliver messages to other 'Network.Endpoints.Endpoint's using TCP/IP.+--+-- Each TCP transport manages both socket bindings and connections on behalf of+-- 'Endpoint's, dynamically opening / closing new sockets as needed to deliver+-- messages to other 'Endpoint's using TCP transports.+-----------------------------------------------------------------------------++module Network.Transport.TCP (+ newTCPTransport+ ) where++-- local imports++import Network.Transport++-- 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.Socket (HostName,ServiceName,Socket,sClose,accept)+import Network.Simple.TCP hiding (accept)++import System.Log.Logger++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------++_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++{-|+Create a new 'Transport' suitable for sending messages over TCP/IP. There can+be multiple instances of these 'Transport's: 'Network.Endpoints.Endpoint' using+different instances will still be able to communicate, provided they use+correct TCP/IP addresses (or hostnames) for communication.+-}+newTCPTransport :: Resolver -> IO Transport+newTCPTransport resolver = do + listeners <- atomically $ newTVar M.empty+ 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+ }+ return Transport {+ scheme = tcpScheme,+ handles = tcpHandles transport,+ bind = tcpBind transport,+ sendTo = tcpSendTo 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 transport name = do + resolved <- resolve (tcpResolver 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 ->+ M.insert name inc bindings+ Just address <- resolve (tcpResolver transport) name+ let (_,port) = parseTCPAddress address+ listener <- async $ do + infoM _log $ "Binding to address " ++ (show address)+ tcpListen address port+ return $ Right Binding {+ bindingName = name,+ unbind = tcpUnbind listener address+ }+ where+ tcpListen address port = listen HostAny port $ \(socket,_) -> do + tcpAccept address socket+ tcpAccept address socket = do+ (client,clientAddress) <- accept socket+ _ <- async $ tcpDispatch address client clientAddress+ tcpAccept address socket+ tcpDispatch address client socketAddress = do+ infoM _log $ "Accepted connection on " ++ (show address)+ identity <- tcpIdentify client socketAddress+ case identity of+ Nothing -> sClose client+ Just (IdentifyMessage clientAddress) -> do+ infoM _log $ "Identified " ++ (show clientAddress)+ msngr <- newMessenger client clientAddress (tcpInbound transport)+ found <- atomically $ do + msngrs <- readTVar $ tcpMessengers transport+ return $ M.lookup clientAddress msngrs+ case found of+ Just _ -> closeMessenger msngr+ Nothing -> do+ addMessenger transport clientAddress msngr+ tcpIdentify client clientAddress = do+ infoM _log $ "Awaiting identity from " ++ (show clientAddress)+ maybeMsg <- receiveMessage client+ case maybeMsg of+ Nothing -> return Nothing+ Just bytes -> do+ let msg = decode bytes+ case msg of+ Left _ -> return Nothing+ Right message -> return $ Just message+ tcpUnbind listener address = do + infoM _log $ "Unbinding from port " ++ (show address)+ cancel listener++tcpSendTo :: TCPTransport -> Name -> Message -> IO ()+tcpSendTo transport name msg = 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+ let (host,port) = parseTCPAddress address+ infoM _log $ "Connecting to " ++ (show address)+ (socket,_) <- connectSock host port+ infoM _log $ "Connected to " ++ (show address)+ msngr <- newMessenger socket address (tcpInbound transport)+ addMessenger transport address msngr+ -- identify all bindings+ identifyAll msngr+ deliver msngr env+ return ()+ Just msngr -> deliver msngr env+ where+ 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++data Messenger = Messenger {+ messengerOut :: Mailbox,+ messengerAddress :: Address,+ messengerSender :: Async (),+ messengerReceiver :: Async (),+ messengerSocket :: Socket+ }+ +instance Show Messenger where+ show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ "," ++ (show $ messengerSocket msngr) ++ ")"+ +newMessenger :: 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+ }+ +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+ sClose $ messengerSocket msngr++sender :: Socket -> Address -> Mailbox -> IO ()+sender socket address mailbox = sendMessages+ where+ sendMessages = do + catch (do+ infoM _log $ "Waiting to send to " ++ (show address)+ msg <- atomically $ readTQueue mailbox+ infoM _log $ "Sending message to " ++ (show address)+ send socket $ encode (B.length msg)+ infoM _log $ "Length sent"+ send socket msg+ infoM _log $ "Message sent to" ++ (show address)+ ) (\e -> do + errorM _log $ "Send error " ++ (show (e :: SomeException))+ throw e)+ sendMessages++dispatcher :: TVar (M.Map Name Mailbox) -> Mailbox -> IO ()+dispatcher bindings mbox = dispatchMessages+ where+ dispatchMessages = do+ infoM _log $ "Dispatching messages"+ env <- atomically $ readTQueue mbox+ dispatchMessage env+ dispatchMessages+ 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 :: Socket -> Address -> Mailbox -> IO ()+receiver socket address mailbox = receiveMessages+ where+ receiveMessages = do+ infoM _log $ "Waiting to receive from " ++ (show address)+ maybeMsg <- receiveMessage socket+ infoM _log $ "Received message from " ++ (show address)+ case maybeMsg of+ Nothing -> return ()+ Just msg -> atomically $ writeTQueue mailbox msg+ +receiveMessage :: Socket -> IO (Maybe Message) +receiveMessage socket = 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+ where+ msgLength (Right size) = size+ msgLength (Left err) = error err
+ tests/TestMemory.hs view
@@ -0,0 +1,76 @@+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 view
@@ -0,0 +1,133 @@+module TestTCP (tests) where++-- local imports++import Network.Endpoints+import Network.Transport.TCP++-- external imports++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+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"++tests :: [Test.Framework.Test]+tests = + [+ testCase "tcp-endpoints+transport" testEndpointTransport,+ testCase "tcp-bind-unbind" testEndpointBindUnbind,+ testCase "tcp-send-receive" testEndpointSendReceive,+ testCase "tcp-send-receive-reply" testEndpointSendReceiveReply+ ] + +testEndpointTransport :: Assertion +testEndpointTransport = do + let name1 = "endpoint1"+ name2 = "endpoint2"+ let resolver = resolverFromList [(name1,"localhost:2000"),+ (name2,"localhost:2001")]+ transport <- newTCPTransport resolver+ _ <- newEndpoint [transport]+ return ()+ +testEndpointBindUnbind :: Assertion+testEndpointBindUnbind = do + let name1 = "endpoint1"+ name2 = "endpoint2"+ let resolver = resolverFromList [(name1,"localhost:2000"),+ (name2,"localhost:2001")]+ transport <- newTCPTransport resolver+ 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,"localhost:2000"),+ (name2,"localhost:2001")]+ bracket (newTCPTransport resolver)+ (\transport1 -> shutdown transport1)+ (\transport1 -> + bracket (newTCPTransport resolver)+ (\transport2 -> shutdown transport2)+ (\transport2 -> do + endpoint1 <- newEndpoint [transport1]+ endpoint2 <- newEndpoint [transport2]+ Right () <- bindEndpoint endpoint1 name1+ Right () <- bindEndpoint endpoint2 name2+ threadDelay $ 1 * 1000000+ _ <- sendMessage endpoint1 name2 $ encode "hello!"+ msgVar <- atomically $ newTVar Nothing+ withAsync (do+ msg <- receiveMessage endpoint2 + atomically $ writeTVar msgVar $ Just msg)+ (\_ -> do + threadDelay (2 * 1000000)+ Just msg <- atomically $ readTVar msgVar+ assertEqual "Received message not same as sent" (Right "hello!") (decode msg))+ return ()))+ +testEndpointSendReceiveReply :: Assertion+testEndpointSendReceiveReply = do+ let name1 = "endpoint1"+ name2 = "endpoint2"+ let resolver = resolverFromList [(name1,"localhost:2002"),+ (name2,"localhost:2003")]+ 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 $ 1 * 1000000+ + infoM _log "Sending message from 1 to 2"+ _ <- sendMessage endpoint1 name2 $ encode "hello!"+ msgVar1 <- atomically $ newTVar Nothing+ withAsync (do+ msg <- receiveMessage endpoint2 + atomically $ writeTVar msgVar1 $ Just msg)+ (\_ -> do + threadDelay (1 * 1000000)+ Just msg <- atomically $ readTVar msgVar1+ assertEqual "Received message not same as sent" (Right "hello!") (decode msg))+ + infoM _log "Sending message from 2 to 1"+ _ <- sendMessage endpoint2 name1 $ encode "hi!"+ msgVar2 <- atomically $ newTVar Nothing+ withAsync (do+ msg <- receiveMessage endpoint1+ atomically $ writeTVar msgVar2 $ Just msg)+ (\_ -> do + threadDelay (1 * 1000000)+ Just msg <- atomically $ readTVar msgVar2+ assertEqual "Received message not same as sent" (Right "hi!") (decode msg))+ + return ()))+
+ tests/Tests.hs view
@@ -0,0 +1,59 @@+module Main where++-- local imports++import Network.Endpoints++-- external imports++import System.Directory+import System.Log.Logger+import System.Log.Handler.Simple++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++-----------------------------------------------------------------------------+-----------------------------------------------------------------------------++main :: IO ()+main = do + initLogging+ defaultMain tests+ +initLogging :: IO () +initLogging = do+ let logFile = "tests.log"+ exists <- doesFileExist 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 = + [+ testCase "hunit" (assertBool "HUnit assertion of truth is false" True),+ testProperty "quickcheck" True,+ testCase "endpoints" testEndpoint+ ] + ++ M.tests+ ++ T.tests+ -- ++ D.tests+ -- ++ S.tests++-- Endpoint tests++testEndpoint :: Assertion+testEndpoint = do+ _ <- newEndpoint []+ return ()+