diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/courier.cabal b/courier.cabal new file mode 100644 --- /dev/null +++ b/courier.cabal @@ -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 diff --git a/src/Network/Endpoints.hs b/src/Network/Endpoints.hs new file mode 100644 --- /dev/null +++ b/src/Network/Endpoints.hs @@ -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 diff --git a/src/Network/Transport.hs b/src/Network/Transport.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport.hs @@ -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) diff --git a/src/Network/Transport/Memory.hs b/src/Network/Transport/Memory.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Memory.hs @@ -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 diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/TCP.hs @@ -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 diff --git a/tests/TestMemory.hs b/tests/TestMemory.hs new file mode 100644 --- /dev/null +++ b/tests/TestMemory.hs @@ -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 () + diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs new file mode 100644 --- /dev/null +++ b/tests/TestTCP.hs @@ -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 ())) + diff --git a/tests/Tests.hs b/tests/Tests.hs new file mode 100644 --- /dev/null +++ b/tests/Tests.hs @@ -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 () +