distributed-process-execution (empty) → 0.1.0

raw patch · 14 files changed

+2852/−0 lines, 14 filesdep +HUnitdep +QuickCheckdep +ansi-terminalsetup-changed

Dependencies added: HUnit, QuickCheck, ansi-terminal, base, binary, bytestring, containers, data-accessor, deepseq, derive, distributed-process, distributed-process-client-server, distributed-process-execution, distributed-process-extras, distributed-process-supervisor, distributed-static, fingertree, ghc-prim, hashable, mtl, network, network-transport, network-transport-tcp, rematch, stm, template-haskell, test-framework, test-framework-hunit, test-framework-quickcheck2, time, transformers, uniplate, unordered-containers

Files

• LICENCE +30/−0
• Setup.lhs +3/−0
• distributed-process-execution.cabal +146/−0
• src/Control/Distributed/Process/Execution.hs +44/−0
• src/Control/Distributed/Process/Execution/EventManager.hs +153/−0
• src/Control/Distributed/Process/Execution/Exchange.hs +149/−0
• src/Control/Distributed/Process/Execution/Exchange/Broadcast.hs +338/−0
• src/Control/Distributed/Process/Execution/Exchange/Internal.hs +276/−0
• src/Control/Distributed/Process/Execution/Exchange/Router.hs +250/−0
• src/Control/Distributed/Process/Execution/Mailbox.hs +744/−0
• tests/MailboxTestFilters.hs +43/−0
• tests/TestExchange.hs +186/−0
• tests/TestMailbox.hs +256/−0
• tests/TestUtils.hs +234/−0

@@ -0,0 +1,30 @@+Copyright Tim Watson, 2012-2013.++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of the author nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain

@@ -0,0 +1,146 @@+name:           distributed-process-execution+version:        0.1.0+cabal-version:  >=1.8+build-type:     Simple+license:        BSD3+license-file:   LICENCE+stability:      experimental+Copyright:      Tim Watson 2012 - 2013+Author:         Tim Watson+Maintainer:     watson.timothy@gmail.com+Stability:      experimental+Homepage:       http://github.com/haskell-distributed/distributed-process-execution+Bug-Reports:    http://github.com/haskell-distributed/distributed-process-execution/issues+synopsis:       Execution Framework for The Cloud Haskell Application Platform+description:    +		The Execution Framework provides tools for load regulation, workload shedding and remote hand-off.+		The currently implementation provides only a subset of the plumbing required, comprising tools+		for event management, mailbox buffering and message routing.+category:       Control+tested-with:    GHC == 7.4.2 GHC == 7.6.2+data-dir:       ""++source-repository head+  type:      git+  location:  https://github.com/haskell-distributed/distributed-process-execution++flag perf+  description: Build with profiling enabled+  default: False++library+  build-depends:+                   base >= 4.4 && < 5,+                   data-accessor >= 0.2.2.3,+                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process-extras >= 0.1.1 && < 0.2,+                   distributed-process-supervisor >= 0.1.1 && < 0.2,+                   distributed-process-client-server >= 0.1.1 && < 0.2,+                   binary >= 0.6.3.0 && < 0.8,+                   deepseq >= 1.3.0.1 && < 1.4,+                   mtl,+                   containers >= 0.4 && < 0.6,+                   hashable >= 1.2.0.5 && < 1.3,+                   unordered-containers >= 0.2.3.0 && < 0.3,+                   fingertree < 0.2,+                   stm >= 2.4 && < 2.5,+                   time > 1.4 && < 1.5,+                   transformers+  if impl(ghc <= 7.5)+    Build-Depends:   template-haskell == 2.7.0.0,+                     derive == 2.5.5,+                     uniplate == 1.6.12,+                     ghc-prim+  extensions:      CPP+  hs-source-dirs:   src+  ghc-options:      -Wall+  exposed-modules:+                   Control.Distributed.Process.Execution,+                   Control.Distributed.Process.Execution.EventManager,+                   Control.Distributed.Process.Execution.Exchange,+                   Control.Distributed.Process.Execution.Mailbox+  other-modules:+                   Control.Distributed.Process.Execution.Exchange.Broadcast,+                   Control.Distributed.Process.Execution.Exchange.Internal,+                   Control.Distributed.Process.Execution.Exchange.Router+++test-suite ExchangeTests+  type:            exitcode-stdio-1.0+--  x-uses-tf:       true+  build-depends:+                   base >= 4.4 && < 5,+                   ansi-terminal >= 0.5 && < 0.7,+                   containers,+                   hashable,+                   unordered-containers >= 0.2.3.0 && < 0.3,+                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process-execution,+                   distributed-process-extras >= 0.1.1 && < 0.2,+                   distributed-static,+                   bytestring,+                   data-accessor,+                   fingertree < 0.2,+                   network-transport >= 0.4 && < 0.5,+                   deepseq >= 1.3.0.1 && < 1.4,+                   mtl,+                   network-transport-tcp >= 0.4 && < 0.5,+                   binary >= 0.6.3.0 && < 0.8,+                   network >= 2.3 && < 2.7,+                   HUnit >= 1.2 && < 2,+                   stm >= 2.3 && < 2.5,+                   time > 1.4 && < 1.5,+                   test-framework >= 0.6 && < 0.9,+                   test-framework-hunit,+                   QuickCheck >= 2.4,+                   test-framework-quickcheck2,+                   transformers,+                   rematch >= 0.2.0.0,+                   ghc-prim+  hs-source-dirs:+                   tests+  ghc-options:     -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog+  extensions:      CPP+  main-is:         TestExchange.hs+  other-modules:   TestUtils+++test-suite MailboxTests+  type:            exitcode-stdio-1.0+--  x-uses-tf:       true+  build-depends:+                   base >= 4.4 && < 5,+                   ansi-terminal >= 0.5 && < 0.7,+                   containers,+                   hashable,+                   unordered-containers >= 0.2.3.0 && < 0.3,+                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process-execution,+                   distributed-process-extras >= 0.1.1 && < 0.2,+                   distributed-static,+                   bytestring,+                   data-accessor,+                   fingertree < 0.2,+                   network-transport >= 0.4 && < 0.5,+                   deepseq >= 1.3.0.1 && < 1.4,+                   mtl,+                   network-transport-tcp >= 0.4 && < 0.5,+                   binary >= 0.6.3.0 && < 0.8,+                   network >= 2.3 && < 2.7,+                   HUnit >= 1.2 && < 2,+                   stm >= 2.3 && < 2.5,+                   time > 1.4 && < 1.5,+                   test-framework >= 0.6 && < 0.9,+                   test-framework-hunit,+                   QuickCheck >= 2.4,+                   test-framework-quickcheck2,+                   transformers,+                   rematch >= 0.2.0.0,+                   ghc-prim+  hs-source-dirs:+                   tests+  ghc-options:     -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog+  extensions:      CPP+  main-is:         TestMailbox.hs+  other-modules:   MailboxTestFilters+                   TestUtils

@@ -0,0 +1,44 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Execution+-- Copyright   :  (c) Tim Watson 2013 - 2014+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Inter-Process Traffic Management]+--+-- The /Execution Framework/ provides tools for load regulation, workload+-- shedding and remote hand-off. The currently implementation provides only+-- a subset of the plumbing required, comprising tools for event management,+-- mailbox buffering and message routing.+--+-----------------------------------------------------------------------------++module Control.Distributed.Process.Execution+  ( -- * Mailbox Buffering+    module Control.Distributed.Process.Execution.Mailbox+    -- * Message Exchanges+  , module Control.Distributed.Process.Execution.Exchange+  ) where++import Control.Distributed.Process.Execution.Exchange hiding (startSupervised)+import Control.Distributed.Process.Execution.Mailbox hiding (startSupervised, post)++{-++Load regulation requires that we apply limits to various parts of the system.+The manner in which they're applied may vary, but the mechanisms are limited+to:++1. rejecting the activity/request+2. accepting the activity immediately+3. blocking some or all requestors+4. blocking some (or all) activities+5. terminiating some (or all) activities++-}++

@@ -0,0 +1,153 @@+{-# LANGUAGE DeriveDataTypeable        #-}+{-# LANGUAGE DeriveGeneric             #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RankNTypes                #-}+{-# LANGUAGE RecordWildCards           #-}+{-# LANGUAGE ScopedTypeVariables       #-}+{-# LANGUAGE TemplateHaskell           #-}+{-# LANGUAGE ImpredicativeTypes        #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Execution.EventManager+-- Copyright   :  (c) Well-Typed / Tim Watson+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Overview]+--+-- The /EventManager/ is a parallel/concurrent event handling tool, built on+-- top of the /Exchange API/. Arbitrary events are published to the event+-- manager using 'notify', and are broadcast simulataneously to a set of+-- registered /event handlers/.+--+-- [Defining and Registering Event Handlers]+--+-- Event handlers are defined as @Serializable m => s -> m -> Process s@,+-- i.e., an expression taking an initial state, an arbitrary @Serializable@+-- event/message and performing an action in the @Process@ monad that evaluates+-- to a new state.+--+-- See "Control.Distributed.Process.Execution.Exchange".+--+-----------------------------------------------------------------------------++module Control.Distributed.Process.Execution.EventManager+  ( EventManager+  , start+  , startSupervised+  , startSupervisedRef+  , notify+  , addHandler+  , addMessageHandler+  ) where++import Control.Distributed.Process hiding (Message, link)+import qualified Control.Distributed.Process as P (Message)+import Control.Distributed.Process.Execution.Exchange+  ( Exchange+  , Message(..)+  , post+  , broadcastExchange+  , broadcastExchangeT+  , broadcastClient+  )+import qualified Control.Distributed.Process.Execution.Exchange as Exchange+  ( startSupervised+  )+import Control.Distributed.Process.Extras.Internal.Primitives+import Control.Distributed.Process.Extras.Internal.Unsafe+  ( InputStream+  , matchInputStream+  )+import Control.Distributed.Process.Supervisor (SupervisorPid)+import Control.Distributed.Process.Serializable hiding (SerializableDict)+import Data.Binary+import Data.Typeable (Typeable)+import GHC.Generics++{- notes++Event manager is implemented over a simple BroadcastExchange. We eschew the+complexities of identifying handlers and allowing them to be removed/deleted+or monitored, since we avoid running them in the exchange process. Instead,+each handler runs as an independent process, leaving handler management up+to the user and allowing all the usual process managemnet techniques (e.g.,+registration, supervision, etc) to be utilised instead.++-}++-- | Opaque handle to an Event Manager.+--+newtype EventManager = EventManager { ex :: Exchange }+  deriving (Typeable, Generic)+instance Binary EventManager where++instance Resolvable EventManager where+  resolve = resolve . ex++-- | Start a new /Event Manager/ process and return an opaque handle+-- to it.+start :: Process EventManager+start = broadcastExchange >>= return . EventManager++startSupervised :: SupervisorPid -> Process EventManager+startSupervised sPid = do+  ex <- broadcastExchangeT >>= \t -> Exchange.startSupervised t sPid+  return $ EventManager ex++startSupervisedRef :: SupervisorPid -> Process (ProcessId, P.Message)+startSupervisedRef sPid = do+  ex <- startSupervised sPid+  Just pid <- resolve ex+  return (pid, unsafeWrapMessage ex)++-- | Broadcast an event to all registered handlers.+notify :: Serializable a => EventManager -> a -> Process ()+notify em msg = post (ex em) msg++-- | Add a new event handler. The handler runs in its own process,+-- which is spawned locally on behalf of the caller.+addHandler :: forall s a. Serializable a+           => EventManager+           -> (s -> a -> Process s)+           -> Process s+           -> Process ProcessId+addHandler m h s =+  spawnLocal $ newHandler (ex m) (\s' m' -> handleMessage m' (h s')) s++-- | As 'addHandler', but operates over a raw @Control.Distributed.Process.Message@.+addMessageHandler :: forall s.+                     EventManager+                  -> (s -> P.Message -> Process (Maybe s))+                  -> Process s+                  -> Process ProcessId+addMessageHandler m h s = spawnLocal $ newHandler (ex m) h s++newHandler :: forall s .+              Exchange+           -> (s -> P.Message -> Process (Maybe s))+           -> Process s+           -> Process ()+newHandler ex handler initState = do+  linkTo ex+  is <- broadcastClient ex+  listen is handler =<< initState++listen :: forall s . InputStream Message+       -> (s -> P.Message -> Process (Maybe s))+       -> s+       -> Process ()+listen inStream handler state = do+  receiveWait [ matchInputStream inStream ] >>= handleEvent inStream handler state+  where+    handleEvent is h s p = do+      r <- h s (payload p)+      let s2 = case r of+                 Nothing -> s+                 Just s' -> s'+      listen is h s2+

@@ -0,0 +1,149 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Execution.Exchange+-- Copyright   :  (c) Tim Watson 2012 - 2014+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Message Exchanges]+--+-- The concept of a /message exchange/ is borrowed from the world of+-- messaging and enterprise integration. The /exchange/ acts like a kind of+-- mailbox, accepting inputs from /producers/ and forwarding these messages+-- to one or more /consumers/, depending on the implementation's semantics.+--+-- This module provides some basic types of message exchange and exposes an API+-- for defining your own custom /exchange types/.+--+-- [Broadcast Exchanges]+--+-- The broadcast exchange type, started via 'broadcastExchange', forward their+-- inputs to all registered consumers (as the name suggests). This exchange type+-- is highly optimised for local (intra-node) traffic and provides two different+-- kinds of client binding, one which causes messages to be delivered directly+-- to the client's mailbox (viz 'bindToBroadcaster'), the other providing a+-- separate stream of messages that can be obtained using the @expect@ and+-- @receiveX@ family of messaging primitives (and thus composed with other forms+-- of input selection, such as typed channels and selective reads on the process+-- mailbox).+--+-- /Important:/ When a @ProcessId@ is registered via 'bindToBroadcaster', only+-- the payload of the 'Message' (i.e., the underlying @Serializable@ datum) is+-- forwarded to the consumer, /not/ the whole 'Message' itself.+--+-- [Router Exchanges]+--+-- The /router/ API provides a means to selectively route messages to one or+-- more clients, depending on the content of the 'Message'. Two modes of binding+-- (and client selection) are provided out of the box, one of which matches the+-- message 'key', the second of which matches on a name and value from the+-- 'headers'. Alternative mechanisms for content based routing can be derived+-- by modifying the 'BindingSelector' expression passed to 'router'+--+-- See 'messageKeyRouter' and 'headerContentRouter' for the built-in routing+-- exchanges, and 'router' for the extensible routing API.+--+-- [Custom Exchange Types]+--+-- Both the /broadcast/ and /router/ exchanges are implemented as custom+-- /exchange types/. The mechanism for defining custom exchange behaviours+-- such as these is very simple. Raw exchanges are started by evaluating+-- 'startExchange' with a specific 'ExchangeType' record. This type is+-- parameterised by the internal /state/ it holds, and defines two API callbacks+-- in its 'configureEx' and 'routeEx' fields. The former is evaluated whenever a+-- client process evaluates 'configureExchange', the latter whenever a client+-- evaluates 'post' or 'postMessage'. The 'configureEx' callback takes a raw+-- @Message@ (from "Control.Distributed.Process") and is responsible for+-- decoding the message and updating its own state (if required). It is via+-- this callback that custom exchange types can receive information about+-- clients and handle it in their own way. The 'routeEx' callback is evaluated+-- with the exchange type's own internal state and the 'Message' originally+-- sent to the exchange process (via 'post') and is responsible for delivering+-- the message to its clients in whatever way makes sense for that exchange+-- type.+--+-----------------------------------------------------------------------------++module Control.Distributed.Process.Execution.Exchange+  ( -- * Fundamental API+    Exchange()+  , Message(..)+    -- * Starting/Running an Exchange+  , startExchange+  , startSupervised+  , startSupervisedRef+  , runExchange+    -- * Client Facing API+  , post+  , postMessage+  , configureExchange+  , createMessage+    -- * Broadcast Exchange+  , broadcastExchange+  , broadcastExchangeT+  , broadcastClient+  , bindToBroadcaster+  , BroadcastExchange+    -- * Routing (Content Based)+  , HeaderName+  , Binding(..)+  , Bindable+  , BindingSelector+  , RelayType(..)+    -- * Starting a Router+  , router+  , supervisedRouter+    -- * Routing (Publishing) API+  , route+  , routeMessage+    -- * Routing via message/binding keys+  , messageKeyRouter+  , bindKey+    -- * Routing via message headers+  , headerContentRouter+  , bindHeader+    -- * Defining Custom Exchange Types+  , ExchangeType(..)+  , applyHandlers+  ) where++import Control.Distributed.Process.Execution.Exchange.Broadcast+  ( broadcastExchange+  , broadcastExchangeT+  , broadcastClient+  , bindToBroadcaster+  , BroadcastExchange+  )+import Control.Distributed.Process.Execution.Exchange.Internal+  ( Exchange()+  , Message(..)+  , ExchangeType(..)+  , startExchange+  , startSupervised+  , startSupervisedRef+  , runExchange+  , post+  , postMessage+  , configureExchange+  , createMessage+  , applyHandlers+  )+import Control.Distributed.Process.Execution.Exchange.Router+  ( HeaderName+  , Binding(..)+  , Bindable+  , BindingSelector+  , RelayType(..)+  , router+  , supervisedRouter+  , route+  , routeMessage+  , messageKeyRouter+  , bindKey+  , headerContentRouter+  , bindHeader+  )+

@@ -0,0 +1,338 @@+{-# LANGUAGE DeriveDataTypeable    #-}+{-# LANGUAGE DeriveGeneric         #-}+{-# LANGUAGE StandaloneDeriving    #-}+{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE PatternGuards         #-}+{-# LANGUAGE RecordWildCards       #-}+{-# LANGUAGE FlexibleInstances     #-}+{-# LANGUAGE EmptyDataDecls        #-}+{-# LANGUAGE TemplateHaskell       #-}+{-# LANGUAGE ImpredicativeTypes    #-}+{-# LANGUAGE UndecidableInstances  #-}+{-# LANGUAGE MultiParamTypeClasses #-}++-- | An exchange type that broadcasts all incomings 'Post' messages.+module Control.Distributed.Process.Execution.Exchange.Broadcast+  (+    broadcastExchange+  , broadcastExchangeT+  , broadcastClient+  , bindToBroadcaster+  , BroadcastExchange+  ) where++import Control.Concurrent.STM (STM, atomically)+import Control.Concurrent.STM.TChan+  ( TChan+  , newBroadcastTChanIO+  , dupTChan+  , readTChan+  , writeTChan+  )+import Control.DeepSeq (NFData)+import Control.Distributed.Process+  ( Process+  , MonitorRef+  , ProcessMonitorNotification(..)+  , ProcessId+  , SendPort+  , processNodeId+  , getSelfPid+  , getSelfNode+  , liftIO+  , newChan+  , sendChan+  , unsafeSend+  , unsafeSendChan+  , receiveWait+  , match+  , matchIf+  , die+  , handleMessage+  , Match+  )+import qualified Control.Distributed.Process as P+import Control.Distributed.Process.Serializable()+import Control.Distributed.Process.Execution.Exchange.Internal+  ( startExchange+  , configureExchange+  , Message(..)+  , Exchange(..)+  , ExchangeType(..)+  , applyHandlers+  )+import Control.Distributed.Process.Extras.Internal.Types+  ( Channel+  , ServerDisconnected(..)+  )+import Control.Distributed.Process.Extras.Internal.Unsafe -- see [note: pcopy]+  ( PCopy+  , pCopy+  , pUnwrap+  , matchChanP+  , InputStream(Null)+  , newInputStream+  )+import Control.Monad (forM_, void)+import Data.Accessor+  ( Accessor+  , accessor+  , (^:)+  )+import Data.Binary+import qualified Data.Foldable as Foldable (toList)+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Typeable (Typeable)+import GHC.Generics++-- newtype RoutingTable r =+--  RoutingTable { routes :: (Map String (Map ProcessId r)) }++-- [note: BindSTM, BindPort and safety]+-- We keep these two /bind types/ separate, since only one of them+-- is truly serializable. The risk of unifying them is that at some+-- later time a maintainer might not realise that BindSTM cannot be+-- sent over the wire due to our use of PCopy.+--++data BindPort = BindPort { portClient :: !ProcessId+                         , portSend   :: !(SendPort Message)+                         } deriving (Typeable, Generic)+instance Binary BindPort where+instance NFData BindPort where++data BindSTM =+    BindSTM  { stmClient :: !ProcessId+             , stmSend   :: !(SendPort (PCopy (InputStream Message)))+             } deriving (Typeable)+{-  | forall r. (Routable r) =>+    BindR    { client :: !ProcessId+             , key    :: !String+             , chanC  :: !r+             }+  deriving (Typeable, Generic)+-}++data OutputStream =+    WriteChan (SendPort Message)+  | WriteSTM  (Message -> STM ())+--  | WriteP    ProcessId+  | NoWrite+  deriving (Typeable)++data Binding = Binding { outputStream :: !OutputStream+                       , inputStream  :: !(InputStream Message)+                       }+             | PidBinding !ProcessId+  deriving (Typeable)++data BindOk = BindOk+  deriving (Typeable, Generic)+instance Binary BindOk where+instance NFData BindOk where++data BindFail = BindFail !String+  deriving (Typeable, Generic)+instance Binary BindFail where+instance NFData BindFail where++data BindPlease = BindPlease+  deriving (Typeable, Generic)+instance Binary BindPlease where+instance NFData BindPlease where++type BroadcastClients = Map ProcessId Binding+data BroadcastEx =+  BroadcastEx { _routingTable   :: !BroadcastClients+              , channel         :: !(TChan Message)+              }++type BroadcastExchange = ExchangeType BroadcastEx++--------------------------------------------------------------------------------+-- Starting/Running the Exchange                                              --+--------------------------------------------------------------------------------++-- | Start a new /broadcast exchange/ and return a handle to the exchange.+broadcastExchange :: Process Exchange+broadcastExchange = broadcastExchangeT >>= startExchange++-- | The 'ExchangeType' of a broadcast exchange. Can be combined with the+-- @startSupervisedRef@ and @startSupervised@ APIs.+--+broadcastExchangeT :: Process BroadcastExchange+broadcastExchangeT = do+  ch <- liftIO newBroadcastTChanIO+  return $ ExchangeType { name        = "BroadcastExchange"+                        , state       = BroadcastEx Map.empty ch+                        , configureEx = apiConfigure+                        , routeEx     = apiRoute+                        }++--------------------------------------------------------------------------------+-- Client Facing API                                                          --+--------------------------------------------------------------------------------++-- | Create a binding to the given /broadcast exchange/ for the calling process+-- and return an 'InputStream' that can be used in the @expect@ and+-- @receiveWait@ family of messaging primitives. This form of client interaction+-- helps avoid cluttering the caller's mailbox with 'Message' data, since the+-- 'InputChannel' provides a separate input stream (in a similar fashion to+-- a typed channel).+-- Example:+--+-- > is <- broadcastClient ex+-- > msg <- receiveWait [ matchInputStream is ]+-- > handleMessage (payload msg)+--+broadcastClient :: Exchange -> Process (InputStream Message)+broadcastClient ex@Exchange{..} = do+  myNode <- getSelfNode+  us     <- getSelfPid+  if processNodeId pid == myNode -- see [note: pcopy]+     then do (sp, rp) <- newChan+             configureExchange ex $ pCopy (BindSTM us sp)+             mRef <- P.monitor pid+             P.finally (receiveWait [ matchChanP rp+                                    , handleServerFailure mRef ])+                       (P.unmonitor mRef)+     else do (sp, rp) <- newChan :: Process (Channel Message)+             configureExchange ex $ BindPort us sp+             mRef <- P.monitor pid+             P.finally (receiveWait [+                           match (\(_ :: BindOk)   -> return $ newInputStream $ Left rp)+                         , match (\(f :: BindFail) -> die f)+                         , handleServerFailure mRef+                         ])+                       (P.unmonitor mRef)++-- | Bind the calling process to the given /broadcast exchange/. For each+-- 'Message' the exchange receives, /only the payload will be sent/+-- to the calling process' mailbox.+--+-- Example:+--+-- (producer)+-- > post ex "Hello"+--+-- (consumer)+-- > bindToBroadcaster ex+-- > expect >>= liftIO . putStrLn+--+bindToBroadcaster :: Exchange -> Process ()+bindToBroadcaster ex@Exchange{..} = do+  us <- getSelfPid+  configureExchange ex $ (BindPlease, us)++--------------------------------------------------------------------------------+-- Exchage Definition/State & API Handlers                                    --+--------------------------------------------------------------------------------++apiRoute :: BroadcastEx -> Message -> Process BroadcastEx+apiRoute ex@BroadcastEx{..} msg = do+  liftIO $ atomically $ writeTChan channel msg+  forM_ (Foldable.toList _routingTable) $ routeToClient msg+  return ex+  where+    routeToClient m (PidBinding p)  = P.forward (payload m) p+    routeToClient m b@(Binding _ _) = writeToStream (outputStream b) m++-- TODO: implement unbind!!?++apiConfigure :: BroadcastEx -> P.Message -> Process BroadcastEx+apiConfigure ex msg = do+  -- for unsafe / non-serializable message passing hacks, see [note: pcopy]+  applyHandlers ex msg $ [ \m -> handleMessage m (handleBindPort ex)+                         , \m -> handleBindSTM ex m+                         , \m -> handleMessage m (handleBindPlease ex)+                         , \m -> handleMessage m (handleMonitorSignal ex)+                         , (const $ return $ Just ex)+                         ]+  where+    handleBindPlease ex' (BindPlease, p) = do+      case lookupBinding ex' p of+        Nothing -> return $ (routingTable ^: Map.insert p (PidBinding p)) ex'+        Just _  -> return ex'++    handleMonitorSignal bx (ProcessMonitorNotification _ p _) =+      return $ (routingTable ^: Map.delete p) bx++    handleBindSTM ex'@BroadcastEx{..} msg' = do+      bind' <- pUnwrap msg' :: Process (Maybe BindSTM) -- see [note: pcopy]+      case bind' of+        Nothing -> return Nothing+        Just s  -> do+          let binding = lookupBinding ex' (stmClient s)+          case binding of+            Nothing -> createBinding ex' s >>= \ex'' -> handleBindSTM ex'' msg'+            Just b  -> sendBinding (stmSend s) b >> return (Just ex')++    createBinding bEx'@BroadcastEx{..} BindSTM{..} = do+      void $ P.monitor stmClient+      nch <- liftIO $ atomically $ dupTChan channel+      let istr = newInputStream $ Right (readTChan nch)+      let ostr = NoWrite -- we write to our own channel, not the broadcast+      let bnd = Binding ostr istr+      return $ (routingTable ^: Map.insert stmClient bnd) bEx'++    sendBinding sp' bs = unsafeSendChan sp' $ pCopy (inputStream bs)++    handleBindPort :: BroadcastEx -> BindPort -> Process BroadcastEx+    handleBindPort x@BroadcastEx{..} BindPort{..} = do+      let binding = lookupBinding x portClient+      case binding of+        Just _  -> unsafeSend portClient (BindFail "DuplicateBinding") >> return x+        Nothing -> do+          let istr = Null+          let ostr = WriteChan portSend+          let bound = Binding ostr istr+          void $ P.monitor portClient+          unsafeSend portClient BindOk+          return $ (routingTable ^: Map.insert portClient bound) x++    lookupBinding BroadcastEx{..} k = Map.lookup k $ _routingTable++{- [note: pcopy]++We rely on risky techniques here, in order to allow for sharing useful+data that is not really serializable. For Cloud Haskell generally, this is+a bad idea, since we want message passing to work both locally and in a+distributed setting. In this case however, what we're really attempting is+an optimisation, since we only use unsafe PCopy based techniques when dealing+with exchange clients residing on our (local) node.++The PCopy mechanism is defined in the (aptly named) "Unsafe" module.++-}++-- TODO: move handleServerFailure into Primitives.hs++writeToStream :: OutputStream -> Message -> Process ()+writeToStream (WriteChan sp) = sendChan sp  -- see [note: safe remote send]+writeToStream (WriteSTM stm) = liftIO . atomically . stm+writeToStream NoWrite        = const $ return ()+{-# INLINE writeToStream #-}++{- [note: safe remote send]++Although we go to great lengths here to avoid serialization and/or copying+overheads, there are some activities for which we prefer to play it safe.+Chief among these is delivering messages to remote clients. Thankfully, our+unsafe @sendChan@ primitive will crash the caller/sender if there are any+encoding problems, however it is only because we /know/ for certain that+our recipient is remote, that we've chosen to write via a SendPort in the+first place! It makes sense therefore, to use the safe @sendChan@ operation+here, since for a remote call we /cannot/ avoid the overhead of serialization+anyway.++-}++handleServerFailure :: MonitorRef -> Match (InputStream Message)+handleServerFailure mRef =+  matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef)+          (\(ProcessMonitorNotification _ _ d) -> die $ ServerDisconnected d)++routingTable :: Accessor BroadcastEx BroadcastClients+routingTable = accessor _routingTable (\r e -> e { _routingTable = r })+

@@ -0,0 +1,276 @@+{-# LANGUAGE DeriveDataTypeable    #-}+{-# LANGUAGE DeriveGeneric         #-}+{-# LANGUAGE StandaloneDeriving    #-}+{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE PatternGuards         #-}+{-# LANGUAGE RecordWildCards       #-}+{-# LANGUAGE FlexibleInstances     #-}+{-# LANGUAGE EmptyDataDecls        #-}+{-# LANGUAGE TemplateHaskell       #-}+{-# LANGUAGE ImpredicativeTypes    #-}+{-# LANGUAGE UndecidableInstances  #-}+{-# LANGUAGE MultiParamTypeClasses #-}++-- | Internal Exchange Implementation+module Control.Distributed.Process.Execution.Exchange.Internal+  ( Exchange(..)+  , Message(..)+  , ExchangeType(..)+  , startExchange+  , startSupervised+  , startSupervisedRef+  , runExchange+  , post+  , postMessage+  , configureExchange+  , createMessage+  , applyHandlers+  ) where++import Control.Concurrent.MVar (MVar, takeMVar, putMVar, newEmptyMVar)+import Control.DeepSeq (NFData)+import Control.Distributed.Process+  ( Process+  , ProcessMonitorNotification(..)+  , ProcessId+  , liftIO+  , spawnLocal+  , unsafeWrapMessage+  )+import qualified Control.Distributed.Process as P (Message, link)+import Control.Distributed.Process.Serializable hiding (SerializableDict)+import Control.Distributed.Process.Extras.Internal.Types+  ( Resolvable(..)+  )+import Control.Distributed.Process.Extras.Internal.Primitives+  ( Linkable(..)+  )+import Control.Distributed.Process.ManagedProcess+  ( channelControlPort+  , handleControlChan+  , handleInfo+  , handleRaw+  , continue+  , defaultProcess+  , InitHandler+  , InitResult(..)+  , ProcessAction+  , ProcessDefinition(..)+  , ControlChannel+  , ControlPort+  )+import qualified Control.Distributed.Process.ManagedProcess as MP+  ( chanServe+  )+import Control.Distributed.Process.ManagedProcess.UnsafeClient+  ( sendControlMessage+  )+import Control.Distributed.Process.Supervisor (SupervisorPid)+import Control.Distributed.Process.Extras.Time (Delay(Infinity))+import Data.Binary+import Data.Typeable (Typeable)+import GHC.Generics+import Prelude hiding (drop)++{- [design notes]++Messages are sent to exchanges and forwarded to clients. An exchange+is parameterised by its routing mechanism, which is responsible for+maintaining its own client state and selecting the clients to which+messages are forwarded.++-}++-- | Opaque handle to an exchange.+--+data Exchange = Exchange { pid   :: !ProcessId+                         , cchan :: !(ControlPort ControlMessage)+                         , xType :: !String+                         } deriving (Typeable, Generic, Eq)+instance Binary Exchange where+instance Show Exchange where+  show Exchange{..} = (xType ++ ":" ++ (show pid))++instance Resolvable Exchange where+  resolve = return . Just . pid++{-+instance Observable Exchange MonitorRef ProcessMonitorNotification where+  observe   = P.monitor . pid+  unobserve = P.unmonitor+  observableFrom ref (ProcessMonitorNotification ref' _ r) =+    return $ if ref' == ref then Just r else Nothing+-}++instance Linkable Exchange where+  linkTo = P.link . pid++-- we communicate with exchanges using control channels+sendCtrlMsg :: Exchange -> ControlMessage -> Process ()+sendCtrlMsg Exchange{..} = sendControlMessage cchan++-- | Messages sent to an exchange can optionally provide a routing+-- key and a list of (key, value) headers in addition to the underlying+-- payload.+data Message =+  Message { key     :: !String  -- ^ a /routing key/ for the payload+          , headers :: ![(String, String)] -- ^ arbitrary key-value headers+          , payload :: !P.Message  -- ^ the underlying @Message@ payload+          } deriving (Typeable, Generic, Show)+instance Binary Message where+instance NFData Message where++data ControlMessage =+    Configure !P.Message+  | Post      !Message+    deriving (Typeable, Generic)+instance Binary ControlMessage where+instance NFData ControlMessage where++-- | Different exchange types are defined using record syntax.+-- The 'configureEx' and 'routeEx' API functions are called during the exchange+-- lifecycle when incoming traffic arrives. Configuration messages are+-- completely arbitrary types and the exchange type author is entirely+-- responsible for decoding them. Messages posted to the exchange (see the+-- 'Message' data type) are passed to the 'routeEx' API function along with the+-- exchange type's own internal state. Both API functions return a new+-- (potentially updated) state and run in the @Process@ monad.+--+data ExchangeType s =+  ExchangeType { name        :: String+               , state       :: s+               , configureEx :: s -> P.Message -> Process s+               , routeEx     :: s -> Message -> Process s+               }++--------------------------------------------------------------------------------+-- Starting/Running an Exchange                                               --+--------------------------------------------------------------------------------++-- | Starts an /exchange process/ with the given 'ExchangeType'.+startExchange :: forall s. ExchangeType s -> Process Exchange+startExchange = doStart Nothing++-- | Starts an exchange as part of a supervision tree.+--+-- Example:+-- > childSpec = toChildStart $ startSupervisedRef exType+--+startSupervisedRef :: forall s . ExchangeType s+                   -> SupervisorPid+                   -> Process (ProcessId, P.Message)+startSupervisedRef t s = do+  ex <- startSupervised t s+  return (pid ex, unsafeWrapMessage ex)++-- | Starts an exchange as part of a supervision tree.+--+-- Example:+-- > childSpec = toChildStart $ startSupervised exType+--+startSupervised :: forall s . ExchangeType s+                -> SupervisorPid+                -> Process Exchange+startSupervised t s = doStart (Just s) t++doStart :: Maybe SupervisorPid -> ExchangeType s -> Process Exchange+doStart mSp t = do+  cchan <- liftIO $ newEmptyMVar+  spawnLocal (maybeLink mSp >> runExchange t cchan) >>= \pid -> do+    cc <- liftIO $ takeMVar cchan+    return $ Exchange pid cc (name t)+  where+    maybeLink Nothing   = return ()+    maybeLink (Just p') = P.link p'++runExchange :: forall s.+               ExchangeType s+            -> MVar (ControlPort ControlMessage)+            -> Process ()+runExchange t tc = MP.chanServe t exInit (processDefinition t tc)++exInit :: forall s. InitHandler (ExchangeType s) (ExchangeType s)+exInit t = return $ InitOk t Infinity++--------------------------------------------------------------------------------+-- Client Facing API                                                          --+--------------------------------------------------------------------------------++-- | Posts an arbitrary 'Serializable' datum to an /exchange/. The raw datum is+-- wrapped in the 'Message' data type, with its 'key' set to @""@ and its+-- 'headers' to @[]@.+post :: Serializable a => Exchange -> a -> Process ()+post ex msg = postMessage ex $ Message "" [] (unsafeWrapMessage msg)++-- | Posts a 'Message' to an /exchange/.+postMessage :: Exchange -> Message -> Process ()+postMessage ex msg = msg `seq` sendCtrlMsg ex $ Post msg++-- | Sends an arbitrary 'Serializable' datum to an /exchange/, for use as a+-- configuration change - see 'configureEx' for details.+configureExchange :: Serializable m => Exchange -> m -> Process ()+configureExchange e m = sendCtrlMsg e $ Configure (unsafeWrapMessage m)++-- | Utility for creating a 'Message' datum from its 'key', 'headers' and+-- 'payload'.+createMessage :: Serializable m => String -> [(String, String)] -> m -> Message+createMessage k h m = Message k h $ unsafeWrapMessage m++-- | Utility for custom exchange type authors - evaluates a set of primitive+-- message handlers from left to right, returning the first which evaluates+-- to @Just a@, or the initial @e@ value if all the handlers yield @Nothing@.+applyHandlers :: a+              -> P.Message+              -> [P.Message -> Process (Maybe a)]+              -> Process a+applyHandlers e _ []     = return e+applyHandlers e m (f:fs) = do+  r <- f m+  case r of+    Nothing -> applyHandlers e m fs+    Just r' -> return r'++--------------------------------------------------------------------------------+-- Process Definition/State & API Handlers                                    --+--------------------------------------------------------------------------------++processDefinition :: forall s.+                     ExchangeType s+                  -> MVar (ControlPort ControlMessage)+                  -> ControlChannel ControlMessage+                  -> Process (ProcessDefinition (ExchangeType s))+processDefinition _ tc cc = do+  liftIO $ putMVar tc $ channelControlPort cc+  return $+    defaultProcess {+        apiHandlers  = [ handleControlChan cc handleControlMessage ]+      , infoHandlers = [ handleInfo handleMonitor+                       , handleRaw convertToCC+                       ]+      } :: Process (ProcessDefinition (ExchangeType s))++handleMonitor :: forall s.+                 ExchangeType s+              -> ProcessMonitorNotification+              -> Process (ProcessAction (ExchangeType s))+handleMonitor ex m = do+  handleControlMessage ex (Configure (unsafeWrapMessage m))++convertToCC :: forall s.+               ExchangeType s+            -> P.Message+            -> Process (ProcessAction (ExchangeType s))+convertToCC ex msg = do+  liftIO $ putStrLn "convert to cc"+  handleControlMessage ex (Post $ Message "" [] msg)++handleControlMessage :: forall s.+                        ExchangeType s+                     -> ControlMessage+                     -> Process (ProcessAction (ExchangeType s))+handleControlMessage ex@ExchangeType{..} cm =+  let action = case cm of+                 Configure msg -> configureEx state msg+                 Post      msg -> routeEx     state msg+  in action >>= \s -> continue $ ex { state = s }+

@@ -0,0 +1,250 @@+{-# LANGUAGE DeriveDataTypeable    #-}+{-# LANGUAGE DeriveGeneric         #-}+{-# LANGUAGE StandaloneDeriving    #-}+{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE PatternGuards         #-}+{-# LANGUAGE RecordWildCards       #-}+{-# LANGUAGE FlexibleInstances     #-}+{-# LANGUAGE EmptyDataDecls        #-}+{-# LANGUAGE TemplateHaskell       #-}+{-# LANGUAGE ImpredicativeTypes    #-}+{-# LANGUAGE UndecidableInstances  #-}+{-# LANGUAGE MultiParamTypeClasses #-}++-- | A simple API for /routing/, using a custom exchange type.+module Control.Distributed.Process.Execution.Exchange.Router+  ( -- * Types+    HeaderName+  , Binding(..)+  , Bindable+  , BindingSelector+  , RelayType(..)+    -- * Starting a Router+  , router+  , supervisedRouter+  , supervisedRouterRef+    -- * Client (Publishing) API+  , route+  , routeMessage+    -- * Routing via message/binding keys+  , messageKeyRouter+  , bindKey+    -- * Routing via message headers+  , headerContentRouter+  , bindHeader+  ) where++import Control.DeepSeq (NFData)+import Control.Distributed.Process+  ( Process+  , ProcessMonitorNotification(..)+  , ProcessId+  , monitor+  , handleMessage+  , unsafeWrapMessage+  )+import qualified Control.Distributed.Process as P+import Control.Distributed.Process.Serializable (Serializable)+import Control.Distributed.Process.Execution.Exchange.Internal+  ( startExchange+  , startSupervised+  , configureExchange+  , Message(..)+  , Exchange+  , ExchangeType(..)+  , post+  , postMessage+  , applyHandlers+  )+import Control.Distributed.Process.Extras.Internal.Primitives+  ( deliver+  , Resolvable(..)+  )+import Control.Distributed.Process.Supervisor (SupervisorPid)+import Data.Binary+import Data.Foldable (forM_)+import Data.Hashable+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as Map+import Data.HashSet (HashSet)+import qualified Data.HashSet as Set+import Data.Typeable (Typeable)+import GHC.Generics++type HeaderName = String++-- | The binding key used by the built-in key and header based+-- routers.+data Binding =+    BindKey    { bindingKey :: !String }+  | BindHeader { bindingKey :: !String+               , headerName :: !HeaderName+               }+  | BindNone+  deriving (Typeable, Generic, Eq, Show)+instance Binary Binding where+instance NFData Binding where+instance Hashable Binding where++-- | Things that can be used as binding keys in a router.+class (Hashable k, Eq k, Serializable k) => Bindable k+instance (Hashable k, Eq k, Serializable k) => Bindable k++-- | Used to convert a 'Message' into a 'Bindable' routing key.+type BindingSelector k = (Message -> Process k)++-- | Given to a /router/ to indicate whether clients should+-- receive 'Message' payloads only, or the whole 'Message' object+-- itself.+data RelayType = PayloadOnly | WholeMessage++data State k = State { bindings  :: !(HashMap k (HashSet ProcessId))+                     , selector  :: !(BindingSelector k)+                     , relayType :: !RelayType+                     }++type Router k = ExchangeType (State k)++--------------------------------------------------------------------------------+-- Starting/Running the Exchange                                              --+--------------------------------------------------------------------------------++-- | A router that matches on a 'Message' 'key'. To bind a client @Process@ to+-- such an exchange, use the 'bindKey' function.+messageKeyRouter :: RelayType -> Process Exchange+messageKeyRouter t = router t matchOnKey -- (return . BindKey . key)+  where+    matchOnKey :: Message -> Process Binding+    matchOnKey m = return $ BindKey (key m)++-- | A router that matches on a specific (named) header. To bind a client+-- @Process@ to such an exchange, use the 'bindHeader' function.+headerContentRouter :: RelayType -> HeaderName -> Process Exchange+headerContentRouter t n = router t (checkHeaders n)+  where+    checkHeaders hn Message{..} = do+      case Map.lookup hn (Map.fromList headers) of+        Nothing -> return BindNone+        Just hv -> return $ BindHeader hn hv++-- | Defines a /router/ exchange. The 'BindingSelector' is used to construct+-- a binding (i.e., an instance of the 'Bindable' type @k@) for each incoming+-- 'Message'. Such bindings are matched against bindings stored in the exchange.+-- Clients of a /router/ exchange are identified by a binding, mapped to+-- one or more 'ProcessId's.+--+-- The format of the bindings, nature of their storage and mechanism for+-- submitting new bindings is implementation dependent (i.e., will vary by+-- exchange type). For example, the 'messageKeyRouter' and 'headerContentRouter'+-- implementations both use the 'Binding' data type, which can represent a+-- 'Message' key or a 'HeaderName' and content. As with all custom exchange+-- types, bindings should be submitted by evaluating 'configureExchange' with+-- a suitable data type.+--+router :: (Bindable k) => RelayType -> BindingSelector k -> Process Exchange+router t s = routerT t s >>= startExchange++supervisedRouterRef :: Bindable k+                    => RelayType+                    -> BindingSelector k+                    -> SupervisorPid+                    -> Process (ProcessId, P.Message)+supervisedRouterRef t sel spid = do+  ex <- supervisedRouter t sel spid+  Just pid <- resolve ex+  return (pid, unsafeWrapMessage ex)++-- | Defines a /router/ that can be used in a supervision tree.+supervisedRouter :: Bindable k+                 => RelayType+                 -> BindingSelector k+                 -> SupervisorPid+                 -> Process Exchange+supervisedRouter t sel spid =+  routerT t sel >>= \t' -> startSupervised t' spid++routerT :: Bindable k+        => RelayType+        -> BindingSelector k+        -> Process (Router k)+routerT t s = do+  return $ ExchangeType { name        = "Router"+                        , state       = State Map.empty s t+                        , configureEx = apiConfigure+                        , routeEx     = apiRoute+                        }++--------------------------------------------------------------------------------+-- Client Facing API                                                          --+--------------------------------------------------------------------------------++-- | Add a binding (for the calling process) to a 'messageKeyRouter' exchange.+bindKey :: String -> Exchange -> Process ()+bindKey k ex = do+  self <- P.getSelfPid+  configureExchange ex (self, BindKey k)++-- | Add a binding (for the calling process) to a 'headerContentRouter' exchange.+bindHeader :: HeaderName -> String -> Exchange -> Process ()+bindHeader n v ex = do+  self <- P.getSelfPid+  configureExchange ex (self, BindHeader v n)++-- | Send a 'Serializable' message to the supplied 'Exchange'. The given datum+-- will be converted to a 'Message', with the 'key' set to @""@ and the+-- 'headers' to @[]@.+--+-- The routing behaviour will be dependent on the choice of 'BindingSelector'+-- given when initialising the /router/.+route :: Serializable m => Exchange -> m -> Process ()+route = post++-- | Send a 'Message' to the supplied 'Exchange'.+-- The routing behaviour will be dependent on the choice of 'BindingSelector'+-- given when initialising the /router/.+routeMessage :: Exchange -> Message -> Process ()+routeMessage = postMessage++--------------------------------------------------------------------------------+-- Exchage Definition/State & API Handlers                                    --+--------------------------------------------------------------------------------++apiRoute :: forall k. Bindable k+         => State k+         -> Message+         -> Process (State k)+apiRoute st@State{..} msg = do+  binding <- selector msg+  case Map.lookup binding bindings of+    Nothing -> return st+    Just bs -> forM_ bs (fwd relayType msg) >> return st+  where+    fwd WholeMessage m = deliver m+    fwd PayloadOnly  m = P.forward (payload m)++-- TODO: implement 'unbind' ???+-- TODO: apiConfigure currently leaks memory if clients die (we don't cleanup)++apiConfigure :: forall k. Bindable k+             => State k+             -> P.Message+             -> Process (State k)+apiConfigure st msg = do+  applyHandlers st msg $ [ \m -> handleMessage m (createBinding st)+                         , \m -> handleMessage m (handleMonitorSignal st)+                         ]+  where+    createBinding s@State{..} (pid, bind) = do+      case Map.lookup bind bindings of+        Nothing -> do _ <- monitor pid+                      return $ s { bindings = newBind bind pid bindings }+        Just ps -> return $ s { bindings = addBind bind pid bindings ps }++    newBind b p bs = Map.insert b (Set.singleton p) bs+    addBind b' p' bs' ps = Map.insert b' (Set.insert p' ps) bs'++    handleMonitorSignal s@State{..} (ProcessMonitorNotification _ p _) =+      let bs  = bindings+          bs' = Map.foldlWithKey' (\a k v -> Map.insert k (Set.delete p v) a) bs bs+      in return $ s { bindings = bs' }+

@@ -0,0 +1,744 @@+{-# LANGUAGE DeriveDataTypeable   #-}+{-# LANGUAGE DeriveGeneric        #-}+{-# LANGUAGE StandaloneDeriving   #-}+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE PatternGuards        #-}+{-# LANGUAGE RecordWildCards      #-}+{-# LANGUAGE FlexibleInstances    #-}+{-# LANGUAGE EmptyDataDecls       #-}+{-# LANGUAGE TemplateHaskell      #-}+{-# LANGUAGE ImpredicativeTypes   #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Execution.Mailbox+-- Copyright   :  (c) Tim Watson 2012 - 2013+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- Generic process that acts as an external mailbox and message buffer.+--+-- [Overview]+--+-- For use when rate limiting is not possible (or desired), this module+-- provides a /buffer process/ that receives mail via its 'post' API, buffers+-- the received messages and delivers them when its /owning process/ asks for+-- them. A mailbox has to be started with a maximum buffer size - the so called+-- /limit/ - and will discard messages once its internal storage reaches this+-- user defined threshold.+--+-- The usual behaviour of the /buffer process/ is to accumulate messages in+-- its internal memory. When a client evaluates 'notify', the buffer will+-- send a 'NewMail' message to the (real) mailbox of its owning process as+-- soon as it has any message(s) ready to deliver. If the buffer already+-- contains undelivered mail, the 'NewMail' message will be dispatched+-- immediately.+--+-- When the owning process wishes to receive mail, evaluating 'deliver' (from+-- any process) will cause the buffer to send its owner a 'Delivery' message+-- containing the accumulated messages and additional information about the+-- number of messages it is delivering, the number of messages dropped since+-- the last delivery and a handle for the mailbox (so that processes can have+-- multiple mailboxes if required, and distinguish between them).+--+-- [Overflow Handling]+--+-- A mailbox handles overflow - when the number of messages it is holding+-- reaches the limit - differently depending on the 'BufferType' selected+-- when it starts. The @Queue@ buffer will, once the limit is reached, drop+-- older messages first (i.e., the head of the queue) to make space for+-- newer ones. The @Ring@ buffer works similarly, but blocks new messages+-- so as to preserve existing ones instead. Finally, the @Stack@ buffer will+-- drop the last (i.e., most recently received) message to make room for new+-- mail.+--+-- Mailboxes can be /resized/ by evaluating 'resize' with a new value for the+-- limit. If the new limit is older that the current/previous one, messages+-- are dropped as though the mailbox had previously seen a volume of mail+-- equal to the difference (in size) between the limits. In this situation,+-- the @Queue@ will drop as many older messages as neccessary to come within+-- the limit, whilst the other two buffer types will drop as many newer messages+-- as needed.+--+-- [Ordering Guarantees]+--+-- When messages are delivered to the owner, they arrive as a list of raw+-- @Message@ entries, given in descending age order (i.e., eldest first).+-- Whilst this approximates the FIFO ordering a process' mailbox would usually+-- offer, the @Stack@ buffer will appear to offer no ordering at all, since+-- it always deletes the most recent message(s). The @Queue@ and @Ring@ buffers+-- will maintain a more queue-like (i.e., FIFO) view of received messages,+-- with the obvious constraint the newer or older data might have been deleted.+--+-- [Post API and Relaying]+--+-- For messages to be properly handled by the mailbox, they can either be sent+-- via the 'post' API or directly to the 'Mailbox'. Messages sent directly to+-- the mailbox will still be handled via the internal buffers and subjected to+-- the mailbox limits. The 'post' API is really just a means to ensure that+-- the conversion from @Serializable a -> Message@ is done in the caller's+-- process and uses the safe @wrapMessage@ variant.+--+-- [Acknowledgements]+--+-- This API is based on the work of Erlang programmers Fred Hebert and+-- Geoff Cant, its design closely mirroring that of the the /pobox/ library+-- application.+--+-----------------------------------------------------------------------------+module Control.Distributed.Process.Execution.Mailbox+  (+    -- * Creating, Starting, Configuring and Running a Mailbox+    Mailbox()+  , startMailbox+  , startSupervised+  , startSupervisedMailbox+  , createMailbox+  , resize+  , statistics+  , monitor+  , Limit+  , BufferType(..)+  , MailboxStats(..)+    -- * Posting Mail+  , post+    -- * Obtaining Mail and Notifications+  , notify+  , deliver+  , active+  , NewMail(..)+  , Delivery(..)+  , FilterResult(..)+  , acceptEverything+  , acceptMatching+    -- * Remote Table+  , __remoteTable+  ) where++import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TChan+  ( TChan+  , newBroadcastTChanIO+  , dupTChan+  , readTChan+  , writeTChan+  )+import Control.Distributed.Process hiding (call, monitor)+import qualified Control.Distributed.Process as P (monitor)+import Control.Distributed.Process.Closure+  ( remotable+  , mkStaticClosure+  )+import Control.Distributed.Process.Serializable hiding (SerializableDict)+import Control.Distributed.Process.Extras.Internal.Types+  ( ExitReason(..)+  , Resolvable(..)+  , Routable(..)+  , Linkable(..)+  )+import Control.Distributed.Process.ManagedProcess+  ( call+  , sendControlMessage+  , channelControlPort+  , handleControlChan+  , handleInfo+  , handleRaw+  , continue+  , defaultProcess+  , UnhandledMessagePolicy(..)+  , InitHandler+  , InitResult(..)+  , ProcessAction+  , ProcessDefinition(..)+  , ControlChannel+  , ControlPort+  )+import qualified Control.Distributed.Process.ManagedProcess as MP+  ( chanServe+  )+import Control.Distributed.Process.ManagedProcess.Server+  ( stop+  )+import Control.Distributed.Process.ManagedProcess.Server.Restricted as Restricted+  ( getState+  , Result+  , RestrictedProcess+  )+import qualified Control.Distributed.Process.ManagedProcess.Server.Restricted as Restricted+  ( handleCall+  , reply+  )+import Control.Distributed.Process.Supervisor (SupervisorPid)+import Control.Distributed.Process.Extras.Time+import Control.Exception (SomeException)+import Data.Accessor+  ( Accessor+  , accessor+  , (^:)+  , (.>)+  , (^=)+  , (^.)+  )+import Data.Binary+import qualified Data.Foldable as Foldable+import Data.Sequence+  ( Seq+  , ViewL(EmptyL, (:<))+  , ViewR(EmptyR, (:>))+  , (<|)+  , (|>)+  )+import qualified Data.Sequence as Seq+import Data.Typeable (Typeable)++import GHC.Generics++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, drop)+#else+import Prelude hiding (drop)+#endif++--------------------------------------------------------------------------------+-- Types                                                                      --+--------------------------------------------------------------------------------++-- external client/configuration API++-- | Opaque handle to a mailbox.+--+data Mailbox = Mailbox { pid   :: !ProcessId+                       , cchan :: !(ControlPort ControlMessage)+                       } deriving (Typeable, Generic, Eq)+instance Binary Mailbox where+instance Show Mailbox where+  show = ("Mailbox:" ++) . show . pid++instance Linkable Mailbox where+  linkTo = link . pid++instance Resolvable Mailbox where+  resolve = return . Just . pid++instance Routable Mailbox where+  sendTo       = post+  unsafeSendTo = post++sendCtrlMsg :: Mailbox+            -> ControlMessage+            -> Process ()+sendCtrlMsg Mailbox{..} = sendControlMessage cchan++-- | Describes the different types of buffer.+--+data BufferType =+    Queue -- ^ FIFO buffer, limiter drops the eldest message (queue head)+  | Stack -- ^ unordered buffer, limiter drops the newest (top) message+  | Ring  -- ^ FIFO buffer, limiter refuses (i.e., drops) new messages+  deriving (Typeable, Eq, Show)++-- TODO: re-implement this process in terms of a limiter expression, i.e.,+--+-- data Limit s = Accept s | Block s+--+-- limit :: forall s. Closure (Message {- new mail -} -> Process (Limit s))++-- | Represents the maximum number of messages the internal buffer can hold.+--+type Limit = Integer++-- | A @Closure@ used to filter messages in /active/ mode.+--+type Filter = Closure (Message -> Process FilterResult)++-- | Marker message indicating to the owning process that mail has arrived.+--+data NewMail = NewMail !Mailbox !Integer+  deriving (Typeable, Generic, Show)+instance Binary NewMail where++-- | Mail delivery.+--+data Delivery = Delivery { box          :: Mailbox -- ^ handle to the sending mailbox+                         , messages     :: [Message] -- ^ list of raw messages+                         , count        :: Integer -- ^ number of messages delivered+                         , totalDropped :: Integer -- ^ total dropped/skipped messages+                         }+  deriving (Typeable, Generic)+instance Binary Delivery where++-- TODO: keep running totals and send them with the stats...++-- | Bundle of statistics data, available on request via+-- the 'mailboxStats' API call.+--+data MailboxStats =+  MailboxStats { pendingMessages :: Integer+               , droppedMessages :: Integer+               , currentLimit    :: Limit+               , owningProcess   :: ProcessId+               } deriving (Typeable, Generic, Show)+instance Binary MailboxStats where++-- internal APIs++data Post = Post !Message+  deriving (Typeable, Generic)+instance Binary Post where++data StatsReq = StatsReq+  deriving (Typeable, Generic)+instance Binary StatsReq where++data FilterResult = Keep | Skip | Send+  deriving (Typeable, Generic)+instance Binary FilterResult++data Mode =+    Active !Filter -- ^ Send all buffered messages (or wait until one arrives)+  | Notify  -- ^ Send a notification once messages are ready to be received+  | Passive -- ^ Accumulate messages in the buffer, dropping them if necessary+  deriving (Typeable, Generic)+instance Binary Mode where+instance Show Mode where+  show (Active _) = "Active"+  show Notify     = "Notify"+  show Passive    = "Passive"++data ControlMessage =+    Resize !Integer+  | SetActiveMode !Mode+  deriving (Typeable, Generic)+instance Binary ControlMessage where++class Buffered a where+  tag    :: a -> BufferType+  push   :: Message -> a -> a+  pop    :: a -> Maybe (Message, a)+  adjust :: Limit -> a -> a+  drop   :: Integer -> a -> a++data BufferState =+  BufferState { _mode    :: Mode+              , _bufferT :: BufferType+              , _limit   :: Limit+              , _size    :: Integer+              , _dropped :: Integer+              , _owner   :: ProcessId+              , ctrlChan :: ControlPort ControlMessage+              }++defaultState :: BufferType+             -> Limit+             -> ProcessId+             -> ControlPort ControlMessage+             -> BufferState+defaultState bufferT limit' pid cc =+  BufferState { _mode    = Passive+              , _bufferT = bufferT+              , _limit   = limit'+              , _size    = 0+              , _dropped = 0+              , _owner   = pid+              , ctrlChan = cc+              }++data State = State { _buffer :: Seq Message+                   , _state  :: BufferState+                   }++instance Buffered State where+  tag q  = _bufferT $ _state q++  -- see note [buffer enqueue/dequeue semantics]+  push m = (state .> size ^: (+1)) . (buffer ^: (m <|))++  -- see note [buffer enqueue/dequeue semantics]+  pop q = maybe Nothing+                (\(s' :> a) -> Just (a, ( (buffer ^= s')+                                        . (state .> size ^: (1-))+                                        $ q))) $ getR (q ^. buffer)++  adjust sz q = (state .> limit ^= sz) $ maybeDrop+    where+      maybeDrop+        | size' <- (q ^. state ^. size),+          size' > sz = (state .> size ^= sz) $ drop (size' - sz) q+        | otherwise  = q++  -- see note [buffer drop semantics]+  drop n q+    | n > 1     = drop (n - 1) $ drop 1 q+    | isQueue q = dropR q+    | otherwise = dropL q+    where+      dropR q' = maybe q' (\(s' :> _) -> dropOne q' s') $ getR (q' ^. buffer)+      dropL q' = maybe q' (\(_ :< s') -> dropOne q' s') $ getL (q' ^. buffer)+      dropOne q' s = ( (buffer ^= s)+                     . (state .> size ^: (\n' -> n' - 1))+                     . (state .> dropped ^: (+1))+                     $ q' )++{- note [buffer enqueue/dequeue semantics]+If we choose to add a message to the buffer, it is always+added to the left hand side of the sequence. This gives+FIFO (enqueue to tail) semantics for queues, LIFO (push+new head) semantics for stacks when dropping messages - note+that dequeueing will always take the eldest (RHS) message,+regardless of the buffer type - and queue-like semantics for+the ring buffer.++We /always/ take the eldest message each time we dequeue,+in an attempt to maintain something approaching FIFO order+when processing the mailbox, for all data structures. Where+we do not achieve this is dropping messages, since the different+buffer types drop messages either on the right (eldest) or left+(youngest).++-- note [buffer drop semantics]++The "stack buffer", when full, only ever attempts to drop the+youngest (leftmost) message, such that it guarantees no ordering+at all, but that is enforced by the code calling 'drop' rather+than the data structure itself. The ring buffer behaves similarly,+since it rejects new messages altogether, which in practise means+dropping from the LHS.++-}++--------------------------------------------------------------------------------+-- Starting/Running a Mailbox                                                 --+--------------------------------------------------------------------------------++-- | Start a mailbox for the calling process.+--+-- > create = getSelfPid >>= start+--+createMailbox :: BufferType -> Limit -> Process Mailbox+createMailbox buffT maxSz =+  getSelfPid >>= \self -> startMailbox self buffT maxSz++-- | Start a mailbox for the supplied @ProcessId@.+--+-- > start = spawnLocal $ run+--+startMailbox :: ProcessId -> BufferType -> Limit -> Process Mailbox+startMailbox = doStartMailbox Nothing++-- | As 'startMailbox', but suitable for use in supervisor child specs.+-- This variant is for use when you want to access to the underlying+-- 'Mailbox' handle in your supervised child refs. See supervisor's+-- @ChildRef@ data type for more information.+--+-- Example:+-- > childSpec = toChildStart $ startSupervised pid bufferType mboxLimit+--+-- See "Control.Distributed.Process.Supervisor"+--+startSupervised :: ProcessId+                -> BufferType+                -> Limit+                -> SupervisorPid+                -> Process (ProcessId, Message)+startSupervised p b l s = do+  mb <- startSupervisedMailbox p b l s+  return (pid mb, unsafeWrapMessage mb)++-- | As 'startMailbox', but suitable for use in supervisor child specs.+--+-- Example:+-- > childSpec = toChildStart $ startSupervisedMailbox pid bufferType mboxLimit+--+-- See "Control.Distributed.Process.Supervisor"+--+startSupervisedMailbox :: ProcessId+                       -> BufferType+                       -> Limit+                       -> SupervisorPid+                       -> Process Mailbox+startSupervisedMailbox p b l s = doStartMailbox (Just s) p b l++doStartMailbox :: Maybe SupervisorPid+               -> ProcessId+               -> BufferType+               -> Limit+               -> Process Mailbox+doStartMailbox mSp p b l = do+  bchan <- liftIO $ newBroadcastTChanIO+  rchan <- liftIO $ atomically $ dupTChan bchan+  spawnLocal (maybeLink mSp >> runMailbox bchan p b l) >>= \pid -> do+    cc <- liftIO $ atomically $ readTChan rchan+    return $ Mailbox pid cc+  where+    maybeLink Nothing   = return ()+    maybeLink (Just p') = link p'++-- | Run the mailbox server loop.+--+runMailbox :: TChan (ControlPort ControlMessage)+           -> ProcessId+           -> BufferType+           -> Limit+           -> Process ()+runMailbox tc pid buffT maxSz = do+  link pid+  tc' <- liftIO $ atomically $ dupTChan tc+  MP.chanServe (pid, buffT, maxSz) (mboxInit tc') (processDefinition pid tc)++--------------------------------------------------------------------------------+-- Mailbox Initialisation/Startup                                             --+--------------------------------------------------------------------------------++mboxInit :: TChan (ControlPort ControlMessage)+         -> InitHandler (ProcessId, BufferType, Limit) State+mboxInit tc (pid, buffT, maxSz) = do+  cc <- liftIO $ atomically $ readTChan tc+  return $ InitOk (State Seq.empty $ defaultState buffT maxSz pid cc) Infinity++--------------------------------------------------------------------------------+-- Client Facing API                                                          --+--------------------------------------------------------------------------------++-- | Monitor a mailbox.+--+monitor :: Mailbox -> Process MonitorRef+monitor = P.monitor . pid++-- | Instructs the mailbox to send a 'NewMail' signal as soon as any mail is+-- available for delivery. Once the signal is sent, it will not be resent, even+-- when further mail arrives, until 'notify' is called again.+--+-- NB: signals are /only/ delivered to the mailbox's owning process.+--+notify :: Mailbox -> Process ()+notify mb = sendCtrlMsg mb $ SetActiveMode Notify++-- | Instructs the mailbox to send a 'Delivery' as soon as any mail is+-- available, or immediately (if the buffer already contains data).+--+-- NB: signals are /only/ delivered to the mailbox's owning process.+--+active :: Mailbox -> Filter -> Process ()+active mb f = sendCtrlMsg mb $ SetActiveMode $ Active f++-- | Alters the mailbox's /limit/ - this might cause messages to be dropped!+--+resize :: Mailbox -> Integer -> Process ()+resize mb sz = sendCtrlMsg mb $ Resize sz++-- | Posts a message to someone's mailbox.+--+post :: Serializable a => Mailbox -> a -> Process ()+post Mailbox{..} m = send pid (Post $ wrapMessage m)++-- | Obtain statistics (from/to anywhere) about a mailbox.+--+statistics :: Mailbox -> Process MailboxStats+statistics mb = call mb StatsReq++--------------------------------------------------------------------------------+-- PRIVATE Filter Implementation(s)                                           --+--------------------------------------------------------------------------------++everything :: Message -> Process FilterResult+everything _ = return Keep++matching :: Closure (Message -> Process FilterResult)+         -> Message+         -> Process FilterResult+matching predicate msg = do+  pred' <- unClosure predicate :: Process (Message -> Process FilterResult)+  res   <- handleMessage msg pred'+  case res of+    Nothing -> return Skip+    Just fr -> return fr++--------------------------------------------------------------------------------+-- Process Definition/State & API Handlers                                    --+--------------------------------------------------------------------------------++processDefinition :: ProcessId+                  -> TChan (ControlPort ControlMessage)+                  -> ControlChannel ControlMessage+                  -> Process (ProcessDefinition State)+processDefinition pid tc cc = do+  liftIO $ atomically $ writeTChan tc $ channelControlPort cc+  return $ defaultProcess { apiHandlers = [+                               handleControlChan     cc handleControlMessages+                             , Restricted.handleCall handleGetStats+                             ]+                          , infoHandlers = [ handleInfo handlePost+                                           , handleRaw  handleRawInputs ]+                          , unhandledMessagePolicy = DeadLetter pid+                          } :: Process (ProcessDefinition State)++handleControlMessages :: State+                      -> ControlMessage+                      -> Process (ProcessAction State)+handleControlMessages st cm+  | (SetActiveMode new) <- cm = activateMode st new+  | (Resize sz')        <- cm = continue $ adjust sz' st+  | otherwise                 = stop $ ExitOther "IllegalState"+  where+    activateMode :: State -> Mode -> Process (ProcessAction State)+    activateMode st' new+      | sz <- (st ^. state ^. size)+      , sz == 0           = continue $ updated st' new+      | otherwise         = do+          let updated' = updated st' new+          case new of+            Notify     -> sendNotification updated' >> continue updated'+            (Active _) -> sendMail updated' >>= continue+            Passive    -> {- shouldn't happen! -} die $ "IllegalState"++    updated s m = (state .> mode ^= m) s++handleGetStats :: StatsReq -> RestrictedProcess State (Result MailboxStats)+handleGetStats _ = Restricted.reply . (^. stats) =<< getState++handleRawInputs :: State -> Message -> Process (ProcessAction State)+handleRawInputs st msg = handlePost st (Post msg)++handlePost :: State -> Post -> Process (ProcessAction State)+handlePost st (Post msg) = do+  let st' = insert msg st+  continue . (state .> mode ^= Passive) =<< forwardIfNecessary st'+  where+    forwardIfNecessary s+      | Notify   <- currentMode = sendNotification s >> return s+      | Active _ <- currentMode = sendMail s+      | otherwise               = return s++    currentMode = st ^. state ^. mode++--------------------------------------------------------------------------------+-- Accessors, State/Stats Management & Utilities                              --+--------------------------------------------------------------------------------++sendNotification :: State -> Process ()+sendNotification st = do+    pid <- getSelfPid+    send ownerPid $ NewMail (Mailbox pid cchan) pending+  where+    ownerPid = st ^. state ^. owner+    pending  = st ^. state ^. size+    cchan    = ctrlChan (st ^. state)++type Count = Integer+type Skipped = Integer++sendMail :: State -> Process State+sendMail st = do+    let Active f = st ^. state ^. mode+    unCl <- catch (unClosure f >>= return . Just)+                  (\(_ :: SomeException) -> return Nothing)+    case unCl of+      Nothing -> return st -- TODO: Logging!?+      Just f' -> do+        (st', cnt, skipped, msgs) <- applyFilter f' st+        us <- getSelfPid+        send ownerPid $ Delivery { box          = Mailbox us (ctrlChan $ st ^. state)+                                 , messages     = Foldable.toList msgs+                                 , count        = cnt+                                 , totalDropped = skipped + droppedMsgs+                                 }+        return $ ( (state .> dropped ^= 0)+                 . (state .> size ^: ((cnt + skipped) -))+                 $ st' )+  where+    applyFilter f s = filterMessages f (s, 0, 0, Seq.empty)++    filterMessages :: (Message -> Process FilterResult)+                   -> (State, Count, Skipped, Seq Message)+                   -> Process (State, Count, Skipped, Seq Message)+    filterMessages f accIn@(buff, cnt, drp, acc) = do+      case pop buff of+        Nothing         -> return accIn+        Just (m, buff') -> do+          res <- f m+          case res of+            Keep -> filterMessages f (buff', cnt + 1, drp, acc |> m)+            Skip -> filterMessages f (buff', cnt, drp + 1, acc)+            Send -> return accIn++    ownerPid    = st ^. state ^. owner+    droppedMsgs = st ^. state ^. dropped++insert :: Message -> State -> State+insert msg st@(State _ BufferState{..}) =+  if _size /= _limit+     then push msg st+     else case _bufferT of+            Ring -> (state .> dropped ^: (+1)) st+            _    -> push msg $ drop 1 st++isQueue :: State -> Bool+isQueue = (== Queue) . _bufferT . _state++isStack :: State -> Bool+isStack = (== Stack) . _bufferT . _state++getR :: Seq a -> Maybe (ViewR a)+getR s =+  case Seq.viewr s of+    EmptyR -> Nothing+    a      -> Just a++getL :: Seq a -> Maybe (ViewL a)+getL s =+  case Seq.viewl s of+    EmptyL -> Nothing+    a      -> Just a++mode :: Accessor BufferState Mode+mode = accessor _mode (\m st -> st { _mode = m })++bufferType :: Accessor BufferState BufferType+bufferType = accessor _bufferT (\t st -> st { _bufferT = t })++limit :: Accessor BufferState Limit+limit = accessor _limit (\l st -> st { _limit = l })++size :: Accessor BufferState Integer+size = accessor _size (\s st -> st { _size = s })++dropped :: Accessor BufferState Integer+dropped = accessor _dropped (\d st -> st { _dropped = d })++owner :: Accessor BufferState ProcessId+owner = accessor _owner (\o st -> st { _owner = o })++buffer :: Accessor State (Seq Message)+buffer = accessor _buffer (\b qb -> qb { _buffer = b })++state :: Accessor State BufferState+state = accessor _state (\s qb -> qb { _state = s })++stats :: Accessor State MailboxStats+stats = accessor getStats (\_ s -> s) -- TODO: use a READ ONLY accessor for this+  where+    getStats (State _ (BufferState _ _ lm sz dr op _)) = MailboxStats sz dr lm op++$(remotable ['everything, 'matching])++-- | A /do-nothing/ filter that accepts all messages (i.e., returns @Keep@+-- for any input).+acceptEverything :: Closure (Message -> Process FilterResult)+acceptEverything = $(mkStaticClosure 'everything)++-- | A filter that takes a @Closure (Message -> Process FilterResult)@ holding+-- the filter function and applies it remotely (i.e., in the mailbox's own+-- managed process).+--+acceptMatching :: Closure (Closure (Message -> Process FilterResult)+                           -> Message -> Process FilterResult)+acceptMatching = $(mkStaticClosure 'matching)++-- | Instructs the mailbox to deliver all pending messages to the owner.+--+deliver :: Mailbox -> Process ()+deliver mb = active mb acceptEverything+

@@ -0,0 +1,43 @@+{-# LANGUAGE TemplateHaskell     #-}+{-# LANGUAGE ScopedTypeVariables #-}++module MailboxTestFilters where++import Control.Distributed.Process+import Control.Distributed.Process.Execution.Mailbox (FilterResult(..))+import Control.Monad (forM)++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, drop)+#else+import Prelude hiding (drop)+#endif+import Data.Maybe (catMaybes)+import Control.Distributed.Process.Closure (remotable, mkClosure, mkStaticClosure)++filterInputs :: (String, Int, Bool) -> Message -> Process FilterResult+filterInputs (s, i, b) msg = do+  rs <- forM [ \m -> handleMessageIf m (\s' -> s' == s) (\_ -> return Keep)+             , \m -> handleMessageIf m (\i' -> i' == i) (\_ -> return Keep)+             , \m -> handleMessageIf m (\b' -> b' == b) (\_ -> return Keep)+             ] $ \h -> h msg+  if (length (catMaybes rs) > 0)+    then return Keep+    else return Skip++filterEvens :: Message -> Process FilterResult+filterEvens m = do+  matched <- handleMessage m (\(i :: Int) -> do+                               if even i then return Keep else return Skip)+  case matched of+    Just fr -> return fr+    _       -> return Skip++$(remotable ['filterInputs, 'filterEvens])++intFilter :: Closure (Message -> Process FilterResult)+intFilter = $(mkStaticClosure 'filterEvens)++myFilter :: (String, Int, Bool) -> Closure (Message -> Process FilterResult)+myFilter = $(mkClosure 'filterInputs)+

@@ -0,0 +1,186 @@+{-# LANGUAGE BangPatterns          #-}+{-# LANGUAGE TemplateHaskell       #-}+{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module Main where++import Control.Distributed.Process hiding (monitor)+import Control.Distributed.Process.Node+import Control.Distributed.Process.Execution.EventManager hiding (start)+import qualified Control.Distributed.Process.Extras+import Control.Distributed.Process.Execution.Exchange+import Control.Distributed.Process.Extras.Internal.Types+import Control.Distributed.Process.Extras.Internal.Primitives+import qualified Control.Distributed.Process.Execution.EventManager as EventManager+  ( start+  )+import Control.Monad (void, forM, forever)+import Control.Rematch (equalTo)++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, drop)+#else+import Prelude hiding (drop)+#endif+import qualified Network.Transport as NT+import Test.Framework as TF (testGroup, Test)+import Test.Framework.Providers.HUnit+import TestUtils++testKeyBasedRouting :: TestResult Bool -> Process ()+testKeyBasedRouting result = do+  (sp, rp) <- newChan :: Process (Channel Int)+  rex <- messageKeyRouter PayloadOnly++  -- Since the /router/ doesn't offer a syncrhonous start+  -- option, we use spawnSignalled to get the same effect,+  -- making it more likely (though it's not guaranteed) that+  -- the spawned process will be bound to the routing exchange+  -- prior to our evaluating 'routeMessage' below.+  void $ spawnSignalled (bindKey "foobar" rex) $ const $ do+    receiveWait [ match (\(s :: Int) -> sendChan sp s) ]++  routeMessage rex (createMessage "foobar" [] (123 :: Int))+  stash result . (== (123 :: Int)) =<< receiveChan rp++testMultipleRoutes :: TestResult () -> Process ()+testMultipleRoutes result = do+  stash result ()    -- we don't rely on the test result for assertions...+  (sp, rp) <- newChan+  rex <- messageKeyRouter PayloadOnly+  let recv = receiveWait [+          match (\(s :: String) -> getSelfPid >>= \us -> sendChan sp (us, Left s))+        , match (\(i :: Int) -> getSelfPid >>= \us -> sendChan sp (us, Right i))+        ]++  us <- getSelfPid+  p1 <- spawnSignalled (link us >> bindKey "abc" rex) (const $ forever recv)+  p2 <- spawnSignalled (link us >> bindKey "def" rex) (const $ forever recv)+  p3 <- spawnSignalled (link us >> bindKey "abc" rex) (const $ forever recv)++  -- publish 2 messages with the routing-key set to 'abc'+  routeMessage rex (createMessage "abc" [] "Hello")+  routeMessage rex (createMessage "abc" [] (123 :: Int))++  -- route another message with the 'abc' value a header (should be ignored)+  routeMessage rex (createMessage "" [("abc", "abc")] "Goodbye")++  received <- forM (replicate (2 * 3) us) (const $ receiveChanTimeout 1000 rp)++  -- all bindings for 'abc' fired correctly+  received `shouldContain` Just (p1, Left "Hello")+  received `shouldContain` Just (p3, Left "Hello")+  received `shouldContain` Just (p1, Right (123 :: Int))+  received `shouldContain` Just (p3, Right (123 :: Int))++  -- however the bindings for 'def' never fired+  received `shouldContain` Nothing+  received `shouldNotContain` Just (p2, Left "Hello")+  received `shouldNotContain` Just (p2, Right (123 :: Int))++  -- none of the bindings should have examined the headers!+  received `shouldNotContain` Just (p1, Left "Goodbye")+  received `shouldNotContain` Just (p2, Left "Goodbye")+  received `shouldNotContain` Just (p3, Left "Goodbye")++testHeaderBasedRouting :: TestResult () -> Process ()+testHeaderBasedRouting result = do+  stash result ()  -- we don't rely on the test result for assertions...+  (sp, rp) <- newChan+  rex <- headerContentRouter PayloadOnly "x-name"+  let recv = const $ forever $ receiveWait [+          match (\(s :: String) -> getSelfPid >>= \us -> sendChan sp (us, Left s))+        , match (\(i :: Int) -> getSelfPid >>= \us -> sendChan sp (us, Right i))+        ]++  us <- getSelfPid+  p1 <- spawnSignalled (link us >> bindHeader "x-name" "yellow" rex) recv+  p2 <- spawnSignalled (link us >> bindHeader "x-name" "red"    rex) recv+  _  <- spawnSignalled (link us >> bindHeader "x-type" "fast"   rex) recv++  -- publish 2 messages with the routing-key set to 'abc'+  routeMessage rex (createMessage "" [("x-name", "yellow")] "Hello")+  routeMessage rex (createMessage "" [("x-name", "yellow")] (123 :: Int))+  routeMessage rex (createMessage "" [("x-name", "red")]    (456 :: Int))+  routeMessage rex (createMessage "" [("x-name", "red")]    (789 :: Int))+  routeMessage rex (createMessage "" [("x-type", "fast")]   "Goodbye")++  -- route another message with the 'abc' value a header (should be ignored)+  routeMessage rex (createMessage "" [("abc", "abc")] "FooBar")++  received <- forM (replicate 5 us) (const $ receiveChanTimeout 1000 rp)++  -- all bindings fired correctly+  received `shouldContain` Just (p1, Left "Hello")+  received `shouldContain` Just (p1, Right (123 :: Int))+  received `shouldContain` Just (p2, Right (456 :: Int))+  received `shouldContain` Just (p2, Right (789 :: Int))+  received `shouldContain` Nothing++  -- simple check that no other bindings have fired+  length received `shouldBe` equalTo (5 :: Int)++testSimpleEventHandling :: TestResult Bool -> Process ()+testSimpleEventHandling result = do+  (sp, rp) <- newChan+  (sigStart, recvStart) <- newChan+  em <- EventManager.start+  Just pid <- resolve em+  void $ monitor pid++  -- Note that in our init (state) function, we write a "start signal"+  -- here; Without a start signal, the message sent to the event manager+  -- (via notify) would race with the addHandler registration.+  pid' <- addHandler em (myHandler sp) (sendChan sigStart ())+  link pid'++  () <- receiveChan recvStart++  notify em ("hello", "event", "manager") -- cast message+  r <- receiveTimeout 100000000 [+      matchChan rp return+    , match (\(ProcessMonitorNotification _ _ _) -> die "ServerDied")+    ]+  case r of+    Just ("hello", "event", "manager") -> stash result True+    _                                  -> stash result False++myHandler :: SendPort (String, String, String)+          -> ()+          -> (String, String, String)+          -> Process ()+myHandler sp s m@(_, _, _) = sendChan sp m >> return s++myRemoteTable :: RemoteTable+myRemoteTable =+  Control.Distributed.Process.Extras.__remoteTable initRemoteTable++tests :: NT.Transport  -> IO [Test]+tests transport = do+  localNode <- newLocalNode transport myRemoteTable+  return [+        testGroup "Event Manager"+        [+          testCase "Simple Event Handlers"+          (delayedAssertion "Expected the handler to run"+           localNode True testSimpleEventHandling)+        ]++      , testGroup "Router"+        [+          testCase "Direct Key Routing"+          (delayedAssertion "Expected the sole matching route to run"+           localNode True testKeyBasedRouting)+        , testCase "Key Based Selective Routing"+          (delayedAssertion "Expected only the matching routes to run"+           localNode () testMultipleRoutes)+        , testCase "Header Based Selective Routing"+          (delayedAssertion "Expected only the matching routes to run"+           localNode () testHeaderBasedRouting)+        ]+    ]++main :: IO ()+main = testMain $ tests+

@@ -0,0 +1,256 @@+{-# LANGUAGE BangPatterns        #-}+{-# LANGUAGE TemplateHaskell     #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Main where++import Control.Distributed.Process+import Control.Distributed.Process.Node+import qualified Control.Distributed.Process.Extras (__remoteTable)+import qualified Control.Distributed.Process.Execution.Mailbox (__remoteTable)+import Control.Distributed.Process.Execution.Mailbox+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer++import Control.Rematch (equalTo)++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, drop)+#else+import Prelude hiding (drop)+#endif++import Data.Maybe (catMaybes)++import Test.Framework as TF (testGroup, Test)+import Test.Framework.Providers.HUnit+import TestUtils+import qualified MailboxTestFilters (__remoteTable)+import MailboxTestFilters (myFilter, intFilter)++import qualified Network.Transport as NT++-- TODO: This whole test suite would be much better off using QuickCheck.+-- The test-framework driver however, doesn't have the API support we'd need+-- to wire in our tests, so we'll have to write a compatibility layer.+-- That should probably go into (or beneath) the C.D.P.P.Test module.++allBuffersShouldRespectFIFOOrdering :: BufferType -> TestResult Bool -> Process ()+allBuffersShouldRespectFIFOOrdering buffT result = do+  let [a, b, c] = ["a", "b", "c"]+  mbox <- createAndSend buffT [a, b, c]+  active mbox acceptEverything+  Just Delivery { messages = msgs } <- receiveTimeout (after 2 Seconds)+                                                         [ match return ]+  let [ma', mb', mc'] = msgs+  Just a' <- unwrapMessage ma' :: Process (Maybe String)+  Just b' <- unwrapMessage mb' :: Process (Maybe String)+  Just c' <- unwrapMessage mc' :: Process (Maybe String)+  let values = [a', b', c']+  stash result $ values == [a, b, c]+--  if values /= [a, b, c]+--     then liftIO $ putStrLn $ "unexpected " ++ ((show buffT) ++ (" values: " ++ (show values)))+--     else return ()++resizeShouldRespectOrdering :: BufferType+           -> TestResult [String]+           -> Process ()+resizeShouldRespectOrdering buffT result = do+  let [a, b, c, d, e] = ["a", "b", "c", "d", "e"]+  mbox <- createAndSend buffT [a, b, c, d, e]+  resize mbox (3 :: Integer)++  active mbox acceptEverything+  Just Delivery{ messages = msgs } <- receiveTimeout (after 2 Seconds) [ match return ]++  let [mc', md', me'] = msgs+  Just c' <- unwrapMessage mc' :: Process (Maybe String)+  Just d' <- unwrapMessage md' :: Process (Maybe String)+  Just e' <- unwrapMessage me' :: Process (Maybe String)+  let values = [c', d', e']+  stash result $ values++bufferLimiting :: BufferType -> TestResult (Integer, [Maybe String]) -> Process ()+bufferLimiting buffT result = do+  let msgs = ["a", "b", "c", "d", "e", "f", "g"]+  mbox <- createMailboxAndPost buffT 4 msgs++  MailboxStats{ pendingMessages = pending'+              , droppedMessages = dropped'+              , currentLimit    = limit' } <- statistics mbox+  pending' `shouldBe` equalTo 4+  dropped' `shouldBe` equalTo 3+  limit'   `shouldBe` equalTo 4++  active mbox acceptEverything+  Just Delivery{ messages = recvd+               , totalDropped = skipped } <- receiveTimeout (after 5 Seconds)+                                                            [ match return ]+  seen <- mapM unwrapMessage recvd+  stash result (skipped, seen)++mailboxIsInitiallyPassive :: TestResult Bool -> Process ()+mailboxIsInitiallyPassive result = do+  mbox <- createMailbox Stack (6 :: Integer)+  mapM_ (post mbox) ([1..5] :: [Int])+  Nothing <- receiveTimeout (after 3 Seconds) [ matchAny return ]+  notify mbox+  inbound <- receiveTimeout (after 3 Seconds) [ match return ]+  case inbound of+    Just (NewMail _ _) -> stash result True+    Nothing            -> stash result False++complexMailboxFiltering :: (String, Int, Bool)+                        -> TestResult (String, Int, Bool)+                        -> Process ()+complexMailboxFiltering inputs@(s', i', b') result = do+  mbox <- createMailbox Stack (10 :: Integer)+  post mbox s'+  post mbox i'+  post mbox b'+  waitForMailboxReady mbox 3++  active mbox $ myFilter inputs+  Just Delivery{ messages = [m1, m2, m3]+               , totalDropped = _ } <- receiveTimeout (after 5 Seconds)+                                                      [ match return ]+  Just s <- unwrapMessage m1 :: Process (Maybe String)+  Just i <- unwrapMessage m2 :: Process (Maybe Int)+  Just b <- unwrapMessage m3 :: Process (Maybe Bool)+  stash result $ (s, i, b)++dropDuringFiltering :: TestResult Bool -> Process ()+dropDuringFiltering result = do+  let rng = [1..50] :: [Int]+  mbox <- createMailbox Stack (50 :: Integer)+  mapM_ (post mbox) rng++  waitForMailboxReady mbox 50+  active mbox $ intFilter++  Just Delivery{ messages = msgs } <- receiveTimeout (after 5 Seconds)+                                                     [ match return ]+  seen <- mapM unwrapMessage msgs+  stash result $ (catMaybes seen) == (filter even rng)++mailboxHandleReUse :: TestResult Bool -> Process ()+mailboxHandleReUse result = do+  mbox <- createMailbox Queue (1 :: Limit)+  post mbox "abc"++  notify mbox+  Just (NewMail mbox' _) <- receiveTimeout (after 2 Seconds)+                                           [ match return ]+  deliver mbox'+  _ <- expect :: Process Delivery+  stash result True++createAndSend :: BufferType -> [String] -> Process Mailbox+createAndSend buffT msgs = createMailboxAndPost buffT 10 msgs++createMailboxAndPost :: BufferType -> Limit -> [String] -> Process Mailbox+createMailboxAndPost buffT maxSz msgs = do+  (cc, cp) <- newChan+  mbox <- createMailbox buffT maxSz+  spawnLocal $ mapM_ (post mbox) msgs >> sendChan cc ()+  () <- receiveChan cp+  waitForMailboxReady mbox $ min (toInteger (length msgs)) maxSz+  return mbox++waitForMailboxReady :: Mailbox -> Integer -> Process ()+waitForMailboxReady mbox sz = do+  sleep $ seconds 1+  notify mbox+  m <- receiveWait [+            matchIf (\(NewMail mbox' sz') -> mbox == mbox' && sz' >= sz)+                    (\_ -> return True)+          , match (\(NewMail _ _) -> return False)+          , matchAny (\_ -> return False)+          ]+  case m of+    True  -> return ()+    False -> waitForMailboxReady mbox sz++myRemoteTable :: RemoteTable+myRemoteTable =+  Control.Distributed.Process.Execution.Mailbox.__remoteTable $+  Control.Distributed.Process.Extras.__remoteTable $+  MailboxTestFilters.__remoteTable initRemoteTable++tests :: NT.Transport  -> IO [Test]+tests transport = do+  {- verboseCheckWithResult stdArgs -}+  localNode <- newLocalNode transport myRemoteTable+  return [+        testGroup "Dequeue/Pop Ordering"+        [+          testCase "Queue Ordering"+          (delayedAssertion+           "Expected the Queue to offer FIFO ordering"+           localNode True (allBuffersShouldRespectFIFOOrdering Queue))+	   +        , testCase "Stack Ordering"+          (delayedAssertion+           "Expected the Queue to offer FIFO ordering"+           localNode True (allBuffersShouldRespectFIFOOrdering Stack))+        , testCase "Ring Ordering"+           (delayedAssertion+            "Expected the Queue to offer FIFO ordering"+            localNode True (allBuffersShouldRespectFIFOOrdering Ring))+        ]+      , testGroup "Resize & Ordering"+        [+          testCase "Queue Drops Eldest"+          (delayedAssertion+           "expected c, d, e"+           localNode ["c", "d", "e"] $ resizeShouldRespectOrdering Queue)+        , testCase "Stack Drops Youngest"+          (delayedAssertion+           "expected a, b, c"+           localNode ["a", "b", "c"] $ resizeShouldRespectOrdering Stack)+        , testCase "Ring Drops Youngest"+          (delayedAssertion+           "expected a, b, c"+           localNode ["a", "b", "c"] $ resizeShouldRespectOrdering Ring)+        ]+      , testGroup "Buffer Limits & Discarded Messages"+        [+          testCase "Queue Drops Eldest and Enqueues New"+          (delayedAssertion+           "expected d, e, f, g"+           localNode ((3 :: Integer), map Just ["d", "e", "f", "g"]) $ bufferLimiting Queue)+        , testCase "Stack Drops Youngest And Pushes New"+          (delayedAssertion+           "expected a, b, c, g"+           localNode ((3 :: Integer), map Just ["a", "b", "c", "g"]) $ bufferLimiting Stack)+        , testCase "Ring Rejects New Entries"+          (delayedAssertion+           "expected a, b, c, d"+           localNode ((3 :: Integer), map Just ["a", "b", "c", "d"]) $ bufferLimiting Ring)+        ]+      , testGroup "Notification, Activation and Delivery"+        [+          testCase "Mailbox is initially Passive"+           (delayedAssertion+            "Expected the Mailbox to remain passive until told otherwise"+            localNode True mailboxIsInitiallyPassive)+        , testCase "Mailbox Notifications include usable control channel"+           (delayedAssertion+            "Expected traffic to be relayed directly to us"+            localNode True mailboxHandleReUse)+        , testCase "Complex Filtering Rules"+           (delayedAssertion+            "Expected the relevant filters to accept our data"+            localNode inputs (complexMailboxFiltering inputs))+        , testCase "Filter out unwanted messages"+           (delayedAssertion+            "Expected only even numbers to be sent delivered"+            localNode True dropDuringFiltering)+        ]+    ]+  where+    inputs = ("hello", 10 :: Int, True)++main :: IO ()+main = testMain $ tests+

@@ -0,0 +1,234 @@+{-# LANGUAGE DeriveDataTypeable        #-}+{-# LANGUAGE TemplateHaskell           #-}+{-# LANGUAGE DeriveGeneric             #-}++module TestUtils+  ( TestResult+    -- ping !+  , Ping(Ping)+  , ping+  , shouldBe+  , shouldMatch+  , shouldContain+  , shouldNotContain+  , shouldExitWith+  , expectThat+  -- test process utilities+  , TestProcessControl+  , startTestProcess+  , runTestProcess+  , testProcessGo+  , testProcessStop+  , testProcessReport+  , delayedAssertion+  , assertComplete+  , waitForExit+  -- logging+  , Logger()+  , newLogger+  , putLogMsg+  , stopLogger+  -- runners+  , mkNode+  , tryRunProcess+  , testMain+  , stash+  ) where++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif+import Control.Concurrent+  ( ThreadId+  , myThreadId+  , forkIO+  )+import Control.Concurrent.STM+  ( TQueue+  , newTQueueIO+  , readTQueue+  , writeTQueue+  )+import Control.Concurrent.MVar+  ( MVar+  , newEmptyMVar+  , takeMVar+  , putMVar+  )++import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable()+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Extras.Internal.Types+import Control.Exception (SomeException)+import qualified Control.Exception as Exception+import Control.Monad (forever)+import Control.Monad.STM (atomically)+import Control.Rematch hiding (match)+import Control.Rematch.Run+import Test.HUnit (Assertion, assertFailure)+import Test.HUnit.Base (assertBool)+import Test.Framework (Test, defaultMain)+import Control.DeepSeq++import Network.Transport.TCP+import qualified Network.Transport as NT++import Data.Binary+import Data.Typeable+import GHC.Generics++--expect :: a -> Matcher a -> Process ()+--expect a m = liftIO $ Rematch.expect a m++expectThat :: a -> Matcher a -> Process ()+expectThat a matcher = case res of+  MatchSuccess -> return ()+  (MatchFailure msg) -> liftIO $ assertFailure msg+  where res = runMatch matcher a++shouldBe :: a -> Matcher a -> Process ()+shouldBe = expectThat++shouldContain :: (Show a, Eq a) => [a] -> a -> Process ()+shouldContain xs x = expectThat xs $ hasItem (equalTo x)++shouldNotContain :: (Show a, Eq a) => [a] -> a -> Process ()+shouldNotContain xs x = expectThat xs $ isNot (hasItem (equalTo x))++shouldMatch :: a -> Matcher a -> Process ()+shouldMatch = expectThat++shouldExitWith :: (Addressable a) => a -> DiedReason -> Process ()+shouldExitWith a r = do+  _ <- resolve a+  d <- receiveWait [ match (\(ProcessMonitorNotification _ _ r') -> return r') ]+  d `shouldBe` equalTo r++waitForExit :: MVar ExitReason+            -> Process (Maybe ExitReason)+waitForExit exitReason = do+    -- we *might* end up blocked here, so ensure the test doesn't jam up!+  self <- getSelfPid+  tref <- killAfter (within 10 Seconds) self "testcast timed out"+  tr <- liftIO $ takeMVar exitReason+  cancelTimer tref+  case tr of+    ExitNormal -> return Nothing+    other      -> return $ Just other++mkNode :: String -> IO LocalNode+mkNode port = do+  Right (transport1, _) <- createTransportExposeInternals+                                    "127.0.0.1" port defaultTCPParameters+  newLocalNode transport1 initRemoteTable++-- | Run the supplied @testProc@ using an @MVar@ to collect and assert+-- against its result. Uses the supplied @note@ if the assertion fails.+delayedAssertion :: (Eq a) => String -> LocalNode -> a ->+                    (TestResult a -> Process ()) -> Assertion+delayedAssertion note localNode expected testProc = do+  result <- newEmptyMVar+  _ <- forkProcess localNode $ testProc result+  assertComplete note result expected++-- | Takes the value of @mv@ (using @takeMVar@) and asserts that it matches @a@+assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()+assertComplete msg mv a = do+  b <- takeMVar mv+  assertBool msg (a == b)++-- synchronised logging++data Logger = Logger { _tid :: ThreadId, msgs :: TQueue String }++-- | Create a new Logger.+-- Logger uses a 'TQueue' to receive and process messages on a worker thread.+newLogger :: IO Logger+newLogger = do+  tid <- liftIO $ myThreadId+  q <- liftIO $ newTQueueIO+  _ <- forkIO $ logger q+  return $ Logger tid q+  where logger q' = forever $ do+        msg <- atomically $ readTQueue q'+        putStrLn msg++-- | Send a message to the Logger+putLogMsg :: Logger -> String -> Process ()+putLogMsg logger msg = liftIO $ atomically $ writeTQueue (msgs logger) msg++-- | Stop the worker thread for the given Logger+stopLogger :: Logger -> IO ()+stopLogger = (flip Exception.throwTo) Exception.ThreadKilled . _tid++-- | Given a @builder@ function, make and run a test suite on a single transport+testMain :: (NT.Transport -> IO [Test]) -> IO ()+testMain builder = do+  Right (transport, _) <- createTransportExposeInternals+                                    "127.0.0.1" "10501" defaultTCPParameters+  testData <- builder transport+  defaultMain testData++-- | Runs a /test process/ around the supplied @proc@, which is executed+-- whenever the outer process loop receives a 'Go' signal.+runTestProcess :: Process () -> Process ()+runTestProcess proc = do+  ctl <- expect+  case ctl of+    Stop -> return ()+    Go -> proc >> runTestProcess proc+    Report p -> receiveWait [matchAny (\m -> forward m p)] >> runTestProcess proc++-- | Starts a test process on the local node.+startTestProcess :: Process () -> Process ProcessId+startTestProcess proc =+   spawnLocal $ do+     getSelfPid >>= register "test-process"+     runTestProcess proc++-- | Control signals used to manage /test processes/+data TestProcessControl = Stop | Go | Report ProcessId+  deriving (Typeable, Generic)++instance Binary TestProcessControl where++-- | A mutable cell containing a test result.+type TestResult a = MVar a++-- | Stashes a value in our 'TestResult' using @putMVar@+stash :: TestResult a -> a -> Process ()+stash mvar x = liftIO $ putMVar mvar x++-- | Tell a /test process/ to stop (i.e., 'terminate')+testProcessStop :: ProcessId -> Process ()+testProcessStop pid = send pid Stop++-- | Tell a /test process/ to continue executing+testProcessGo :: ProcessId -> Process ()+testProcessGo pid = send pid Go++-- | A simple @Ping@ signal+data Ping = Ping+  deriving (Typeable, Generic, Eq, Show)++instance Binary Ping where+instance NFData Ping where++ping :: ProcessId -> Process ()+ping pid = send pid Ping+++tryRunProcess :: LocalNode -> Process () -> IO ()+tryRunProcess node p = do+  tid <- liftIO myThreadId+  runProcess node $ catch p (\e -> liftIO $ Exception.throwTo tid (e::SomeException))++-- | Tell a /test process/ to send a report (message)+-- back to the calling process+testProcessReport :: ProcessId -> Process ()+testProcessReport pid = do+   self <- getSelfPid+   send pid $ Report self