diff --git a/changes.md b/changes.md --- a/changes.md +++ b/changes.md @@ -1,3 +1,17 @@ +0.1.1.1 + +There are significant changes in this release, due to a complete rewrite of TCP connection handling, and, +in so doing, a rewrite of the basic model for how transports and endpoints interact. This rewrite fixes +bug #10. + +Endpoints now simply have inbound and outbound mailboxes, and a set of names to which the endpoint is bound +via a transport. The intent is that endpoints are discrete entities orthogonal to the transport which moves +messages between them. Its the job of the transport to move messages from the outbound mailbox of one endpoint +to the inbound mailbox of the endpoint bound to the name specified in the envelope for a given message. + +New helpers for managing the lifecycle of bindings and connections now exist, and also a helper for setting the name +of a client. + 0.1.0.15 * Initial implementation of IPv6 transports (TCP and UDP) @@ -102,16 +116,16 @@ types of messages. Each message pump can operate independently of one another. * Fixed issue #2 and other interim bugs all resulting from multiple name bindings that resolve - the same underlying address. Now, bound sockets are reused (with reference counting) so + the same underlying address. Now, bound sockets are reused (with reference counting) so that if there are multiple bindings to the same address only 1 socket is created and user. - * Unit tests passing on Mac OS X again, mostly due to correct management of sockets and + * Unit tests passing on Mac OS X again, mostly due to correct management of sockets and appropriate reuse. - + 0.1.0.7 * First inclusion of changelog in package - + 0.1.0.6 * Added simple implementation of transport for UDP diff --git a/courier.cabal b/courier.cabal --- a/courier.cabal +++ b/courier.cabal @@ -1,10 +1,10 @@ name: courier -version: 0.1.0.15 +version: 0.1.1.1 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 +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. . A primary driver of this package's design is separation of concerns: many algorithms @@ -14,12 +14,12 @@ the distributed algorithm itself. With this in mind, this package aims to provide a variety of transports as well as support for common communication idioms (e.g., in order message delivery, selective out of order message delivery, - synchronous RPC, etc.). + synchronous RPC, etc.). . Applications may process messages in the order received at an endpoint, or use selective message reception to process the first message arriving at an endpoint that also matches a provided selection function. Through selective message reception, - applications may approximate the style of an Erlang application, and enjoy better + applications may approximate the style of an Erlang application, and enjoy better composability of message reception with multiple independent dispatch routines or message pumps. @@ -30,8 +30,8 @@ tests/TestMemory.hs tests/TestRPC.hs tests/TestTCP.hs - tests/TestTransports.hs tests/TestUDP.hs + tests/TransportTestSuite.hs tests/TestUtils.hs homepage: http://github.com/hargettp/courier @@ -56,15 +56,21 @@ Network.Endpoints Network.Transport Network.Transport.Memory - Network.Transport.TCP - Network.Transport.UDP + Network.Transport.Sockets + Network.Transport.Sockets.TCP + Network.Transport.Sockets.UDP Control.Concurrent.Mailbox Network.RPC - ghc-options: -Wall -fno-state-hack - other-modules: - Network.Transport.Internal - Network.Transport.Sockets + ghc-options: + -Wall + -fno-state-hack + -threaded + -- -O2 + -- -fprof-auto + + -- other-modules: + build-depends: base >=4.6 && <5, async, bytestring, @@ -76,16 +82,38 @@ text, uuid +executable echo-server + hs-source-dirs: echo + main-is: EchoServer.hs + ghc-options: + -Wall + -fno-state-hack + -threaded + build-depends: base >=4.6 && <5, + cereal, + courier + +executable echo-client + hs-source-dirs: echo + main-is: EchoClient.hs + ghc-options: + -Wall + -fno-state-hack + -threaded + build-depends: base >=4.6 && <5, + cereal, + courier + Test-Suite test-courier type: exitcode-stdio-1.0 hs-source-dirs: tests main-is: Tests.hs - build-depends: + build-depends: -- base modules - base, + base, -- testing support - HUnit, + HUnit, test-framework, test-framework-hunit, -- Cabal, @@ -100,4 +128,9 @@ -- this project's modules courier - ghc-options: -Wall -threaded + ghc-options: + -Wall + -threaded + -- -O2 + -- -fprof-auto + -- "-with-rtsopts=-N -p -s -h -i0.1" diff --git a/echo/EchoClient.hs b/echo/EchoClient.hs new file mode 100644 --- /dev/null +++ b/echo/EchoClient.hs @@ -0,0 +1,58 @@ +----------------------------------------------------------------------------- +-- | +-- Module : EchoClient +-- Copyright : (c) Phil Hargett 2015 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Simple echo client for testing out courier. +-- +----------------------------------------------------------------------------- + +module Main where + +-- local imports + +import Network.Endpoints +import Network.RPC +import Network.Transport.Sockets.TCP + +-- external imports + +import Control.Exception + +import Data.Serialize + +import System.Environment +import System.IO + +----------------------------------------------------------------------------- +----------------------------------------------------------------------------- + +main :: IO () +main = do + [serverNameStr,clientNameStr] <- getArgs + let server = Name serverNameStr + client = Name clientNameStr + endpoint <- newEndpoint + withTransport (newTCPTransport4 tcpSocketResolver4) $ \transport -> + withEndpoint transport endpoint $ + withName endpoint client $ + withConnection transport endpoint server $ do + hPutStrLn stdout $ "Started echo client on " ++ (show client) ++ " for " ++ (show server) + finally (echo endpoint server client) + (hPutStrLn stdout $ "\nStopped echo client on " ++ (show client) ++ " for " ++ (show server)) + +echo :: Endpoint -> Name -> Name -> IO () +echo endpoint server client = do + text <- hGetLine stdin + let cs = newCallSite endpoint client + response <- call cs server "echo" $ encode text + case decode response of + Left _ -> error "Could not decode message" + Right responseText -> do + hPutStrLn stdout $ "> " ++ responseText + echo endpoint server client diff --git a/echo/EchoServer.hs b/echo/EchoServer.hs new file mode 100644 --- /dev/null +++ b/echo/EchoServer.hs @@ -0,0 +1,55 @@ +----------------------------------------------------------------------------- +-- | +-- Module : EchoServer +-- Copyright : (c) Phil Hargett 2015 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Simple echo server for testing out courier. +-- +----------------------------------------------------------------------------- + +module Main where + +-- local imports + +import Network.Endpoints +import Network.RPC +import Network.Transport.Sockets.TCP + +-- external imports + +import Control.Exception + +import Data.Serialize + +import System.Environment +import System.IO + +----------------------------------------------------------------------------- +----------------------------------------------------------------------------- + +main :: IO () +main = do + [nameStr] <- getArgs + let server = Name nameStr + endpoint <- newEndpoint + withTransport (newTCPTransport4 tcpSocketResolver4) $ \transport -> + withEndpoint transport endpoint $ + withBinding transport endpoint server $ do + hPutStrLn stdout $ "Started echo server on " ++ (show server) + finally (echo endpoint server) + (hPutStrLn stdout $ "\nStopped echo server on " ++ (show server)) + +echo :: Endpoint -> Name -> IO () +echo endpoint server = do + (bytes,reply) <- hear endpoint server "echo" + case decode bytes of + Left _ -> error "Could not decode message" + Right text -> do + hPutStrLn stdout text + reply $ encode text + echo endpoint server diff --git a/examples/HelloWorld.hs b/examples/HelloWorld.hs --- a/examples/HelloWorld.hs +++ b/examples/HelloWorld.hs @@ -1,31 +1,54 @@ -module HelloWorld ( - main -) where +----------------------------------------------------------------------------- +-- | +-- Module : HelloWorld +-- Copyright : (c) Phil Hargett 2015 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- Basic code sample for explaining how to use courier. +-- +----------------------------------------------------------------------------- - -- Just import this package to access the primary APIs +module Main where + + -- Import this package to manage endpoints import Network.Endpoints -- A specific transport is necessary, however -import Network.Transport.TCP +import Network.Transport.Sockets.TCP -- Needed for serialization import Data.Serialize main :: IO () main = do - let name1 = "endpoint1" - name2 = "endpoint2" - resolver = resolverFromList [(name1,"localhost:2000"), - (name2,"localhost:2001")] - transport <- newTCPTransport resolver - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - sendMessage_ endpoint1 name2 $ encode "hello world!" - msg <- receiveMessage endpoint2 - let Right txt = decode msg - in print (txt :: String) - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - shutdown transport + -- each endpoint needs a name; since we're using TCP + -- as our transport, they need to be host/port pairs + let name1 = Name "localhost:9001" + name2 = Name "localhost:9002" + -- the default resolvers just pull apart a name into separate + -- host and port components; more elaborate resolvers could + -- perform name lookups or other translations + resolver = tcpSocketResolver4 + -- we need endpoints for each end of the communication + endpoint1 <- newEndpoint + endpoint2 <- newEndpoint + -- we need a transport to move messages between endpoints + withTransport (newTCPTransport4 resolver) $ \transport -> + withEndpoint transport endpoint1 $ + withEndpoint transport endpoint2 $ + -- the first endpoint is just a client, so it needs a name to receive + -- responses, but does not need a binding since it isn't accept connections + withName endpoint1 name1 $ + -- the second endpoint is a server, so it needs a binding + withBinding transport endpoint2 name2 $ + -- a connection between the first endpoint and the name of the second + -- creates a bi-directional path for messages to flow between the endpoints + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello world!" + msg <- receiveMessage endpoint2 + let Right txt = decode msg + in print (txt :: String) diff --git a/src/Control/Concurrent/Mailbox.hs b/src/Control/Concurrent/Mailbox.hs --- a/src/Control/Concurrent/Mailbox.hs +++ b/src/Control/Concurrent/Mailbox.hs @@ -3,7 +3,7 @@ -- Module : Control.Concurrent.Mailbox -- Copyright : (c) Phil Hargett 2014 -- License : MIT (see LICENSE file) --- +-- -- Maintainer : phil@haphazardhouse.net -- Stability : experimental -- Portability : non-portable (uses STM) @@ -11,7 +11,7 @@ -- A 'Mailbox' is a drop-in replacement for 'TQueue' in "Control.Concurrent.STM", except that -- it also supports selective out of order message reception: that is, it allows the caller -- to dequeue the first message among the messages available in the queue that matches --- a supplied test function, or block if no such match is possible with the messages currently +-- a supplied test function, or block if no such match is possible with the messages currently -- in the queue. -- -- As 'Mailbox' implements the same basic @read / write / peek@ group of functions as a 'TQueue', @@ -76,7 +76,10 @@ -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- -data Mailbox m = Mailbox +{-| +A 'Mailbox' is a variation of a 'Control.Concurrent.TQueue', with the added feature of selective message retrieval. +-} +data Mailbox m = Mailbox {-# UNPACK #-} !(TVar [m]) {-# UNPACK #-} !(TVar [m]) @@ -234,7 +237,7 @@ Nothing -> return Nothing {-| -Put a data item back onto a channel, where it will be the next item read. +Put a data item back onto a mailbox, where it will be the next item read. -} unGetMailbox :: Mailbox m -> m -> STM () unGetMailbox (Mailbox _read _write) msg = do @@ -259,7 +262,7 @@ -- {-| -Extract the first element from a list matching the +Extract the first element from a list matching the provided test and return a new list without the matching element. -} @@ -272,7 +275,7 @@ Just v -> (Just v,xs) {-| -Find the first element from a list matching the +Find the first element from a list matching the provided test, or Nothing if there is no match. -} find :: (m -> Maybe v) -> [m] -> Maybe v diff --git a/src/Network/Endpoints.hs b/src/Network/Endpoints.hs --- a/src/Network/Endpoints.hs +++ b/src/Network/Endpoints.hs @@ -1,11 +1,14 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE OverloadedStrings #-} ----------------------------------------------------------------------------- -- | -- Module : Network.Endpoints -- Copyright : (c) Phil Hargett 2013 -- License : MIT (see LICENSE file) --- +-- -- Maintainer : phil@haphazardhouse.net -- Stability : experimental -- Portability : non-portable (requires STM) @@ -14,29 +17,27 @@ -- 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 +-- by enabling their 'Endpoint's with different 'Network.Transport's without modifying the logic of their -- application that sends / receives 'Message's. -- ----------------------------------------------------------------------------- module Network.Endpoints ( - + -- * How to use courier in an application -- $use - + -- * Primary API - Endpoint, + Endpoint(..), newEndpoint, - - bindEndpoint, - bindEndpoint_, - unbindEndpoint, - unbindEndpoint_, - + + withName, + bindName, + unbindName, + BindException(..), + sendMessage, - sendMessage_, broadcastMessage, - broadcastMessage_, receiveMessage, receiveMessageTimeout, postMessage, @@ -46,193 +47,196 @@ selectMessageTimeout, detectMessage, detectMessageTimeout, - dispatchMessage, - dispatchMessageTimeout, - - -- * 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 - + + -- * Other types + Envelope(..), + Message, + Name(..), + ) where -- local imports -import Network.Transport - -- external imports import Control.Concurrent import Control.Concurrent.Async +import Control.Concurrent.Mailbox import Control.Concurrent.STM +import Control.Exception -import qualified Data.Map as M +import qualified Data.ByteString as B +import Data.Serialize +import qualified Data.Set as S +import Data.Typeable +import GHC.Generics + -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- +_log :: String +_log = "transport.endpoints" + -- $use --- +-- -- A sample of how to use this library: --- --- > module HelloWorld ( --- > main --- > ) where +-- +-- > module Main where -- > --- > -- Just import this package to access the primary APIs +-- > -- Import this package to manage endpoints -- > import Network.Endpoints -- > --- > -- A specific transport is necessary, however --- > import Network.Transport.TCP +-- > -- A specific transport is necessary, however +-- > import Network.Transport.Sockets.TCP -- > -- > -- Needed for serialization -- > import Data.Serialize -- > -- > main :: IO () -- > main = do --- > let name1 = "endpoint1" --- > name2 = "endpoint2" --- > resolver = resolverFromList [(name1,"localhost:2000"), --- > (name2,"localhost:2001")] --- > transport <- newTCPTransport resolver --- > endpoint1 <- newEndpoint [transport] --- > endpoint2 <- newEndpoint [transport] --- > Right () <- bindEndpoint endpoint1 name1 --- > Right () <- bindEndpoint endpoint2 name2 --- > sendMessage_ endpoint1 name2 $ encode "hello world!" --- > msg <- receiveMessage endpoint2 --- > let Right txt = decode msg --- > in print (txt :: String) --- > Right () <- unbindEndpoint endpoint1 name1 --- > Right () <- unbindEndpoint endpoint2 name2 --- > shutdown transport +-- > -- each endpoint needs a name; since we're using TCP +-- > -- as our transport, they need to be host/port pairs +-- > let name1 = Name "localhost:9001" +-- > name2 = Name "localhost:9002" +-- > -- the default resolvers just pull apart a name into separate +-- > -- host and port components; more elaborate resolvers could +-- > -- perform name lookups or other translations +-- > resolver = tcpSocketResolver4 +-- > -- we need endpoints for each end of the communication +-- > endpoint1 <- newEndpoint +-- > endpoint2 <- newEndpoint +-- > -- we need a transport to move messages between endpoints +-- > withTransport (newTCPTransport4 resolver) $ \transport -> +-- > withEndpoint transport endpoint1 $ +-- > withEndpoint transport endpoint2 $ +-- > -- the first endpoint is just a client, so it needs a name to receive +-- > -- responses, but does not need a binding since it isn't accept connections +-- > withName endpoint1 name1 $ +-- > -- the second endpoint is a server, so it needs a binding +-- > withBinding transport endpoint2 name2 $ +-- > -- a connection between the first endpoint and the name of the second +-- > -- creates a bi-directional path for messages to flow between the endpoints +-- > withConnection transport endpoint1 name2 $ do +-- > sendMessage endpoint1 name2 $ encode "hello world!" +-- > msg <- receiveMessage endpoint2 +-- > let Right txt = decode msg +-- > in print (txt :: String) +-- > +-- {-| 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 Message + -- | The 'Transport' used by this 'Endpoint' + -- endpointTransport :: Transport, + -- | The 'Mailbox' where inbound 'Message's that the 'Endpoint' will receive are queued + endpointInbound :: Mailbox Message, + endpointOutbound :: Mailbox Envelope, + -- | The 'Name's to which the 'Endpoint' is bound + boundEndpointNames :: TVar (S.Set Name ) } {-| -Create a new 'Endpoint' using the provided transports. +Messages are containers for arbitrary data that may be sent to other 'Network.Endpoints.Endpoint's. -} -newEndpoint :: [Transport] -> IO Endpoint -newEndpoint trans = do - transports <- atomically $ newTVar trans - bindings <- atomically $ newTVar M.empty - mailbox <- atomically $ newMailbox - return Endpoint { - endpointTransports = transports, - endpointBindings = bindings, - endpointMailbox = mailbox - } +type Message = B.ByteString {-| -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@. +Name for uniquely identifying an 'Endpoint'; suitable for identifying +the target destination for a 'Message'. The specific interpretation of a name +is left to each 'Network.Transport.Transport' -} -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 () +newtype Name = Name String deriving (Eq,Ord,Show,Generic) +instance Serialize Name + {-| -Invoke 'bindEndpoint', but ignore any returned result (success or failure). + An 'Envelope' wraps a 'Message' with the 'Name's of the destination for the message and (optionally) +the origin. -} -bindEndpoint_ :: Endpoint -> Name -> IO () -bindEndpoint_ endpoint name = do - _ <- bindEndpoint endpoint name - return () +data Envelope = Envelope { + messageOrigin :: Maybe Name, + messageDestination :: Name, + envelopeMessage :: Message +} deriving (Eq,Show) + {-| -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@. +Create a new 'Endpoint' using the provided transports. -} -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 () +newEndpoint :: IO Endpoint +newEndpoint = atomically $ do + inbound <- newMailbox + outbound <- newMailbox + names <- newTVar S.empty + return Endpoint { + endpointInbound = inbound, + endpointOutbound = outbound, + boundEndpointNames = names + } {-| -Invoke 'unbindEndpoint', but ignore any returned result (success or failure). +Declare an 'Endpoint' as having the specified 'Name' while the supplied function executes. This can +often be useful for establishing the 'Name' of a client or initiator of a 'Network.Transport.Connection', +without requiring the client also have a 'Network.Transport.Binding'. + -} -unbindEndpoint_ :: Endpoint -> Name -> IO () -unbindEndpoint_ endpoint name = do - _ <- unbindEndpoint endpoint name - return () +withName :: Endpoint -> Name -> IO () -> IO () +withName endpoint origin actor = do + atomically $ bindName endpoint origin + finally actor $ atomically $ unbindName endpoint origin {-| -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@. +Establish 'Name' as one of the 'boundEndpointNames' for an 'Endpoint'. Throws 'BindingExists' if +the 'Endpoint' is already bound to the 'Name'. -} -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 () +bindName :: Endpoint -> Name -> STM () +bindName endpoint name = do + bindings <- readTVar $ boundEndpointNames endpoint + if S.member name bindings + then throw $ BindingExists name + else modifyTVar (boundEndpointNames endpoint) $ S.insert name {-| -A variant of 'sendMessage' for use when the return value can be ignored. - +Remove 'Name' as one of the 'boundEndpointNames' for an 'Endpoint'. Throws 'BindingDoesNotExist' +if the 'Endpoint' is not bound to the 'Name'. -} -sendMessage_ :: Endpoint -> Name -> Message -> IO () -sendMessage_ endpoint name msg = do - _ <- sendMessage endpoint name msg - return () +unbindName :: Endpoint -> Name -> STM () +unbindName endpoint name = modifyTVar (boundEndpointNames endpoint) $ \bindings -> do + case S.member name bindings of + False -> throw $ BindingDoesNotExist name + True -> S.delete name bindings {-| -Helper for sending a single 'Message' to several 'Endpoint's. +Exceptions generated when `Network.Transport.bind`ing a 'Name'. -} -broadcastMessage :: Endpoint -> [Name] -> Message -> IO [(Either String ())] -broadcastMessage endpoint names msg = do - mapM (\name -> sendMessage endpoint name msg) names +data BindException = + BindingExists Name | + BindingDoesNotExist Name + deriving (Show,Typeable) +instance Exception BindException + {-| -Variant of 'broadcastMessage' that ignores the results of sending. +Send a 'Message' to specific 'Name' via the indicated 'Endpoint'. +-} +sendMessage :: Endpoint -> Name -> Message -> IO () +sendMessage endpoint name msg = atomically $ + writeMailbox (endpointOutbound endpoint) $ Envelope Nothing name msg +{-| +Helper for sending a single 'Message' to several 'Endpoint's. -} -broadcastMessage_ :: Endpoint -> [Name] -> Message -> IO () -broadcastMessage_ endpoint names msg = do - _ <- broadcastMessage endpoint names msg - return () +broadcastMessage :: Endpoint -> [Name] -> Message -> IO () +broadcastMessage endpoint names msg = mapM_ (\name -> sendMessage endpoint name msg) names {-| Receive the next 'Message' sent to the 'Endpoint', blocking until a message is available. -} receiveMessage :: Endpoint -> IO Message -receiveMessage endpoint = atomically $ readMailbox $ endpointMailbox endpoint +receiveMessage endpoint = atomically $ readMailbox $ endpointInbound endpoint {-| Wait for a message to be received within the timeout, blocking until either a message @@ -251,9 +255,8 @@ may be useful for applications that prefer to use the 'Endpoint''s 'Mailbox' as a general queue of ordered messages. -} -postMessage :: Endpoint -> Message -> IO () -postMessage endpoint message = do - atomically $ writeMailbox (endpointMailbox endpoint) message +postMessage :: Endpoint -> Message -> STM () +postMessage endpoint = writeMailbox (endpointInbound endpoint) {-| Select the next available message in the 'Endpoint' 'Mailbox' matching @@ -261,9 +264,7 @@ differs from 'receiveMessage' in that it supports out of order message reception. -} selectMessage :: Endpoint -> (Message -> Maybe v) -> IO v -selectMessage endpoint testFn = do - msg <- atomically $ selectMailbox (endpointMailbox endpoint) testFn - return msg +selectMessage endpoint testFn = atomically $ selectMailbox (endpointInbound endpoint) testFn {-| Wait for a message to be selected within the timeout, blocking until either a message @@ -285,9 +286,7 @@ message if it is not consumed immediately. -} detectMessage :: Endpoint -> (Message -> Maybe v) -> IO v -detectMessage endpoint testFn = do - msg <- atomically $ findMailbox (endpointMailbox endpoint) testFn - return msg +detectMessage endpoint testFn = atomically $ findMailbox (endpointInbound endpoint) testFn {-| Find a 'Message' in the 'Endpoint' 'Mailbox' matching the supplied @@ -301,33 +300,3 @@ case resultOrTimeout of Left result -> return $ Just result Right () -> return Nothing - -{-| -Dispatch the next available message in the 'Endpoint' 'Mailbox' matching -the supplied test function, or blocking until one is available. Once a -matching message is found, handle the message with the supplied handler -and return any result obtained. This function differs from 'receiveMessage' -in that it supports out of order message reception. --} -dispatchMessage :: Endpoint -> (Message -> Maybe v) -> (v -> IO r) -> IO r -dispatchMessage endpoint = handleMailbox (endpointMailbox endpoint) - -dispatchMessageTimeout :: Endpoint -> Int -> (Message -> Maybe v) -> (v -> IO r) -> IO (Maybe r) -dispatchMessageTimeout endpoint delay testFn handleFn = do - resultOrTimeout <- race (dispatchMessage endpoint testFn handleFn) (threadDelay delay) - case resultOrTimeout of - Left result -> return $ Just result - Right () -> return Nothing - -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/RPC.hs b/src/Network/RPC.hs --- a/src/Network/RPC.hs +++ b/src/Network/RPC.hs @@ -83,13 +83,17 @@ -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- +{-| +An identifier for what method to invoke on the receiving 'Endpoint'. If 'hear' has been invoked on the 'Endpoint' +with a matching identifier, then calls will be delivered to that invocation of 'hear'. +-} type Method = String data RPCMessageType = Req | Rsp deriving (Eq,Show,Enum,Generic) instance Serialize RPCMessageType -{- +{-| A unique identifier for a 'Request' -} newtype RequestId = RequestId (Word32, Word32, Word32, Word32) deriving (Generic,Eq,Show) @@ -104,13 +108,17 @@ ruuid <- nextRandom return $ RequestId $ toWords ruuid +{-| +Encapsulates the initiating side of a 'call': every invocation of 'call' produces a 'Request' that is sent +to the destination 'Endpoint', where the 'hear'ing side will generate a 'Response' after completing the request' +-} data Request = Request { requestId :: RequestId, requestCaller :: Name, requestMethod :: Method, requestArgs :: Message } deriving (Eq,Show) - + instance Serialize Request where put req = do put Req @@ -125,7 +133,10 @@ method <- get args <- get return $ Request rid caller method args - +{-| +Encapsulates the completion side of a 'call': every invocation of 'call' produces a 'Request' that is sent +to the destination 'Endpoint', where the 'hear'ing side will generate a 'Response' after completing the request' +-} data Response = Response { responseId :: RequestId, responseFrom :: Name, @@ -168,7 +179,7 @@ call (CallSite endpoint from) name method args = do rid <- mkRequestId let req = Request {requestId = rid,requestCaller = from,requestMethod = method, requestArgs = args} - sendMessage_ endpoint name $ encode req + sendMessage endpoint name $ encode req selectMessage endpoint $ \msg -> do case decode msg of Left _ -> Nothing @@ -207,7 +218,7 @@ recvAll req M.empty where sendAll req = do - forM_ names $ \name -> sendMessage_ endpoint name $ encode req + forM_ names $ \name -> sendMessage endpoint name $ encode req recv req = selectMessage endpoint $ \msg -> do case decode msg of Left _ -> Nothing @@ -248,7 +259,7 @@ return $ complete partialResults where sendAll req = do - forM_ names $ \name -> sendMessage_ endpoint name $ encode req + forM_ names $ \name -> sendMessage endpoint name $ encode req recv req = selectMessage endpoint $ \msg -> do case decode msg of Left _ -> Nothing @@ -287,7 +298,7 @@ recvAny req where sendAll req = do - forM_ names $ \name -> sendMessage_ endpoint name $ encode req + forM_ names $ \name -> sendMessage endpoint name $ encode req recvAny req = selectMessage endpoint $ \msg -> do case decode msg of Left _ -> Nothing @@ -332,7 +343,7 @@ return (args, reply caller rid) where reply caller rid result = do - sendMessage_ endpoint caller $ encode $ Response rid name result + sendMessage endpoint caller $ encode $ Response rid name result {-| @@ -348,7 +359,7 @@ Nothing -> return Nothing where reply caller rid result = do - sendMessage_ endpoint caller $ encode $ Response rid name result + sendMessage endpoint caller $ encode $ Response rid name result {-| A 'HandleSite' is a just reference to the actual handler of a specific method. diff --git a/src/Network/Transport.hs b/src/Network/Transport.hs --- a/src/Network/Transport.hs +++ b/src/Network/Transport.hs @@ -1,105 +1,195 @@ -{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DeriveDataTypeable #-} ----------------------------------------------------------------------------- -- | -- Module : Network.Transport --- Copyright : (c) Phil Hargett 2013 +-- Copyright : (c) Phil Hargett 2015 -- 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. +-- that specific 'Transport' implementations should provide in order to move messages +-- between 'Endpoint's. -- -- The definition of a transport is deliberately low-level in nature. Unless a specific -- transport describes itself as supporting features like guaranteed delivery, applications -- should NOT assume that message delivery is reliable. -- --- For example, if a sender sends a message to a name that has not yet been bound, then --- immediately waits on the response for that message, then the application may hang, +-- For example, if a sender sends a message to a name that has not yet been bound, then +-- immediately waits on the response for that message, then the application may hang, -- as the original message may have been dropped. -- -- However, many application may find it easier to push features such as reliable -- message delivery into a custom transport, leaving the application only having -- to concern itself with the messages being delivered rather than how they arrive. -- +-- The main abstractions common to all transports: +-- +-- * 'Endpoint's may not receive messages until either 'bind' has been called on an available +-- 'Transport', or a 'bindName' has been called on the 'Endpoint'. The latter is typically useful +-- for 'Endpoint's that originate connections but do not accept them directly. +-- * A 'Connection' defines a bi-directional pathway for messages to flow between 'Endpoint's. The initiator +-- of the connection is a client, and the destination for the connection is a server. Server 'Endpoint's +-- may have to have called 'bind' on the 'Transport' before being able to accept connections or receive messages; +-- a client only has to have called 'bindName' in order to receive responses from the server. +-- * Client's proactively attempt to maintain 'Connection's to the server; in the event a server breaks the connection +-- (other because of a crash or deliberate exit), the client will continue to restore the connection. +-- * 'Connection's essentially define the topology over which messages will propagate between 'Endpoint's. +-- ----------------------------------------------------------------------------- - module Network.Transport ( - Address, - Binding(..), - Envelope(..), - Message, - Name, - Resolver, - resolve, - resolverFromList, - Scheme, + + -- * Transport Transport(..), + TransportException(..), - module Control.Concurrent.Mailbox - - ) where + Dispatcher(..), + dispatcher, + Mailboxes, + pullMessage, + dispatchMessage, + + withTransport, + + withEndpoint, + + Binding(..), + withBinding, + withBinding2, + withBinding3, + withBinding4, + + Connection(..), + withConnection, + withConnection2, + withConnection3, + withConnection4, + +) where + -- local imports -import Control.Concurrent.Mailbox +import Network.Endpoints + -- external imports -import Data.ByteString as B -import Data.Serialize +import Control.Concurrent.Async +import Control.Concurrent.Mailbox +import Control.Concurrent.STM +import Control.Exception -import GHC.Generics +import qualified Data.Map as M +import qualified Data.Set as S +import Data.Typeable -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- {-| -Messages are containers for arbitrary data that may be sent to other 'Network.Endpoints.Endpoint's. +A 'Transport' defines a specific method for establishing connections +between 'Endpoint's. -} -type Message = B.ByteString +data Transport = Transport { + bind :: Endpoint -> Name -> IO Binding, + dispatch :: Endpoint -> IO Dispatcher, + connect :: Endpoint -> Name -> IO Connection, + shutdown :: IO () + } {-| -Name for uniquely identifying an 'Endpoint'; suitable for identifying -the target destination for a 'Message'. +A 'Dispatcher' encapsulates a 'Transport's interaction with an 'Endpoint' for outbound message flow. +-} +data Dispatcher = Dispatcher { + stop :: IO () +} +{-| +An exception encountered by a 'Transport'. -} -type Name = String +data TransportException = + NoDataRead | + DataUnderflow + deriving (Show,Typeable) +instance Exception TransportException + {-| -An 'Envelope' is a container for a 'Message' with the 'Address' of the 'Message''s destination. +Wraps in a continually repeating call to 'dispatchMessage' in a 'Dispatcher' so that dispatching +can be stopped when no longer needed. +-} +dispatcher :: Mailboxes -> Endpoint -> IO Dispatcher +dispatcher mailboxes endpoint = do + d <- async disp + return Dispatcher { + stop = cancel d + } + where + disp = do + atomically $ do + envelope <- readMailbox $ endpointOutbound endpoint + let name = messageDestination envelope + msg = envelopeMessage envelope + dispatchMessage mailboxes name msg + disp +{-| +A mutable 'Map' of 'Name's to 'Mailbox'es of 'Message's. -} -data Envelope = Envelope { - envelopeDestination :: Name, - envelopeContents :: Message - } deriving (Eq,Show,Generic) +type Mailboxes = TVar (M.Map Name (Mailbox Message)) -instance Serialize Envelope +{-| +Pull a 'Message' intended for a destination name from a 'Mailboxes' directly, +without the use of a 'Transport' +-} +pullMessage :: Mailboxes -> Name -> STM Message +pullMessage mailboxes destination = do + outbound <- readTVar mailboxes + case M.lookup destination outbound of + Nothing -> retry + Just mailbox -> readMailbox mailbox {-| -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'. +A simple function to multiplex messages (each wrapped in an 'Envelope') in the 'endpointOutbound' +mailbox of an 'Endpoint' to one or more 'Mailboxes' by extract the 'messageDestination' from the 'Envelope' and +finding or creating a 'Mailbox' containing messages only for that destination. This is often useful for 'Transport's, +who then only have to monitor a specific 'Mailbox' to know when there are messages to send to a particular destination. -} +dispatchMessage :: Mailboxes -> Name -> Message -> STM () +dispatchMessage mailboxes name message = do + outbound <- readTVar mailboxes + mailbox <- case M.lookup name outbound of + Nothing -> do + mailbox <- newMailbox + modifyTVar mailboxes $ M.insert name mailbox + return mailbox + Just mailbox -> return mailbox + writeMailbox mailbox message {-| -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. +Within the body of the function, ensures that 'Message's are dispatched as necessary. -} -type Address = String +withTransport :: IO Transport -> (Transport -> IO ()) -> IO () +withTransport factory fn = do + transport <- factory + finally (fn transport) + (shutdown transport) {-| -A scheme is an identifier for a discrete type of transport. +Within the body of the function, ensure that there is a 'Dispatcher' for the 'Endpoint'. + -} -type Scheme = String +withEndpoint :: Transport -> Endpoint -> IO () -> IO () +withEndpoint transport endpoint fn = do + d <- dispatch transport endpoint + finally fn + (stop d) {-| -Bindings are a site for receiving messages on a particular 'Address' +Bindings are a site for receiving messages on a particular 'Name' through a 'Transport'. -} data Binding = Binding { @@ -108,35 +198,91 @@ } {-| -A 'Transport' defines a specific method for establishing connections -between 'Endpoint's. +A helper for ensuring there is a 'Binding' of a specific 'Endpoint' to a specific 'Name' +on the provided 'Transport' during a function. + -} -data Transport = Transport { - scheme :: String, - handles :: Name -> IO Bool, - bind :: Mailbox Message -> Name -> IO (Either String Binding), - sendTo :: Name -> Message -> IO (), - shutdown :: IO () - } +withBinding :: Transport -> Endpoint -> Name -> IO () -> IO () +withBinding transport endpoint name actor = do + atomically $ do + bindings <- readTVar $ boundEndpointNames endpoint + if S.member name bindings + then throw $ BindingExists name + else modifyTVar (boundEndpointNames endpoint) $ S.insert name + binding <- bind transport endpoint name + finally actor $ do + unbind binding + atomically $ modifyTVar (boundEndpointNames endpoint) $ S.delete name {-| -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'. +A helper for ensuring there are 'Binding's of a specific 'Endpoint' to specific 'Name's +on the provided 'Transport' during a function. -} -newtype Resolver = Resolver (Name -> IO (Maybe Address)) +withBinding2 :: Transport -> (Endpoint,Name) -> (Endpoint,Name) -> IO () -> IO () +withBinding2 transport (endpoint1,name1) (endpoint2,name2) fn = + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 fn {-| -Ask the 'Resolver' to find one or more 'Address'es for the provided -'Name', if any are available from this resolver. +A helper for ensuring there are 'Binding's of a specific 'Endpoint' to specific 'Name's +on the provided 'Transport' during a function. +-} +withBinding3 :: Transport -> (Endpoint,Name) -> (Endpoint,Name) -> (Endpoint,Name) -> IO () -> IO () +withBinding3 transport (endpoint1,name1) (endpoint2,name2) (endpoint3,name3) fn = + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ + withBinding transport endpoint3 name3 fn +{-| +A helper for ensuring there are 'Binding's of a specific 'Endpoint' to specific 'Name's +on the provided 'Transport' during a function. -} -resolve :: Resolver -> Name -> IO (Maybe Address) -resolve (Resolver resolver) name = resolver name +withBinding4 :: Transport -> (Endpoint,Name) -> (Endpoint,Name) -> (Endpoint,Name) -> (Endpoint,Name) -> IO () -> IO () +withBinding4 transport (endpoint1,name1) (endpoint2,name2) (endpoint3,name3) (endpoint4,name4) fn = + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ + withBinding transport endpoint3 name3 $ + withBinding transport endpoint4 name4 fn {-| -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. +Connections are bi-directional pathways for exchanging 'Message's with another 'Endpoint' +that is bound to a specific 'Name' on a shared 'Transport'. -} -resolverFromList :: [(Name,Address)] -> Resolver -resolverFromList addresses = Resolver (\name -> return $ lookup name addresses) +data Connection = Connection { + disconnect :: IO () +} + +{-| +A helper for ensuring that a 'Connection' is maintained during execution of a function. +-} +withConnection :: Transport -> Endpoint -> Name -> IO () -> IO () +withConnection transport endpoint name fn = do + connection <- connect transport endpoint name + finally fn $ disconnect connection + +{-| +A helper for ensuring that 2 'Connection's are maintained during execution of a function. +-} +withConnection2 :: Transport -> Endpoint -> Name -> Name -> IO () -> IO () +withConnection2 transport endpoint name1 name2 fn = + withConnection transport endpoint name1 $ + withConnection transport endpoint name2 fn + +{-| +A helper for ensuring that 3 'Connection's are maintained during execution of a function. +-} +withConnection3 :: Transport -> Endpoint -> Name -> Name -> Name -> IO () -> IO () +withConnection3 transport endpoint name1 name2 name3 fn = + withConnection transport endpoint name1 $ + withConnection transport endpoint name2 $ + withConnection transport endpoint name3 fn + +{-| +A helper for ensuring that 4 'Connection's are maintained during execution of a function. +-} +withConnection4 :: Transport -> Endpoint -> Name -> Name -> Name -> Name -> IO () -> IO () +withConnection4 transport endpoint name1 name2 name3 name4 fn = + withConnection transport endpoint name1 $ + withConnection transport endpoint name2 $ + withConnection transport endpoint name3 $ + withConnection transport endpoint name4 fn diff --git a/src/Network/Transport/Internal.hs b/src/Network/Transport/Internal.hs deleted file mode 100644 --- a/src/Network/Transport/Internal.hs +++ /dev/null @@ -1,46 +0,0 @@ ------------------------------------------------------------------------------ --- | --- 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 diff --git a/src/Network/Transport/Memory.hs b/src/Network/Transport/Memory.hs --- a/src/Network/Transport/Memory.hs +++ b/src/Network/Transport/Memory.hs @@ -3,7 +3,7 @@ -- 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) @@ -11,7 +11,7 @@ -- 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. +-- 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. @@ -19,66 +19,78 @@ ----------------------------------------------------------------------------- module Network.Transport.Memory ( - newMemoryTransport - ) where + newMemoryTransport, + module Network.Transport + +) where + -- local imports +import Control.Concurrent.Mailbox +import Network.Endpoints import Network.Transport -- external imports +import Control.Concurrent.Async import Control.Concurrent.STM +import Control.Exception + import qualified Data.Map as M -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- -memoryScheme :: Scheme -memoryScheme = "mem" - -data MemoryTransport = MemoryTransport { - boundMailboxes :: TVar (M.Map Name (Mailbox Message)) - } - {-| 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 - } +newMemoryTransport = do + vBindings <- atomically $ newTVar M.empty return Transport { - scheme = memoryScheme, - handles = memoryHandles transport, - bind = memoryBind transport, - sendTo = memorySendTo transport, + bind = memoryBind vBindings, + dispatch = memoryDispatcher vBindings, + connect = memoryConnect, shutdown = return () } -memoryBind :: MemoryTransport -> Mailbox Message -> 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 +memoryDispatcher :: Bindings -> Endpoint -> IO Dispatcher +memoryDispatcher vBindings endpoint = do + d <- async disp + return Dispatcher { + stop = cancel d + } + where + disp = do + atomically $ do + bindings <- readTVar vBindings + env <- readMailbox $ endpointOutbound endpoint + case M.lookup (messageDestination env) bindings of + Nothing -> return () + Just destination -> postMessage destination (envelopeMessage env) + disp -memorySendTo :: MemoryTransport -> Name -> Message -> IO () -memorySendTo transport name msg = do - mailboxes <- atomically $ readTVar $ boundMailboxes transport - case M.lookup name mailboxes of - Just mailbox -> atomically $ writeMailbox mailbox msg - Nothing -> return () -- error $ "No mailbox for " ++ name +memoryBind :: Bindings -> Endpoint -> Name -> IO Binding +memoryBind vBindings endpoint name = atomically $ do + bindings <- readTVar vBindings + case M.lookup name bindings of + Nothing -> do + modifyTVar vBindings $ M.insert name endpoint + return Binding { + bindingName = name, + unbind = memoryUnbind vBindings endpoint name + } + Just _ -> throw $ BindingExists name -memoryUnbind :: MemoryTransport -> Name -> IO () -memoryUnbind transport name = do - atomically $ modifyTVar (boundMailboxes transport) deleteBinding - where deleteBinding m = M.delete name m +memoryUnbind :: Bindings -> Endpoint -> Name -> IO () +memoryUnbind vBindings _ name = atomically $ do + modifyTVar vBindings $ M.delete name + +type Bindings = TVar (M.Map Name Endpoint) + +memoryConnect :: Endpoint -> Name -> IO Connection +memoryConnect _ _ = + return Connection { + disconnect = return () + } diff --git a/src/Network/Transport/Sockets.hs b/src/Network/Transport/Sockets.hs --- a/src/Network/Transport/Sockets.hs +++ b/src/Network/Transport/Sockets.hs @@ -1,512 +1,229 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-} ----------------------------------------------------------------------------- -- | -- Module : Network.Transport.Sockets --- Copyright : (c) Phil Hargett 2013 +-- Copyright : (c) Phil Hargett 2015 -- License : MIT (see LICENSE file) --- +-- -- Maintainer : phil@haphazardhouse.net -- Stability : experimental --- Portability : non-portable (uses STM) +-- Portability : non-portable (requires STM) -- --- Common facilities for socket-based transports, such as UDP and TCP/IP. +-- Helpers for socket transports. -- ----------------------------------------------------------------------------- - module Network.Transport.Sockets ( - - Bindings, - - newSocketBindings, - SocketBindings, - bindAddress, - unbindAddress, - closeBindings, - - SocketTransport(..), - SocketConnectionFactory, - SocketMessengerFactory, - SocketBindingFactory, - newSocketTransport, - - Connection(..), + SocketConnection(..), + Connections, + socketConnect, - IdentifyMessage(..), + socketListen, - Messenger(..), - newMessenger, - addMessenger, - replaceMessenger, - deliver, - closeMessenger, + messenger, + connector, - dispatcher, - sender, - socketSendTo, - receiver, - receiveSocketBytes, - receiveSocketMessage, - receiveSocketMessages, - SocketSend, + Resolver, + ResolverException, + resolve1, + socketResolver4, + socketResolver6, + wildcard, - parseSocketAddress, - lookupAddresses, - lookupAddress, - lookupWildcardAddress + module Network.Transport - ) where +) where -- local imports +import Network.Endpoints import Network.Transport -import Network.Transport.Internal -- external imports + import Control.Concurrent.Async import Control.Concurrent.STM - import Control.Exception +import Control.Monad -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 Data.Serialize +import Data.Typeable import GHC.Generics -import Network.Socket hiding (recv,socket,bind,sendTo,shutdown) -import qualified Network.Socket.ByteString as NSB - -import System.Log.Logger +import qualified Network.Socket as NS -------------------------------------------------------------------------------- -------------------------------------------------------------------------------- -_log :: String -_log = "transport.sockets" +type Connect = Endpoint -> Name -> IO SocketConnection -type SocketConnectionFactory = Address -> IO Connection -type SocketMessengerFactory = Bindings -> Resolver -> Connection -> Mailbox Message -> IO Messenger -type SocketBindingFactory = SocketTransport -> SocketBindings -> Mailbox Message -> Name -> IO (Either String Binding) +{-| +A 'Resolver' translates a name into a list of 'NS.SockAddr' for use with a socket-based 'Transport'. +-} +type Resolver = Name -> IO [NS.SockAddr] -newSocketTransport :: Resolver -> Scheme -> SocketBindingFactory -> SocketConnectionFactory -> SocketMessengerFactory -> IO Transport -newSocketTransport resolver socketScheme binder connectionFactory messengerFactory = do - messengers <- atomically $ newTVar M.empty - bindings <- atomically $ newTVar M.empty - sockets <- newSocketBindings - inbound <- atomically $ newMailbox - dispatch <- async $ dispatcher bindings inbound - let transport = SocketTransport { - socketMessengers = messengers, - socketBindings = bindings, - socketConnection = connectionFactory, - socketMessenger = messengerFactory bindings resolver, - socketInbound = inbound, - socketDispatchers = S.fromList [dispatch], - socketResolver = resolver - } - return Transport { - scheme = socketScheme, - handles = socketTransportHandles transport, - bind = binder transport sockets, - sendTo = socketSendTo transport, - shutdown = socketTransportShutdown transport sockets - } -type Bindings = TVar (M.Map Name (Mailbox Message)) - -data SocketBinding = SocketBinding { - socketCount :: TVar Int, - socketSocket :: TMVar Socket, - socketListener :: TMVar (Async ()) -} - -socketTransportHandles :: SocketTransport -> Name -> IO Bool -socketTransportHandles transport name = do - resolved <- resolve (socketResolver transport) name - return $ isJust resolved - where - isJust (Just _) = True - isJust _ = False - -type SocketBindings = TVar (M.Map Address SocketBinding) - -newSocketBindings :: IO SocketBindings -newSocketBindings = atomically $ newTVar M.empty - -bindAddress :: SocketBindings -> Address -> IO (Socket,Async ()) -> IO () -bindAddress bindings address factory = do - (count,binding) <- atomically $ do - bmap <- readTVar bindings - binding <- case M.lookup address bmap of - Nothing -> do - count <- newTVar 1 - listener <- newEmptyTMVar - sock <- newEmptyTMVar - let binding = SocketBinding { - socketCount = count, - socketListener = listener, - socketSocket = sock - } - modifyTVar bindings $ \bs -> M.insert address binding bs - return binding - Just binding -> do - modifyTVar (socketCount binding) $ \c -> c + 1 - return binding - count <- readTVar $ socketCount binding - return (count,binding) - if count == 1 - then do - infoM _log $ "Opening binding for " ++ (show address) - (sock,listener) <- factory - infoM _log $ "Opened binding for " ++ (show address) - atomically $ do - putTMVar (socketSocket binding) sock - putTMVar (socketListener binding) listener - return () - else return () - -unbindAddress :: SocketBindings -> Address -> IO () -unbindAddress bindings address = do - (count,maybeBinding) <- atomically $ do - bmap <- readTVar bindings - case M.lookup address bmap of - Nothing -> return (0,Nothing) - Just binding -> do - modifyTVar (socketCount binding) $ \count -> count - 1 - count <- readTVar (socketCount binding) - if count == 0 - then do - modifyTVar bindings $ \bm -> M.delete address bm - sock <- takeTMVar $ socketSocket binding - listener <- takeTMVar $ socketListener binding - return (0,Just (sock,listener)) - else return (count,Nothing) - case maybeBinding of - -- no binding to shutdown; can just return - Nothing -> do - infoM _log $ "No binding to shutdown for " ++ (show address) ++ "; count is " ++ (show count) - return () - -- we're the last, so close the binding - Just (sock,listener) -> do - infoM _log $ "Closing binding for " ++ (show address) ++ "; count is " ++ (show count) - cancel listener - sClose sock - infoM _log $ "Closed binding for " ++ (show address) - -data SocketTransport = SocketTransport { - socketMessengers :: TVar (M.Map Address Messenger), - socketBindings :: Bindings, - socketConnection :: Address -> IO Connection, - socketMessenger :: Connection -> Mailbox Message -> IO Messenger, - socketInbound :: Mailbox Message, - socketDispatchers :: S.Set (Async ()), - socketResolver :: Resolver +{-| +A 'SocketConnection' encapsulates the state of a single outbound connection. +-} +data SocketConnection = SocketConnection { + connectionDestination :: TMVar Name, + sendSocketMessage :: Message -> IO (), + receiveSocketMessage :: IO Message, + disconnectSocket :: IO () } {-| -A connection specializes the use of a transport for a particular -destination. +Exceptions thrown by a 'Resolver'. -} -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 () - } +data ResolverException = CannotResolveName Name + deriving (Show,Typeable) +instance Exception ResolverException + {-| -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. +Socket-based protocols need a way to unambiguously identify who is communicating on a particular 'SocketConnection'; +this type accomodates both identifying the sender and sending an actual 'Message'. -} -data Messenger = Messenger { - messengerDone :: TVar Bool, - messengerOut :: Mailbox Message, - messengerAddress :: Address, - messengerSender :: Async (), - messengerReceiver :: Async (), - messengerConnection :: Connection - } +data SocketMessage = + IdentifySender Name + | SocketMessage Message + deriving (Generic) -data IdentifyMessage = IdentifyMessage Address deriving (Generic) +instance Serialize SocketMessage -instance Serialize IdentifyMessage +{-| +Helper to resolve a 'Name' to a single 'NS.SockAddr' for use with a socket-based 'Transport'. +-} +resolve1 :: Resolver -> Name -> IO NS.SockAddr +resolve1 resolve name = do + addresses <- resolve name + case addresses of + [] -> throw $ CannotResolveName name + (address:_) -> return address {-| -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@. +Simple 'Resolver' for socket-based 'Transport's that just parse's the 'String' inside the 'Name', with the expected +format "host:port" -} -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") +socketResolver :: NS.Family -> NS.SocketType -> Name -> IO [NS.SockAddr] +socketResolver family socketType name = + let (host,port) = split nm + Name nm = name + hints = NS.defaultHints { NS.addrFlags = [NS.AI_ADDRCONFIG, NS.AI_CANONNAME, NS.AI_NUMERICSERV] } + in do + addresses <- NS.getAddrInfo (Just hints) (Just host) (Just port) + return $ map NS.addrAddress $ filter (\addrInfo -> NS.addrFamily addrInfo == family && NS.addrSocketType addrInfo == socketType) addresses where - host h = if h == "" then - "localhost" - else h - port p = p - -lookupAddresses :: Family -> SocketType -> Address -> IO [SockAddr] -lookupAddresses family socketType address = - let (host,port) = parseSocketAddress address - hints = defaultHints { addrFlags = [AI_ADDRCONFIG, AI_CANONNAME, AI_NUMERICSERV] } - in do - addresses <- getAddrInfo (Just hints) (Just host) (Just port) - return $ map addrAddress $ filter (\addrInfo -> addrFamily addrInfo == family && addrSocketType addrInfo == socketType) addresses - -lookupAddress :: Family -> SocketType -> Address -> IO SockAddr -lookupAddress family socketType address = do - addresses <- lookupAddresses family socketType address - return $ addresses !! 0 - -lookupWildcardAddress :: Family -> SocketType -> Address -> IO SockAddr -lookupWildcardAddress family socketType address = do - sockAddr <- lookupAddress family socketType address - case sockAddr of - SockAddrInet port _ -> return $ SockAddrInet port iNADDR_ANY - SockAddrInet6 port flow _ scope -> return $ SockAddrInet6 port flow iN6ADDR_ANY scope - _ -> return sockAddr - -type SocketSend = Socket -> B.ByteString -> IO () - -instance Show Messenger where - show msngr = "Messenger(" ++ (show $ messengerAddress msngr) ++ ")" - -newMessenger :: Connection -> Mailbox Message -> IO Messenger -newMessenger conn inc = do - out <- atomically $ newMailbox - done <- atomically $ newTVar False - sndr <- async $ sender conn done out - rcvr <- async $ receiver conn done inc - return Messenger { - messengerDone = done, - 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) - -replaceMessenger :: SocketTransport -> Address -> Messenger -> IO () -replaceMessenger transport address msngr = do - found <- atomically $ do - msngrs <- readTVar $ socketMessengers transport - return $ M.lookup address msngrs - case found of - Just _ -> do - infoM _log $ "Already have messenger for " ++ (show address) - closeMessenger msngr - Nothing -> do - addMessenger transport address msngr + split [] = ("","") + split (w:ws) = case w of + ':' -> ("",ws) + _ -> let (w',ws') = split ws + in (w:w',ws') +{-| +Variation of 'socketResolve' for standard IP addresses. +-} +socketResolver4 :: NS.SocketType -> Name -> IO [NS.SockAddr] +socketResolver4 = socketResolver NS.AF_INET +{-| +Variation of 'socketResolve' for standard IPv6 addresses. +-} +socketResolver6 ::NS.SocketType -> Name -> IO [NS.SockAddr] +socketResolver6 = socketResolver NS.AF_INET6 -deliver :: Messenger -> Message -> IO () -deliver msngr message = atomically $ writeMailbox (messengerOut msngr) message +{-| +Returns the wildcard 'NS.SockAddr' for the specified 'NS.SockAddr', if possible. +-} +wildcard :: NS.SockAddr -> IO NS.SockAddr +wildcard address = + case address of + NS.SockAddrInet port _ -> return $ NS.SockAddrInet port NS.iNADDR_ANY + NS.SockAddrInet6 port flow _ scope -> return $ NS.SockAddrInet6 port flow NS.iN6ADDR_ANY scope + _ -> return address -dispatcher :: TVar (M.Map Name (Mailbox Message)) -> Mailbox Message -> IO () -dispatcher bindings mbox = dispatchMessages - where - dispatchMessages = catchExceptions (do - infoM _log $ "Dispatching messages" - env <- atomically $ readMailbox 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 - writeMailbox dest msg - return () +{-| +Listens for new connections on a new 'NS.Socket', and return the 'NS.Socket'. +-} +socketListen :: NS.Family -> NS.SocketType -> Resolver -> Name -> IO NS.Socket +socketListen family socketType resolver name = do + address <- resolve1 resolver name + socket <- NS.socket family socketType NS.defaultProtocol + NS.setSocketOption socket NS.NoDelay 1 + NS.setSocketOption socket NS.ReuseAddr 1 + when (NS.isSupportedSocketOption NS.ReusePort) + $ NS.setSocketOption socket NS.ReusePort 1 + wildcard address >>= NS.bind socket + NS.listen socket 2048 -- TODO think about a configurable backlog + return socket -sender :: Connection -> TVar Bool -> Mailbox Message -> IO () -sender conn done mailbox = sendMessages - where - sendMessages = do - reconnect - catchExceptions (do - infoM _log $ "Waiting to send to " ++ (show $ connAddress conn) - msg <- atomically $ readMailbox 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) - isDone <- atomically $ readTVar done - if isDone - then return () - else sendMessages - reconnect = do - -- TODO need a timeout here, in case connecting always fails - infoM _log $ "Reconnecting to " ++ (show $ connAddress conn) - connected <- atomically $ tryReadTMVar $ connSocket conn - case connected of - Just _ -> do - infoM _log $ "Reconnected to " ++ (show $ connAddress conn) - 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 () +{-| +Mutble 'M.Map' of 'Name's to 'Connection's. +-} +type Connections = TVar (M.Map Name Connection) +{-| +Establish a 'connector' for exchanging messages from 'Endpoint' with a destination specified by 'Name'. +-} +socketConnect :: Mailboxes -> Connect -> Endpoint -> Name -> IO Connection +socketConnect mailboxes sConnect endpoint name = do + connr <- async $ connector mailboxes endpoint name sConnect + let conn = Connection { + disconnect = cancel connr + } + return conn -socketSendTo :: SocketTransport -> Name -> Message -> IO () -socketSendTo transport name msg = do - isLocal <- local - if isLocal - then return () - else remote +{-| +Maintain a 'Connection' between an 'Endpoint' and a destination specified by 'Name'. Should the connection, +the connection will automatically be re-attempted. +-} +connector :: Mailboxes -> Endpoint -> Name -> Connect -> IO () +connector mailboxes endpoint name transportConnect = loopUntilKilled $ do + connection <- transportConnect endpoint name + origins <- atomically $ do + putTMVar (connectionDestination connection) name + readTVar $ boundEndpointNames endpoint + forM_ (S.elems origins) $ \origin -> + -- atomically $ dispatchMessage mailboxes name $ encode $ IdentifySender origin + sendSocketMessage connection $ encode $ IdentifySender origin + finally (messenger mailboxes endpoint connection) $ + disconnectSocket connection 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 $ writeMailbox 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 -> TVar Bool -> Mailbox Message -> IO () -receiver conn done mailbox = do - socket <- atomically $ readTMVar $ connSocket conn - receiveSocketMessages socket done (connAddress conn) mailbox - -receiveSocketMessages :: Socket -> TVar Bool -> Address -> Mailbox Message -> IO () -receiveSocketMessages sock done addr mailbox = do - 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 $ writeMailbox mailbox msg - isDone <- atomically $ readTVar done - if isDone - then return () - else receiveSocketMessages sock done addr mailbox) - (\e -> do - isDone <- atomically $ readTVar done - if isDone - then return () - -- Dropping this message to info, as even well-behaved applications - -- may generate it...even though it is benign - else infoM _log $ "Receive error: " ++ (show (e :: SomeException))) + loopUntilKilled fn = + catch (catch fn untilKilled) + (loop fn) + loop :: IO () -> SomeException -> IO () + loop fn _ = loopUntilKilled fn + untilKilled :: AsyncException -> IO () + untilKilled _ = return () -receiveSocketMessage :: Socket -> IO (Maybe B.ByteString) -receiveSocketMessage socket = do - maybeLen <- receiveSocketBytes 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 <- receiveSocketBytes socket $ msgLength (decode len) - infoM _log $ "Received message" - return maybeMsg +{-| +Exchange 'Messages' with the 'Endpoint' on the other side of a 'SocketConnection'. +-} +messenger :: Mailboxes -> Endpoint -> SocketConnection -> IO () +messenger mailboxes endpoint connection = + -- counting on race_ to kill reader & writer + -- if messenger is killed; since it uses withAsync, it should + race_ receiver sender where - msgLength (Right size) = size - msgLength (Left err) = error err - -receiveSocketBytes :: Socket -> Int -> IO (Maybe B.ByteString) -receiveSocketBytes sock maxBytes = do - bs <- NSB.recv sock maxBytes - if B.null bs - then return Nothing - else return $ Just bs - -closeMessenger :: Messenger -> IO () -closeMessenger msngr = do - infoM _log $ "Closing mesenger to " ++ (messengerAddress msngr) - atomically $ modifyTVar (messengerDone msngr) (\_ -> True) - cancel $ messengerSender msngr - cancel $ messengerReceiver msngr - connClose $ messengerConnection msngr - infoM _log $ "Closed messenger to " ++ (messengerAddress msngr) - -closeBindings :: SocketBindings -> IO () -closeBindings sockets = do - infoM _log $ "Closing bindings" - bindings <- atomically $ readTVar sockets - mapM_ (unbindAddress sockets) $ M.keys bindings - infoM _log $ "Closed bindings" - -socketTransportShutdown :: SocketTransport -> SocketBindings -> IO () -socketTransportShutdown transport sockets = do - closeBindings sockets - 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 + receiver = do + smsg <- receiveSocketMessage connection + -- TODO consider a way of using a message to identify the name of + -- the endpoint on the other end of the connection + case decode smsg of + Left _ -> return () + Right (IdentifySender name) -> atomically $ putTMVar (connectionDestination connection) name + Right (SocketMessage msg) -> atomically $ postMessage endpoint msg + receiver + sender = do + msg <- atomically $ do + -- this basically means we wait until we have a name + name <- readTMVar $ connectionDestination connection + pullMessage mailboxes name + sendSocketMessage connection $ encode $ SocketMessage msg + sender diff --git a/src/Network/Transport/Sockets/TCP.hs b/src/Network/Transport/Sockets/TCP.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Sockets/TCP.hs @@ -0,0 +1,163 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Network.Transport.Sockets.TCP +-- Copyright : (c) Phil Hargett 2015 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- TCP transport. +-- +----------------------------------------------------------------------------- +module Network.Transport.Sockets.TCP ( + newTCPTransport4, + newTCPTransport6, + + tcpSocketResolver4, + tcpSocketResolver6, + + module Network.Transport +) where + +-- local imports + +import Network.Endpoints +import Network.Transport +import Network.Transport.Sockets + +-- external imports + +import Control.Concurrent.Async +import Control.Concurrent.STM +import Control.Exception +import Control.Monad + +import qualified Data.ByteString as BS +import qualified Data.Map as M +import Data.Serialize + +import qualified Network.Socket as NS +import qualified Network.Socket.ByteString as NSB + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +type SocketConnections = TVar (M.Map NS.SockAddr Connection) + +newTCPTransport :: NS.Family -> Resolver -> IO Transport +newTCPTransport family resolver = atomically $ do + vPeers <- newTVar M.empty + mailboxes <- newTVar M.empty + return Transport { + bind = tcpBind mailboxes family resolver, + dispatch = dispatcher mailboxes, + connect = socketConnect mailboxes $ tcpConnect family resolver, + shutdown = tcpShutdown vPeers + } + +{-| +Create a 'Transport' for exchanging 'Message's between endpoints via TCP over IP +-} +newTCPTransport4 :: Resolver -> IO Transport +newTCPTransport4 = newTCPTransport NS.AF_INET + +{-| +Create a 'Transport' for exchanging 'Message's between endpoints via TCP over IPv6 +-} +newTCPTransport6 :: Resolver -> IO Transport +newTCPTransport6 = newTCPTransport NS.AF_INET6 + +{-| +Create a 'Resolve' for resolving 'Name's for use with TCP over IP. +-} +tcpSocketResolver4 :: Name -> IO [NS.SockAddr] +tcpSocketResolver4 = socketResolver4 NS.Stream + +{-| +Create a 'Resolve' for resolving 'Name's for use with TCP over IPv6. +-} +tcpSocketResolver6 :: Name -> IO [NS.SockAddr] +tcpSocketResolver6 = socketResolver6 NS.Stream + +tcpBind :: Mailboxes -> NS.Family -> Resolver -> Endpoint -> Name -> IO Binding +tcpBind mailboxes family resolver endpoint name = do + vConnections <- atomically $ newTVar M.empty + listener <- async $ tcpListen mailboxes family resolver vConnections endpoint name + return Binding { + bindingName = name, + unbind = cancel listener + } + +tcpListen :: Mailboxes -> NS.Family -> Resolver -> SocketConnections -> Endpoint -> Name -> IO () +tcpListen mailboxes family resolver vConnections endpoint name = do + socket <- socketListen family NS.Stream resolver name + finally (accept mailboxes socket vConnections endpoint) + (tcpUnbind socket) + +accept :: Mailboxes -> NS.Socket -> SocketConnections -> Endpoint -> IO () +accept mailboxes socket vConnections endpoint = do + (peer,peerAddress) <- NS.accept socket + connection <- tcpConnection peer + msngr <- async $ messenger mailboxes endpoint connection + let conn = Connection { + disconnect = cancel msngr + } + maybeOldConn <- atomically $ do + connections <- readTVar vConnections + let oldConn = M.lookup peerAddress connections + modifyTVar vConnections $ M.insert peerAddress conn + return oldConn + case maybeOldConn of + Just oldConn -> disconnect oldConn + Nothing -> return () + accept mailboxes socket vConnections endpoint + +tcpUnbind :: NS.Socket -> IO () +tcpUnbind = NS.close + +tcpConnect :: NS.Family -> Resolver -> Endpoint -> Name -> IO SocketConnection +tcpConnect family resolver _ name = do + socket <- NS.socket family NS.Stream NS.defaultProtocol + address <- resolve1 resolver name + NS.connect socket address + tcpConnection socket + +tcpConnection :: NS.Socket -> IO SocketConnection +tcpConnection socket = do + vName <- atomically newEmptyTMVar + return SocketConnection { + connectionDestination = vName, + sendSocketMessage = tcpSend socket, + receiveSocketMessage = tcpReceive socket, + disconnectSocket = tcpDisconnect socket + } + +tcpSend :: NS.Socket -> Message -> IO () +tcpSend socket message = do + let len = BS.length message + NSB.sendAll socket $ encode len + NSB.sendAll socket message + +tcpReceive :: NS.Socket -> IO Message +tcpReceive socket = readBytesWithLength + where + readBytesWithLength = do + lengthBytes <- readBytes 8 -- TODO must figure out what defines length of an integer in bytes + case decode lengthBytes of + Left _ -> throw NoDataRead + Right len -> readBytes len + readBytes = NSB.recv socket + +tcpDisconnect :: NS.Socket -> IO () +tcpDisconnect = NS.close + +tcpShutdown :: SocketConnections -> IO () +tcpShutdown vPeers = do + peers <- atomically $ readTVar vPeers + -- this is how we disconnect incoming connections + -- we don't have to disconnect outbound connectinos, because + -- they should already be disconnected before here + forM_ (M.elems peers) disconnect + return () diff --git a/src/Network/Transport/Sockets/UDP.hs b/src/Network/Transport/Sockets/UDP.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Sockets/UDP.hs @@ -0,0 +1,161 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Network.Transport.Sockets.UDP +-- Copyright : (c) Phil Hargett 2015 +-- License : MIT (see LICENSE file) +-- +-- Maintainer : phil@haphazardhouse.net +-- Stability : experimental +-- Portability : non-portable (requires STM) +-- +-- UDP transport. +-- +----------------------------------------------------------------------------- +module Network.Transport.Sockets.UDP ( + newUDPTransport4, + newUDPTransport6, + + udpSocketResolver4, + udpSocketResolver6, + + module Network.Transport +) where + +-- local imports + +import Network.Endpoints +import Network.Transport +import Network.Transport.Sockets + +-- external imports + +import Control.Concurrent.Async +import Control.Concurrent.Mailbox +import Control.Concurrent.STM +import Control.Exception +import Control.Monad + +import qualified Data.Map as M + +import qualified Network.Socket as NS +import qualified Network.Socket.ByteString as NSB + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- +type SocketConnections = TVar (M.Map NS.SockAddr Connection) + +newUDPTransport :: NS.Family -> Resolver -> IO Transport +newUDPTransport family resolver = do + socket <- NS.socket family NS.Datagram NS.defaultProtocol + atomically $ do + vPeers <- newTVar M.empty + mailboxes <- newTVar M.empty + vSocket <- newTMVar socket + return Transport { + bind = udpBind family resolver, + dispatch = udpDispatcher vSocket resolver, + connect = udpConnect mailboxes family resolver, + shutdown = udpShutdown vPeers + } + +{-| +Create a 'Transport' for exchanging 'Message's between endpoints via UDP over IP +-} +newUDPTransport4 :: Resolver -> IO Transport +newUDPTransport4 = newUDPTransport NS.AF_INET + +{-| +Create a 'Transport' for exchanging 'Message's between endpoints via UDP over IPv6 +-} +newUDPTransport6 :: Resolver -> IO Transport +newUDPTransport6 = newUDPTransport NS.AF_INET6 + +{-| +Create a 'Resolve' for resolving 'Name's for use with UDP over IP. +-} +udpSocketResolver4 :: Name -> IO [NS.SockAddr] +udpSocketResolver4 = socketResolver4 NS.Datagram + +{-| +Create a 'Resolve' for resolving 'Name's for use with UDP over IPv6. +-} +udpSocketResolver6 :: Name -> IO [NS.SockAddr] +udpSocketResolver6 = socketResolver6 NS.Datagram + +udpBind :: NS.Family -> Resolver -> Endpoint -> Name -> IO Binding +udpBind family resolver endpoint name = do + socket <- NS.socket family NS.Datagram NS.defaultProtocol + address <- resolve1 resolver name + NS.setSocketOption socket NS.ReuseAddr 1 + when (NS.isSupportedSocketOption NS.ReusePort) + $ NS.setSocketOption socket NS.ReusePort 1 + NS.bindSocket socket address + listener <- async $ + finally (receiver socket) + (udpUnbind socket) + return Binding { + bindingName = name, + unbind = cancel listener + } + where + receiver socket = do + msg <- udpReceive socket + -- TODO consider a way of using a message to identify the name of + -- the endpoint on the other end of the connection + atomically $ postMessage endpoint msg + receiver socket + +udpDispatcher :: TMVar NS.Socket -> Resolver -> Endpoint -> IO Dispatcher +udpDispatcher vSocket resolver endpoint = do + d <- async disp + return Dispatcher { + stop = cancel d + } + where + disp = do + (socket,name,msg) <- atomically $ do + envelope <- readMailbox $ endpointOutbound endpoint + let name = messageDestination envelope + msg = envelopeMessage envelope + socket <- readTMVar vSocket + return (socket,name,msg) + address <- resolve1 resolver name + udpSend socket address msg + disp + +udpUnbind :: NS.Socket -> IO () +udpUnbind = NS.close + +udpConnect :: Mailboxes -> NS.Family -> Resolver -> Endpoint -> Name -> IO Connection +udpConnect mailboxes family resolver _ name = do + socket <- NS.socket family NS.Datagram NS.defaultProtocol + address <- resolve1 resolver name + sender <- async $ finally (writer socket address) (udpDisconnect socket) + return Connection { + disconnect = cancel sender + } + where + writer socket address = do + msg <- atomically $ pullMessage mailboxes name + udpSend socket address msg + writer socket address + +udpSend :: NS.Socket -> NS.SockAddr -> Message -> IO () +udpSend socket address message = NSB.sendAllTo socket message address + +udpReceive :: NS.Socket -> IO Message +udpReceive socket = do + (msg,_) <- NSB.recvFrom socket 512 + return msg + +udpDisconnect :: NS.Socket -> IO () +udpDisconnect = NS.close + +udpShutdown :: SocketConnections -> IO () +udpShutdown vPeers = do + peers <- atomically $ readTVar vPeers + -- this is how we disconnect incoming connections + -- we don't have to disconnect outbound connectinos, because + -- they should already be disconnected before here + forM_ (M.elems peers) disconnect + return () diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs deleted file mode 100644 --- a/src/Network/Transport/TCP.hs +++ /dev/null @@ -1,203 +0,0 @@ -{-# 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, - newTCPTransport6, - - lookupAddresses, - lookupTCPAddress, - lookupWildcardTCPAddress - ) 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 Data.Serialize -import qualified Data.Set as S - -import qualified Network.Socket as NS -import qualified Network.Socket.ByteString as NSB - -import System.Log.Logger - --------------------------------------------------------------------------------- --------------------------------------------------------------------------------- - -_log :: String -_log = "transport.tcp" - -tcpScheme :: Scheme -tcpScheme = "tcp" - -lookupTCPAddress :: Address -> NS.Family -> IO NS.SockAddr -lookupTCPAddress address family = lookupAddress family NS.Stream address - -lookupWildcardTCPAddress :: Address -> NS.Family -> IO NS.SockAddr -lookupWildcardTCPAddress address family = lookupWildcardAddress family NS.Stream address - -{-| -Create a new 'Transport' suitable for sending messages over TCP/IP (IPv4). 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 = let family = NS.AF_INET in newSocketTransport resolver tcpScheme (tcpBind family) (newTCPConnection family) newTCPMessenger - -{-| -Create a new 'Transport' suitable for sending messages over TCP/IP (IPv4). 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. --} -newTCPTransport6 :: Resolver -> IO Transport -newTCPTransport6 resolver = let family = NS.AF_INET6 in newSocketTransport resolver tcpScheme (tcpBind family) (newTCPConnection family) newTCPMessenger - -tcpBind :: NS.Family -> SocketTransport -> SocketBindings -> Mailbox Message -> Name -> IO (Either String Binding) -tcpBind family transport sockets inc name = do - atomically $ modifyTVar (socketBindings transport) $ \ bindings -> - M.insert name inc bindings - Just address <- resolve (socketResolver transport) name - bindAddress sockets address $ do - sock <- NS.socket family NS.Stream NS.defaultProtocol - listener <- do - infoM _log $ "Binding to address " ++ (show address) - tcpListen transport family address sock - return (sock,listener) - return $ Right Binding { - bindingName = name, - unbind = tcpUnbind sockets address - } - -tcpListen :: SocketTransport -> NS.Family -> Address -> NS.Socket -> IO (Async ()) -tcpListen transport family address sock = do - catchExceptions (do - NS.setSocketOption sock NS.NoDelay 1 - NS.setSocketOption sock NS.ReuseAddr 1 - sockAddr <- lookupWildcardTCPAddress address family - NS.bind sock sockAddr - NS.listen sock 2048) -- TODO think about a configurable backlog - (\e -> do - warningM _log $ "Listen error on port " ++ address ++ ": " ++ (show (e :: SomeException)) - NS.sClose sock) - async $ tcpAccept transport family address sock - -tcpAccept :: SocketTransport -> NS.Family -> Address -> NS.Socket -> IO () -tcpAccept transport family address sock = do - infoM _log $ "Listening for connections on " ++ (show address) ++ ": " ++ (show sock) - (client,clientAddress) <- NS.accept sock - _ <- async $ tcpDispatch transport family address client clientAddress - tcpAccept transport family address sock - -tcpDispatch :: SocketTransport -> NS.Family -> Address -> NS.Socket -> NS.SockAddr -> IO () -tcpDispatch transport family address client socketAddress = do - infoM _log $ "Accepted connection on " ++ (show address) - identity <- tcpIdentify client socketAddress - case identity of - Nothing -> NS.sClose client - Just (IdentifyMessage clientAddress) -> do - infoM _log $ "Identified " ++ (show clientAddress) - newConn <- (newTCPConnection family) clientAddress - atomically $ putTMVar (connSocket newConn) client - msngr <- newMessenger newConn (socketInbound transport) - replaceMessenger transport clientAddress msngr - -tcpIdentify :: NS.Socket -> NS.SockAddr -> IO (Maybe IdentifyMessage) -tcpIdentify client clientAddress = do - infoM _log $ "Awaiting identity from " ++ (show clientAddress) - maybeMsg <- receiveSocketMessage 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 :: SocketBindings -> Address -> IO () -tcpUnbind sockets address = do - infoM _log $ "Unbinding from TCP port " ++ (show address) - unbindAddress sockets address - infoM _log $ "Unbound from TCP port " ++ (show address) - -newTCPConnection :: NS.Family -> Address -> IO Connection -newTCPConnection family address = do - sock <- atomically $ newEmptyTMVar - return Connection { - connAddress = address, - connSocket = sock, - connConnect = do - socket <- NS.socket family NS.Stream NS.defaultProtocol - sockAddr <- lookupTCPAddress address family - NS.connect socket sockAddr - -- NS.connect socket $ NS.addrAddress $ head ipv4Addrs - atomically $ putTMVar sock socket - -- infoM _log $ "Initiated socket connection to " ++ (show $ head ipv4Addrs) - return socket, - connSend = tcpSend address, - connReceive = receiveSocketBytes, - connClose = do - infoM _log $ "Closing connection to " ++ address - open <- atomically $ tryTakeTMVar sock - case open of - Just socket -> do - infoM _log $ "Closing socket " ++ (show socket) ++ " for " ++ address - NS.sClose socket - infoM _log $ "Closed socket " ++ (show socket) ++ " for " ++ address - Nothing -> do - infoM _log $ "No socket to close for " ++ address - return () - infoM _log $ "Connection to " ++ address ++ " closed" - } - -newTCPMessenger :: Bindings -> Resolver -> Connection -> Mailbox Message -> 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 - -tcpSend :: Address -> NS.Socket -> B.ByteString -> IO () -tcpSend addr sock bs = do - NSB.sendAll sock $ encode (B.length bs) - infoM _log $ "Length sent" - NSB.sendAll sock bs - infoM _log $ "Message sent to" ++ (show addr) diff --git a/src/Network/Transport/UDP.hs b/src/Network/Transport/UDP.hs deleted file mode 100644 --- a/src/Network/Transport/UDP.hs +++ /dev/null @@ -1,154 +0,0 @@ -{-# 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, - newUDPTransport6, - - lookupAddresses, - lookupUDPAddress, - lookupWildcardUDPAddress - ) 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 Network.Socket as NS -import Network.Socket.ByteString(sendAllTo,recvFrom) - -import System.Log.Logger - --------------------------------------------------------------------------------- --------------------------------------------------------------------------------- - -_log :: String -_log = "transport.udp" - -udpScheme :: Scheme -udpScheme = "udp" - -lookupUDPAddress :: Address -> NS.Family -> IO NS.SockAddr -lookupUDPAddress address family = lookupAddress family NS.Datagram address - -lookupWildcardUDPAddress :: Address -> NS.Family -> IO NS.SockAddr -lookupWildcardUDPAddress address family = lookupWildcardAddress family NS.Datagram address - -newUDPTransport :: Resolver -> IO Transport -newUDPTransport resolver = newSocketTransport resolver udpScheme (udpBind NS.AF_INET) (newUDPConnection NS.AF_INET) newUDPMessenger - -newUDPTransport6 :: Resolver -> IO Transport -newUDPTransport6 resolver = newSocketTransport resolver udpScheme (udpBind NS.AF_INET6) (newUDPConnection NS.AF_INET6) newUDPMessenger - -udpBind :: NS.Family -> SocketTransport -> SocketBindings -> Mailbox Message -> Name -> IO (Either String Binding) -udpBind family transport sockets inc name = do - atomically $ modifyTVar (socketBindings transport) $ \bindings -> - M.insert name inc bindings - Just address <- resolve (socketResolver transport) name - bindAddress sockets address $ do - sockaddr <- lookupWildcardUDPAddress address family - sock <- NS.socket family NS.Datagram NS.defaultProtocol - -- have to set this option in case we frequently rebind sockets - infoM _log $ "Binding to " ++ (show address) ++ " over UDP" - NS.setSocketOption sock NS.ReuseAddr 1 - NS.bindSocket sock sockaddr - infoM _log $ "Bound to " ++ (show address) ++ " over UDP" - rcvr <- async $ udpReceiveSocketMessages sock address (socketInbound transport) - return (sock,rcvr) - return $ Right Binding { - bindingName = name, - unbind = do - infoM _log $ "Unbinding from UDP port " ++ (show address) - unbindAddress sockets address - infoM _log $ "Unbound from UDP port " ++ (show address) - } - -newUDPConnection :: NS.Family -> Address -> IO Connection -newUDPConnection family address = do - sock <- atomically newEmptyTMVar - return Connection { - connAddress = address, - connSocket = sock, - connConnect = NS.socket family NS.Datagram NS.defaultProtocol, - connSend = (\s bs -> do - addr <- lookupUDPAddress address family - 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 -> NS.sClose s - Nothing -> return () - return () - } - -newUDPMessenger :: Bindings -> Resolver -> Connection -> Mailbox Message -> IO Messenger -newUDPMessenger _ _ conn mailbox = do - msngr <- newMessenger conn mailbox - return msngr - -udpReceiveSocketMessages :: NS.Socket -> Address -> Mailbox Message -> 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 - NS.sClose sock - return () - Just msg -> do - atomically $ writeMailbox 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 :: NS.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 diff --git a/tests/TestMemory.hs b/tests/TestMemory.hs --- a/tests/TestMemory.hs +++ b/tests/TestMemory.hs @@ -5,79 +5,25 @@ import Network.Endpoints import Network.Transport.Memory --- external imports - -import Control.Exception +import TransportTestSuite -import Data.Serialize +-- external imports 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 - finally (do - endpoint <- newEndpoint [transport] - Right () <- bindEndpoint endpoint name1 - return ()) - (shutdown transport) - -testEndpointBindUnbind :: Assertion -testEndpointBindUnbind = do - let name1 = "endpoint1" - transport <- newMemoryTransport - finally (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 ()) - (shutdown transport) +tests = transportTestSuite + newMemoryTransport + "mem" + (Name "name1") + (Name "name2") + (Name "name3") + (Name "name4") -testEndpointSendReceive :: Assertion -testEndpointSendReceive = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - finally (do - 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 ()) - (shutdown transport) +_log :: String +_log = "_test_memory" -- Memory tests - -testMemoryTransport :: Assertion -testMemoryTransport = do - transport <- newMemoryTransport - shutdown transport - diff --git a/tests/TestRPC.hs b/tests/TestRPC.hs --- a/tests/TestRPC.hs +++ b/tests/TestRPC.hs @@ -3,7 +3,7 @@ -- Module : TestRPC -- Copyright : (c) Phil Hargett 2014 -- License : MIT (see LICENSE file) --- +-- -- Maintainer : phil@haphazardhouse.net -- Stability : experimental -- Portability : non-portable (requires STM) @@ -22,11 +22,11 @@ import Network.RPC import Network.Transport.Memory +import TestUtils + -- external imports -import Control.Applicative import Control.Concurrent -import Control.Concurrent.Async import qualified Data.Map as M import Data.Serialize @@ -43,253 +43,72 @@ tests :: [Test.Framework.Test] tests = [ - testCase "call-one-hear-call" testOneHearCall, - testCase "call-one-call-hear" testOneCallHear, - testCase "call-concurrent-call-hear" testConcurrentCallHear, - testCase "call-one-handler" testOneHandler, - testCase "call-two-handlers" testTwoHandlers, - testCase "gcall-three-handlers" testGroupCall, - testCase "anycall-three-handlers" testAnyCall, testCase "call-one-with-timeout" testOneHandlerWithTimeout, testCase "gcall-three-handlers-with-timeout"testGroupCallWithTimeout ] -testOneHearCall :: Assertion -testOneHearCall = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - _ <- async $ do - (bytes,reply) <- hear endpoint2 name2 "foo" - let Right msg = decode bytes - reply $ encode $ msg ++ "!" - let cs = newCallSite endpoint1 name1 - bytes <- call cs name2 "foo" $ encode "hello" - let Right result = decode bytes - assertEqual "Result not expected value" "hello!" result - -testOneCallHear :: Assertion -testOneCallHear = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - let cs = newCallSite endpoint1 name1 - acall <- async $ call cs name2 "foo" $ encode "hello" - _ <- async $ do - (bytes,reply) <- hear endpoint2 name2 "foo" - let Right msg = decode bytes - reply $ encode $ msg ++ "!" - bytes <- wait acall - let Right result = decode bytes - assertEqual "Result not expected value" "hello!" result - -testConcurrentCallHear :: Assertion -testConcurrentCallHear = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - let cs1 = newCallSite endpoint1 name1 - cs2 = newCallSite endpoint2 name2 - let call1 = call cs1 name2 "foo" $ encode "hello" - hear1 = do - (bytes,reply) <- hear endpoint2 name2 "foo" - let Right msg = decode bytes - reply $ encode $ msg ++ "!" - call2 = call cs2 name1 "bar" $ encode "ciao" - hear2 = do - (bytes,reply) <- hear endpoint1 name1 "bar" - let Right msg = decode bytes - reply $ encode $ msg ++ "!" - (result1,(),result2,()) <- runConcurrently $ (,,,) - <$> Concurrently call1 - <*> Concurrently hear1 - <*> Concurrently call2 - <*> Concurrently hear2 - assertEqual "Result not expected value" (Right "hello!") (decode result1) - assertEqual "Result not expected value" (Right "ciao!") (decode result2) - -testOneHandler :: Assertion -testOneHandler = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - h <- handle endpoint2 name2 "foo" $ \bytes -> - let Right msg = decode bytes - in return $ encode $ msg ++ "!" - let cs = newCallSite endpoint1 name1 - bytes <- call cs name2 "foo" $ encode "hello" - let Right result = decode bytes - assertEqual "Result not expected value" "hello!" result - hangup h - -testTwoHandlers :: Assertion -testTwoHandlers = do - let name1 = "endpoint1" - name2 = "endpoint2" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - h1 <- handle endpoint2 name2 "foo" $ \bytes -> - let Right msg = decode bytes - in return $ encode $ msg ++ "!" - h2 <- handle endpoint2 name2 "bar" $ \bytes -> - let Right msg = decode bytes - in return $ encode $ msg ++ "?" - let cs = newCallSite endpoint1 name1 - bytes1 <- call cs name2 "foo" $ encode "hello" - let Right result1 = decode bytes1 - assertEqual "Result not expected value" "hello!" result1 - bytes2 <- call cs name2 "bar" $ encode "hello" - let Right result2 = decode bytes2 - assertEqual "Result not expected value" "hello?" result2 - hangup h1 - hangup h2 - -testGroupCall :: Assertion -testGroupCall = do - let name1 = "endpoint1" - name2 = "endpoint2" - name3 = "endpoint3" - name4 = "endpoint4" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - endpoint3 <- newEndpoint [transport] - endpoint4 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - Right () <- bindEndpoint endpoint3 name3 - Right () <- bindEndpoint endpoint4 name4 - h2 <- handle endpoint2 name2 "foo" $ \bytes -> let Right msg = decode bytes in - return $ encode $ if msg == "hello" then "foo" else "" - h3 <- handle endpoint3 name3 "foo" $ \bytes -> let Right msg = decode bytes in - return $ encode $ if msg == "hello" then "bar" else "" - h4 <- handle endpoint4 name4 "foo" $ \bytes -> let Right msg = decode bytes in - return $ encode $ if msg == "hello" then "baz" else "" - let cs = newCallSite endpoint1 name1 - results <- (gcall cs [name2,name3,name4] "foo" $ encode "hello") - assertBool "Foo not present in results" (elem (encode "foo") $ M.elems results) - assertBool "Bar not present in results" (elem (encode "bar") $ M.elems results) - assertBool "Bar not present in results" (elem (encode "baz") $ M.elems results) - assertEqual "Unxpected number of results" 3 (M.size results) - hangup h2 - hangup h3 - hangup h4 - -testAnyCall :: Assertion -testAnyCall = do - let name1 = "endpoint1" - name2 = "endpoint2" - name3 = "endpoint3" - name4 = "endpoint4" - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - endpoint3 <- newEndpoint [transport] - endpoint4 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - Right () <- bindEndpoint endpoint3 name3 - Right () <- bindEndpoint endpoint4 name4 - h2 <- handle endpoint2 name2 "foo" $ \bytes -> let Right msg = decode bytes in - return $ encode $ if msg == "hello" then "foo" else "" - h3 <- handle endpoint3 name3 "foo" $ \bytes -> let Right msg = decode bytes in - return $ encode $ if msg == "hello" then "foo" else "" - h4 <- handle endpoint4 name4 "foo" $ \bytes -> let Right msg = decode bytes in - return $ encode $ if msg == "hello" then "foo" else "" - let cs = newCallSite endpoint1 name1 - (result,responder) <- (anyCall cs [name2,name3,name4] "foo" $ encode "hello") - assertEqual "Response should have been 'foo'" (encode "foo") result - assertBool "Responder was not in original list of names" $ elem responder [name2,name3,name4] - hangup h2 - hangup h3 - hangup h4 - testOneHandlerWithTimeout :: Assertion testOneHandlerWithTimeout = do - let name1 = "endpoint1" - name2 = "endpoint2" - longer = 500 * 1000 -- half a second - shorter = 250 * 1000 -- quarter second - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - -- first call with caller waiting longer than handler - h1 <- handle endpoint2 name2 "foo" $ \bytes -> do - let Right msg = decode bytes - threadDelay shorter - return $ encode $ msg ++ "!" - let cs1 = newCallSite endpoint1 name1 - Just bytes1 <- callWithTimeout cs1 name2 "foo" longer $ encode "hello" - let Right result1 = decode bytes1 - assertEqual "Result not expected value" (Just "hello!") (Just result1) - hangup h1 - -- now call with handler waiting longer than caller - h2 <- handle endpoint2 name2 "foo" $ \bytes -> do - let Right msg = decode bytes - threadDelay longer - return $ encode $ msg ++ "!" - let cs2 = newCallSite endpoint1 name1 - bytes2 <- (callWithTimeout cs2 name2 "foo" shorter $ encode "hello") - assertEqual "Result not expected value" Nothing bytes2 - hangup h2 + let name1 = Name "endpoint1" + name2 = Name "endpoint2" + longer = (3 * 1000000 :: Int) + shorter = (1 * 1000000 :: Int) + withTransport newMemoryTransport $ \transport -> + withNewEndpoint2 transport $ \endpoint1 endpoint2 -> + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ + withConnection transport endpoint1 name2 $ do + -- first call with caller waiting longer than handler + h1 <- handle endpoint2 name2 "foo" $ \bytes -> do + let Right msg = decode bytes + threadDelay shorter + return $ encode $ msg ++ "!" + let cs1 = newCallSite endpoint1 name1 + Just bytes1 <- callWithTimeout cs1 name2 "foo" longer $ encode "hello" + let Right result1 = decode bytes1 + assertEqual "Result not expected value" (Just "hello!") (Just result1) + hangup h1 + -- now call with handler waiting longer than caller + h2 <- handle endpoint2 name2 "foo" $ \bytes -> do + let Right msg = decode bytes + threadDelay longer + return $ encode $ msg ++ "!" + let cs2 = newCallSite endpoint1 name1 + bytes2 <- (callWithTimeout cs2 name2 "foo" shorter $ encode "hello") + assertEqual "Result not expected value" Nothing bytes2 + hangup h2 testGroupCallWithTimeout :: Assertion testGroupCallWithTimeout = do - let name1 = "endpoint1" - name2 = "endpoint2" - name3 = "endpoint3" - name4 = "endpoint4" - longest = 750 * 1000 -- three quarters of a second - longer = 500 * 1000 -- half a second - shorter = 250 * 1000 -- quarter second - transport <- newMemoryTransport - endpoint1 <- newEndpoint [transport] - endpoint2 <- newEndpoint [transport] - endpoint3 <- newEndpoint [transport] - endpoint4 <- newEndpoint [transport] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - Right () <- bindEndpoint endpoint3 name3 - Right () <- bindEndpoint endpoint4 name4 - h2 <- handle endpoint2 name2 "foo" $ \bytes -> do - let Right msg = decode bytes - threadDelay shorter - return $ encode $ if msg == "hello" then "foo" else "" - h3 <- handle endpoint3 name3 "foo" $ \bytes -> do - threadDelay shorter - let Right msg = decode bytes - return $ encode $ if msg == "hello" then "bar" else "" - h4 <- handle endpoint4 name4 "foo" $ \bytes -> do - threadDelay longest - let Right msg = decode bytes - return $ encode $ if msg == "hello" then "baz" else "" - let cs = newCallSite endpoint1 name1 - results <- gcallWithTimeout cs [name2,name3,name4] "foo" longer $ encode "hello" - assertEqual "Foo not present in results" (Just $ Just $ encode "foo") (M.lookup name2 results) - assertEqual "Bar not present in results" (Just $ Just $ encode "bar") (M.lookup name3 results) - assertEqual "Baz shouldn't be present in results" (Just $ Nothing) (M.lookup name4 results) - assertEqual "Unxpected number of results" 3 (M.size results) - hangup h2 - hangup h3 - hangup h4 + let name1 = Name "endpoint1" + name2 = Name "endpoint2" + name3 = Name "endpoint3" + name4 = Name "endpoint4" + longest = 750 * 1000 -- three quarters of a second + longer = 500 * 1000 -- half a second + shorter = 250 * 1000 -- quarter second + withTransport newMemoryTransport $ \transport -> + withNewEndpoint4 transport $ \endpoint1 endpoint2 endpoint3 endpoint4 -> do + withBinding4 transport (endpoint1,name1) (endpoint2,name2) (endpoint3,name3) (endpoint4,name4) $ + withConnection3 transport endpoint1 name2 name3 name4 $ do + h2 <- handle endpoint2 name2 "foo" $ \bytes -> do + let Right msg = decode bytes + threadDelay shorter + return $ encode $ if msg == "hello" then "foo" else "" + h3 <- handle endpoint3 name3 "foo" $ \bytes -> do + threadDelay shorter + let Right msg = decode bytes + return $ encode $ if msg == "hello" then "bar" else "" + h4 <- handle endpoint4 name4 "foo" $ \bytes -> do + threadDelay longest + let Right msg = decode bytes + return $ encode $ if msg == "hello" then "baz" else "" + let cs = newCallSite endpoint1 name1 + results <- gcallWithTimeout cs [name2,name3,name4] "foo" longer $ encode "hello" + assertEqual "Foo not present in results" (Just $ Just $ encode "foo") (M.lookup name2 results) + assertEqual "Bar not present in results" (Just $ Just $ encode "bar") (M.lookup name3 results) + assertEqual "Baz shouldn't be present in results" (Just $ Nothing) (M.lookup name4 results) + assertEqual "Unxpected number of results" 3 (M.size results) + hangup h2 + hangup h3 + hangup h4 diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -1,58 +1,49 @@ -module TestTCP (tests) where +module TestTCP ( + tests4 + ,tests6 + ) where -- local imports -import Network.Endpoints -import Network.Transport.TCP +import Network.Transport.Sockets.TCP -import TestTransports -import TestUtils +import TransportTestSuite -- external imports -import Control.Exception - import Test.Framework -import Test.HUnit -import Test.Framework.Providers.HUnit --- import Test.Framework.Providers.QuickCheck2 ----------------------------------------------------------------------------- ----------------------------------------------------------------------------- -_log :: String -_log = "test.transport.tcp" +tests4 :: IO [Test.Framework.Test] +tests4 = do + name1 <- newTCPAddress + name2 <- newTCPAddress + name3 <- newTCPAddress + name4 <- newTCPAddress + return $ transportTestSuite + (newTCPTransport4 tcpSocketResolver4) + "tcp4" + name1 + name2 + name3 + name4 -tests :: [Test.Framework.Test] -tests = - [ - testCase "tcp-endpoints+transport" $ testEndpointTransport newTCPTransport, - testCase "tcp-bind-unbind" $ endpointBindUnbind _log newTCPTransport newTCPAddress, - testCase "tcp-send-receive" $ endpointSendReceive _log newTCPTransport newTCPAddress, - testCase "tcp-double-send-receive" $ endpointDoubleSendReceive _log newTCPTransport newTCPAddress, - testCase "tcp-send-receive-reply" $ endpointSendReceiveReply _log newTCPTransport newTCPAddress, - testCase "tcp-multiple-send-receive-reply" $ endpointMultipleSendReceiveReply _log newTCPTransport newTCPAddress, - testCase "tcp-local-send-receive-reply" $ endpointLocalSendReceiveReply _log newTCPTransport newTCPAddress, +tests6 :: IO [Test.Framework.Test] +tests6 = do + name1 <- newTCPAddress6 + name2 <- newTCPAddress6 + name3 <- newTCPAddress6 + name4 <- newTCPAddress6 + return $ transportTestSuite + (newTCPTransport6 tcpSocketResolver6) + "tcp6" + name1 + name2 + name3 + name4 - testCase "tcp6-endpoints+transport" $ whenIPv6 $ testEndpointTransport newTCPTransport6, - testCase "tcp6-bind-unbind" $ whenIPv6 $ endpointBindUnbind _log newTCPTransport6 newTCPAddress6, - testCase "tcp6-send-receive" $ whenIPv6 $ endpointSendReceive _log newTCPTransport6 newTCPAddress6, - testCase "tcp6-double-send-receive" $ whenIPv6 $ endpointDoubleSendReceive _log newTCPTransport6 newTCPAddress6, - testCase "tcp6-send-receive-reply" $ whenIPv6 $ endpointSendReceiveReply _log newTCPTransport6 newTCPAddress6, - testCase "tcp6-multiple-send-receive-reply" $ whenIPv6 $ endpointMultipleSendReceiveReply _log newTCPTransport6 newTCPAddress6, - testCase "tcp6-local-send-receive-reply" $ whenIPv6 $ endpointLocalSendReceiveReply _log newTCPTransport6 newTCPAddress6 - ] -testEndpointTransport :: (Resolver -> IO Transport) -> Assertion -testEndpointTransport transportFactory = do - let name1 = "endpoint1" - name2 = "endpoint2" - address1 <- newTCPAddress - address2 <- newTCPAddress - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (transportFactory resolver) - shutdown - (\transport -> do - _ <- newEndpoint [transport] - return ()) +_log :: String +_log = "_test_TCP" diff --git a/tests/TestTransports.hs b/tests/TestTransports.hs deleted file mode 100644 --- a/tests/TestTransports.hs +++ /dev/null @@ -1,252 +0,0 @@ -module TestTransports ( - testDelay, - verifiedSend, - whenIPv6, - - -- Common tests - endpointTransport, - endpointBindUnbind, - endpointSendReceive, - endpointDoubleSendReceive, - endpointSendReceiveReply, - endpointLocalSendReceiveReply, - endpointMultipleSendReceiveReply -) where - --- local imports - -import Network.Endpoints - -import Network.Transport.TCP - --- external imports - -import Control.Concurrent -import Control.Exception - -import Data.Serialize - -import qualified Network.Socket as NS - -import System.Log.Logger - -import Test.HUnit - ------------------------------------------------------------------------------ ------------------------------------------------------------------------------ - -_log :: String -_log = "test.transports" - -testDelay :: Int -testDelay = 1 * 1000000 - -pause :: IO () -pause = threadDelay testDelay - -whenIPv6 :: Assertion -> Assertion -whenIPv6 assn = do - addresses <- lookupAddresses NS.AF_INET6 NS.Stream "localhost:1" - case addresses of - [] -> do - warningM _log $ "IPv6 not available" - return () - _ -> assn - -{- -Common tests--just supply the transport factory and an address generator --} - -endpointTransport :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointTransport _log newTransport newAddress = do - address1 <- newAddress - address2 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTransport resolver) - shutdown - (\transport -> do - _ <- newEndpoint [transport] - return ()) - -endpointBindUnbind :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointBindUnbind _log newTransport newAddress = do - infoM _log "Starting bind-unbind test" - address1 <- newAddress - address2 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newTransport 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 ()) - -endpointSendReceive :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointSendReceive _log newTransport newAddress = do - infoM _log "Starting send-receive test" - address1 <- newAddress - address2 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracketTest _log resolver newTransport $ \transport1 transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - pause - - verifiedSend _log endpoint1 endpoint2 name1 name2 "hello" - - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - return () - infoM _log "Finished send-receive test" - -endpointDoubleSendReceive :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointDoubleSendReceive _log newTransport newAddress = do - infoM _log "Starting double-send-receive test" - address1 <- newAddress - address2 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - name3 = "endpoint3" - let resolver = resolverFromList [(name1,address1), - (name2,address2), - (name3,address1)] - bracketTest _log resolver newTransport $ \transport1 transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - endpoint3 <- newEndpoint [transport1] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - Right () <- bindEndpoint endpoint3 name3 - pause - - verifiedSend _log endpoint1 endpoint2 name1 name2 "hello" - verifiedSend _log endpoint3 endpoint2 name3 name2 "ciao!" - - Right () <- unbindEndpoint endpoint1 name1 - - verifiedSend _log endpoint3 endpoint2 name3 name2 "hi!" - - Right () <- unbindEndpoint endpoint2 name2 - Right () <- unbindEndpoint endpoint3 name3 - return () - infoM _log "Finished double-send-receive test" - -endpointSendReceiveReply :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointSendReceiveReply _log newTransport newAddress = do - infoM _log "Starting send-receive-reply test" - address1 <- newAddress - address2 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracketTest _log resolver newTransport $ \transport1 transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - pause - - verifiedSend _log endpoint1 endpoint2 name1 name2 "hello" - verifiedSend _log endpoint2 endpoint1 name2 name1 "hi!" - - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - - return () - infoM _log "Finished send-receive-reply test" - -endpointLocalSendReceiveReply :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointLocalSendReceiveReply _log newTransport newAddress = do - infoM _log "Starting local-send-receive-reply test" - address1 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address1)] - bracket (newTransport resolver) - shutdown - (\transport1 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport1] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - pause - - verifiedSend _log endpoint1 endpoint2 name1 name2 "hello" - verifiedSend _log endpoint2 endpoint1 name2 name1 "hi!" - - Right () <- unbindEndpoint endpoint1 name1 - Right () <- unbindEndpoint endpoint2 name2 - - return ()) - infoM _log "Finished local-send-receive-reply test" - -endpointMultipleSendReceiveReply :: String -> (Resolver -> IO Transport) -> IO Address -> Assertion -endpointMultipleSendReceiveReply _log newTransport newAddress = do - infoM _log "Starting multiple-send-receive-reply test" - address1 <- newAddress - address2 <- newAddress - let name1 = "endpoint1" - name2 = "endpoint2" - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracketTest _log resolver newTransport $ \transport1 transport2 -> do - endpoint1 <- newEndpoint [transport1] - endpoint2 <- newEndpoint [transport2] - Right () <- bindEndpoint endpoint1 name1 - Right () <- bindEndpoint endpoint2 name2 - pause - - 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 () - infoM _log "Finished multiple-send-receive-reply test" - where - roundtrip endpoint1 endpoint2 name1 name2 = do - verifiedSend _log endpoint1 endpoint2 name1 name2 "hello" - verifiedSend _log endpoint2 endpoint1 name2 name1 "hi!" - --- helpers - -bracketTest :: String -> Resolver -> (Resolver -> IO Transport) -> (Transport -> Transport -> Assertion) -> Assertion -bracketTest _log resolver newTransport blk = do - bracket (newTransport resolver) - shutdown - (\transport1 -> - bracket (newTransport resolver) - shutdown - (\transport2 -> catch (blk transport1 transport2) - (\e -> do - errorM _log $ "Encountered error running test: " ++ (show (e :: SomeException)) - assertFailure "Unexpected error"))) - -verifiedSend :: String -> Endpoint -> Endpoint -> Name -> Name -> String -> Assertion -verifiedSend _log endpoint1 endpoint2 name1 name2 msg = do - infoM _log $ "Sending message from " ++ name1 ++ " to " ++ name2 - sendMessage_ endpoint1 name2 $ encode msg - maybeMsg <- receiveMessageTimeout endpoint2 testDelay - case maybeMsg of - Nothing -> assertFailure "No message received" - Just msg1 -> assertEqual "Received message not same as sent" (Right msg) (decode msg1) diff --git a/tests/TestUDP.hs b/tests/TestUDP.hs --- a/tests/TestUDP.hs +++ b/tests/TestUDP.hs @@ -1,59 +1,50 @@ -module TestUDP where +module TestUDP ( + tests4 + ,tests6 +) where -- local imports -import Network.Endpoints -import Network.Transport.UDP +import Network.Transport.Sockets.UDP -import TestTransports -import TestUtils +import TransportTestSuite -- external imports -import Control.Exception import Test.Framework -import Test.HUnit -import Test.Framework.Providers.HUnit --- import Test.Framework.Providers.QuickCheck2 ----------------------------------------------------------------------------- ----------------------------------------------------------------------------- -_log :: String -_log = "test.transport.udp" - -tests :: [Test.Framework.Test] -tests = - [ - testCase "udp-endpoints+transport" testEndpointTransport, +tests4 :: IO [Test.Framework.Test] +tests4 = do + name1 <- newUDPAddress + name2 <- newUDPAddress + name3 <- newUDPAddress + name4 <- newUDPAddress + return $ transportTestSuite + (newUDPTransport4 udpSocketResolver4) + "udp4" + name1 + name2 + name3 + name4 - testCase "udp-bind-unbind" $ endpointBindUnbind _log newUDPTransport newUDPAddress, - testCase "udp-send-receive" $ endpointSendReceive _log newUDPTransport newUDPAddress, - testCase "udp-double-send-receive" $ endpointDoubleSendReceive _log newUDPTransport newUDPAddress, - testCase "udp-send-receive-reply" $ endpointSendReceiveReply _log newUDPTransport newUDPAddress, - testCase "udp-multiple-send-receive-reply" $ endpointMultipleSendReceiveReply _log newUDPTransport newUDPAddress, - testCase "udp-local-send-receive-reply" $ endpointLocalSendReceiveReply _log newUDPTransport newUDPAddress, - - testCase "udp6-endpoints+transport" $ whenIPv6 $ testEndpointTransport, +tests6 :: IO [Test.Framework.Test] +tests6 = do + name1 <- newUDPAddress6 + name2 <- newUDPAddress6 + name3 <- newUDPAddress6 + name4 <- newUDPAddress6 + return $ transportTestSuite + (newUDPTransport6 udpSocketResolver6) + "udp6" + name1 + name2 + name3 + name4 - testCase "udp6-bind-unbind" $ whenIPv6 $ endpointBindUnbind _log newUDPTransport6 newUDPAddress6, - testCase "udp6-send-receive" $ whenIPv6 $ endpointSendReceive _log newUDPTransport6 newUDPAddress6, - testCase "udp6-double-send-receive" $ whenIPv6 $ endpointDoubleSendReceive _log newUDPTransport6 newUDPAddress6, - testCase "udp6-send-receive-reply" $ whenIPv6 $ endpointSendReceiveReply _log newUDPTransport6 newUDPAddress6, - testCase "udp6-multiple-send-receive-reply" $ whenIPv6 $ endpointMultipleSendReceiveReply _log newUDPTransport6 newUDPAddress6, - testCase "udp6-local-send-receive-reply" $ whenIPv6 $ endpointLocalSendReceiveReply _log newUDPTransport6 newUDPAddress6 - ] +_log :: String +_log = "_test_UDP" -testEndpointTransport :: Assertion -testEndpointTransport = do - let name1 = "endpoint1" - name2 = "endpoint2" - address1 <- newUDPAddress - address2 <- newUDPAddress - let resolver = resolverFromList [(name1,address1), - (name2,address2)] - bracket (newUDPTransport resolver) - shutdown - (\transport -> do - _ <- newEndpoint [transport] - return ()) +-- UDP tests diff --git a/tests/TestUtils.hs b/tests/TestUtils.hs --- a/tests/TestUtils.hs +++ b/tests/TestUtils.hs @@ -1,34 +1,83 @@ module TestUtils - ( newTCPAddress - , newUDPAddress - , newTCPAddress6 - , newUDPAddress6) - where + ( + withNewEndpoint, + withNewEndpoint2, + withNewEndpoint3, + withNewEndpoint4, + newTCPAddress, + newUDPAddress, + newTCPAddress6, + newUDPAddress6, + isIPv6Available, + + timeBound, + troubleshoot, + + ) + where + +-- local imports + +import Network.Endpoints +import Network.Transport.Sockets.TCP + +-- external imports +import Control.Concurrent +import Control.Concurrent.Async import Control.Exception + import qualified Network.Socket as NS -newTCPAddress :: IO String +import System.Log.Logger + +import Test.HUnit + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +withNewEndpoint :: Transport -> (Endpoint -> IO ()) -> IO () +withNewEndpoint transport fn = do + endpoint <- newEndpoint + withEndpoint transport endpoint $ fn endpoint + +withNewEndpoint2 :: Transport -> (Endpoint -> Endpoint -> IO ()) -> IO () +withNewEndpoint2 transport fn = do + withNewEndpoint transport $ \endpoint1 -> + withNewEndpoint transport $ \endpoint2 -> + fn endpoint1 endpoint2 + +withNewEndpoint3 :: Transport -> (Endpoint -> Endpoint -> Endpoint -> IO ()) -> IO () +withNewEndpoint3 transport fn = withNewEndpoint transport $ \endpoint1 -> + withNewEndpoint2 transport $ \endpoint2 endpoint3 -> + fn endpoint1 endpoint2 endpoint3 + +withNewEndpoint4 :: Transport -> (Endpoint -> Endpoint -> Endpoint -> Endpoint -> IO ()) -> IO () +withNewEndpoint4 transport fn = withNewEndpoint2 transport $ \endpoint1 endpoint2 -> + withNewEndpoint2 transport $ \endpoint3 endpoint4 -> + fn endpoint1 endpoint2 endpoint3 endpoint4 + +newTCPAddress :: IO Name newTCPAddress = do - NS.SockAddrInet (NS.PortNum p) _ <- availablePort NS.AF_INET NS.Stream - return $ "localhost:" ++ show p + NS.SockAddrInet p _ <- availablePort NS.AF_INET NS.Stream + return $ Name $ "localhost:" ++ show p -newUDPAddress :: IO String +newUDPAddress :: IO Name newUDPAddress = do - NS.SockAddrInet (NS.PortNum p) _ <- availablePort NS.AF_INET NS.Datagram - return $ "localhost:" ++ show p + NS.SockAddrInet p _ <- availablePort NS.AF_INET NS.Datagram + return $ Name $ "localhost:" ++ show p -newTCPAddress6 :: IO String +newTCPAddress6 :: IO Name newTCPAddress6 = do - NS.SockAddrInet6 (NS.PortNum p) _ _ _ <- availablePort NS.AF_INET6 NS.Stream - return $ "localhost:" ++ show p + NS.SockAddrInet6 p _ _ _ <- availablePort NS.AF_INET6 NS.Stream + return $ Name $ "localhost:" ++ show p -newUDPAddress6 :: IO String +newUDPAddress6 :: IO Name newUDPAddress6 = do - NS.SockAddrInet6 (NS.PortNum p) _ _ _ <- availablePort NS.AF_INET6 NS.Datagram - return $ "localhost:" ++ show p + NS.SockAddrInet6 p _ _ _ <- availablePort NS.AF_INET6 NS.Datagram + return $ Name $ "localhost:" ++ show p -availablePort :: NS.Family -> NS.SocketType -> IO NS.SockAddr +availablePort :: NS.Family -> NS.SocketType -> IO NS.SockAddr availablePort f t = do let hints = NS.defaultHints { NS.addrFamily = f , NS.addrSocketType = t @@ -47,7 +96,25 @@ else return addr ) +isIPv6Available :: IO Bool +isIPv6Available = do + addresses <- tcpSocketResolver6 (Name "localhost:1") + case addresses of + [] -> return False + _ -> return True + isPrivileged :: NS.SockAddr -> Bool -isPrivileged (NS.SockAddrInet (NS.PortNum p) _) = p < 1025 -isPrivileged (NS.SockAddrInet6 (NS.PortNum p) _ _ _) = p < 1025 -isPrivileged (NS.SockAddrUnix _) = False +isPrivileged (NS.SockAddrInet p _) = p < 1025 +isPrivileged (NS.SockAddrInet6 p _ _ _) = p < 1025 +isPrivileged _ = False + +timeBound :: Int -> IO () -> IO () +timeBound delay action = do + outcome <- race (threadDelay delay) action + assertBool "Test should not block" $ outcome == Right () + +troubleshoot :: IO () -> IO () +troubleshoot fn = do + finally (do + updateGlobalLogger rootLoggerName (setLevel INFO) + fn) (updateGlobalLogger rootLoggerName (setLevel WARNING)) diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -2,7 +2,7 @@ -- local imports -import Network.Endpoints +import TestUtils -- external imports @@ -13,15 +13,13 @@ import System.Log.Logger import Test.Framework -import Test.HUnit -import Test.Framework.Providers.HUnit -- Test modules import qualified TestMailbox as MB import qualified TestMemory as M +import qualified TestRPC as R import qualified TestTCP as T import qualified TestUDP as U -import qualified TestRPC as R ----------------------------------------------------------------------------- ----------------------------------------------------------------------------- @@ -29,7 +27,15 @@ main :: IO () main = do initLogging - defaultMain tests + ipv6 <- isIPv6Available + t4 <- T.tests4 + t6 <- T.tests6 + u4 <- U.tests4 + u6 <- U.tests6 + let allTests = tests ++ t4 ++ u4 + if ipv6 + then defaultMain (allTests ++ t6 ++ u6) + else defaultMain allTests initLogging :: IO () initLogging = do @@ -40,19 +46,6 @@ tests :: [Test.Framework.Test] tests = - [ - testCase "hunit" (assertBool "HUnit assertion of truth is false" True), - testCase "endpoints" testEndpoint - ] - ++ MB.tests + MB.tests ++ M.tests - ++ T.tests - ++ U.tests ++ R.tests - --- Endpoint tests - -testEndpoint :: Assertion -testEndpoint = do - _ <- newEndpoint [] - return () diff --git a/tests/TransportTestSuite.hs b/tests/TransportTestSuite.hs new file mode 100644 --- /dev/null +++ b/tests/TransportTestSuite.hs @@ -0,0 +1,275 @@ +module TransportTestSuite + ( + transportTestSuite, + + testTransportEndpointSendReceive, + testTransportEndpointSend2Receive2, + testTransportEndpointSendReceive2SerialServers, + testTransportEndpointSendReceive2SerialClients, + testTransportOneHearCall, + testTransportOneCallHear, + testTransportConcurrentCallHear, + testTransportOneHandler, + testTransportTwoHandlers, + testTransportGroupCall, + testTransportAnyCall, + + module TestUtils + ) + where + +-- local imports + +import Network.Endpoints +import Network.RPC +import Network.Transport +import TestUtils + +-- external imports + +import Control.Applicative +import Control.Concurrent.Async + +import qualified Data.Map as M +import Data.Serialize + +import Test.Framework +import Test.HUnit +import Test.Framework.Providers.HUnit + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- + +transportTestSuite :: IO Transport -> String -> Name -> Name -> Name -> Name -> [Test.Framework.Test] +transportTestSuite transport transportLabel name1 name2 name3 name4 = [ + testCase (transportLabel ++ "-sendReceive") $ + testTransportEndpointSendReceive transport name1 name2, + testCase (transportLabel ++ "-send2Receive2") $ + testTransportEndpointSend2Receive2 transport name1 name2, + testCase (transportLabel ++ "-sendReceive-2-serial-servers") $ + testTransportEndpointSendReceive2SerialServers transport name1 name2, + testCase (transportLabel ++ "-sendReceive-2-serial-clients") $ + testTransportEndpointSendReceive2SerialClients transport name1 name2, + testCase (transportLabel ++ "-rpc-one-hear-call") $ + testTransportOneHearCall transport name1 name2, + testCase (transportLabel ++ "-rpc-one-call-hear") $ + testTransportOneCallHear transport name1 name2, + testCase (transportLabel ++ "-rpc-concurrent-call-hear") $ + testTransportConcurrentCallHear transport name1 name2, + testCase (transportLabel ++ "-rpc-one-handler") $ + testTransportOneHandler transport name1 name2, + testCase (transportLabel ++ "-rpc-two-handlers") $ + testTransportTwoHandlers transport name1 name2, + testCase (transportLabel ++ "-rpc-group-call") $ + testTransportGroupCall transport name1 name2 name3 name4, + testCase (transportLabel ++ "-rpc-any-call") $ + testTransportAnyCall transport name1 name2 name3 name4 + ] + +timeLimited :: Assertion -> Assertion +timeLimited assn = timeBound (1 * 1000000 :: Int) assn + +testTransportEndpointSendReceive :: IO Transport -> Name -> Name -> Assertion +testTransportEndpointSendReceive transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint2 transport $ \endpoint1 endpoint2 -> do + withBinding2 transport (endpoint1,name1) (endpoint2,name2) $ do + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello!" + msg <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "hello!") (decode msg) + return () + +testTransportEndpointSend2Receive2 :: IO Transport -> Name -> Name -> Assertion +testTransportEndpointSend2Receive2 transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint2 transport $ \endpoint1 endpoint2 -> do + withBinding2 transport (endpoint1,name1) (endpoint2,name2) $ do + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello!" + msg1 <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "hello!") (decode msg1) + sendMessage endpoint1 name2 $ encode "ciao!" + msg2 <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "ciao!") (decode msg2) + return () + +testTransportEndpointSendReceive2SerialServers :: IO Transport -> Name -> Name -> Assertion +testTransportEndpointSendReceive2SerialServers transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint2 transport $ \endpoint1 endpoint2 -> do + withName endpoint1 name1 $ do + withBinding transport endpoint2 name2 $ do + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello!" + msg <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "hello!") (decode msg) + return () + withBinding transport endpoint2 name2 $ do + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello!" + msg <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "hello!") (decode msg) + return () + +testTransportEndpointSendReceive2SerialClients :: IO Transport -> Name -> Name -> Assertion +testTransportEndpointSendReceive2SerialClients transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint2 transport $ \endpoint1 endpoint2 -> + withBinding transport endpoint2 name2 $ do + withName endpoint1 name1 $ + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello!" + msg <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "hello!") (decode msg) + return () + withName endpoint1 name1 $ + withConnection transport endpoint1 name2 $ do + sendMessage endpoint1 name2 $ encode "hello!" + msg <- receiveMessage endpoint2 + assertEqual "Received message not same as sent" (Right "hello!") (decode msg) + return () + +testTransportOneHearCall :: IO Transport -> Name -> Name -> Assertion +testTransportOneHearCall transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint2 transport $ \endpoint1 endpoint2 -> + withBinding2 transport (endpoint1,name1) (endpoint2,name2) $ do + withConnection transport endpoint1 name2 $ do + _ <- async $ do + (bytes,reply) <- hear endpoint2 name2 "foo" + let Right msg = decode bytes + reply $ encode $ msg ++ "!" + let cs = newCallSite endpoint1 name1 + bytes <- call cs name2 "foo" $ encode "hello" + let Right result = decode bytes + assertEqual "Result not expected value" "hello!" result + +testTransportOneCallHear :: IO Transport -> Name -> Name -> Assertion +testTransportOneCallHear transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint transport $ \endpoint1 -> + withNewEndpoint transport $ \endpoint2 -> + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ do + withConnection transport endpoint1 name2 $ do + let cs = newCallSite endpoint1 name1 + acall <- async $ call cs name2 "foo" $ encode "hello" + _ <- async $ do + (bytes,reply) <- hear endpoint2 name2 "foo" + let Right msg = decode bytes + reply $ encode $ msg ++ "!" + bytes <- wait acall + let Right result = decode bytes + assertEqual "Result not expected value" "hello!" result + +testTransportConcurrentCallHear :: IO Transport -> Name -> Name -> Assertion +testTransportConcurrentCallHear transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint transport $ \endpoint1 -> + withNewEndpoint transport $ \endpoint2 -> + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ + withConnection transport endpoint1 name2 $ do + let cs1 = newCallSite endpoint1 name1 + cs2 = newCallSite endpoint2 name2 + let call1 = call cs1 name2 "foo" $ encode "hello" + hear1 = do + (bytes,reply) <- hear endpoint2 name2 "foo" + let Right msg = decode bytes + reply $ encode $ msg ++ "!" + call2 = call cs2 name1 "bar" $ encode "ciao" + hear2 = do + (bytes,reply) <- hear endpoint1 name1 "bar" + let Right msg = decode bytes + reply $ encode $ msg ++ "!" + (result1,(),result2,()) <- runConcurrently $ (,,,) + <$> Concurrently call1 + <*> Concurrently hear1 + <*> Concurrently call2 + <*> Concurrently hear2 + assertEqual "Result not expected value" (Right "hello!") (decode result1) + assertEqual "Result not expected value" (Right "ciao!") (decode result2) + +testTransportOneHandler :: IO Transport -> Name -> Name -> Assertion +testTransportOneHandler transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint transport $ \endpoint1 -> + withNewEndpoint transport $ \endpoint2 -> + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ + withConnection transport endpoint1 name2 $ do + h <- handle endpoint2 name2 "foo" $ \bytes -> + let Right msg = decode bytes + in return $ encode $ msg ++ "!" + let cs = newCallSite endpoint1 name1 + bytes <- call cs name2 "foo" $ encode "hello" + let Right result = decode bytes + assertEqual "Result not expected value" "hello!" result + hangup h + +testTransportTwoHandlers :: IO Transport -> Name -> Name -> Assertion +testTransportTwoHandlers transportFactory name1 name2 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint transport $ \endpoint1 -> + withNewEndpoint transport $ \endpoint2 -> + withBinding transport endpoint1 name1 $ + withBinding transport endpoint2 name2 $ + withConnection transport endpoint1 name2 $ do + h1 <- handle endpoint2 name2 "foo" $ \bytes -> + let Right msg = decode bytes + in return $ encode $ msg ++ "!" + h2 <- handle endpoint2 name2 "bar" $ \bytes -> + let Right msg = decode bytes + in return $ encode $ msg ++ "?" + let cs = newCallSite endpoint1 name1 + bytes1 <- call cs name2 "foo" $ encode "hello" + let Right result1 = decode bytes1 + assertEqual "Result not expected value" "hello!" result1 + bytes2 <- call cs name2 "bar" $ encode "hello" + let Right result2 = decode bytes2 + assertEqual "Result not expected value" "hello?" result2 + hangup h1 + hangup h2 + +testTransportGroupCall :: IO Transport -> Name -> Name -> Name -> Name -> Assertion +testTransportGroupCall transportFactory name1 name2 name3 name4 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint4 transport $ \endpoint1 endpoint2 endpoint3 endpoint4 -> do + withBinding4 transport (endpoint1,name1) (endpoint2,name2) (endpoint3,name3) (endpoint4,name4) $ + withConnection3 transport endpoint1 name2 name3 name4 $ do + h2 <- handle endpoint2 name2 "foo" $ \bytes -> let Right msg = decode bytes in + return $ encode $ if msg == "hello" then "foo" else "" + h3 <- handle endpoint3 name3 "foo" $ \bytes -> let Right msg = decode bytes in + return $ encode $ if msg == "hello" then "bar" else "" + h4 <- handle endpoint4 name4 "foo" $ \bytes -> let Right msg = decode bytes in + return $ encode $ if msg == "hello" then "baz" else "" + let cs = newCallSite endpoint1 name1 + results <- (gcall cs [name2,name3,name4] "foo" $ encode "hello") + assertBool "Foo not present in results" (elem (encode "foo") $ M.elems results) + assertBool "Bar not present in results" (elem (encode "bar") $ M.elems results) + assertBool "Bar not present in results" (elem (encode "baz") $ M.elems results) + assertEqual "Unxpected number of results" 3 (M.size results) + hangup h2 + hangup h3 + hangup h4 + +testTransportAnyCall :: IO Transport -> Name -> Name -> Name -> Name -> Assertion +testTransportAnyCall transportFactory name1 name2 name3 name4 = timeLimited $ do + withTransport transportFactory $ \transport -> + withNewEndpoint4 transport $ \endpoint1 endpoint2 endpoint3 endpoint4 -> do + withBinding4 transport (endpoint1,name1) (endpoint2,name2) (endpoint3,name3) (endpoint4,name4) $ + withConnection3 transport endpoint1 name2 name3 name4 $ do + h2 <- handle endpoint2 name2 "foo" $ \bytes -> let Right msg = decode bytes in + return $ encode $ if msg == "hello" then "foo" else "" + h3 <- handle endpoint3 name3 "foo" $ \bytes -> let Right msg = decode bytes in + return $ encode $ if msg == "hello" then "foo" else "" + h4 <- handle endpoint4 name4 "foo" $ \bytes -> let Right msg = decode bytes in + return $ encode $ if msg == "hello" then "foo" else "" + let cs = newCallSite endpoint1 name1 + (result,responder) <- (anyCall cs [name2,name3,name4] "foo" $ encode "hello") + assertEqual "Response should have been 'foo'" (encode "foo") result + assertBool "Responder was not in original list of names" $ elem responder [name2,name3,name4] + hangup h2 + hangup h3 + hangup h4