courier 0.1.0.15 → 0.1.1.1

raw patch · 24 files changed

+1674/−1882 lines, 24 filesnew-component:exe:echo-clientnew-component:exe:echo-serverPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• changes.md +18/−4
• courier.cabal +51/−18
• echo/EchoClient.hs +58/−0
• echo/EchoServer.hs +55/−0
• examples/HelloWorld.hs +44/−21
• src/Control/Concurrent/Mailbox.hs +9/−6
• src/Network/Endpoints.hs +144/−175
• src/Network/RPC.hs +20/−9
• src/Network/Transport.hs +212/−66
• src/Network/Transport/Internal.hs +0/−46
• src/Network/Transport/Memory.hs +54/−42
• src/Network/Transport/Sockets.hs +167/−450
• src/Network/Transport/Sockets/TCP.hs +163/−0
• src/Network/Transport/Sockets/UDP.hs +161/−0
• src/Network/Transport/TCP.hs +0/−203
• src/Network/Transport/UDP.hs +0/−154
• tests/TestMemory.hs +11/−65
• tests/TestRPC.hs +63/−244
• tests/TestTCP.hs +34/−43
• tests/TestTransports.hs +0/−252
• tests/TestUDP.hs +35/−44
• tests/TestUtils.hs +88/−21
• tests/Tests.hs +12/−19
• tests/TransportTestSuite.hs +275/−0

@@ -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

@@ -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"

@@ -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

@@ -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

@@ -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)

@@ -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

@@ -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

@@ -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.

@@ -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

@@ -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

@@ -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 ()+  }

@@ -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

@@ -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 ()

@@ -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 ()

@@ -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)

@@ -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

@@ -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-

@@ -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

@@ -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"

@@ -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)

@@ -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

@@ -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))

@@ -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 ()

@@ -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