patterns 0.0.3 → 0.1.0

raw patch · 22 files changed

+3406/−3127 lines, 22 filesdep +conduitdep −enumeratorPVP ok

version bump matches the API change (PVP)

Dependencies added: conduit

Dependencies removed: enumerator

API changes (from Hackage documentation)

Files

• patterns.cabal +39/−10
• src/Network/Mom/Patterns.hs +105/−107
• src/Network/Mom/Patterns/Basic.hs +76/−1091
• src/Network/Mom/Patterns/Basic/Client.hs +92/−0
• src/Network/Mom/Patterns/Basic/Publisher.hs +110/−0
• src/Network/Mom/Patterns/Basic/Puller.hs +94/−0
• src/Network/Mom/Patterns/Basic/Pusher.hs +50/−0
• src/Network/Mom/Patterns/Basic/Server.hs +115/−0
• src/Network/Mom/Patterns/Basic/Subscriber.hs +84/−0
• src/Network/Mom/Patterns/Broker.hs +102/−0
• src/Network/Mom/Patterns/Broker/Broker.hs +172/−0
• src/Network/Mom/Patterns/Broker/Client.hs +139/−0
• src/Network/Mom/Patterns/Broker/Common.hs +298/−0
• src/Network/Mom/Patterns/Broker/Server.hs +105/−0
• src/Network/Mom/Patterns/Device.hs +0/−559
• src/Network/Mom/Patterns/Enumerator.hs +0/−418
• src/Network/Mom/Patterns/Streams.hs +987/−0
• src/Network/Mom/Patterns/Types.hs +165/−0
• src/Service.hs +0/−378
• src/Types.hs +0/−564
• src/broker/Heartbeat.hs +91/−0
• src/broker/Registry.hs +582/−0

@@ -1,7 +1,7 @@ Name:            patterns-Version:         0.0.3+Version:         0.1.0 Cabal-Version:   >= 1.14.0-Copyright:       Copyright (c) Tobias Schoofs, 2011 - 2012+Copyright:       Copyright (c) Tobias Schoofs, 2011 - 2013 License:         LGPL license-file:    license/lgpl-3.0.txt Author:          Tobias Schoofs@@ -26,25 +26,54 @@   More examples and a test suite are available    on <http://github.com/toschoo/mom>. +  .++  Release History:++  .++  [0.1.0] Major Review:+          Conduits replace enumerators,+          Interfaces simplified to ByteString,+          Interfaces simplified in general,+          Advanced patterns:+          Majordomo Broker.++  .++  [0.0.1] Initial Release.++-- with libzmq 2.1, 2.2 is broken! Library   Build-Depends:   base        >= 4.5 && <= 5.0,                    bytestring  >= 0.9.2.1,                    utf8-string >= 0.3.7,                    containers >= 0.4.2.1,-                   zeromq-haskell >= 0.8.4,-                   enumerator >= 0.4.18,-                   mtl         >= 2.0.1.0,-                   time        >= 1.4+                   zeromq-haskell >= 0.8.4, +                   conduit >= 1.0.7.1,+                   mtl     >= 2.0.1.0,+                   time    >= 1.4    default-language: Haskell98 -  hs-source-dirs: src/Network/Mom, src+  hs-source-dirs: src/Network/Mom, src, src/broker                       Exposed-Modules: Network.Mom.Patterns, +                   Network.Mom.Patterns.Streams,+                   Network.Mom.Patterns.Types,                    Network.Mom.Patterns.Basic,-                   Network.Mom.Patterns.Device,-                   Network.Mom.Patterns.Enumerator-  other-modules: Factory, Types, Service+                   Network.Mom.Patterns.Basic.Client,+                   Network.Mom.Patterns.Basic.Server,+                   Network.Mom.Patterns.Basic.Publisher,+                   Network.Mom.Patterns.Basic.Subscriber,+                   Network.Mom.Patterns.Basic.Puller,+                   Network.Mom.Patterns.Basic.Pusher,+                   Network.Mom.Patterns.Broker,+                   Network.Mom.Patterns.Broker.Common,+                   Network.Mom.Patterns.Broker.Broker,+                   Network.Mom.Patterns.Broker.Server,+                   Network.Mom.Patterns.Broker.Client+  other-modules: Factory, Heartbeat, Registry    ghc-options: -Wall 

@@ -6,40 +6,55 @@ -- Stability  : experimental -- Portability: portable ----- This package implements communication patterns--- that are often used in distributed applications.--- The package implements a set of basic patterns--- and a device to connect basic patterns through--- routers, brokers, load balancers, /etc./ --- The package is based on the /zeromq/ library,--- but, in some cases, deviates from the /zeromq/ terminology.+-- In distributed message-oriented applications,+-- the same communication patterns show up+-- over and over again.+-- This package implements some of these patterns+-- based on the /zeromq/ library.+-- /Patterns/ uses the /zeromq-haskell/ package,+-- but goes beyond in several aspects:+-- +-- * It uses /conduits/ to stream incoming and+--   outgoing message segments;+--+-- * It defines libraries of basic patterns+--   to enforce coherent use of /zeromq/ sockets;+--+-- * It implements modules for advanced patterns;+--   currently the majordomo pattern (broker) is implemented.+--  -- More information on /zeromq/ can be found at -- <http://www.zeromq.org>. ------------------------------------------------------------------------------- module Network.Mom.Patterns (+           -- * Patterns           -- $patterns +          -- * Streams+          -- $streams++          -- | The type module defines some fundamental types:+          module Network.Mom.Patterns.Types,++          -- | The streams module defines a streaming device+          --   and some useful operations:+          module Network.Mom.Patterns.Streams,+           -- * Basic Patterns           -- $basic            module Network.Mom.Patterns.Basic, -          -- * Devices-          -- $device--          module Network.Mom.Patterns.Device,--          -- * Enumerators-          -- $enumerator+          -- * Advanced Patterns+          -- $advanced -          module Network.Mom.Patterns.Enumerator           ) where +  import Network.Mom.Patterns.Types+  import Network.Mom.Patterns.Streams   import Network.Mom.Patterns.Basic-  import Network.Mom.Patterns.Device-  import Network.Mom.Patterns.Enumerator    {- $patterns      Instead of a centralised message broker@@ -51,33 +66,32 @@      threads, processes and network nodes      according to certain protocol patterns.      -     The Patterns package hides details-     about sockets and instead -     provides higher-level abstractions,-     in particular a set of basic patterns.-     Additionally, it implements a /device/ to connect patterns-     with each other to form more complex patterns.-     Devices are stream transformers that may be used as -     routers, brokers or load balancers.+     The /Patterns/ package hides details+     about sockets and +     provides instead higher-level abstractions,+     in particular an event-driven streaming device +     and a set of basic patterns.+     Streams come in handy when building more complex pattern,+     such as routers, brokers or load balancers.+     Currently included in this version+     is the Majordomo pattern, +     a service broker and load balancer+     for the /client\/server/ pattern.       The interfaces provided by the Patterns package      support the separation of different concerns      involved with application design.      Most of the interfaces are higher-order functions      that accept -     data converters, stream processors, error handlers-     and control actions.-     The goal is to ease the implementation-     of general purpose and domain-specific libraries-     providing those building blocks.-     Distributed application components can then be built-     by bundling converters, stream processors and error handlers+     stream processors and control actions.+     Distributed application components can be built+     by bundling stream processors       together to request or provide services,       publish or subscribe data or to       allocate work to processing nodes.       Note that, since the patterns package is based on ZMQ,-     applications based on patterns must be compiled with the +     applications based on patterns must be linked with the       /-threaded/ flag.          -}@@ -103,13 +117,9 @@        and worker processes that connect to the pipeline;        jobs are work-balanced among workers; -     * Exclusive Pair (a.k.a. Peer-to-Peer)-       consisting of two peer processes -       that freely exchange messages between each other.-      All of these basic patterns consist of two parts-     that, roughly, can be described as a client and a server side.-     Only parts of the same pattern+     which can, roughly, be described as a client and a server side.+     Only those sides belonging to the same pattern      can communicate with each other.       Since communication may - and usually does -      cross processes and network nodes,@@ -137,86 +147,74 @@        can receive messages only from a publisher;       * a pusher cannot receive messages and-       can send message only to a pipe;+       can send message only to a puller;       * a puller cannot send messages and-       can receive messages only from a pipe;+       can receive messages only from a pusher; -     * peers can exchange messages only with other peers.+     Note that the /peer-to-peer/ pattern+     defined in /zeromq/ +     is not provided by the /patterns/ library.+     For this kind of communication the basic+     /zeromq/ package schould be used.   -} -  {- $device-     Devices relay messages between compatible services, /i.e./-     Servers and Clients,-     Publishers and Subscribers,-     Pipes and Pullers and-     Peers and Peers.+  {- $streams+     The /patterns/ package uses conduits for message processing.+     All message endpoints create or receive messages as +     streams of message segments +     (even if there is only one segment in the message).+     Endpoint like /clients/, /server/ and so on+     use producers, consumers and conduits from the /conduit/ package+     to handle messages. -     A device polls over a list of access points;-     the list is passed in when the device is started, but-     the application may, at any time,-     add or remove access points.+     There is additionally a streaming device+     that relays messages between compatible services, /i.e./+     Servers and Clients,+     Publishers and Subscribers and+     Pusher and Pullers. +     A streaming device polls over a list of access points.      When data is available,-     the device activates an application-defined stream transformer.+     the an application-defined stream transformer+     is invoked.      The outgoing stream may be directed to      one or several access points      including the source itself.-     Note, however, that the basic patterns-     restrict possible combinations. -     -     Devices are more general than basic patterns-     and could even be used to simulate them,-     which may indeed be usefull in some situations.-     It is preferrable, however, to use basic patterns instead of devices-     where ever possible.-     The main purpose of devices-     is to link topologies for -     load-balancing, routing or scaling;-     they can be seen as a kind of smart software switch-     connecting basic patterns.+     Note, however, that the +     combination of local access point and remote target socket +     must adhere to the restrictions of possible peer combinations. ++     Streams are mainly thought to implement+     more complex patterns,+     brokers, load balancers, /etc./+     In basic patterns, they are used +     to implement background processes,+     for instance in the /server/ module.   -} -  {- $enumerator-     Basic patterns and devices exchange messages.-     Messages are composed of one or more segments;-     The underlying library guarantees that-     a message, consisting of many segments,-     is sent and received as a whole, -     /i.e./ all segments are sent and received-     or the message as a whole is not sent or-     received at all.-     -     Messages may be segmented into -     parts with different functional purpose, -     such as an envelope and a body;-     segments may also be used to -     split a message into single elements of -     the same type, /e.g./ the rows of a huge -     result set of a dababase query.-     -     To uniform the message handling,-     patterns and devices treat all messages-     as message streams. Streams are processed-     using /enumerator/s (to create streams)-     and /iteratee/s (to receive streams).-     One half of the stream processing-     is implemented by application code,-     /e.g./, how a publisher creates its outgoing stream-             is defined by an application-specific enumerator;-             the stream is then sent to subscribers by-             a package-implemented standard iteratee.-             Subscribers, on the other hand, receive-             the stream by means of an internal enumerator-             and call an application-defined iteratee.+  {- $advanced+     The zeromq design +     encourages to build new complex patterns,+     some of which are described on the zeromq website.+     The main idea, here, is that the design of the middleware+     should not statically impose one topology +     on all possible communication scenarios;+     therefore, zeromq does not place a broker+     at the very core of its architecture.+     Instead, application components can connect freely+     using the architectural pattern +     that best solves the problem at hand. -     The Enumerator module provides generic-     enumerators (called /fetchers/) and-     iteratees   (called /dumps/)-     that handle common stream patterns,-     such as single segment messages, -     messages with a fixed number of segments,-     streams generated by lists and-     transformation into strings, lists, monoids, /etc./- +     Advanced communication patterns use+     message exchange protocols and other means,+     even brokers, to make communication reliably and scalable.++     The 'patterns' library aims to provide+     advanced topics as libraries that can be used +     flexibly to solve concrete application problems.+     Currently, as a proof-of-concept,+     the majordomo pattern is implemented,+     a service broker, providing a central access point+     for clients, load balancing and service discovery.   -} 

@@ -4,1094 +4,79 @@ -- Copyright  : (c) Tobias Schoofs -- License    : LGPL  -- Stability  : experimental--- Portability: portable--- --- Basic communication patterns---------------------------------------------------------------------------------module Network.Mom.Patterns.Basic (-          -- * Server/Client-          withServer,-          Client, clientContext, setClientOptions, withClient,-          request, askFor, checkFor,-          -- * Publish/Subscribe-          Pub, pubContext, setPubOptions, withPub, issue,-          withPeriodicPub,-          withSub,-          Sub, subContext, setSubOptions,-          withSporadicSub, checkSub, waitSub, unsubscribe, resubscribe,-          -- * Pipeline-          Pipe, pipeContext, setPipeOptions, withPipe, push, -          withPuller,-          -- * Exclusive Pair-          Peer, peerContext, setPeerOptions, withPeer, send, receive,-          -- * Service Access Point-          AccessPoint(..), LinkType(..), parseLink,-          -- * Converters-          InBound, OutBound,-          idIn, idOut, inString, outString, inUTF8, outUTF8,-          -- * Errors and Error Handlers-          Criticality(..),-          OnError, OnError_,-          chainIO, chainIOe, tryIO, tryIOe,-          -- * Generic Serivce-          Service, srvName, srvContext, pause, resume, -          changeParam, changeOption,-          -- * ZMQ Context-          Z.Context, Z.withContext,-          Z.SocketOption(..),-          -- * Helpers-          Topic, alltopics, notopic,-          Timeout, Parameter, noparam)-where--  import           Types-  import           Service-  import           Factory--  import           Network.Mom.Patterns.Device--  import qualified Data.ByteString.Char8  as B-  import qualified Data.Enumerator        as E-  import           Data.Enumerator (($$))--  import           Control.Concurrent -  import           Control.Applicative ((<$>))-  import           Control.Monad-  import           Prelude hiding (catch)-  import           Control.Exception (SomeException,-                                      catch, try, finally)--  import qualified System.ZMQ as Z--  -------------------------------------------------------------------------  -- | Starts one or more server threads-  --   and executes an action that-  --   receives a 'Service' to control the server.-  --   The 'Service' is a thread local resource.-  --   It must not be passed to threads forked -  --   from the thread that has started the service.-  --   The 'Service' is valid only in the scope of the action.-  --   When the action terminates, the server is automatically stopped.-  --   During the action, the server can be paused and restarted.-  --   Also, the 'SocketOption' of the underlying ZMQ 'Z.Socket' -  --   can be changed.-  --   Please refer to 'pause', 'resume' and 'changeOption' for more details.-  ---  --   The application may implement control parameters.-  --   Control parameters are mere strings that are passed-  --   to the application call-backs. -  --   It is up to the application to enquire these strings-  --   and to implement different behaviour for the possible settings.-  --   Control parameter can be changed during run-time-  --   by means of 'changeParam'.-  ---  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ context;-  ---  --   * 'String': The name of the server, useful for debugging;-  ---  --   * 'Parameter': The initial value of the control parameter -  --                  passed to all application call-backs;-  ---  --   * 'Int': The number of worker threads;-  --            note that a server with only one thread-  --            handles client requests sequentially.-  --            The number of threads -  --            (together with the number of hardware processing resources)-  --            defines how many client requests can be processed in parallel. -  ---  --   * 'AccessPoint': The access point, -  --                    through which this server can be reached;-  ---  --   * 'LinkType': The link type; -  --                 standalone servers usually bind their access point,-  --                 whereas clients connect to it.-  --                 Instead, a server may also connect-  --                 to a load-balancing device,-  --                 to which other servers and clients connect-  --                 (see 'withDevice' and 'withQueue').-  ---  --   * 'InBound': The converter to convert the incoming-  --                data stream (of type 'B.ByteString') -  --                into a client request component.-  --                Note that the converter converts-  --                single message segments to components of type /c/.-  --                The 'E.Iteratee', receiving this /c/-typed-  --                elements shall combine them -  --                to a complete request of type /i/,-  --                which is then processed by an 'E.Enumerator'-  --                to create the server response.-  ---  --   * 'OutBound': The converter to convert the results of type /o/-  --                 to a 'B.ByteString', which then is sent-  --                 back to the client.-  ---  --   * 'OnError': The error handler-  ---  --   * 'String' -> 'E.Iteratee': The 'E.Iteratee' that processes-  --                             request components of type /c/-  --                             and yields a request of type /i/.-  --                             The 'String' argument is-  --                             the control parameter,-  --                             whose logic is implemented -  --                             by the application.-  ---  --   * 'Fetch': The 'E.Enumerator' that processes-  --              the request of type /i/ to produce-  --              results of type /o/.-  ---  --   * 'Service' -> IO (): The action to invoke, -  --                         when the server has been started; -  --                         the service is used to control the server.-  --   -  --   The following code fragment shows a -  --   simple server to process data base queries -  --   using standard converters and error handlers-  --   not further defined here:-  ---  --   @-  --   withContext 1 $ \\ctx -> do-  --      c <- connectODBC \"DSN=xyz\"  -- some ODBC connection-  --      s <- prepare c \"select ...\" -- some database query -  --      withServer ctx -  --          \"MyQuery\" -- name of the server is \"MyQuery\"-  --          noparam     -- no parameter-  --          5           -- five worker threads-  --          (Address \"tcp:\/\/*:5555\" []) Bind -- bind to this address-  --          iconv oconv -- some standard converters-  --          onErr       -- some standard error handler-  --          (\\_  -> one []) -- 'E.Iteratee' for single segment messages;-  --                           -- refer to 'Enumerator' for details-  --          (dbFetcher s) $ \\srv -> -- the 'E.Enumerator';-  --            untilInterrupt $ do -- install a signal handler for /SIGINT/-  --                                -- and repeat the following action-  --                                -- until /SIGINT/ is received;-  --              putStrLn $ \"server \" ++ srvName srv ++ -  --                         \" up and running...\"-  --              threadDelay 1000000-  --   @-  ---  --   The untilInterrupt loop may be implemented as follows:-  ---  --   @-  ---  --     untilInterrupt :: IO () -> IO ()-  --     untilInterrupt run = do-  --       continue <- newMVar True-  --       _ <- installHandler sigINT (Catch $ handler continue) Nothing-  --       go continue -  --      where handler m = modifyMVar_ m (\\_ -> return False)-  --            go      m = do run-  --                           continue <- readMVar m-  --                           when continue $ go m-  --   @-  -- -  --   Finally, a simple dbFetcher:-  ---  --   @-  --     dbFetcher :: SQL.Statement -> Fetch [SQL.SqlValue] String-  --     dbFetcher s _ _ _ stp = tryIO (SQL.execute s []) >>= \\_ -> go stp-  --       where go step = -  --               case step of-  --                 E.Continue k -> do-  --                   mbR <- tryIO $ SQL.fetchRow s-  --                   case mbR of-  --                     Nothing -> E.continue k-  --                                        -- convRow is not defined here-  --                     Just r  -> go $$ k (E.Chunks [convRow r]) -  --                 _ -> E.returnI step-  --   @-  -------------------------------------------------------------------------  withServer :: Z.Context   -> String          -> -                Parameter   -> Int             ->-                AccessPoint                    -> -                LinkType                       ->-                InBound c   -> OutBound o      -> -                OnError                        ->-                (String  -> E.Iteratee c IO i) ->-                Fetch i o                      -> -                (Service -> IO a)              -> IO a-  withServer ctx name param n ac t iconv oconv onerr build fetch =-    withService ctx name param service -    where service = serve n ac t iconv oconv onerr -                          build fetch--  -------------------------------------------------------------------------  -- the server implementation-  -------------------------------------------------------------------------  serve :: Int                            ->-           AccessPoint                    ->-           LinkType                       -> -           InBound c                      ->-           OutBound o                     ->-           OnError                        ->-           (String  -> E.Iteratee c IO i) ->-           Fetch i o                      -> -           Z.Context -> String -> String  -> String -> IO () -> IO ()-  serve n ac t iconv oconv onerr -        build fetch ctx name sockname param ready-  -------------------------------------------------------------------------  -- prepare service for single client-  -------------------------------------------------------------------------    | n <= 1 = -      Z.withSocket ctx Z.Rep $ \client -> do-        link t ac client-        Z.withSocket ctx Z.Sub $ \cmd -> do-          conCmd cmd sockname ready-          poll False [Z.S cmd Z.In, Z.S client Z.In] (go client) param-      `catch` (\e -> onerr Fatal e name param >>= \_ -> return ())-  -------------------------------------------------------------------------  -- prepare service for multiple clients -  -------------------------------------------------------------------------    | otherwise = (do-        add <- ("inproc://wrk_" ++) <$> show <$> mkUniqueId-        as  <- replicateM n newEmptyMVar -        zs  <- replicateM n newEmptyMVar-        withQueue ctx ("Queue " ++ name)-                      (ac, t) (Address add [], Bind) onQErr $ \_ -> do-          _ <- mapM (uncurry $ start add) (zip as zs)-          mapM_ takeMVar as  -- wait for workers to start-          ready              -- report state to service-          mapM_ takeMVar zs) -- wait for workers to terminate-        `catch` (\e -> onerr Fatal e name param >>= \_ -> return ())-  -------------------------------------------------------------------------  -- start thread-  -------------------------------------------------------------------------    where start add a z = forkIO (startWork add a `finally` putMVar z ())-  -------------------------------------------------------------------------  -- start worker for multiple clients -  -------------------------------------------------------------------------          startWork add starter = Z.withSocket ctx Z.Rep $ \worker -> (do-            trycon worker add retries-            Z.withSocket ctx Z.Sub $ \cmd -> do-              trycon cmd sockname retries-              Z.subscribe cmd noparam-              putMVar starter ()-              poll False [Z.S cmd Z.In, Z.S worker Z.In] (go worker) param)-            `catch` (\e -> onerr Critical e name param >>= \_ -> return ())-  -------------------------------------------------------------------------  -- receive requests and do the job-  -------------------------------------------------------------------------          go worker p = do-              ei <- E.run (rcvEnum worker iconv $$ build p)-              ifLeft ei (\e -> handle worker e p) $ \i ->-                        catch (body worker p i)-                              (\e -> handle worker e p)-          body worker p i = do-               eiR <- E.run (fetch ctx p i $$ itSend worker oconv)-               ifLeft eiR-                 (\e -> handle worker e p)-                 (\_ -> return ())-  -------------------------------------------------------------------------  -- generic error handler-  -------------------------------------------------------------------------          handle sock e p = onerr Error e name p >>= \mbX ->-              case mbX of-                Nothing -> -                  Z.send sock B.empty []-                Just x  -> do -                  Z.send sock x [Z.SndMore]-                  Z.send sock B.empty []-          onQErr c e nm _ = onerr c e nm noparam >>= \_ -> return ()--  -------------------------------------------------------------------------  -- | Client data type-  -------------------------------------------------------------------------  data Client i o = Client {-                         cliCtx  :: Z.Context,-                         cliSock :: Z.Socket Z.Req,-                         cliAdd  :: AccessPoint,-                         cliOut  :: OutBound o,-                         cliIn   :: InBound  i}--  -------------------------------------------------------------------------  -- | Obtaining the 'Z.Context' from 'Client'-  -------------------------------------------------------------------------  clientContext :: Client i o -> Z.Context-  clientContext = cliCtx--  -------------------------------------------------------------------------  -- | Setting 'Z.SocketOption' to the underlying ZMQ 'Z.Socket'-  -------------------------------------------------------------------------  setClientOptions :: Client i o -> [Z.SocketOption] -> IO ()-  setClientOptions c = setSockOs (cliSock c)--  -------------------------------------------------------------------------  -- | Creates a 'Client';-  --   a client is not a background process like a server,-  --   but a data type that provides functions-  --   to interoperate with a server.-  --   'withClient' creates a client and -  --   invokes the application-defined action,-  --   which receives a 'Client' argument.-  --   The lifetime of the 'Client' is limited-  --   to the invoked action.-  --   When the action terminates, the 'Client' /dies/.-  ---  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ Context;-  ---  --   * 'AccessPoint': The access point, to which the client connects;-  ---  --   * 'OutBound': Converter to convert a request from type /o/-  --                 to the wire format 'B.ByteString'.-  --                 Note that, as for servers, the request-  --                 may be composed of components that together-  --                 form the request. The type /o/-  --                 corresponds to one of these request components,-  --                 not necessarily to the request type as a whole,-  --                 which is determined when issuing a request.-  ---  --   * 'InBound': Converter to convert a reply ('B.ByteString')-  --                into type 'i'.-  --                Note again that the reply may consist of many-  --                message segments. The type /i/ relates to one-  --                reply component, -  --                not necessarily to the reply type as a whole,-  --                which is determined when issuing a request.-  ---  --   * 'Client' -> IO a: The action to perform with this client.-  -------------------------------------------------------------------------  withClient :: Z.Context  -> AccessPoint -> -                OutBound o -> InBound i   -> -                (Client i o -> IO a)      -> IO a-  withClient ctx ac oconv iconv act = Z.withSocket ctx Z.Req $ \s -> do -    trycon s (acAdd ac) retries-    act Client {-        cliCtx  = ctx,-        cliSock = s,-        cliAdd  = ac,-        cliOut  = oconv,-        cliIn   = iconv}--  -------------------------------------------------------------------------  -- | Synchronously requesting a service;-  --   the function blocks the current thread,-  --   until a reply is received.-  --   -  --   Parameters:-  ---  --   * 'Client': The client that performs the request-  ---  --   * 'E.Enumerator': Enumerator to create the request message stream-  ---  --   * 'E.Iteratee': Iteratee to process the reply message stream-  ---  --   A simple client that just writes the results to 'stdout':-  ---  --   @-  --     rcv :: String -> IO ()-  --     rcv req = withContext 1 $ \\ctx -> -  --       withClient ctx -  --         (Address \"tcp:\/\/localhost:5555\" []) -- connect to this address-  --         (return . B.pack) (return . B.unpack) $ -- string converters-  --         \\s -> do-  --           -- request with enum and outit-  --           ei <- request s (enum req) outit      -  --           case ei of-  --             Left e  -> putStrLn $ \"Error: \" ++ show (e::SomeException)-  --             Right _ -> return ()-  --   @-  ---  --   @-  --     -- Enumerator that returns just one string-  --     enum :: String -> E.Enumerator String IO ()-  --     enum = once (return . Just)-  --   @-  ---  --   @-  --     -- Iteratee that just writes to stdout-  --     outit :: E.Iteratee String IO ()-  --     outit = do-  --       mbi <- EL.head-  --       case mbi of-  --         Nothing -> return ()-  --         Just i  -> liftIO (putStrLn i) >> outit-  --   @-  ---  --   Note that this code just issues one request,-  --   which is not the most typical use case.-  --   It is more likely that the action will loop for ever-  --   and receive requests, for instance, from a user interface.-  -------------------------------------------------------------------------  request :: Client i o           ->   -             E.Enumerator o IO () ->-             E.Iteratee i IO a    -> IO (Either SomeException a) -  request c enum it = tryout ?> reicv-    where tryout    = try $ askFor    c enum-          reicv  _  =       rcvClient c it--  -------------------------------------------------------------------------  -- | Asynchronously requesting a service;-  --   the function sends a request to the server -  --   without waiting for a result.-  --   -  --   Parameters:-  ---  --   * 'Client': The client that performs the request-  ---  --   * 'E.Enumerator': Enumerator to create the request message stream-  -------------------------------------------------------------------------  askFor :: Client i o -> E.Enumerator o IO () -> IO ()-  askFor c enum = E.run_ (enum $$ itSend (cliSock c) (cliOut c))--  -------------------------------------------------------------------------  -- | Polling for a reply;-  --   the function polls for a server request.-  --   If nothing has been received, it returns 'Nothing';-  --   otherwise it returns 'Just' the result or an error.-  --   -  --   Parameters:-  ---  --   * 'Client': The client that performs the request-  ---  --   * 'E.Iteratee': Iteratee to process the reply message stream-  ---  --   The synchronous request (see 'request') -  --   could be implemented asynchronously like:-  ---  --   @-  --     rcv :: String -> IO ()-  --     rcv req = withContext 1 $ \\ctx -> do-  --       let ap = address l \"tcp:\/\/localhost:5555\" []-  --       withClient ctx ap -  --         (return . B.pack) (return . B.unpack) -  --         $ \\s -> do-  --           ei <- try $ askFor s (enum req)-  --           case ei of-  --             Left  e -> putStrLn $ \"Error: \" ++ show (e::SomeException)-  --             Right _ -> wait s-  --       -- check for results periodically -  --       where wait s = checkFor s outit >>= \\mbei ->-  --               case mbei of-  --                 Nothing        -> do putStrLn \"Waiting...\"-  --                                      threadDelay 10000 >> wait s-  --                 Just (Left e)  -> putStrLn $ \"Error: \" ++ show e-  --                 Just (Right _) -> putStrLn \"Ready!\"-  --   @-  -------------------------------------------------------------------------  checkFor :: Client i o -> E.Iteratee i IO a -> -              IO (Maybe (Either SomeException a))-  checkFor c it = Z.poll [Z.S (cliSock c) Z.In] 0 >>= \[s] ->-    case s of-      Z.S _ Z.In -> Just <$> rcvClient c it-      _          -> return Nothing--  -------------------------------------------------------------------------  -- The real working horse behind the scene-  -------------------------------------------------------------------------  rcvClient :: Client i o -> E.Iteratee i IO a -> IO (Either SomeException a)-  rcvClient c it = E.run (rcvEnum (cliSock c) (cliIn c) $$ it)--  -------------------------------------------------------------------------  -- | Publisher-  -------------------------------------------------------------------------  data Pub o = Pub {-                 pubCtx  :: Z.Context,-                 pubSock :: Z.Socket Z.Pub,-                 pubAdd  :: AccessPoint,-                 pubOut  :: OutBound o}--  -------------------------------------------------------------------------  -- | Obtaining the 'Z.Context' from 'Pub'-  -------------------------------------------------------------------------  pubContext :: Pub o -> Z.Context-  pubContext = pubCtx--  -------------------------------------------------------------------------  -- | Setting 'Z.SocketOption' to the underlying ZMQ 'Z.Socket'-  -------------------------------------------------------------------------  setPubOptions :: Pub o -> [Z.SocketOption] -> IO ()-  setPubOptions p = setSockOs (pubSock p)--  -------------------------------------------------------------------------  -- | Creates a publisher;-  --   A publisher is a data type-  --   that provides an interface to publish data to subscribers.-  --   'withPub' creates a publisher and invokes-  --   an application-defined action,-  --   which receives a 'Pub' argument.-  --   The lifetime of the publisher is limited to the action.-  --   When the action terminates, the publisher /dies/.-  ---  --   Parameter:-  ---  --   * 'Z.Context': The ZMQ Context-  ---  --   * 'AccessPoint': The access point the publisher will bind-  ---  --   * 'OutBound': A converter to convert from type /o/ -  --                 to the wire format 'B.ByteString'.-  --                 Note that a publisher may create-  --                 a data stream; the type /o/ is then-  --                 the type of one segment of this stream,-  --                 not of the stream as a whole.-  ---  --   * 'Pub' -> IO (): The action to invoke-  -------------------------------------------------------------------------  withPub :: Z.Context -> AccessPoint -> OutBound o -> -             (Pub o -> IO a) -> IO a-  withPub ctx ac oconv act = Z.withSocket ctx Z.Pub $ \s -> do-    Z.bind s (acAdd ac)-    act Pub {-          pubCtx  = ctx,-          pubSock = s,-          pubAdd  = ac,-          pubOut  = oconv}--  -------------------------------------------------------------------------  -- | Publishes the data stream created by an enumerator;-  ---  --   Parameters:-  ---  --   * 'Pub': The publisher-  ---  --   * 'E.Enumerator': The enumerator to create an outgoing -  --                     data stream.-  ---  --   A simple weather report publisher:-  ---  --   @-  --     withContext 1 $ \\ctx -> withPub ctx-  --         (Address \"tcp:\/\/*:5555\" [])-  --         (return . B.pack) $ \\pub -> untilInterrupt $ do-  --           issue pub (once weather noparam)-  --           threadDelay 10000 -- update every 10ms-  --   @-  ---  --   @-  --     -- fake weather report with some random values-  --     weather :: String -> IO (Maybe String)-  --     weather _ = do-  --         zipcode     <- randomRIO (10000, 99999) :: IO Int-  --         temperature <- randomRIO (-10, 30) :: IO Int-  --         humidity    <- randomRIO ( 10, 60) :: IO Int-  --         return $ Just (unwords [show zipcode, -  --                                 show temperature, -  --                                 show humidity])-  --   @-  -------------------------------------------------------------------------  issue :: Pub o -> E.Enumerator o IO () -> IO ()-  issue p enum = E.run_ (enum $$ itSend (pubSock p) (pubOut p))-             -  -------------------------------------------------------------------------  -- | Creates a background process that-  --   periodically publishes data;-  ---  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ Context-  ---  --   * 'String': Name of this Publisher; -  --               useful for debugging-  ---  --   * 'Parameter': The initial value of the control parameter-  ---  --   * 'Z.Timeout': The period of the publisher in microseconds;-  --                  the process will issue the publisher data -  --                  every n microseconds.-  ---  --   * 'AccessPoint': Bind address -  ---  --   * 'OutBound': A converter that converts one segment-  --                 of the data stream from type /o/-  --                 to the wire format 'B.ByteString'-  ---  --   * 'OnError_': Error Handler-  ---  --   * 'String' -> 'Fetch': 'E.Enumerator' to create-  --                          the outgoing data stream;-  --                          the string argument is the parameter.-  ---  --   * 'Service' -> IO (): The user action to perform-  ---  --   The weather report publisher introduced above (see 'withPub')-  --   can be implemented by means of 'withPeriodicPub' as:-  ---  --   @-  --     withPeriodicPub ctx \"Weather Report\" noparam -  --       100000 -- publish every 100ms-  --       (Address \"tcp:\/\/*:5555\" []) -  --       (return . B.pack) -- string converter-  --       onErr_            -- standard error handler-  --       (\\_ -> fetch1 fetch) -- creates one instance-  --                             -- of the return of \"fetch\";-  --                             -- see 'Enumerator' for details-  --       $ \\pub -> -  --         untilInterrupt $ do -- until /SIGINT/, see 'withServer' for details-  --           threadDelay 100000-  --           putStrLn $ \"I am doing nothing \" ++ srvName pub-  --   @-  -------------------------------------------------------------------------  withPeriodicPub :: Z.Context           -> -                     String -> Parameter ->-                     Z.Timeout           ->-                     AccessPoint         -> -                     OutBound o          ->-                     OnError_            ->-                     Fetch_ o            -> -                     (Service -> IO a)   -> IO a-  withPeriodicPub ctx name param period ac oconv onerr fetch =-    withService ctx name param service -    where service = publish period ac oconv onerr fetch--  -------------------------------------------------------------------------  -- PeriodicPub implementation-  -------------------------------------------------------------------------  publish :: Z.Timeout            ->-             AccessPoint          -> -             OutBound o           ->-             OnError_             ->-             Fetch_  o            -> -             Z.Context -> String  -> -             String -> String     -> IO () -> IO ()-  publish period ac oconv onerr -          fetch ctx name sockname param ready = -    Z.withSocket ctx Z.Pub $ \sock -> do-      Z.bind sock (acAdd ac)-      Z.withSocket ctx Z.Sub $ \cmd -> do-        conCmd cmd sockname ready-        periodicSend False period cmd (go sock) param-    `catch` (\e -> onerr Fatal e name param)-  -------------------------------------------------------------------------  -- do the job periodically-  -------------------------------------------------------------------------    where go sock p   = catch (body sock p) -                              (\e -> onerr Error e name p)-          body sock p = do-            eiR <- E.run (fetch ctx p () $$ itSend sock oconv)-            ifLeft eiR-              (\e -> onerr Error e name p)-              (\_ -> return ())--  -------------------------------------------------------------------------  -- | A subscription is a background service-  --   that receives and processes data streams-  --   from a publisher.-  --   A typical use case is an application-  --   that operates on periodically updated data;-  --   the subscriber would receive these data and-  --   and make them accessible to other threads in the process-  --   through an 'MVar'.-  --   -  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ Context-  ---  --   * 'String': The subscriber's name -  ---  --   * 'Parameter': The initial value of the control parameter-  ---  --   * ['Topic']:  The topics to subscribe to;-  --                 in the example above ('withPub'),-  --                 the publisher publishes the weather report-  --                 per zip code; the zip code, in this example,-  --                 could be a meaningful topic for a subscriber.-  --                 It is good practice to send the topic-  --                 in an initial message segment,-  --                 the envelope, to avoid that the subscriber-  --                 matches on some arbitrary part of the message.-  ---  --   * 'InBound': A converter that converts one segment-  --                of the incoming data stream to type /o/-  ---  --   * 'OnError_': Error handler-  ---  --   * 'Dump': 'E.Iteratee' to process the incoming data stream.-  ---  --   * 'Service' -> IO (): Application-defined action to control-  --                         the service. Note that 'Service' is-  --                         a thread-local resource and must not-  --                         be passed to threads forked from the action.-  ---  --  Weather Report Subscriber:-  --  -  --  @-  --     withContext 1 $ \\ctx -> -  --       withSub ctx \"Weather Report\" noparam -  --               [\"10001\"] -- zipcode to subscribe to-  --               (Address \"tcp:\/\/localhost:5555\" []) -  --               (return . B.unpack) -  --               onErr_ output -- Iteratee that just writes to stdout-  --               $ \\s -> untilInterrupt $ do-  --                 putStrLn $ \"Doing nothing \" ++ srvName s-  --                 threadDelay 1000000-  --  @-  -------------------------------------------------------------------------  withSub :: Z.Context              -> -             String                 -> -             Parameter              -> -             [Topic]                -> -             AccessPoint            -> -             InBound i -> OnError_  ->-             Dump i                 -> -             (Service -> IO a)      -> IO a-  withSub ctx name param sub ac iconv onErr dump =-    withService ctx name param service -    where service = subscribe sub ac iconv onErr dump--  subscribe :: [Topic]     -> -               AccessPoint -> -               InBound i   -> -               OnError_    -> -               Dump i      -> -               Z.Context   -> -               String      -> -               String -> Parameter -> IO () -> IO ()-  subscribe sub ac iconv onerr dump -            ctx name sockname param ready =-    Z.withSocket ctx Z.Sub $ \sock -> do-      trycon      sock (acAdd ac) retries-      mapM_ (Z.subscribe sock) sub-      Z.withSocket ctx Z.Sub $ \cmd -> do-        conCmd cmd sockname ready-        poll False [Z.S cmd Z.In, Z.S sock Z.In] (go sock) param-    `catch` (\e -> onerr Fatal e name param)-    where go sock p = do-            eiR <- E.run (rcvEnum sock iconv $$ dump ctx p)-            ifLeft eiR-              (\e -> onerr Error e name p)-              (\_ -> return ())--  -------------------------------------------------------------------------  -- | An alternative to the background subscriber (see 'withSub');-  -------------------------------------------------------------------------  data Sub i = Sub {-               subCtx   :: Z.Context,-               subSock  :: Z.Socket Z.Sub,-               subAdd   :: AccessPoint,-               subIn    :: InBound i}--  -------------------------------------------------------------------------  -- | Obtaining the 'Z.Context' from 'Sub'-  -------------------------------------------------------------------------  subContext :: Sub i -> Z.Context-  subContext = subCtx--  -------------------------------------------------------------------------  -- | Setting 'Z.SocketOption' to the underlying ZMQ 'Z.Socket'-  -------------------------------------------------------------------------  setSubOptions :: Sub i -> [Z.SocketOption] -> IO ()-  setSubOptions s = setSockOs (subSock s)--  -------------------------------------------------------------------------  -- | Similar to 'Pub', a 'Sub' is a data type-  --   that provides an interface to subscribe data.-  --   'withSporadicSub' creates a subscriber and invokes-  --   an application-defined action,-  --   which receives a 'Sub' argument.-  --   The lifetime of the subscriber is limited to the action.-  --   When the action terminates, the subscriber /dies/.-  -------------------------------------------------------------------------  withSporadicSub :: Z.Context -> AccessPoint -> InBound i -> [Topic] ->-                     (Sub i -> IO a) -> IO a-  withSporadicSub ctx ac iconv topics act = Z.withSocket ctx Z.Sub $ \s -> do-    trycon      s (acAdd ac) retries-    mapM_ (Z.subscribe s) topics-    act Sub {-          subCtx   = ctx,-          subSock  = s,-          subAdd   = ac,-          subIn    = iconv}--  -------------------------------------------------------------------------  -- | Polling for data;-  --   If nothing has been received, the function returns 'Nothing';-  --   otherwise it returns 'Just' the result or an error.-  --   -  --   Parameters:-  ---  --   * 'Sub': The subscriber-  ---  --   * 'E.Iteratee': Iteratee to process the data-  -------------------------------------------------------------------------  checkSub :: Sub i -> E.Iteratee i IO a -> -              IO (Maybe (Either SomeException a))-  checkSub s it = Z.poll [Z.S (subSock s) Z.In] 0 >>= \[p] ->-    case p of-      Z.S _ Z.In -> Just <$> rcvSub s it-      _          -> return Nothing--  -------------------------------------------------------------------------  -- | Waiting for data;-  --   the function blocks the current thread,-  --   until data are being received from the publisher.-  --   It returns either 'SomeException' or the result.-  --   -  --   Parameters:-  ---  --   * 'Sub': The subscriber-  ---  --   * 'E.Iteratee': Iteratee to process the data stream-  -------------------------------------------------------------------------  waitSub :: Sub i -> E.Iteratee i IO a -> IO (Either SomeException a)-  waitSub = rcvSub--  -------------------------------------------------------------------------  -- | Unsubscribe a topic-  -------------------------------------------------------------------------  unsubscribe :: Sub i -> Topic -> IO ()-  unsubscribe s = Z.unsubscribe (subSock s)--  -------------------------------------------------------------------------  -- | Subscribe another topic-  -------------------------------------------------------------------------  resubscribe :: Sub i -> Topic -> IO ()-  resubscribe s = Z.subscribe (subSock s)--  -------------------------------------------------------------------------  -- The working horse behind the scene-  -------------------------------------------------------------------------  rcvSub :: Sub i -> E.Iteratee i IO a -> IO (Either SomeException a)-  rcvSub s it = E.run (rcvEnum (subSock s) (subIn s) $$ it)--  -------------------------------------------------------------------------  -- | A puller is a background service -  --   that receives and processes data streams from a pipeline.-  --   -  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ Context-  ---  --   * 'String': The service name-  ---  --   * 'Parameter': The initial value of the control parameter-  ---  --   * 'AccessPoint': The address to connect to-  ---  --   * 'InBound': A converter to convert -  --                segments of the incoming data stream-  --                from the wire format 'B.ByteString'-  --                to the type /i/-  ---  --   * 'OnError_': Error Handler-  ---  --   * 'Dump': 'E.Iteratee' to process-  --              the incoming data stream-  ---  --   * 'Service' -> IO (): Application-defined action-  ---  --   A worker that just writes the incoming stream to /stdout/:-  ---  --   @-  --     withContext 1 $ \\ctx -> -  --       withPuller ctx \"Worker\" noparam -  --             (Address \"tcp:\/\/localhost:5555\" [])-  --             (return . B.unpack)-  --             onErr_ output-  --             $ \\s -> untilInterrupt $ do-  --               putStrLn \"Doing nothing \" ++ srvName s-  --               threadDelay 100000-  --   @-  -------------------------------------------------------------------------  withPuller :: Z.Context           ->-                String -> Parameter ->-                AccessPoint         ->-                InBound i           ->  -                OnError_            ->-                Dump   i            -> -                (Service -> IO a)   -> IO a-  withPuller ctx name param ac iconv onerr dump =-    withService ctx name param service -    where service = pull ac iconv onerr dump --  pull :: AccessPoint          ->-          InBound i            ->-          OnError_             ->-          Dump   i             ->-          Z.Context -> String  -> -          String    -> String  -> IO () -> IO ()-  pull ac iconv onerr dump ctx name sockname param ready = -    Z.withSocket ctx Z.Pull $ \sock -> do-      trycon sock (acAdd ac) retries-      Z.withSocket ctx Z.Sub $ \cmd -> do-        conCmd cmd sockname ready-        poll False [Z.S cmd Z.In, Z.S sock Z.In] (go sock) param-    `catch` (\e -> onerr Fatal e name param)-  -------------------------------------------------------------------------  -- do the job -  -------------------------------------------------------------------------    where go sock p = E.run_  (rcvEnum sock iconv $$ dump ctx p)-                      `catch` (\e -> onerr Error e name p)- -  -------------------------------------------------------------------------  -- | A pipeline consists of a \"pusher\" -  --   and a set of workers (\"pullers\").-  --   The pusher sends jobs down the pipeline that will be-  --   assigned to one of the workers. -  --   The pipeline pattern is, thus, a work-balancing scheme.-  -------------------------------------------------------------------------  data Pipe o = Pipe {-                  pipCtx  :: Z.Context,-                  pipSock :: Z.Socket Z.Push,-                  pipAdd  :: AccessPoint,-                  pipOut  :: OutBound o-                }--  -------------------------------------------------------------------------  -- | Obtaining the 'Z.Context' from 'Pipe'-  -------------------------------------------------------------------------  pipeContext :: Pipe o -> Z.Context-  pipeContext = pipCtx--  -------------------------------------------------------------------------  -- | Setting 'Z.SocketOption' to the underlying ZMQ 'Z.Socket'-  -------------------------------------------------------------------------  setPipeOptions :: Pipe o -> [Z.SocketOption] -> IO ()-  setPipeOptions p = setSockOs (pipSock p)--  -------------------------------------------------------------------------  -- | Creates a pipeline;-  --   a 'Pipe' is a data type -  --   that provides an interface to /push/ a data stream-  --   to workers connected to the other side of the pipe.-  --   'withPipe' creates a pipeline and invokes an application-defined-  --   action which receives a 'Pipe' argument.-  --   The lifetime of the 'Pipe' is limited to the action.-  --   When the action terminates, the 'Pipe' /dies/.-  ---  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ Context-  ---  --   * 'AccessPoint': The bind address-  ---  --   * 'OutBound': A converter to convert message segments-  --                 of type /o/ to the wire format 'B.ByteString'-  ---  --   * 'Pipe' -> IO (): The action to invoke-  -------------------------------------------------------------------------  withPipe :: Z.Context  -> AccessPoint -> -              OutBound o -> -              (Pipe o -> IO a)     -> IO a-  withPipe ctx ac oconv act = Z.withSocket ctx Z.Push $ \s -> do -    Z.bind s (acAdd ac)-    act Pipe {-        pipCtx  = ctx,-        pipSock = s,-        pipAdd  = ac,-        pipOut  = oconv}--  -------------------------------------------------------------------------  -- | Sends a job down the pipeline;-  --   -  --   Parameters:-  ---  --   * 'Pipe': The pipeline-  ---  --   * 'E.Enumerator': enumerator to create the data stream-  --                     that constitutes the /job/-  ---  --  A simple pusher:-  --  -  --  @-  --    sendF :: FilePath -> IO ()-  --    sendF f = withContext 1 $ \\ctx -> do-  --     let ap = Address \"tcp:\/\/*:5555\" []-  --     withPipe ctx ap return $ \\p ->-  --       push pu (EB.enumFile f) -- file enumerator-  --                               -- see Data.Enumerator.Binary (EB)-  --  @-  -------------------------------------------------------------------------  push :: Pipe o -> E.Enumerator o IO () -> IO () -  push p enum = E.run_ (enum $$ itSend (pipSock p) (pipOut p))--  -------------------------------------------------------------------------  -- | An Exclusive Pair is a general purpose pattern-  --   of two equal peers that communicate with each other-  --   by sending ('send') and receiving ('receive') data.-  --   One of the peers has to 'Z.bind' the 'AccessPoint'-  --   the other 'Z.connect's to it.-  -------------------------------------------------------------------------  data Peer a = Peer {-                  peeCtx  :: Z.Context,-                  peeSock :: Z.Socket Z.Pair,-                  peeAdd  :: AccessPoint,-                  peeIn   :: InBound a,-                  peeOut  :: OutBound a-                }--  -------------------------------------------------------------------------  -- | Obtains the 'Z.Context' from a 'Peer'-  -------------------------------------------------------------------------  peerContext :: Peer a -> Z.Context-  peerContext = peeCtx--  -------------------------------------------------------------------------  -- | Sets 'Z.SocketOption' -  -------------------------------------------------------------------------  setPeerOptions :: Peer a -> [Z.SocketOption] -> IO ()-  setPeerOptions p = setSockOs (peeSock p)--  -------------------------------------------------------------------------  -- | Creates a 'Peer';-  --   a peer is a data type -  --   that provides an interface to exchange data with another peer.-  --   'withPeer' creates the peer and invokes an application-defined-  --   action that receives a 'Peer' argument.-  --   The lifetime of the 'Peer' is limited to the action.-  --   When the action terminates, the 'Peer' /dies/.-  ---  --   Parameters:-  ---  --   * 'Z.Context': The ZMQ Context-  ---  --   * 'AccessPoint': The address, to which this peer either-  --                    binds or connects-  ---  --   * 'LinkType': One of the peers has to bind the address,-  --                 the other has to connect.-  ---  --   * 'InBound': A converter to convert message segments-  --                from the wire format 'B.ByteString' to type /i/-  ---  --   * 'OutBound': A converter to convert message segments-  --                 of type /o/ to the wire format 'B.ByteString'-  ---  --   * 'Peer' -> IO (): The action to invoke-  -------------------------------------------------------------------------  withPeer :: Z.Context -> AccessPoint -> LinkType ->-              InBound a -> OutBound a  ->-              (Peer a -> IO b) -> IO b-  withPeer ctx ac t iconv oconv act = Z.withSocket ctx Z.Pair $ \s -> -    link t ac s >> act Peer {-                         peeCtx  = ctx,-                         peeSock = s,-                         peeAdd  = ac,-                         peeIn   = iconv,-                         peeOut  = oconv}--  -------------------------------------------------------------------------  -- | Sends a data stream to another peer;-  ---  --   Parameters:-  --  -  --   * 'Peer': The peer-  ---  --   * 'E.Enumerator': Enumerator to create the outoing data stream-  -------------------------------------------------------------------------  send :: Peer o -> E.Enumerator o IO () -> IO ()-  send p enum = E.run_ (enum $$ itSend (peeSock p) (peeOut p))--  -------------------------------------------------------------------------  -- | Receives a data stream from another peer;-  ---  --   Parameters:-  --  -  --   * 'Peer': The peer-  ---  --   * 'E.Iteratee': Iteratee to process the incoming data stream-  -------------------------------------------------------------------------  receive :: Peer i -> E.Iteratee i IO a -> IO (Either SomeException a)-  receive p it = E.run (rcvEnum (peeSock p) (peeIn p) $$ it)--  -------------------------------------------------------------------------  -- connect to command socket-  -------------------------------------------------------------------------  conCmd :: Z.Socket Z.Sub -> String -> IO () -> IO ()-  conCmd cmd sockname ready = do-    trycon      cmd sockname retries-    Z.subscribe cmd ""-    ready+-- Portability: non-portable+-- +-- Basic communication patterns+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic (++                -- * Basic communication patterns+                -- $basic++                -- * Client Server+                -- $cliser++                module Network.Mom.Patterns.Basic.Client,+                module Network.Mom.Patterns.Basic.Server,++                -- * Pub Sub+                -- $pubsub++                module Network.Mom.Patterns.Basic.Publisher,+                module Network.Mom.Patterns.Basic.Subscriber,++                -- * Pipeline+                -- $pipe+                module Network.Mom.Patterns.Basic.Pusher,+                module Network.Mom.Patterns.Basic.Puller)++where++  import           Network.Mom.Patterns.Basic.Client+  import           Network.Mom.Patterns.Basic.Server+  import           Network.Mom.Patterns.Basic.Publisher+  import           Network.Mom.Patterns.Basic.Subscriber+  import           Network.Mom.Patterns.Basic.Pusher+  import           Network.Mom.Patterns.Basic.Puller++  {- $basic+      The Basic module provides the basic communication patterns+      defined in the zeromq library.+      The Basic module is just a convenience reexport of +      the single modules containing the patterns.+      The basic functionality is split into small modules,+      since usually one application module will only need+      one side of a communication pattern. For the case,+      an application needs several patterns at the same location,+      the basic module eases life providing a single import+      for all patterns.+  -}++  {- $cliser+     Clients request a service from a server.+     When a server receives a client request,+     it sends a response to this client.+     This implies that clients have to start the communication+     by sending a request.+     Usually, many clients connect to one server.+  -}++  {- $pubsub+     Publishers issue data under some topic,+     to which interested parties can subscribe.+     Subscribers do not need to request data explicitly,+     they receive the newest updates when available,+     once they have subscribed to the topic.+     Usually, many subscribers connect to one publisher.+  -}++  {- $pipe+     Pushers and pullers form a pipeline+     to send work packages downstream;+     pushers can only send data,+     which represent the work to be done,+     pullers can only receive data.+     Usually, the pusher binds the address,+     to which many pullers connect.+  -}+

@@ -0,0 +1,92 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Basic/Client.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: portable+-- +-- Client side of Client/Server+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic.Client (+         Client, clService, withClient, request)+where++  import qualified System.ZMQ            as Z+  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams++  ------------------------------------------------------------------------+  -- | Client data type+  ------------------------------------------------------------------------+  data Client = Client {clSock    :: Z.Socket Z.Req,+                        clService :: Service} -- currently unused++  ------------------------------------------------------------------------+  -- | Create a client+  --   with name 'Service',+  --   linking to address 'String',+  --   connecting or binding the address according to 'LinkType'+  --   and finally entering the action, in whose scope+  --   the client lives.+  ------------------------------------------------------------------------+  withClient :: Context          ->+                Service          -> +                String           ->+                LinkType         ->+                (Client -> IO a) -> IO a+  withClient ctx srv add lt act =+    Z.withSocket ctx Z.Req $ \s -> do+      link lt s add []+      act $ Client s srv++  ------------------------------------------------------------------------+  -- | Request a service:+  --+  --   * 'Client' - The client, through which the service is requested+  --+  --   * 'Timeout' - Timeout in microseconds, -1 to wait eternally.+  --                 With timeout = 0, the function returns immediately+  --                 with 'Nothing'.+  --                 When the timeout expires, request is abandoned. +  --                 In this case, the result of the request+  --                 is Nothing.+  --               +  --   * 'Source'  - The source of the request stream;+  --                 the format of the request will probably comply+  --                 with some communication protocol,+  --                 as, for instance, in the majordomo pattern.+  --+  --   * 'SinkR'   - The sink receiving the reply. The result of the sink+  --                 is returned as the request's overall result.+  --                 Note that the sink may perform different +  --                 actions on the segments of the resulting stream,+  --                 /e.g./ storing data in a database,+  --                 and return the number of records received.+  --+  -- A \'hello world\' Example:+  --+  -- >  import qualified Data.Conduit          as C+  -- >  import qualified Data.ByteString.Char8 as B+  -- >  import           Network.Mom.Patterns.Basic.Client+  -- >  import           Network.Mom.Patterns.Types+  --+  -- >  main :: IO ()+  -- >  main = withContext 1 $ \ctx -> +  -- >           withClient ctx "test" +  -- >               "tcp://localhost:5555" Connect $ \c -> do+  -- >             mbX <- request c (-1) src snk+  -- >             case mbX of+  -- >               Nothing -> putStrLn "No Result"+  -- >               Just x  -> putStrLn $ "Result: " ++ x+  -- >    where src = C.yield (B.pack "hello world")+  -- >          snk = do mbX <- C.await +  -- >                   case mbX of+  -- >                     Nothing -> return Nothing+  -- >                     Just x  -> return $ Just $ B.unpack x+  ------------------------------------------------------------------------+  request :: Client -> Timeout -> Source -> SinkR (Maybe a) -> IO (Maybe a)+  request c tmo src snk = do+    runSender   (clSock c) src+    if tmo /= 0 then runReceiver (clSock c) tmo snk+                else return Nothing

@@ -0,0 +1,110 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Basic/Publisher.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Publish side of 'Publish/Subscribe'+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic.Publisher (+                 +                     -- * Publisher++                     Pub, withPub, issue, ++                     -- * Forwarder++                     withForwarder+                   )+where++  import qualified Data.ByteString.Char8 as B+  import           Data.List (intercalate)+  import qualified System.ZMQ            as Z++  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams++  ------------------------------------------------------------------------+  -- | Publisher data type+  ------------------------------------------------------------------------+  newtype Pub = Pub {pubSock :: Z.Socket Z.Pub}++  ------------------------------------------------------------------------+  -- | Create and link a publisher:+  --+  --   * 'Context'     -  The zeromq context+  --+  --   * 'String'      -  The service address+  --+  --   * 'LinkType'    -  How to link (bind or connect)+  --+  --   * (Pub -> IO a) -  The action, in whose scope the publisher lives+  ------------------------------------------------------------------------+  withPub :: Context       ->+             String        -> +             LinkType      ->+             (Pub -> IO a) -> IO a+  withPub ctx add lt act = +    Z.withSocket ctx Z.Pub $ \s -> +      link lt s add [] >> act (Pub s)++  ------------------------------------------------------------------------+  -- | Publish data:+  --+  --   * 'Pub'       - The publisher+  --+  --   * ['Service'] - List of topics, to which these data should be+  --                   published+  --+  --   * 'Source'    - Create the stream to publish.+  --                   The first message segment+  --                   contains the subscription header,+  --                   /i.e./ the comma-separated list of topics+  ------------------------------------------------------------------------+  issue :: Pub -> [Service] -> Source -> IO ()+  issue p topics src = runSender (pubSock p) pubSrc+    where pubSrc = let ts = B.pack $ intercalate "," topics+                    in streamList [ts] >> src++  ------------------------------------------------------------------------+  -- | A simple forwarder,+  --   /i.e./ a device that connects to a publisher+  --   and provides an additional endpoint +  --   for more subscribers to connect to.+  --   A forwarder, hence, is a means to extend +  --   the capacity of a publisher.+  --+  --   * 'Context'            - The zeromq context+  --+  --   * 'Service'            - The name of the forwarder+  --+  --   * (String, 'LinkType') - access point for subscribers;+  --                            usually, you want to bind+  --                            the address, such that subscribers+  --                            connect to it.+  --+  --   * (String, 'LinkType') - access point for the publisher;+  --                            usually, you want to connect +  --                            to the publisher.+  --+  --   * 'OnError_'           - Error handler+  --+  --   * 'Control' a          - Control loop+  ------------------------------------------------------------------------+  withForwarder :: Context            ->+                   Service            ->+                   [Service]          ->+                   (String, LinkType) ->  -- subscribers+                   (String, LinkType) ->  -- publishers+                   OnError_           ->+                   Control a          -> IO a+  withForwarder ctx srv topics (pub, pubt)+                               (sub, subt) onErr =+    withStreams ctx srv (-1)+                [Poll "sub" sub SubT subt topics [],+                 Poll "pub" pub PubT pubt []     []]+                (\_ -> return ()) onErr job+    where job s = passAll s ["pub"]

@@ -0,0 +1,94 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Basic/Puller.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Puller side of \'Pipeline\'+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic.Puller (+          -- * Puller++          withPuller, ++          -- * Pipeline+ +          withPipe)+where++  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams++  ------------------------------------------------------------------------+  -- | Start a puller as a background service:+  --+  --   * 'Context'   - The zeromq context+  --+  --   * 'Service'   - Service name of this worker+  --+  --   * 'String'    - The address to link to+  --+  --   * 'LinkType'  - Whether to connect to or to bind the address;+  --                   usually you want to connect many workers+  --                   to one pusher+  --   +  --   * 'OnError_'  - Error handler+  --  +  --   * 'Sink'      - The application-defined sink+  --                   that does the job sent down the pipeline+  --+  --   * 'Control' a - Control loop+  ------------------------------------------------------------------------+  withPuller :: Context              ->+                Service              -> +                String               ->+                LinkType             ->+                OnError_             ->+                Sink                 ->+                (Controller -> IO a) -> IO a+  withPuller ctx srv add lt onErr snk =+    withStreams ctx srv (-1) +                [Poll "pusher" add PullT lt [] []]+                (\_ -> return())+                onErr+                (\_ -> snk)++  ------------------------------------------------------------------------+  -- | A pipeline extends the capacity of the +  --   pusher-puller chain;+  --   a pipeline connects to a pusher+  --   and provides an access point to a set of pullers.+  --  +  --   * 'Context'            - The zeromq context+  --   +  --   * 'Service'            - The service name of this queue+  --+  --   * (String, 'LinkType') - Address and link type, to where pullers+  --                            connect. Note: if pullers connect,+  --                            the pipeline must bind the address!+  --+  --   * (String, 'LinkType') - Address and link type that pushers bind.+  --                            Note, again, that +  --                            if pusher bind, the pipeline must+  --                            connect to the address!+  --+  --   * 'OnError_'           - Error handler+  --+  --   * 'Control' a          - 'Controller' action+  ------------------------------------------------------------------------+  withPipe :: Context              ->+              Service              ->+              (String, LinkType)   ->  -- for pullers +              (String, LinkType)   ->  -- for pushers+              OnError_             ->+              (Controller -> IO a) -> IO a+  withPipe ctx srv (pus, pust)+                   (pul, pult) onErr =+    withStreams ctx srv (-1)+                [Poll "pusher" pus PipeT pust [] [],+                 Poll "puller" pul PullT pult [] []]+                (\_ -> return ()) onErr job+    where job s = passAll s ["pusher"]+

@@ -0,0 +1,50 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Basic/Pusher.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Pusher side of \'Pipeline\'+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic.Pusher (+                     Pusher, withPusher, push)+where++  import qualified System.ZMQ            as Z+  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams++  ------------------------------------------------------------------------+  -- | The pusher data type+  ------------------------------------------------------------------------+  newtype Pusher = Pusher {pipSock :: Z.Socket Z.Push}++  ------------------------------------------------------------------------+  -- | The function in whose scope the pusher lives:+  --+  --   * 'Context'          - The zeromq Context+  --+  --   * 'String'           - The address+  --+  --   * 'LinkType'         - Link type; usually, you want to bind+  --                          a pusher to its address+  --+  --   * ('Pusher' -> IO a) - Action in whose scope the pusher lives+  ------------------------------------------------------------------------+  withPusher :: Context          ->+                String           -> +                LinkType         ->+                (Pusher -> IO a) -> IO a+  withPusher ctx add lt act = +    Z.withSocket ctx Z.Push $ \s -> link lt s add [] >> act (Pusher s)++  ------------------------------------------------------------------------+  -- | Push a job down the pipeline;+  --   the 'Source' creates the outgoing stream.+  ------------------------------------------------------------------------+  push :: Pusher -> Source -> IO ()+  push p = runSender (pipSock p) ++

@@ -0,0 +1,115 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Basic/Server.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Server side of \'Client\/Server\'+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic.Server (++         -- * Server++         withServer, ++         -- * Queue+         +         withQueue)+where++  import Data.Conduit ((=$))+  import Network.Mom.Patterns.Types+  import Network.Mom.Patterns.Streams++  ------------------------------------------------------------------------+  -- | Start a server as a background process+  -- +  --   * 'Context'   - The zeromq context+  --  +  --   * 'Service'   - Service name+  --+  --   * 'String'    - The address to link to+  --+  --   * 'LinkType'  - Whether to connect to or to bind the address+  --   +  --   * 'OnError_'  - Error handler+  --  +  --   * 'Conduit_'  - The application-defined stream transformer;+  --                   the conduit receives the request as input stream+  --                   and should create the output stream that is+  --                   internally sent back to the client+  --+  --   * 'Control' a - Control action+  --+  -- A very simple example, which just sends the incoming stream+  -- back to the client ('bounce'):+  --+  -- >  import           Control.Monad (forever)+  -- >  import           Control.Concurrent+  -- >  import           Network.Mom.Patterns.Basic.Server+  -- >  import           Network.Mom.Patterns.Types+  --+  -- >  main :: IO ()+  -- >  main = withContext 1 $ \ctx -> +  -- >             withServer ctx "Bouncer" "tcp://*:5555" Bind+  -- >                        (\_ _ _ -> return ()) -- ignore error+  -- >                        bounce $ \_ -> forever $ threadDelay 100000+  -- >    where bounce = passThrough+  ------------------------------------------------------------------------+  withServer :: Context    ->+                Service    -> +                String     ->+                LinkType   ->+                OnError_   ->+                Conduit_   ->+                Control a  -> IO a+  withServer ctx srv add lt onErr serve =+    withStreams ctx srv (-1) +                [Poll "client" add DealerT lt [] []]+                igTmo+                onErr+                job +    where job s   = serve =$ passAll s ["client"]+          igTmo _ = return ()++  ------------------------------------------------------------------------+  -- | A simple load balancer device to link clients and servers.+  --+  --   * 'Context'            - The zeromq context+  --   +  --   * 'Service'            - The service name of this queue+  --+  --   * (String, 'LinkType') - Address and link type, to where clients+  --                            connect. Note if clients connect,+  --                            the queue must bind the address!+  --+  --   * (String, 'LinkType') - Address and link type, to where servers+  --                            connect. Note, again, that +  --                            if servers connect, the queue must+  --                            bind the address!+  --+  --   * 'OnError_'           - Error handler+  --+  --   * 'Control' a          - 'Controller' action+  ------------------------------------------------------------------------+  withQueue :: Context            ->+               Service            ->+               (String, LinkType) ->+               (String, LinkType) ->+               OnError_           ->+               Control a          -> IO a+  withQueue ctx srv (rout, routl)+                    (deal, deall) onErr =+    withStreams ctx srv (-1) +                [Poll "client" rout RouterT routl [] [],+                 Poll "server" deal DealerT deall [] []]+                onTmo+                onErr+                job+    where job s = let target | getSource s == "client" = "server"+                             | otherwise               = "client"+                   in passAll s [target]+          onTmo _ = return ()+

@@ -0,0 +1,84 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Basic/Subscriber.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Subscriber side of \'Publish Subscribe\'+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Basic.Subscriber (+         Sub, withSub, subscribe, checkSub)+         +where++  import qualified Data.Conduit          as C+  import qualified System.ZMQ            as Z+  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams++  ------------------------------------------------------------------------+  -- | Subscription data type+  ------------------------------------------------------------------------+  newtype Sub = Sub {subSock :: Z.Socket Z.Sub}++  ------------------------------------------------------------------------+  -- | Create a subscription and start the action, in which it lives+  --+  --   * 'Context'       - The zeromq context+  --+  --   * 'String'        - The address +  --+  --   * 'LinkType'      - The link type, usually Connect+  --+  --   * ('Sub' -> IO a) - The action, in which the subscription lives+  ------------------------------------------------------------------------+  withSub :: Context       ->+             String        -> +             LinkType      ->+             (Sub -> IO a) -> IO a+  withSub ctx add lt act = +    Z.withSocket ctx Z.Sub $ \s -> +      link lt s add [] >> act (Sub s)++  ------------------------------------------------------------------------+  -- | Subscribe to a list of topics;+  --   Note that a subscriber has to subscribe to at least one topic+  --   to receive any data.+  --+  --   * 'Sub'       - The subscriber+  --+  --   * ['Service'] - The list of topics to subscribe to+  ------------------------------------------------------------------------+  subscribe :: Sub -> [Service] -> IO ()+  subscribe s = mapM_ (Z.subscribe $ subSock s) ++  ------------------------------------------------------------------------+  -- | Check for new data:+  --+  --   * 'Sub'     - The subscriber+  --+  --   * 'Timeout' - When timeout expires,+  --                 the function returns 'Nothing'.+  --                 Timeout may be +  --                 -1  - listen eternally,+  --                 0   - return immediately,+  --                 \> 0 - timeout in microseconds+  --+  --   * 'SinkR'   - Sink the result stream.+  --                 Note that the subscription header,+  --                 /i.e./ a message segment containing+  --                        a comma-separated list +  --                        of the topics, to which+  --                        the data belong,+  --                 is dropped.+  ------------------------------------------------------------------------+  checkSub :: Sub -> Timeout -> SinkR (Maybe a) -> IO (Maybe a)+  checkSub s tmo snk = runReceiver (subSock s) tmo subSnk+    where subSnk = do+            mb <- C.await -- subscription header is filtered out!+            case mb of+              Nothing -> return Nothing+              Just _  -> snk+

@@ -0,0 +1,102 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Broker.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Majordomo Service Broker +-------------------------------------------------------------------------------+module Network.Mom.Patterns.Broker (++         -- * The Majordomo Pattern+         -- $majordomo+        +         -- * Majordomo \'Worker\'+         -- $worker++         module Network.Mom.Patterns.Broker.Server,++         -- * Majordomo \'Client\'+         -- $client++         module Network.Mom.Patterns.Broker.Client,++         -- * Majordomo \'Broker\'+         -- $broker++         module Network.Mom.Patterns.Broker.Broker,++         -- * Majordomo common definitions+         -- $common++         module Network.Mom.Patterns.Broker.Common)+where++  import Network.Mom.Patterns.Broker.Server+  import Network.Mom.Patterns.Broker.Client+  import Network.Mom.Patterns.Broker.Broker+  import Network.Mom.Patterns.Broker.Common++  {- $majordomo+     The Majordomo pattern+     defines a communication protocol+     between clients and servers based+     on a service broker that provides++     * A central access point for all services++     * A service infrastructure for services like+       service discovery, configurations, data storage, /etc./++     * Reliable communication based on heart-beats.++     The broker package does not provide a pre-defined borker,+     /i.e./ it does not contain an application.+     Instead, it provides /API/s that can be used+     to build brokers, even within the same process+     as clients and workers. ++     The present module provides a convenience reexport of the three modules,+     making up the majordomo pattern: server, client and broker.+     There is also a \"common\" module with definitions relevant+     for all three parties. +  -}++  {- $worker+     The server module contains the server-side of the majordomo pattern.+     Majordomo servers look like ordinary servers,+     but use a different protocol for communicating +     with the broker intervening between clients and servers.+  -}++  {- $client+     The client module contains the client-side of the majordomo pattern.+     Majordomo clients look very similar to ordinary clients,+     but use a different protocol for communicating +     with the broker intervening between clients and servers.+     The code provided to a majordomo clients, however,+     is exactly the same as the code provided to an ordinary client.+  -}++  {- $broker+     The broker module contains the broker code.+     The broker is implemented by means of streams,+     very similar to servers and clients.+     The functionality provided by the broker module+     can be used to implement a standalone broker+     or an in-process broker linked with its clients and servers.+     Note, however, +     that brokers must not share the same process+     with other brokers, since, internally,+     the broker module uses a global registry,+     which would be shared among all brokers +     in the same process as well. +  -}++  {- $common+     The \"common\" module contains +     protocol building blocks and+     MDP-specific exceptions.+  -}

@@ -0,0 +1,172 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Broker/Broker.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Majordomo Broker+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Broker.Broker (withBroker)+where++  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams+  import           Network.Mom.Patterns.Broker.Common++  import qualified Registry  as R+  import           Heartbeat (hbPeriodReached)     ++  import qualified Data.ByteString        as B+  import qualified Data.ByteString.Char8  as BC+  import qualified Data.Conduit           as C+  import           Data.Conduit ((=$), ($$))+  import           Data.Time.Clock++  import           Control.Applicative ((<$>))+  import           Control.Monad.Trans (liftIO)+  import           Control.Monad       (when)+  import           Prelude hiding (catch)+  import           Control.Exception (throwIO, finally)+  import           Control.Concurrent.MVar++  import           System.IO.Unsafe++  ------------------------------------------------------------------------+  -- | Start a broker as a background process+  -- +  --   * 'Context'   - The zeromq context+  --  +  --   * 'Service'   - Service name -+  --                   the service name is for debugging only,+  --                   there is no relation whatsoever+  --                   to the service of the Majordomo Protocol.+  --+  --   * 'Msec'      - The heartbeat interval in milliseconds,+  --                   which should be equal +  --                   for all workers and the broker +  --+  --   * 'String'    - The address clients connect to+  --+  --   * 'String'    - The address servers connect to+  --+  --   * 'OnError_'  - Error handler+  --  +  --   * 'Control' a - Control action+  ------------------------------------------------------------------------+  withBroker :: Context  -> Service -> Msec -> String -> String -> +                OnError_ -> (Controller -> IO r)                -> IO r+  withBroker ctx srv tmo aClients aServers onerr ctrl | tmo <= 0  = +    throwIO $ MDPExc "Heartbeat is mandatory"+                                                      | otherwise = +    lockBroker $ \_ -> do+      R.clean+      R.setHbPeriod tmo+      t <- getCurrentTime+      m <- newMVar t+      withStreams ctx srv (1000 * fromIntegral tmo)+                  [Poll "servers" aServers RouterT Bind [] [],+                  Poll "clients" aClients RouterT Bind [] []]+                  (handleTmo    m tmo) onerr +                  (handleStream m tmo) ctrl ++  -- handle heartbeat -------------------------------------------------------+  handleTmo :: MVar UTCTime -> Msec -> Streamer -> IO ()+  handleTmo m tmo s = hbPeriodReached m tmo >>= \x -> +                         when x $ R.checkWorker >>= mapM_ sndHb+    where hbM   i   = [i, B.empty, mdpW01, xHeartBeat]+          sndHb i   = C.runResourceT $ streamList (hbM i) $$  +                                       passAll s ["servers"]+ +  ------------------------------------------------------------------------+  -- Handle incoming Streams+  ------------------------------------------------------------------------+  handleStream :: MVar UTCTime -> Msec -> StreamSink+  handleStream m x s = +    let action | getSource s == "clients" = recvClient s+               | getSource s == "servers" = recvWorker s+               | otherwise                = return ()+        -- and handle heartbeat afterwards--------------------------------+     in action >> liftIO (handleTmo m x s)++  ------------------------------------------------------------------------+  -- Receive stream from client+  ------------------------------------------------------------------------+  recvClient :: StreamSink+  recvClient s = mdpCRcvReq >>= uncurry go+    where go i sn | hdr sn == mmiHdr = handleMMI i sn+                  | otherwise        = handleReq i sn+          handleReq i sn = do+            mbW <- liftIO $ R.getWorker sn+            case mbW of+              Nothing -> noWorker sn+              Just w  -> sendRequest w [i] s+          handleMMI i sn | srvc sn /= mmiSrv = C.yield mmiNimpl =$ +                                                 sendReply sn [i] s+                         | otherwise = do+            mbX <- C.await+            case mbX of+              Nothing -> liftIO (throwIO $ MMIExc +                                   "No ServiceName in mmi.service request")+              Just x  -> do+                m <- bool2MMI <$> liftIO (R.lookupService x)+                C.yield m =$ sendReply sn [i] s+          bool2MMI True  = mmiFound+          bool2MMI False = mmiNotFound+          hdr            = B.take 4+          srvc           = B.drop 4+          noWorker sn    = liftIO (throwIO $ BrokerExc $+                                  "No Worker for service " ++ BC.unpack sn)++  ------------------------------------------------------------------------+  -- Receive stream from worker +  ------------------------------------------------------------------------+  recvWorker :: StreamSink +  recvWorker s = do+    f <- mdpWRcvRep+    case f of+      WBeat  w    -> liftIO (R.updWorkerHb w) -- update his heartbeat +      WReady w sn -> liftIO $ R.insert w sn   -- insert new worker+      WReply w is -> handleReply w is s       -- handle reply from worker+      WDisc  w    -> liftIO $ R.remove w      -- disconnect from worker+      _           -> liftIO (throwIO $ Ouch "Unexpected Frame from Worker!")++  ------------------------------------------------------------------------+  -- Handle reply+  ------------------------------------------------------------------------+  handleReply :: Identity -> [Identity] -> StreamSink+  handleReply w is s = do+    mbS <- liftIO $ R.getServiceName w+    case mbS of+      Nothing -> liftIO (throwIO $ ServerExc "Unknown Worker")+      Just sn -> sendReply sn is s >> liftIO (R.freeWorker w)++  ------------------------------------------------------------------------+  -- Send request to worker+  ------------------------------------------------------------------------+  sendRequest :: Identity -> [Identity] -> StreamSink+  sendRequest w is s = mdpWSndReq w is =$ passAll s ["servers"]++  ------------------------------------------------------------------------+  -- Send reply to client+  ------------------------------------------------------------------------+  sendReply :: B.ByteString -> [Identity] -> StreamSink+  sendReply sn is s = mdpCSndRep sn is =$ passAll s ["clients"]++  ------------------------------------------------------------------------+  -- Mechanism to exclude two brokers +  -- from running in the same process+  ------------------------------------------------------------------------+  {-# NOINLINE _brk #-}+  _brk :: MVar () +  _brk = unsafePerformIO $ newMVar ()++  lockBroker :: (() -> IO b) -> IO b+  lockBroker act = do+    mb_ <- tryTakeMVar _brk -- withMVar _brk +    case mb_ of+      Nothing -> throwIO $ SingleBrokerExc "Another broker is running!"+      Just _  -> finally (act ()) (putMVar _brk ())++

@@ -0,0 +1,139 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Broker/Client.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Majordomo Client+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Broker.Client (+                   Client, withClient, checkService,+                   request)+where++  import qualified Data.ByteString.Char8  as B++  import           Control.Monad.Trans (liftIO)+  import           Control.Exception   (throwIO)+  import           Data.Conduit (($=), (=$))+  import qualified Data.Conduit          as C+  import qualified System.ZMQ as Z++  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams+  import           Network.Mom.Patterns.Broker.Common ++  -------------------------------------------------------------------------+  -- | Client data type+  -------------------------------------------------------------------------+  data Client = Client {clSock    :: Z.Socket Z.XReq,+                        clService :: Service}++  ------------------------------------------------------------------------+  -- | Create a client and start the action, +  --   in whose scope the client lives;++  --   * 'Context'        - The zeromq context+  --+  --   * 'Service'        - The client service name;+  --                        the client will be specialised+  --                        for requesting this service.+  --+  --   * 'String'         - The address to connect to.+  --+  --   * 'LinkType'       - This parameter is ignored;+  --                        the majordomo protocol expects clients+  --                        to connect to the broker.+  --+  --   * 'Client' -> IO a - The action loop+  ------------------------------------------------------------------------+  withClient :: Context          ->+                Service          -> +                String           ->+                LinkType         ->+                (Client -> IO a) -> IO a+  withClient ctx srv add _ act =+    Z.withSocket ctx Z.XReq $ \s -> do+      link Connect s add []+      act $ Client s srv++  ------------------------------------------------------------------------+  -- | Service discovery:+  --   The function checks whether the client's service +  --   is provided by the broker.+  --   +  --   Return values:+  --+  --   * Nothing: The broker timed out+  --+  --   * Just False: The service is not available+  --+  --   * Just True: The service is available+  ------------------------------------------------------------------------+  checkService :: Client -> Timeout -> IO (Maybe Bool)+  checkService c tmo = do+    runSender   (clSock c) $ mdpSrc mmi (C.yield $ B.pack $ clService c)+    runReceiver (clSock c) tmo $ mdpSnk mmi mmiResponse +    where mmi = B.unpack $ mmiHdr `B.append` mmiSrv+          mmiResponse = do+            mbX <- C.await+            case mbX of+              Nothing -> liftIO (putStrLn "Timeout!") >> return Nothing+              Just x  | x == mmiFound    -> return $ Just True+                      | x == mmiNotFound -> return $ Just False+                      | x == mmiNimpl    -> liftIO (throwIO $+                          MMIExc "MMI Service Request not available")+                      | otherwise        -> liftIO (throwIO $+                          MMIExc $ "Unexpected response code "  +++                                   "from mmi.service request: " +++                                        B.unpack x)+                        +  ------------------------------------------------------------------------+  -- | Request a service:+  --+  --   * 'Client' - The client, through which the service is requested+  --+  --   * 'Timeout' - Timeout in microseconds, -1 to wait eternally.+  --                 With timeout = 0, the function returns immediately+  --                 with 'Nothing'.+  --                 When the timeout expires, request is abandoned. +  --                 In this case, the result of the request+  --                 is Nothing.+  --               +  --   * 'Source'  - The source of the request stream;+  --                 the format of the request will probably comply+  --                 with some communication protocol,+  --                 as, for instance, in the majordomo pattern.+  --+  --   * 'SinkR'   - The sink receiving the reply. The result of the sink+  --                 is returned as the request's overall result.+  --                 Note that the sink may perform different +  --                 actions on the segments of the resulting stream,+  --                 /e.g./ storing data in a database,+  --                 and return the number of records received.+  ------------------------------------------------------------------------+  request :: Client -> Timeout -> Source -> SinkR (Maybe a) -> IO (Maybe a)+  request c tmo src snk = do+    runSender   (clSock c)     $ mdpSrc (clService c) src+    runReceiver (clSock c) tmo $ mdpSnk (clService c) snk++  {- currently not provided+  checkReceive :: Client -> Timeout -> SinkR (Maybe a) -> IO (Maybe a)+  checkReceive c tmo snk = +    runReceiver (clSock c) tmo $ mdpSnk (clService c) snk+  -}+    +  ------------------------------------------------------------------------+  -- Add MDP headers to a stream created by source+  ------------------------------------------------------------------------+  mdpSrc :: Service -> Source -> Source+  mdpSrc sn src = src $= mdpCSndReq sn ++  ------------------------------------------------------------------------+  -- Parse and remove the MDP headers +  -- before the stream is passed over to a sink+  ------------------------------------------------------------------------+  mdpSnk :: Service -> SinkR (Maybe a) -> SinkR (Maybe a)+  mdpSnk sn snk = mdpCRcvRep sn =$ snk

@@ -0,0 +1,298 @@+{-# LANGUAGE DeriveDataTypeable,RankNTypes #-}+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Broker/Common.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Majordomo common definitions+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Broker.Common+where++  import           Network.Mom.Patterns.Types++  import qualified Data.ByteString.Char8  as B+  import qualified Data.ByteString        as BB+  import qualified Data.Conduit as C+  import           Control.Monad (unless)+  import           Control.Monad.Trans (liftIO)+  import           Control.Exception (Exception, throwIO) +  import           Data.Typeable (Typeable)++  import           Control.Applicative ((<$>))++  ------------------------------------------------------------------------+  -- | Majordomo protocol client/worker version 1+  ------------------------------------------------------------------------+  mdpC01, mdpW01 :: B.ByteString+  mdpC01 = B.pack "MDPC01"+  mdpW01 = B.pack "MDPW01"++  ------------------------------------------------------------------------+  -- | Message types (ready, request, reply, heartbeat, disconnect)+  ------------------------------------------------------------------------+  xReady, xRequest, xReply, xHeartBeat, xDisc :: B.ByteString+  xReady     = BB.pack [0x01]+  xRequest   = BB.pack [0x02]+  xReply     = BB.pack [0x03]+  xHeartBeat = BB.pack [0x04]+  xDisc      = BB.pack [0x05]++  ------------------------------------------------------------------------+  -- | Service name +  ------------------------------------------------------------------------+  type ServiceName = String++  ------------------------------------------------------------------------+  -- | Majordomo Management Interface (MMI) -+  --   \"mmi.service\" +  ------------------------------------------------------------------------+  mmiHdr, mmiSrv :: B.ByteString +  mmiHdr = B.pack "mmi."+  mmiSrv = B.pack "service"+  +  ------------------------------------------------------------------------+  -- | Majordomo Management Interface -- responses:+  --   Found (\"200\"), NotFound (\"404\"), NotImplemented (\"501\")+  ------------------------------------------------------------------------+  mmiFound, mmiNotFound, mmiNimpl :: B.ByteString+  mmiFound    = B.pack "200"+  mmiNotFound = B.pack "404"+  mmiNimpl    = B.pack "501"++  ------------------------------------------------------------------------+  -- | Client -> Broker: send request+  ------------------------------------------------------------------------+  mdpCSndReq :: ServiceName -> Conduit B.ByteString ()+  mdpCSndReq sn = mapM_ C.yield [B.empty, mdpC01, B.pack sn] >> passThrough++  ------------------------------------------------------------------------+  -- | Client -> Broker: receive request+  ------------------------------------------------------------------------+  mdpCRcvReq :: Conduit o (Identity, B.ByteString)+  mdpCRcvReq = do i <- identity+                  protocol+                  sn <- getChunk +                  return (i, sn)+    where  protocol = chunk mdpC01 "Unknown Protocol - expected Client 0.1"++  ------------------------------------------------------------------------+  -- | Broker -> Client: send reply+  ------------------------------------------------------------------------+  mdpCSndRep :: B.ByteString -> [Identity] -> Conduit B.ByteString ()+  mdpCSndRep sn is = mapM_ C.yield hdr >> passThrough+    where hdr = toIs is ++ [mdpC01, sn]++  ------------------------------------------------------------------------+  -- | Broker -> Client: receive reply+  ------------------------------------------------------------------------+  mdpCRcvRep :: ServiceName -> Conduit B.ByteString ()+  mdpCRcvRep sn = empty >> protocol >> serviceName >> passThrough+    where protocol    = chunk mdpC01 "Unknown Protocol - expected Client 0.1"+          serviceName = chunk (B.pack sn) ("Wrong service - expected: " ++ sn)++  ------------------------------------------------------------------------+  -- | Broker -> Server: send request +  ------------------------------------------------------------------------+  mdpWSndReq :: Identity -> [Identity] -> Conduit B.ByteString ()+  mdpWSndReq w is = mapM_ C.yield hdr >> passThrough+    where hdr = [w, B.empty, mdpW01, xRequest] ++ toIs is ++ [B.empty]++  ------------------------------------------------------------------------+  -- | Broker -> Server: receive request +  ------------------------------------------------------------------------+  mdpWRcvReq :: Conduit o WFrame+  mdpWRcvReq = do +    empty >> protocol+    t <- frameType+    case t of+      HeartBeatT  -> return $ WBeat B.empty+      DisconnectT -> return $ WDisc B.empty+      RequestT    -> WRequest <$> envelope+      x           -> liftIO $ throwIO $ MDPExc $+                       "Unexpected Frame from Broker: " ++ show x+    where protocol = chunk mdpW01 ("Unknown Protocol from Broker " +++                                   " -- expected: Worker 0.1")++  ------------------------------------------------------------------------+  -- | Server -> Broker: send reply+  ------------------------------------------------------------------------+  mdpWSndRep :: [Identity] -> Conduit B.ByteString ()+  mdpWSndRep is = streamList hdr >> passThrough+    where hdr = [B.empty, -- identity delimiter+                 mdpW01,+                 xReply] ++ toIs is ++ [B.empty]+ +  ------------------------------------------------------------------------+  -- | Server -> Broker: receive reply+  ------------------------------------------------------------------------+  mdpWRcvRep :: Conduit o WFrame+  mdpWRcvRep = do+    w <- identity+    protocol+    t <- frameType+    case t of+      HeartBeatT  -> return $ WBeat w+      DisconnectT -> return $ WDisc w+      ReadyT      -> WReady w <$> getSrvName+      ReplyT      -> getRep w+      x           -> liftIO $ throwIO $ MDPExc $+                       "Unexpected Frame from Worker: " ++ show x+    where protocol   = chunk mdpW01 "Unknown Protocol from Worker"+          getSrvName = getChunk+          getRep w   = do is <- envelope+                          return $ WReply w is ++  ------------------------------------------------------------------------ +  -- | Broker \<-\> Server: send heartbeat +  ------------------------------------------------------------------------+  mdpWBeat :: Conduit B.ByteString ()+  mdpWBeat = streamList [B.empty,+                         mdpW01,+                         xHeartBeat]++  ------------------------------------------------------------------------ +  -- | Server -> Broker: send connect request (ready)+  ------------------------------------------------------------------------+  mdpWConnect :: ServiceName -> Source+  mdpWConnect sn = streamList [B.empty, -- identity delimiter+                               mdpW01, +                               xReady, +                               B.pack sn]++  ------------------------------------------------------------------------ +  -- | Server -\> Broker: disconnect +  ------------------------------------------------------------------------+  mdpWDisconnect :: Source+  mdpWDisconnect = streamList [B.empty, -- identity delimiter+                               mdpW01,+                               xDisc]++  ------------------------------------------------------------------------ +  -- | Broker -> Server: disconnect+  ------------------------------------------------------------------------+  mdpWBrkDisc :: Identity -> Source+  mdpWBrkDisc i = streamList [i] >> mdpWDisconnect++  ------------------------------------------------------------------------+  -- | Broker / Server protocol:+  --  Heartbeat, Ready, Reply, Request, Disconnect+  ------------------------------------------------------------------------+  data WFrame = WBeat    Identity+              | WReady   Identity B.ByteString+              | WReply   Identity [Identity]+              | WRequest          [Identity]+              | WDisc    Identity+    deriving (Eq, Show)++  ------------------------------------------------------------------------+  -- | Worker Frame Type+  ------------------------------------------------------------------------+  data FrameType = ReadyT | RequestT | ReplyT | HeartBeatT | DisconnectT+    deriving (Eq, Show, Read)++  ------------------------------------------------------------------------+  -- | Get frame type+  ------------------------------------------------------------------------+  frameType :: Conduit o FrameType+  frameType = do+    mbT <- C.await+    case mbT of+      Nothing -> liftIO (throwIO $ MDPExc+                           "Incomplete Message: No Frame Type")+      Just t  | t == xHeartBeat -> return HeartBeatT+              | t == xReady     -> return ReadyT+              | t == xReply     -> return ReplyT+              | t == xDisc      -> return DisconnectT+              | t == xRequest   -> return RequestT+              | otherwise       -> liftIO (throwIO $ MDPExc $ +                                     "Unknown Frame: " ++ B.unpack t)++  ------------------------------------------------------------------------+  -- | Get empty segment+  ------------------------------------------------------------------------+  empty :: Conduit o ()+  empty = chunk B.empty "Missing Separator"++  ------------------------------------------------------------------------+  -- | Check segment contents+  ------------------------------------------------------------------------+  chunk :: B.ByteString -> String -> Conduit o ()+  chunk p e = do+    mb <- C.await+    case mb of+      Nothing -> liftIO (throwIO $ MDPExc $ "Incomplete Message: " ++ e)+      Just x  -> unless (x == p) $ liftIO (throwIO $ MDPExc (e ++ ": " ++ +                                                                  show x))+  ------------------------------------------------------------------------+  -- | Get segment contents+  ------------------------------------------------------------------------+  getChunk :: Conduit o B.ByteString+  getChunk = do+    mb <- C.await+    case mb of+      Nothing -> liftIO (throwIO $ MDPExc "Incomplete Message")+      Just x  -> return x++  ------------------------------------------------------------------------+  -- | Get identity+  ------------------------------------------------------------------------+  identity :: Conduit o Identity+  identity = do+    mbI <- C.await+    case mbI of+      Nothing -> liftIO (throwIO $ MDPExc+                           "Incomplete Message: No Identity")+      Just i  | B.null i  -> +                  liftIO (throwIO $ MDPExc+                           "Incomplete Message: Empty Identity")+              | otherwise -> empty >> return i++  ------------------------------------------------------------------------+  -- | Get block of identities ("envelope")+  ------------------------------------------------------------------------+  envelope :: Conduit o [Identity]+  envelope = go []+    where go is = do+            mbI <- C.await+            case mbI of+              Nothing -> liftIO (throwIO $ MDPExc+                                   "Incomplete Message: No Identity")+              Just i  | B.null i  -> +                          if null is +                            then liftIO (throwIO $ MDPExc+                                      "Incomplete Message: No identities")+                            else return is+                      | otherwise -> empty >> go (i:is) ++  ------------------------------------------------------------------------+  -- | Create envelope [(identity, B.empty)]+  ------------------------------------------------------------------------+  toIs :: [Identity] -> [B.ByteString]+  toIs = foldr toI []+    where toI i is  = [i, B.empty] ++ is++  -------------------------------------------------------------------------+  -- | MDP Exception+  -------------------------------------------------------------------------+  data MDPException = +         -- | Server-side exception+         ServerExc   String+         -- | Client-side exception+         | ClientExc   String+         -- | Broker exception+         | BrokerExc   String+         -- | Generic Protocol+         | MDPExc   String+         -- | MMI Protocol+         | MMIExc   String+         -- | SingleBroker error+         --   (another broker is already running in the same process)+         | SingleBrokerExc String -- move to Broker.Common+    deriving (Show, Read, Typeable, Eq)++  instance Exception MDPException+

@@ -0,0 +1,105 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Broker/Server.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Majordomo Server+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Broker.Server (withServer)+where++  import           Prelude hiding (catch)+  import           Control.Exception (throwIO, finally)+  import           Control.Monad.Trans (liftIO)+  import           Control.Monad (when)+  import           Control.Concurrent.MVar+  import           Data.Conduit ((=$), (=$=), ($$))+  import qualified Data.Conduit    as C+  import qualified Data.ByteString as B++  import           Network.Mom.Patterns.Types+  import           Network.Mom.Patterns.Streams+  import           Network.Mom.Patterns.Broker.Common+  import           Heartbeat -- time arithmetics+  import           Data.Time.Clock ++  ------------------------------------------------------------------------+  -- | Start a server as a background process+  -- +  --   * 'Context'   - The zeromq context+  --  +  --   * 'Service'   - Service name; +  --                   the service name is used to register+  --                   at the broker.+  -- +  --   * 'Msec'      - Heartbeat in Milliseconds;+  --                   must be synchronised with the broker heartbeat+  --+  --   * 'String'    - The address to link to+  --+  --   * 'OnError_'  - Error handler+  --  +  --   * 'Conduit_'  - The application-defined stream transformer;+  --                   the conduit receives the request as input stream+  --                   and should create the output stream that is+  --                   internally sent back to the client+  --+  --   * 'Control' a - Control action+  ------------------------------------------------------------------------+  withServer :: Context     ->+                Service     -> +                Msec        ->+                String      ->+                OnError_    ->+                Conduit_    ->+                (Control a) -> IO a+  withServer ctx srv tmo add onErr serve act | tmo <= 0  =+    throwIO $ MDPExc "Heartbeat is mandatory"+                                             | otherwise = do+    t <- getCurrentTime+    m <- newMVar t                             -- my  heartbeat +    h <- newMVar (timeAdd t (tolerance * tmo)) -- his heartbeat+    withStreams ctx srv (1000 * fromIntegral tmo)+                [Poll "client" add DealerT Connect [] []]+                (handleTmo m h tmo) onErr+                (job       m h) $ \c ->+      -- connect message, main loop, disconnect message ------------------+      finally (send c ["client"] (mdpWConnect srv) >> act c)+              (send c ["client"]  mdpWDisconnect)++          -- receiv message -----------------------------------------------    +    where job m h s +            | getSource s == "client" = mdpServe m h s +            | otherwise               = return ()+          mdpServe m h s = do+            f <- mdpWRcvReq+            case f of+              WRequest is -> liftIO (updBeat h) >>+                             serve =$= mdpWSndRep is =$ passAll s ["client"]+              WBeat    _  -> liftIO (updBeat h)+              WDisc    _  -> liftIO (throwIO $ BrokerExc +                                           "Broker disconnects")+              _           -> liftIO (throwIO $ Ouch +                                   "Unknown frame from Broker!")+            liftIO (handleTmo m h tmo s) -- send heartbeat if it's time++          -- update his heartbeat -----------------------------------------+          updBeat h = modifyMVar_ h $ \_ -> do+                        now <- getCurrentTime+                        return (now `timeAdd` (tolerance * tmo))++  ------------------------------------------------------------------------+  -- Send heartbeat if it's time and check broker's state+  ------------------------------------------------------------------------+  handleTmo :: MVar UTCTime -> MVar UTCTime -> Msec -> Streamer -> IO ()+  handleTmo m h tmo s = do hbPeriodReached m tmo >>= \x -> when x sndHb+                           hbDelay               >>= \x -> when x $ throwIO $+                                                     BrokerExc $ "Missing heartbeat"+    where sndHb   = C.runResourceT $ +                      streamList [B.empty, mdpW01, xHeartBeat] $$  +                      passAll s ["client"]+          hbDelay = getCurrentTime >>= \now -> +                        readMVar h >>= \t   -> return (now > t)

@@ -1,559 +0,0 @@-module Network.Mom.Patterns.Device (-         -- * Device Services-         withDevice,-         withQueue, -         withForwarder, -         withPipeline, -         -- * Polling-         PollEntry, pollEntry,-         -- * Access Types-         AccessType(..),-         -- * Device Service Commands-         addDevice, remDevice, changeTimeout,-         -- * Streamer -         Streamer, getStreamSource, filterTargets,-         -- * Transformer-         Transformer,-         putThrough, ignoreStream, continueHere,--         -- * Transformer Combinators-         -- $recursive_helpers--         emit, emitPart, pass, passBy, end, absorb, merge,-         -- * Helpers-         Identifier, OnTimeout)-where--  import           Types-  import           Service--  import qualified Data.Enumerator        as E-  import           Data.Enumerator (($$))-  import qualified Data.Enumerator.List   as EL-  import           Data.Monoid -  import qualified Data.Map               as Map-  import           Data.Map (Map)-  import qualified Data.Sequence          as S-  import           Data.Sequence ((|>), ViewR(..), ViewL(..))-  import           Prelude hiding (catch)-  import           Control.Exception (catch, finally, throwIO,-                                      bracketOnError)-  import           Control.Concurrent-  import qualified System.ZMQ as Z--  -------------------------------------------------------------------------  -- | Starts a device and executes an action that receives a 'Service'-  --   to control the device-  ---  --   Parameters:-  ---  --   * 'Z.Context' - The /ZMQ/ context-  ---  --   * 'String' - The device name-  ---  --   * 'Parameter' - The initial value of the control parameter-  ---  --   * 'Timeout' - The polling timeout:-  --     /< 0/ - listens eternally,-  --     /0/ - returns immediately,-  --     /> 0/ - timeout in microseconds;-  --     when the timeout expires, the 'OnTimeout' action is invoked.-  ---  --   * 'PollEntry' - List of 'PollEntry';-  --                   the device will polll over -  --                   all list members and direct-  --                   streams to a subset of this list-  --                   determined by the stream transformer.-  ---  --   * 'InBound' - in-bound converter;-  --                 the stream is presented to the transformer-  --                 as chunks of type /o/.-  -- -  --   * 'OutBound' - out-bound converter-  -- -  --   * 'OnError_' - Error handler-  -- -  --   * 'Parameter' -> 'OnTimeout' - Action to perform on timeout-  -- -  --   * 'Parameter' -> 'Transformer' - The stream transformer-  -- -  --   * 'Service' -> IO () - The action to invoke,-  --                          when the device has been started;-  --                          The 'Service' is used to control the device.-  -------------------------------------------------------------------------  withDevice :: Z.Context -> String -> Parameter ->-                Timeout                          -> -                [PollEntry]                      -> -                InBound o -> OutBound o          -> -                OnError_                         ->-                (Parameter -> OnTimeout)         ->-                (Parameter -> Transformer o)     ->-                (Service -> IO a)                -> IO a-  withDevice ctx name param tmo acs iconv oconv onerr ontmo trans =-    withService ctx name param service -    where service = device_ tmo acs iconv oconv onerr ontmo trans--  -------------------------------------------------------------------------  -- | Starts a queue;-  --   a queue connects clients with a dealer ('XDealer'), -  --                   /i.e./ a load balancer for requests,-  --             and servers with a router ('XRouter') that routes responses-  --                   back to the client.-  --  -  --  Parameters:-  ---  --  * 'Z.Context': the /ZMQ/ Context-  ---  --  * 'String': the queue name-  ---  --  * ('AccessPoint', 'LinkType'):-  --                       the access point of the /dealer/ ('XDealer')-  --                       and its link type;-  --                       you usually want to bind the dealer-  --                       so that many clients can connect to it.-  ---  --  * ('AccessPoint', 'LinkType'):-  --                       the access point of the /router/ ('XRouter');-  --                       and its link type;-  --                       you usually want to bind the router-  --                       so that many servers can connect to it.-  ---  --  * 'OnError_': the error handler-  ---  --  * 'Service' -> IO (): the action to run-  ---  --   'withQueue' is implemented by means of 'withDevice' as:-  --   -  --   @  -  --      withQueue ctx name (dealer, ld) (router, lr) onerr act = -  --        withDevice ctx name noparam (-1)-  --              [pollEntry \"clients\" XDealer dealer ld [],-  --               pollEntry \"server\"  XRouter router lr []]-  --              return return onerr (\_ -> return ()) (\_ -> putThrough) act-  --   @  -  -------------------------------------------------------------------------  withQueue :: Z.Context                  -> -               String                     ->-               (AccessPoint, LinkType)    ->-               (AccessPoint, LinkType)    ->-               OnError_                   -> -               (Service -> IO a)          -> IO a-  withQueue ctx name (dealer, l1) (router, l2) onerr = -    withDevice ctx name noparam (-1)-          [pollEntry "clients" XDealer dealer l1 [],-           pollEntry "servers" XRouter router l2 []]-          return return onerr (\_ -> return ()) (\_ -> putThrough)--  -------------------------------------------------------------------------  -- | Starts a Forwarder;-  --   a forwarder connects a publisher and its subscribers.-  --   Note that the forwarder uses a /subscriber/ ('XSub') -  --   to conntect to the /publisher/ and-  --   a /publisher/ ('XPub') to bind the /subscribers/.-  --  -  --  Parameters:-  ---  --  * 'Z.Context': the /ZMQ/ Context-  ---  --  * 'String': the forwarder name-  ---  --  * 'Topic': the subscription topic-  ---  --  * ('AccessPoint', 'AccessPoint'):-  --                       the access points;-  --                       the first is the /subscriber/ ('XSub'),-  --                       the second is the /publisher/ ('XPub');-  --                       this rule is not enforced -  --                       by the type system;  -  --                       you have to take care of it on your own!-  ---  --  * 'OnError_': the error handler-  ---  --  * 'Service' -> IO (): the action to run-  --   -  --   'withForwarder' is implemented by means of 'withDevice' as:-  ---  --   @  -  --      withForwarder ctx name topics (sub, pub) onerr act = -  --        withDevice ctx name noparam (-1)-  --              [pollEntry \"subscriber\" XSub router Connect topics,-  --               pollEntry \"publisher\"  XPub dealer Bind    []]-  --              return return onerr (\_ -> return ()) (\_ -> putThrough) act-  --   @  -  -------------------------------------------------------------------------  withForwarder :: Z.Context                  -> -                   String -> [Topic]          -> -                   (AccessPoint, LinkType)    ->-                   (AccessPoint, LinkType)    ->-                   OnError_                   -> -                   (Service -> IO a)          -> IO a-  withForwarder ctx name topics (sub, l1) (pub, l2) onerr = -    withDevice ctx name noparam (-1)-          [pollEntry "subscriber" XSub sub l1 topics, -           pollEntry "publisher"  XPub pub l2 []]-          return return onerr (\_ -> return ()) (\_ -> putThrough)--  -------------------------------------------------------------------------  -- | Starts a pipeline;-  --   a pipeline connects a /pipe/-  --   and its /workers/.-  --   Note that the pipeline uses a /puller/ ('XPull')-  --   to conntect to the /pipe/ and-  --   a /pipe/ ('XPipe') to bind the /pullers/.-  ---  --  Parameters:-  ---  --  * 'Z.Context': the /ZMQ/ Context-  ---  --  * 'String': the pipeline name-  ---  --  * ('AccessPoint', 'LinkType'):-  --                       the access point of the /puller/ ('XPull')-  --                       and its link type;-  --                       you usually want to connect the puller -  --                       to one pipe so that it appears -  --                       as one puller among others,-  --                       to which the pipe may send jobs.-  ---  --  * ('AccessPoint', 'LinkType'):-  --                       the access point of the /pipe/ ('XPipe');-  --                       and its link type;-  --                       you usually want to bind the pipe-  --                       so that many pullers can connect to it.-  ---  --  * 'OnError_': the error handler-  ---  --  * 'Service' -> IO (): the action to run-  ---  --   'withPipeline' is implemented by means of 'withDevice' as:-  --   -  --   @  -  --      withPipeline ctx name topics (puller, l1) (pusher, l2) onerr act =-  --        withDevice ctx name noparam (-1)-  --              [pollEntry \"pull\"  XPull puller l1 [],-  --               pollEntry \"push\"  XPush pusher l2 []]-  --              return return onerr (\_ -> return ()) (\_ -> putThrough) act-  --   @  -  -------------------------------------------------------------------------  withPipeline :: Z.Context                   -> -                   String                     ->-                   (AccessPoint, LinkType)    ->-                   (AccessPoint, LinkType)    ->-                   OnError_                   -> -                   (Service -> IO a)          -> IO a-  withPipeline ctx name (puller, l1) (pusher, l2) onerr = -    withDevice ctx name noparam (-1)-          [pollEntry "pull"  XPull puller l1 [], -           pollEntry "push"  XPipe pusher l2 []]-          return return onerr (\_ -> return ()) (\_ -> putThrough)--  -------------------------------------------------------------------------  -- | A transformer is an 'E.Iteratee'-  --   to transform streams.-  --   It receives two arguments:-  --   -  --   * a 'Streamer' which provides information on access points;-  --   -  --   * a 'Sequence' which may be used to store chunks of an incoming-  --     stream before they are sent to the target.-  ---  --   Streamer and sequence keep track of the current transformation.-  --   The streamer knows where the stream comes from and -  --   may be queried about other streams in the device.-  -------------------------------------------------------------------------  type Transformer o = Streamer o -> S.Seq o -> E.Iteratee o IO ()--  -------------------------------------------------------------------------  -- | Holds information on streams and the current state of the device;-  --   streamers are passed to transformers.-  -------------------------------------------------------------------------  data Streamer o = Streamer {-                      strmSrc    :: (Identifier, Z.Poll),-                      strmIdx    :: Map Identifier Z.Poll,-                      strmPoll   :: [Z.Poll],-                      strmOut    :: OutBound o}--  -------------------------------------------------------------------------  -- | Retrieves the identifier of the source of the current stream-  -------------------------------------------------------------------------  getStreamSource :: Streamer o -> Identifier-  getStreamSource = fst . strmSrc--  -------------------------------------------------------------------------  -- | Filters target streams;-  --   the function resembles /filter/ of 'Data.List':-  --   it receives the property of an 'Identifier';-  --   if a 'PollEntry' has this property, it is added to the result set.-  ---  --   The function is intended to select targets for an out-going stream,-  --   typically based on the identifier of the source stream.-  --   The following example selects all poll entries, but the source:-  ---  --   @-  --     broadcast :: Streamer o -> [Identifier]-  --     broadcast s = filterTargets s notSource-  --       where notSource = (/=) (getStreamSource s)-  --   @-  -------------------------------------------------------------------------  filterTargets :: Streamer o -> (Identifier -> Bool) -> [Identifier]-  filterTargets s f = map fst $ Map.toList $ Map.filterWithKey flt $ strmIdx s-    where flt k _ = f k--  -------------------------------------------------------------------------  -- $recursive_helpers-  -- The following functions are building blocks-  -- for defining transformers.-  -- The building blocks operate on sequences, stream targets and -  -- transformers.-  -- They manipulate streams, send them to targets and enter -  -- a transformer.-  -------------------------------------------------------------------------  -- | Sends all sequence elements to the targets identified-  --   by the list of 'Identifier' and terminates the outgoing stream.-  --   The transformation continues with the transformer-  --   passed in and an empty sequence. -  -------------------------------------------------------------------------  emit :: Streamer o    -> [Identifier] -> S.Seq o -> -          Transformer o -> E.Iteratee o IO ()-  emit s is os go = tryIO sender >> go s S.empty-    where sender = mapM_ (\i -> sendStreamer s i (sendseq s os True)) is--  -------------------------------------------------------------------------  -- | Sends all sequence elements to the targets identified-  --   by the list of 'Identifier', but unlike 'emit', -  --   does not terminate the outgoing stream.-  --   The transformation continues with the transformer-  --   passed in and an empty sequence. -  ---  --   Note that all outgoing streams, once started, -  --   have to be terminated before the transformer ends. -  --   Otherwise, a protocol error will occur.-  -------------------------------------------------------------------------  emitPart :: Streamer o    -> [Identifier] -> S.Seq o -> -              Transformer o -> E.Iteratee o IO ()-  emitPart s is os go = tryIO sender >> go s S.empty-    where sender = mapM_ (\i -> sendStreamer s i (sendseq s os False)) is--  -------------------------------------------------------------------------  -- | Sends one element (/o/) to the targets and continues -  --   with an empty sequence;-  --   the Boolean parameter determines whether this is the last message-  --   to send. -  ---  --   Note that all outgoing streams, once started, -  --   have to be terminated before the transformer ends. -  --   Otherwise, a protocol error will occur.-  -------------------------------------------------------------------------  pass :: Streamer o    -> [Identifier] -> o -> Bool ->-          Transformer o -> E.Iteratee o IO ()-  pass s is o lst go = tryIO sender >> go s S.empty-    where sender = mapM_ (\i -> sendStreamer s i -                                 (sendseq s (S.singleton o) lst)) is--  -------------------------------------------------------------------------  -- | Sends one element (/o/) to the targets, -  --   but, unlike 'pass', passes the sequence to the transformer. -  --   'passBy' does not terminate the outgoing stream.-  -------------------------------------------------------------------------  passBy :: Streamer o    -> [Identifier] -> o -> S.Seq o -> -           Transformer o -> E.Iteratee o IO ()-  passBy s is o os go = tryIO sender >> go s os-    where sender = mapM_ (\i -> sendStreamer s i -                                 (sendseq s (S.singleton o) False)) is--  -------------------------------------------------------------------------  -- | Terminates the outgoing stream by sending the new element-  --   as last segment to all targets and ends the transformer-  --   by ignoring the rest of the incoming stream.-  -------------------------------------------------------------------------  end :: Streamer o -> [Identifier] -> o -> E.Iteratee o IO ()-  end s is o = pass s is o True (\_ _ -> go)-    where go = EL.head >>= \mbO ->-               case mbO of-                 Nothing -> return ()-                 Just _  -> go--  -------------------------------------------------------------------------  -- | Adds a new element to the sequence-  --   and calls the transformer without sending anything-  -------------------------------------------------------------------------  absorb :: Streamer o    -> o -> S.Seq o ->-            Transformer o -> E.Iteratee o IO ()-  absorb s o os go = go s (os |> o)--  -------------------------------------------------------------------------  -- | Merges the new element with the last element of the sequence;-  --   if the sequence is currently empty, the new element-  --   will be its only member.-  --   Merged elements appear as one element of the sequence -  --   in the continuation of the transformation.-  --   The type /o/ must be a 'Monoid', /i.e./,-  --   it must implement /mappend/ and /mempty/.-  --   The function does not send anything.-  -------------------------------------------------------------------------  merge :: Monoid o => Streamer o    -> o -> S.Seq o ->-                       Transformer o -> E.Iteratee o IO ()-  merge s o os go = -    let os' = case S.viewr os of-                EmptyR  -> S.singleton o-                xs :> x -> xs |> (x `mappend` o)-     in go s os'--  -------------------------------------------------------------------------  -- | Transformer that-  --   passes messages one-to-one to all poll entries -  --   but the current source-  -------------------------------------------------------------------------  putThrough :: Transformer a-  putThrough s' os' = EL.head >>= \mbo -> go mbo s' os'-    where go mbo s _ = do-            mbo' <- EL.head-            case mbo of-              Nothing -> return ()-              Just x  -> do-                let lst = case mbo' of-                            Nothing -> True-                            Just _  -> False-                let trg = filterTargets s (/= getStreamSource s)-                pass s trg x lst (go mbo')--  -------------------------------------------------------------------------  -- | Transformer that-  --   ignores the remainder of the current stream;-  --   it is usually used to terminate a transformer.-  -------------------------------------------------------------------------  ignoreStream :: Transformer a-  ignoreStream _ _ = EL.consume >>= \_ -> return ()--  -------------------------------------------------------------------------  -- | Transformer that-  --   does nothing but continuing the transformer, from which it is called-  --   and, hence, is identical to /return ()/;-  --   it is usually passed to a transformer combinator,-  --   like 'emit', to continue processing right here -  --   instead of recursing into another transformer.-  -------------------------------------------------------------------------  continueHere :: Transformer a-  continueHere _ _ = return ()--  -------------------------------------------------------------------------  -- Internal-  -------------------------------------------------------------------------  sendStreamer :: Streamer o -> Identifier -> (Z.Poll -> IO ()) -> IO ()-  sendStreamer s i act = case Map.lookup i (strmIdx s) of-                           Nothing  -> return ()-                           Just p   -> act p--  mapIOSeq :: (a -> IO ()) -> S.Seq a -> IO ()-  mapIOSeq f os = case S.viewl os of-                    EmptyL  -> return ()-                    x :< xs -> f x >> mapIOSeq f xs--  sendseq :: Streamer o -> S.Seq o -> Bool -> Z.Poll -> IO ()-  sendseq s os lst p -    | lst       = case S.viewr os of-                    EmptyR  -> return ()-                    xs :> x -> mapIOSeq (\o -> dosend s p o False) xs-                               >>              dosend s p x True -    | otherwise = mapIOSeq (\o -> dosend s p o False) os--  dosend :: Streamer o -> Z.Poll -> o -> Bool -> IO ()-  dosend  s (Z.S sock _) o lst = -    let flg = if lst then [] else [Z.SndMore]-     in strmOut s o >>= \x -> Z.send sock x flg-  dosend  _ _ _ _ = error "Ouch!"--  -------------------------------------------------------------------------  -- Creates poll list and enters runDevice -  -------------------------------------------------------------------------  device_ :: Timeout                   ->-             [PollEntry]               -> -             InBound o -> OutBound o   -> -             OnError_                  ->-             (String -> OnTimeout)     ->-             (String -> Transformer o) ->-             Z.Context -> String       -> -             String -> String -> IO () -> IO ()-  device_ tmo acs iconv oconv onerr ontmo trans-          ctx name sockname param imReady = do -    xp <- catch (mkPoll ctx tmo acs Map.empty [] [])-                (\e -> onerr Fatal e name param >> throwIO e)-    m  <- newMVar xp-    finally (runDevice name m iconv oconv onerr ontmo trans -                       sockname param imReady)-            (withMVar m (mapM_ closeS . xpPoll) >> return ())--  closeS :: Z.Poll -> IO ()-  closeS p = case p of -               Z.S s _ -> safeClose s-               _       -> return ()--  -------------------------------------------------------------------------  -- creates and binds or connects all sockets recursively;-  -- on its way, creates the Map from Identifiers to PollItems,-  --             a list of PollItems-  --             and a list of Identifiers with the same order;-  -- finally executes "run"-  -------------------------------------------------------------------------  mkPoll :: Z.Context -> Timeout     -> -            [PollEntry]              -> -            Map Identifier Z.Poll    ->-            [Identifier]             ->-            [Z.Poll]                 -> IO XPoll -  mkPoll ctx t []     m is ps = return XPoll{xpCtx  = ctx,-                                             xpTmo  = t,-                                             xpMap  = m,-                                             xpIds  = is,-                                             xpPoll = ps}-  mkPoll ctx t (k:ks) m is ps = bracketOnError-    (access ctx (pollType k)-                (pollLink k) -                (pollOs   k) -                (pollAdd  k) -                (pollSub  k))-    (\p -> closeS p >> return [])-    (\p -> do let m'  = Map.insert (pollId k) p m-              let is' = pollId k : is-              let ps' = p:ps-              mkPoll ctx t ks m' is' ps')--  -------------------------------------------------------------------------  -- finally start the device entering Service.xpoll-  -------------------------------------------------------------------------  runDevice :: String -> MVar XPoll      -> -               InBound o -> OutBound o   -> -               OnError_                  -> -               (String -> OnTimeout)     ->-               (String -> Transformer o) -> -               String -> String -> IO () -> IO ()-  runDevice name mxp iconv oconv onerr ontmo trans sockname param imReady = (do-      xp <- readMVar mxp-      Z.withSocket (xpCtx xp) Z.Sub $ \cmd -> do-        trycon      cmd sockname retries-        Z.subscribe cmd ""-        let p = Z.S cmd Z.In-        modifyMVar_ mxp $ \_ -> return xp {xpPoll = p : xpPoll xp}-        imReady-        finally (xpoll False mxp ontmo go param)-                (modifyMVar_ mxp $ \xp' -> -                   return xp' {xpPoll = tail (xpPoll xp')})) -      `catch` (\e -> onerr Fatal e name param) -- >> throwIO e)-    where go i poller p =-            case poller of -              Z.S s _ -> do-                xp <- readMVar mxp-                let strm = Streamer {-                             strmSrc  = (i, poller), -                             strmIdx  = xpMap  xp,-                             strmPoll = xpPoll xp,-                             strmOut  = oconv}-                eiR <- E.run (rcvEnum s iconv $$ -                              trans p strm S.empty) -                case eiR of-                  Left e  -> consumeOnErr s >> onerr Error e name p -                  Right _ -> return ()-              _ -> error "Ouch!"--  consumeOnErr :: Z.Socket a -> IO ()-  consumeOnErr s = E.run_ (rcvEnum s idIn $$ EL.consume >>= \_ -> return ())-

@@ -1,418 +0,0 @@-{-# LANGUAGE CPP #-}----------------------------------------------------------------------------------- |--- Module     : Network/Mom/Patterns/Enumerator.hs--- Copyright  : (c) Tobias Schoofs--- License    : LGPL --- Stability  : experimental--- Portability: portable--- --- Enumerators for basic patterns---------------------------------------------------------------------------------module Network.Mom.Patterns.Enumerator (-          -- * Enumerators-          -- $enums--          -- ** Raw Enumerators-          enumWith, enumFor, once, just,-          -- ** Fetchers-          Fetch, Fetch_, -          FetchHelper, FetchHelper',-          FetchHelper_, FetchHelper_',-          fetcher, fetcher_,-          fetch1, fetch1_,-          fetchFor, fetchFor_,-          fetchJust, fetchJust_,-          listFetcher, listFetcher_,-#ifdef _TEST-          err,-#endif-          -- * Iteratees-          -- $its--          -- ** Raw Iteratees-          one, mbOne, toList, toString, append,-          store,-          -- ** Dumps -          Dump, sink, sinkI, nosink) -where--  import           Types--  import qualified Data.Enumerator        as E-  import           Data.Enumerator (($$))-  import qualified Data.Enumerator.List   as EL-  import qualified Data.Monoid            as M-  import           Data.List (foldl', intercalate)--  import           Control.Applicative ((<$>))-  import           Control.Monad-  import           Prelude hiding (catch)--  import qualified System.ZMQ as Z--  -------------------------------------------------------------------------  -- $enums-  -- Enumerators generate streams-  -- and pass chunks of the stream for further processing-  -- to Iteratees.-  -- The Enumerator-Iteratee abstraction is very powerful-  -- and is by far not discussed exhaustively here.-  -- For more details, please refer to the documentation-  -- of the Enumerator package.-  -- -  -- The Patterns package provides a small set-  -- of enumerators that may be useful for many messaging patterns.-  -- Enumerators are split into raw enumerators,-  -- which can be used with patterns under direct control-  -- of application code such as 'Client', 'Pub' and 'Peer',-  -- and Fetchers, which are used with services, /i.e./-  -- 'withServer' and 'withPeriodicPub'.-  -------------------------------------------------------------------------  -- | Calls an application-defined /getter/ function-  --   until this returns 'Nothing';-  --   if the getter throws an exception,-  --   the enumerator returns 'E.Error'.-  -------------------------------------------------------------------------  enumWith :: (i -> IO (Maybe o)) -> i -> E.Enumerator o IO ()-  enumWith get i step = -    case step of-      E.Continue k -> chainIOe (get i) $ \mbO ->-                      case mbO of-                        Nothing -> E.continue k-                        Just o  -> enumWith get i $$ k (E.Chunks [o])-      _            -> E.returnI step--  -------------------------------------------------------------------------  -- | Calls the application-defined /getter/ function /n/ times;-  --   The enumerator receives a pair ('Int', 'Int'),-  --   where the first integer is a counter and -  --   the second is the upper bound.-  --   /n/ is defined as /snd - fst/, /i.e./-  --   the counter is incremented until it reaches the value-  --   of the bound. The counter must be a value less than the bound-  --   to avoid protocol errors, /i.e./ the /getter/ must be called-  --   at least once.-  --   The current value of the counter and additional input-  --   are passed to the /getter/.-  --   if the getter throws an exception,-  --   the enumerator returns 'E.Error'.-  -------------------------------------------------------------------------  enumFor :: (Int -> i -> IO o) -> (Int, Int) -> i -> E.Enumerator o IO ()-  enumFor get runner i = go runner-    where go (c,e) step = -            case step of-              E.Continue k -> -                if c >= e then E.continue k-                          else chainIOe (get c i) $ \o ->-                               go (c+1,e) $$ k (E.Chunks [o])-              _            -> E.returnI step--  -------------------------------------------------------------------------  -- | Calls the application-defined /getter/ function once;-  --   the enumerator must return a value -  --   (the result type is not 'Maybe'),-  --   otherwise, the sending iteratee has nothing to send -  --   which would most likely result in a protocol error.-  --   if the getter throws an exception,-  --   the enumerator returns 'E.Error'.-  -------------------------------------------------------------------------  once :: (i -> IO o) -> i -> E.Enumerator o IO ()-  once = go True-    where go first get i step =-            case step of-              E.Continue k -> -                if first then chainIOe (get i) $ \o ->-                     go False get i $$ k (E.Chunks [o])-                else E.continue k-              _ -> E.returnI step--  -------------------------------------------------------------------------  -- | Passes just the input value to iteratee;-  --   -  --   > just "hello world"-  ---  --   hence, reduces to just "hello world" sent over the wire.-  -------------------------------------------------------------------------  just :: o -> E.Enumerator o IO ()-  just = go True-    where go first o step = -            case step of-              E.Continue k -> -                if first then go False o $$ k (E.Chunks [o])-                  else E.continue k-              _ -> E.returnI step--  -------------------------------------------------------------------------  -- | Calls the application-defined 'FetchHelper'-  --   until it returns 'Nothing';-  --   note that the 'FetchHelper' shall return at least one 'Just'-  --   value to avoid a protocol error.-  --   If the 'FetchHelper' throws an exception,-  --   the 'fetcher' returns 'E.Error'.-  -------------------------------------------------------------------------  fetcher :: FetchHelper i o -> Fetch i o -  fetcher fetch ctx p i step =-    case step of-      (E.Continue k) -> chainIOe (fetch ctx p i) $ \mbo ->-        case mbo of -          Nothing -> E.continue k-          Just o  -> fetcher fetch ctx p i $$ k (E.Chunks [o]) -      _ -> E.returnI step--  -------------------------------------------------------------------------  -- | A variant of 'fetcher' without input;-  -------------------------------------------------------------------------  fetcher_ :: FetchHelper_ o -> Fetch_ o-  fetcher_ = fetcher--  -------------------------------------------------------------------------  -- | Calls the application-defined 'FetchHelper'' once;-  --   If the 'FetchHelper'' throws an exception,-  --   the 'fetcher' returns 'E.Error'.-  -------------------------------------------------------------------------  fetch1 :: FetchHelper' i o -> Fetch i o-  fetch1 = go True -    where go first fetch ctx p i step =-            case step of-              (E.Continue k) -> -                if first then chainIOe (fetch ctx p i) $ \o ->-                     go False fetch ctx p i $$ k (E.Chunks [o])-                else E.continue k-              _ -> E.returnI step--  -------------------------------------------------------------------------  -- | A variant of 'fetch1' without input;-  -------------------------------------------------------------------------  fetch1_ :: FetchHelper_' o -> Fetch_ o-  fetch1_ = fetch1--  -------------------------------------------------------------------------  -- | Calls the iteratee for each element of the input list-  -------------------------------------------------------------------------  listFetcher :: Fetch [o] o-  listFetcher ctx p os step = -    case step of-      (E.Continue k) -> -        if null os then E.continue k-                   else listFetcher ctx p (tail os) $$ k (E.Chunks [head os])-      _ -> E.returnI step--  -------------------------------------------------------------------------  -- | A variant of 'listFetcher' for services without input;-  --   the list, in this case, is passed as an additional argument-  --   to the fetcher.-  -------------------------------------------------------------------------  listFetcher_ :: [o] -> Fetch_ o-  listFetcher_ l ctx p _ step =-    case step of-      (E.Continue k) -> -        if null l then E.continue k-                  else listFetcher_ (tail l) ctx p () $$ k (E.Chunks [head l])-      _ -> E.returnI step--  -------------------------------------------------------------------------  -- | Calls the application-defined /getter/ /n/ times;-  --   The /getter/ is a variant of 'FetchHelper'' -  --   with the current value of the counter as additional argument.-  --   For more details, refer to 'enumFor'.-  -------------------------------------------------------------------------  fetchFor :: (Z.Context -> Parameter -> Int -> i -> IO o) -> -              (Int, Int) -> Fetch i o-  fetchFor fetch (c,e) ctx p i step =-    case step of-      (E.Continue k) ->-         if c >= e then E.continue k-                   else chainIOe (fetch ctx p c i) $ \x -> -                        fetchFor fetch (c+1, e) ctx p i $$ k (E.Chunks [x])-      _ -> E.returnI step--  -------------------------------------------------------------------------  -- | A variant of 'fetchFor' without input-  -------------------------------------------------------------------------  fetchFor_ :: (Z.Context -> Parameter -> Int -> () -> IO o) -> -               (Int, Int) -> Fetch_ o-  fetchFor_ = fetchFor--  -------------------------------------------------------------------------  -- | Passes just the input value to the iteratee;-  --   -  --   > fetchJust "hello world"-  ---  --   hence, reduces to just \"hello world\" sent over the wire.-  --   Note that the input /i/ is ignored.-  -------------------------------------------------------------------------  fetchJust :: o -> Fetch i o-  fetchJust o _ _ _ = go True-    where go first step = -            case step of-              (E.Continue k) ->-                if first then go False $$ k (E.Chunks [o]) -- yield?-                  else E.continue k-              _ -> E.returnI step--  -------------------------------------------------------------------------  -- | A variant of 'fetchJust' without input-  -------------------------------------------------------------------------  fetchJust_ :: o -> Fetch_ o-  fetchJust_ = fetchJust--  -------------------------------------------------------------------------  -- For testing only-  -------------------------------------------------------------------------#ifdef _TEST-  err :: Fetch_ o-  err _ _ _ s = do-    ei <- liftIO $ catch -            (throwIO (AssertionFailed "Test") >>= \_ -> return $ Right ())-            (\e -> return $ Left e)-    case ei of-      Left e  -> E.returnI (E.Error e)-      Right _ -> E.returnI s-#endif--  -------------------------------------------------------------------------  -- $its-  -- Iteratees process chunks of streams-  -- passed in by an enumerator.-  -- The Enumerator-Iteratee abstraction is very powerful-  -- and is by far not discussed exhaustively here.-  -- For more details, please refer to the documentation-  -- of the Enumerator package.-  -- -  -- The Patterns package provides a small set-  -- of iteratees that may be useful for many messaging patterns.-  -- Iteratees are split into raw iteratees,-  -- which can be used with patterns under direct control-  -- of application code such as 'Client', 'Pub', 'Peer'-  -- and, for obtaining the request, 'withServer',-  -- and Dumps, which are used with services, /i.e./-  -- 'withSub' and 'withPuller'.-  -------------------------------------------------------------------------  -- | Calls the application-defined IO action-  --   for each element of the stream;-  --   The IO action could, for instance, -  --   write to an already opened file,-  --   store values in an 'MVar' or-  --   send them through a 'Chan' to another thread -  --   for further processing.-  --   An exception thrown in the IO action-  --   is re-thrown by 'E.throwError'.-  -------------------------------------------------------------------------  store :: (i -> IO ()) -> E.Iteratee i IO ()-  store save = do-    mbi <- EL.head-    case mbi of-      Nothing -> return ()-      Just i  -> tryIO (save i) >> store save--  -------------------------------------------------------------------------  -- | Returns one value of type /i/;-  --   if the enumerator creates a value, this value is returned;-  --   otherwise, the input value is returned.-  -------------------------------------------------------------------------  one :: i -> E.Iteratee i IO i-  one x = do-    mbi <- EL.head-    case mbi of-      Nothing -> return x-      Just i  -> return i--  -------------------------------------------------------------------------  -- | Returns one value of type 'Maybe' /i/;-  --   equal to 'Data.Enumerator.List.head'-  -------------------------------------------------------------------------  mbOne :: E.Iteratee i IO (Maybe i)-  mbOne = EL.head--  -------------------------------------------------------------------------  -- | Returns a list containing all chunks of the stream;-  --   equal to 'Data.Enumerator.List.consume';-  --   note that this iteratee causes a space leak-  --   and is not suitable for huge streams or streams of unknown size.-  -------------------------------------------------------------------------  toList :: E.Iteratee i IO [i] -- equal to EL.consume-  toList = EL.consume--  -------------------------------------------------------------------------  -- | Returns a string containing all chunks of the stream-  --   intercalated with the input string, /e.g./:-  --   if the stream consists of the two elements \"hello\" and \"world\"-  ---  --   > toString " " -  ---  --   returns "hello world".-  --   Note that this iteratee causes a space leak-  --   and is not suitable for huge streams or streams of unknown size.-  -------------------------------------------------------------------------  toString :: String -> E.Iteratee String IO String -  toString s = intercalate s <$> EL.consume--  -------------------------------------------------------------------------  -- | Merges the elements of a stream using 'M.mappend';-  --   if the stream is empty, 'append' returns 'M.mempty'.-  --   The type /i/ must be instance of 'M.Monoid'.-  --   Note that this iteratee causes a space leak-  --   and is not suitable for huge streams or streams of unknown size.-  -------------------------------------------------------------------------  append :: M.Monoid i => E.Iteratee i IO i-  append = foldl' M.mappend M.mempty <$> EL.consume--  -------------------------------------------------------------------------  -- | Opens a data sink, dumps the stream into this sink-  --   and closes the sink when the stream terminates-  --   or when an error occurs;-  --   the first IO action is used to open the sink (of type /s/),-  --   the second closes the sink and-  --   the third writes one element into the sink.-  -------------------------------------------------------------------------  sink :: (Z.Context -> String ->           IO s ) -> -          (Z.Context -> String -> s ->      IO ()) -> -          (Z.Context -> String -> s -> i -> IO ()) -> Dump i-  sink op cl sv ctx p = tryIO (op ctx p) >>= go-    where go s   = E.catchError (body s) (onerr s)-          body s = do-            mbi <- EL.head-            case mbi of-              Nothing -> tryIO (cl ctx p s)-              Just i  -> tryIO (sv ctx p s i) >> body s-          onerr  s e  =  tryIO (cl ctx p s)   >> E.throwError e--  -------------------------------------------------------------------------  -- | Variant of 'sink' that uses the first segment of the stream-  --   as input parameter to open the sink.-  --   The first segment, which could contain -  --   a file name or parameters for an /SQL/ query,-  --   is not written to the sink.-  --   As with 'sink', the sink is closed when the stream terminates or-  --   when an error occurs.-  -------------------------------------------------------------------------  sinkI :: (Z.Context -> String ->      i -> IO s ) -> -           (Z.Context -> String -> s      -> IO ()) -> -           (Z.Context -> String -> s -> i -> IO ()) -> Dump i-  sinkI op cl sv ctx p = do-          mbi <- EL.head-          case mbi of-            Nothing -> return ()-            Just i  -> do-              s <- tryIO (op ctx p i)-              E.catchError (body s) (onerr s)-    where body s = do-            mbi <- EL.head-            case mbi of-              Nothing -> tryIO (cl ctx p s)-              Just i  -> tryIO (sv ctx p s i) >> body s-          onerr s e   =  tryIO (cl ctx p s)   >> E.throwError e--  -------------------------------------------------------------------------  -- | Similar to 'sink', but uses a data sink that is opened and closed-  --   outside the scope of the service or does not need to be -  --   opened and closed at all;-  --   examples may be services that write to 'MVar' or 'Chan'.-  --   'nosink' is implemented as a closure of 'store':-  ---  --   > nosink save ctx p = store (save ctx p)-  -------------------------------------------------------------------------  nosink :: (Z.Context -> String -> i -> IO ()) -> Dump i-  nosink sv ctx p = store (sv ctx p)-

@@ -0,0 +1,987 @@+{-# LANGUAGE CPP, DeriveDataTypeable, RankNTypes #-}+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Streams.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Stream processing services+-------------------------------------------------------------------------------+module Network.Mom.Patterns.Streams (+         -- * Processing Streams+         -- $procStreams++         withStreams,+         runReceiver, runSender,+         PollEntry(..),+         AccessType(..), parseAccess,+         LinkType(..), link, parseLink, ++         -- * Streamer+         -- $streamer++         Streamer, +         StreamConduit, StreamSink, StreamAction,+         filterStreams, getSource,++         -- * StreamSinks+         -- $sinks++         stream, part, +         passAll, pass1, passN, passWhile, +         ignoreStream,++         -- * Controller+         -- $controller++         Controller, Control, internal,+         stop, pause, resume, send, receive++         -- * Complete Example+         -- $example++       )+where++  import           Control.Monad.Trans (liftIO)+  import           Control.Monad (when)+  import           Control.Applicative ((<$>))+  import           Prelude hiding (catch)+  import           Control.Exception (SomeException, +                                      bracket, bracketOnError, finally,+                                      catch, throwIO)+  import           Control.Concurrent+  import           Data.Conduit (($$))+  import qualified Data.Conduit          as C+  import qualified Data.ByteString.Char8 as B+  import           Data.Char (toLower)+  import           Data.Map (Map)+  import qualified Data.Map as Map+  import qualified System.ZMQ            as Z++  import           Factory+  import           Network.Mom.Patterns.Types++  {- $procStreams+     This module provides functions to automate stream processing,+     in particular the function 'withStreams' that starts +     a background action polling on a set of streams.+     The function uses uses application-defined callbacks+     to manipulate streams.++     The functions 'runReceiver' and 'runSender' are intended mainly+     for testing. They send or receive respectively streams, +     which are handled or created by a conduit 'Sink' and 'Source'.+  -}+  ------------------------------------------------------------------------+  -- | Starts polling on a set of streams.+  --   The actual polling will be run in another thread.+  --   The current thread continues with the action passed in.+  --   When this action terminates, the streamer stops polling.+  --+  --   Parameters:+  --+  --   * 'Context' - The /ZMQ/ context+  --+  --   * 'Service' - The service name +  --                 indicated for instance in error messages.+  --+  --   * 'Timeout' - The polling timeout:+  --     /< 0/ - listens eternally,+  --     /0/ - returns immediately,+  --     /> 0/ - timeout in microseconds;+  --     when the timeout expires, the 'StreamAction' is invoked.+  --+  --   * 'PollEntry' - List of 'PollEntry';+  --                   the streamer will poll over +  --                   all list members.+  --                   When input is available,+  --                   it is directed to the 'StreamSink'.+  --+  --   * 'StreamAction' - Invoked when timeout expires.+  --+  --   * 'OnError_' - Error handler+  -- +  --   * 'StreamSink' - The sink, to which the stream is sent.+  --                    Note that the sink must terminate +  --                    the outgoing stream +  --                    (using one of the terminating sinks+  --                     described below).+  --                    Not terminating the stream properly+  --                    will result in a zeromq socket error.+  -- +  --   * 'Control' a - The action to invoke,+  --                   when the streamer has been started;+  --                   The 'Control' is used to control the device.+  ------------------------------------------------------------------------+  withStreams :: Context      -> +                 Service      ->+                 Timeout      ->+                 [PollEntry]  ->+                 StreamAction -> -- on timeout+                 OnError_     -> -- error handler+                 StreamSink   -> -- stream handler+                 Control a    -> IO a+  withStreams ctx sn tmo pes+              onTmo+              onErr+              onStream+              ctrl = do+      running <- newEmptyMVar+      ready   <- newEmptyMVar+      catch (startService ready running)+            (\e -> do onErr Fatal e "Service is going down"+                      throwIO (e::SomeException))+          ----------------------------------------------------------------+          -- start the polling thread+          -- create the controller+          -- invoke the control action+          ----------------------------------------------------------------+    where startService ready running = do+            cmdN <- cmdName sn+            _    <- forkIO $ finally (runStreams cmdN ready `catch` +                                       \e -> do onErr Fatal e +                                                  "Service is going down"+                                                throwIO (e::SomeException))+                                     (putMVar running ())+            Z.withSocket ctx Z.XReq $ \cmd -> do+              _ <- readMVar ready+              Z.connect cmd cmdN+              let c = Controller {ctrlCtx  = ctx,+                                  ctrlCmd  = cmd,+                                  ctrlOpts = []}+              finally (ctrl c)+                      (finally (stop c)+                               (takeMVar running))+          ----------------------------------------------------------------+          -- The polling thread+          ----------------------------------------------------------------+          runStreams cmdN ready = +            Z.withSocket ctx Z.XReq $ \cmd -> bracket+                         (do Z.bind cmd cmdN+                             let c = Z.S cmd Z.In+                             (m, is, ps) <- mkPoll ctx pes Map.empty [] []+                             return (m, c, is, ps))+                         (\(_, _, _,  ps) -> mapM_ closeS ps)+                         (\(m, c, is, ps) -> do xp <- newMVar PollSt {+                                                        polMap  = m,+                                                        polCmd  = c,+                                                        polIs   = is,+                                                        polPs   = ps,+                                                        polTmo  = tmo,+                                                        polCont = True+                                                }+                                                putMVar ready () +                                                poll xp)++          ----------------------------------------------------------------+          -- Poll+          ----------------------------------------------------------------+          poll :: PollT ()+          poll xp = do+            c <- psCmd xp+            (_, ps) <- psPolls xp++            (x:ss)  <- Z.poll (c:ps) tmo+            case x of+              Z.S _ Z.In -> do catch (handleCmd xp)+                                     (\e -> do onErr Critical e +                                                     "Can't handle command"+                                               cleanStream c)+                               q <- psContinue xp+                               when q $ poll xp+              _            -> handleStream ss xp >> poll xp+           +          ----------------------------------------------------------------+          -- Handle Streams+          ----------------------------------------------------------------+          handleStream :: [Z.Poll] -> PollT () +          handleStream ss xp = do+            m <- psMap xp+            c <- psCmd xp+            (is, ps) <- psPolls xp+            case getStream is ss of+              Nothing -> onTmo Streamer {strmSrc  = Nothing, +                                         strmIdx  = m,+                                         strmCmd  = c,+                                         strmPoll = ps} `catch` +                           \e -> onErr Error e "Timeout Action failed"+              Just (i, p) -> catch (+                C.runResourceT $ readPoll p $$+                                   onStream Streamer {+                                               strmSrc  = Just (i, p),+                                               strmIdx  = m,+                                               strmCmd  = c,+                                               strmPoll = ps}) (+                \e -> onErr Error e "Stream handling failed" >> cleanStream p)++          ----------------------------------------------------------------+          -- Handle Commands+          ----------------------------------------------------------------+          handleCmd :: PollT ()+          handleCmd xp = do+            m <- psMap xp+            c <- psCmd xp+            (_, ps) <- psPolls xp+            case c of+              Z.S s _ -> do +                x <- B.unpack <$> Z.receive s []+                case x of+                  "send" -> cmdSend s Streamer {strmSrc  = Nothing,+                                                strmIdx  = m,+                                                strmCmd  = c,+                                                strmPoll = ps}+                  "stop"   -> cmdStop ps >> setPsContinue False xp+                  "pause"  -> cmdPause s+                  "resume" -> return () -- ignore+                  "test"   -> putStrLn "test successful"+                  _        -> undefined+              _ -> throwIO $ Ouch "Ouch! Not a poller in handleCmd!"++  ------------------------------------------------------------------------+  -- Simplify parameter passing in streams+  ------------------------------------------------------------------------+  data PollState = PollSt {+                     polMap  :: Map Identifier Z.Poll,+                     polCmd  :: Z.Poll,+                     polIs   :: [Identifier],+                     polPs   :: [Z.Poll],+                     polTmo  :: Timeout,+                     polCont :: Bool}++  type PollT r = MVar PollState -> IO r++  psPolls :: PollT ([Identifier], [Z.Poll])+  psPolls m = withMVar m $ \p -> return (polIs p, polPs p)++  psMap :: PollT (Map Identifier Z.Poll)+  psMap m = withMVar m (return . polMap)++  psCmd :: PollT Z.Poll+  psCmd m = withMVar m (return . polCmd)++  psContinue :: PollT Bool+  psContinue m = withMVar m (return . polCont)++  setPsContinue :: Bool -> PollT ()+  setPsContinue t m = modifyMVar_ m $ \p -> return p{polCont = t}++  ------------------------------------------------------------------------+  -- Get first stream with input+  ------------------------------------------------------------------------+  getStream :: [Identifier] -> [Z.Poll] -> Maybe (Identifier, Z.Poll)+  getStream _ [] = Nothing+  getStream [] _ = Nothing+  getStream (i:is) (p:ps) = case p of+                              Z.S _ Z.In -> Just (i, p)+                              _          -> getStream is ps++  ------------------------------------------------------------------------+  -- Receive from poll entry+  ------------------------------------------------------------------------+  readPoll :: Z.Poll -> Source+  readPoll p = case p of+                 Z.S s _ -> recv s+                 _       -> liftIO $ throwIO $ Ouch "Ouch! No socket in poll"++  ------------------------------------------------------------------------+  -- Remove all messages from stream+  ------------------------------------------------------------------------+  cleanStream :: Z.Poll -> IO ()+  cleanStream p = case p of+                    Z.S s _ -> go s+                    _       -> return ()+    where go s = do+            m <- Z.moreToReceive s+            when m $ Z.receive s [] >>= \_ -> go s++  ------------------------------------------------------------------------+  -- Traditional receive+  ------------------------------------------------------------------------+  recv :: Z.Socket a -> Source +  recv s = liftIO (Z.receive s []) >>= \x -> do+           C.yield x+           m <- liftIO $ Z.moreToReceive s+           when m $ recv s++  ------------------------------------------------------------------------+  -- Receive with tmo+  ------------------------------------------------------------------------+  waitForRecv :: Z.Socket a -> Z.Timeout -> IO (Maybe (Z.Socket a))+  waitForRecv s tmo = Z.poll [Z.S s Z.In] tmo >>= \[s'] ->+                        case s' of+                          Z.S _ Z.In -> return $ Just s+                          _          -> return Nothing++  ------------------------------------------------------------------------+  -- | Receiver Sink: +  --   Internally a zeromq socket is waiting for input;+  --   when input is available, it is send to the sink.+  -- +  --   * 'Z.Socket a' - The source socket+  --+  --   * 'Timeout' - receiver timeout+  --     /< 0/ - listens eternally,+  --     /0/ - returns immediately,+  --     /> 0/ - timeout in microseconds;+  --     when the timeout expires, the stream terminates+  --      and the return value is Nothing.+  ------------------------------------------------------------------------+  runReceiver :: Z.Socket a -> Timeout -> SinkR (Maybe o) -> IO (Maybe o)+  runReceiver s tmo snk = do+    mb_ <- waitForRecv s tmo+    case mb_ of+      Nothing -> return Nothing+      Just  _ -> C.runResourceT $ recv s $$ snk++  ------------------------------------------------------------------------+  -- | Sender Source:+  --   The 'Source' generates a stream,+  --   which is relayed to the 'Z.Socket'.+  ------------------------------------------------------------------------+  runSender :: Z.Socket a -> Source -> IO ()+  runSender s src = C.runResourceT $ src $$ relay s++  ------------------------------------------------------------------------+  -- Create cmdName+  ------------------------------------------------------------------------+  cmdName :: Service -> IO String+  cmdName sn = do+    u <- show <$> mkUniqueId+    return $ "inproc://_" ++ sn ++ "_" ++ u+            +  ------------------------------------------------------------------------+  -- Creates a socket, binds or links it and sets the socket options+  ------------------------------------------------------------------------+  access :: Context -> AccessType -> LinkType -> [Z.SocketOption] ->+            String  -> [Service]  -> IO Z.Poll+  access ctx a l os u ts = +    case a of +      ServerT -> Z.socket ctx Z.Rep  >>= go+      ClientT -> Z.socket ctx Z.Req  >>= go+      DealerT -> Z.socket ctx Z.XReq >>= go+      RouterT -> Z.socket ctx Z.XRep >>= go+      PubT    -> Z.socket ctx Z.Pub  >>= go+      PipeT   -> Z.socket ctx Z.Push >>= go+      PullT   -> Z.socket ctx Z.Pull >>= go+      PeerT   -> Z.socket ctx Z.Pair >>= go+      SubT    -> Z.socket ctx Z.Sub  >>= \s -> +                   mapM_ (Z.subscribe s) ts >> go s+   where go s = do setSockOs s os+                   case l of+                     Bind    -> Z.bind s u+                     Connect -> trycon s u retries+                   return $ Z.S s Z.In++  ------------------------------------------------------------------------+  -- creates and binds or connects all sockets recursively;+  -- on its way, creates the Map from Identifiers to PollItems,+  --             a list of PollItems+  --             and a list of Identifiers with the same order;+  -- finally executes "run"+  ------------------------------------------------------------------------+  mkPoll :: Context               -> +            [PollEntry]           -> +            Map Identifier Z.Poll ->+            [Identifier]          ->+            [Z.Poll]              -> IO (Map Identifier Z.Poll, +                                         [Identifier], [Z.Poll])+  mkPoll _   []     m is ps = return (m, is, ps)+  mkPoll ctx (k:ks) m is ps = bracketOnError+    (access ctx (pollType k)+                (pollLink k) +                (pollOs   k) +                (pollAdd  k) +                (pollSub  k))+    (\p -> closeS p >> return (m, [], []))+    (\p -> mkPoll ctx ks (Map.insert (pollId k) p m) +                         (pollId k:is) (p:ps)) ++  ------------------------------------------------------------------------+  -- Close socket in a poll entry+  ------------------------------------------------------------------------+  closeS :: Z.Poll -> IO ()+  closeS p = case p of +               Z.S s _ -> safeClose s+               _       -> return ()++  ------------------------------------------------------------------------+  -- safely close a socket +  ------------------------------------------------------------------------+  safeClose :: Z.Socket a -> IO ()+  safeClose s = catch (Z.close s)+                      (\e -> let _ = (e::SomeException)+                              in return ())++  ------------------------------------------------------------------------+  -- handle stop command+  ------------------------------------------------------------------------+  cmdStop :: [Z.Poll] -> IO ()+  cmdStop = mapM_ closeS ++  ------------------------------------------------------------------------+  -- handle pause command+  ------------------------------------------------------------------------+  cmdPause :: Z.Socket a -> IO ()+  cmdPause s = do+    x <- B.unpack <$> Z.receive s []+    case x of+      "resume" -> return ()+      _        -> cmdPause s++  ------------------------------------------------------------------------+  -- handle send command+  ------------------------------------------------------------------------+  cmdSend :: Z.Socket a -> Streamer -> IO ()+  cmdSend cmd s = do+    ok <- Z.moreToReceive cmd+    if ok then do ds  <- getDest cmd []+                  C.runResourceT $ recv cmd $$ passAll s ds+          else throwIO $ ProtocolExc "Abrupt end of send command"++  ------------------------------------------------------------------------+  -- get destination to send to+  ------------------------------------------------------------------------+  getDest :: Z.Socket a -> [Identifier] -> IO [Identifier]+  getDest cmd is = do+    i <- B.unpack <$> Z.receive cmd []+    if null i then if null is then throwIO $ ProtocolExc "No identifiers" +                              else return is+              else do ok <- Z.moreToReceive cmd+                      if ok then getDest cmd (i:is)+                            else throwIO $ ProtocolExc +                                   "Incomplete identifier list"++  ------------------------------------------------------------------------+  -- relay stream to a socket+  ------------------------------------------------------------------------+  relay :: Z.Socket a -> Sink+  relay s = do+    mbX <- C.await+    case mbX of+      Nothing -> return ()+      Just x  -> go x+     where go x = do+             mbN <- C.await +             case mbN of+               Nothing -> liftIO (Z.send s x [])+               Just n  -> liftIO (Z.send s x [Z.SndMore]) >> go n+  +  ------------------------------------------------------------------------+  -- send one message to many streams+  ------------------------------------------------------------------------+  multiSend :: [(Identifier, Z.Poll)] -> B.ByteString -> [Z.Flag] -> IO ()+  multiSend ps m os = go ps+    where go []         = return ()+          go ((_,p):pp) = sndPoll p >> go pp+          sndPoll p = case p of+                        Z.S s _ -> Z.send s m os+                        _       -> throwIO $ Ouch "Ouch! Not a Poll!"++  ------------------------------------------------------------------------+  -- Find sockets corresponding to identifiers+  ------------------------------------------------------------------------+  idsToSocks :: Streamer -> [Identifier] -> Either String [(Identifier, Z.Poll)]+  idsToSocks s = getSocks +    where getSock i | i == internal = Right $ strmCmd s+                    | otherwise     =+            case Map.lookup i (strmIdx s) of+              Nothing -> Left $ "Unknown identifier " ++ i +              Just p  -> Right p+          getSocks []     = Right []+          getSocks (i:is) = +            case getSock i of+              Left  e -> Left e+              Right p -> +                case getSocks is of+                  Left  e  -> Left e+                  Right ps -> Right ((i,p):ps)+       +  {- $streamer+     A streamer represents the current state of the streaming device+     started by means of 'withStreams'. It is passed in to +     application-defined callbacks, namely the timeout action +     ('StreamAction') and the 'Sink' ('StreamSink').++     There is a bunch of useful sinks that receive a streamer as input+     (see below). ++  -}+  ------------------------------------------------------------------------+  -- | Holds information on streams and the current state of the streamer,+  --   /i.e./ the current source.+  --   Streamers are passed to processing conduits.+  ------------------------------------------------------------------------+  data Streamer = Streamer {+                     strmSrc  :: Maybe (Identifier, Z.Poll),+                     strmIdx  :: Map Identifier Z.Poll,+                     strmCmd  :: Z.Poll,+                     strmPoll :: [Z.Poll]}++  ------------------------------------------------------------------------+  -- | Conduit with Streamer+  ------------------------------------------------------------------------+  type StreamConduit = Streamer -> Conduit B.ByteString ()++  ------------------------------------------------------------------------+  -- | Sink with Streamer+  ------------------------------------------------------------------------+  type StreamSink    = Streamer -> Sink++  ------------------------------------------------------------------------+  -- | IO Action with Streamer (/e.g./ Timeout action)+  ------------------------------------------------------------------------+  type StreamAction  = Streamer -> IO ()++  ------------------------------------------------------------------------+  -- | Get current source+  ------------------------------------------------------------------------+  getSource :: Streamer -> Identifier+  getSource s = case strmSrc s of+                  Nothing    -> ""+                  Just (i,_) -> i++  ------------------------------------------------------------------------+  -- | Filter subset of streams; usually you want to filter+  --   a subset of streams to which to relay an incoming stream.+  --   Note that the result is just a list of stream identifiers,+  --   which of course could be used directly in the first place.+  --   A meaningful use of filterstreams would be, for instance:+  --+  --   > let targets = filterStreams s (/= getSource s)+  --+  --   Where all streams but the source are selected.+  ------------------------------------------------------------------------+  filterStreams :: Streamer -> (Identifier -> Bool) -> [Identifier]+  filterStreams s p = map fst $ Map.toList $ +                                Map.filterWithKey (\k _ -> p k) $ strmIdx s++  {- $sinks+     To manipulate and relay incoming streams,+     the application passes a 'StreamSink' to 'withStreams'.+     The following sinks are building blocks for more +     application-focused manipulations.+     +     The peculiarities of the zeromq library,+     in particular the fact that messages are sent+     entirely, /i.e./ with all segments belonging to the same message,+     or not at all,+     require some care in designing zeromq sinks.+     The sink must ensure to mark the last segment sent +     (see 'Z.SndMore').+     Also, the incoming stream should be exhausted +     to avoid message segements lingering around in the pipe.++     Applications can construct new sinks by either+     calling a building block in the their own sink code, /e.g./:++     > example :: [B.ByteString] -> StreamSink+     > example headers s is = do+     >   mbX <- C.await+     >   case mbX of +     >     Nothing -> return ()+     >     Just x  -> do stream  s is headers+     >                   passAll s is++     or by combining a sink with a conduit forming+     a more complex sink, /e.g./:++     > example :: StreamSink+     > example s is = sourceList headers =$ passAll s is+  -}++  ------------------------------------------------------------------------+  -- | Send the 'ByteString' segments to the outgoing streams+  --   identified by ['Identifier'].+  --   The stream is terminated.+  ------------------------------------------------------------------------+  stream :: Streamer -> [Identifier] -> [B.ByteString] -> Sink+  stream s ds ms = +    case idsToSocks s ds of+      Left e   -> liftIO $ throwIO $ ProtocolExc e +      Right ss -> go ms ss+    where go [] _      = return ()+          go [x] ss    = liftIO (multiSend ss x [])+          go (x:xs) ss = liftIO (multiSend ss x [Z.SndMore]) >> go xs ss++  ------------------------------------------------------------------------+  -- | Send the 'ByteString' segments to the outgoing streams+  --   identified by ['Identifier']+  --   without terminating the stream,+  --   /i.e./ more segments must be sent.+  ------------------------------------------------------------------------+  part :: Streamer -> [Identifier] -> [B.ByteString] -> Sink+  part s ds ms = +    case idsToSocks s ds of+      Left  e  -> liftIO $ throwIO $ ProtocolExc e+      Right ss -> go ss+    where go ss = liftIO $ mapM_ (\x -> multiSend ss x [Z.SndMore]) ms++  ------------------------------------------------------------------------+  -- | Pass all segments of an incoming stream +  --   to a list of outgoing streams.+  --   The stream is terminated.+  ------------------------------------------------------------------------+  passAll :: Streamer -> [Identifier] -> Sink+  passAll s ds = +    case idsToSocks s ds of +      Left e   -> liftIO $ throwIO $ ProtocolExc e +      Right ss -> do mbI <- C.await -- C.awaitForever $ \i -> do+                     case mbI of+                       Nothing -> return ()+                       Just i  -> go ss i+    where go ss i = do +            mbN <- C.await+            case mbN of+              Nothing -> liftIO (multiSend ss i [])+              Just n  -> liftIO (multiSend ss i [Z.SndMore]) >> go ss n++  ------------------------------------------------------------------------+  -- | Pass one segment and ignore the remainder of the stream.+  --   The stream is terminated.+  ------------------------------------------------------------------------+  pass1 :: Streamer -> [Identifier] -> Sink+  pass1 s ds = +    case idsToSocks s ds of+      Left e   -> liftIO $ throwIO $ ProtocolExc e +      Right ss -> do+        mbX <- C.await +        case mbX of+          Nothing -> return ()+          Just x  -> liftIO (multiSend ss x []) >> ignoreStream++  ------------------------------------------------------------------------+  -- | Pass n segments and ignore the remainder of the stream.+  --   The stream is terminated.+  ------------------------------------------------------------------------+  passN :: Streamer -> [Identifier] -> Int -> Sink+  passN s ds n | n <= 0    = ignoreStream+               | otherwise =  +    case idsToSocks s ds of+      Left e   -> liftIO $ throwIO $ ProtocolExc e+      Right ss -> do mbX <- C.await +                     case mbX of+                       Nothing -> return ()+                       Just x  -> go ss n x+    where go ss i x | i <= 0 = return ()+                    | i == 1 = do liftIO (multiSend ss x []) +                                  ignoreStream+                    | otherwise = do+                        mbY <- C.await+                        case mbY of+                          Nothing -> liftIO (multiSend ss x [])+                          Just y  -> do liftIO (multiSend ss x [Z.SndMore])+                                        go ss (i-1) y++  ------------------------------------------------------------------------+  -- | Pass while condition is true and ignore the remainder of the stream.+  --   The stream is terminated.+  ------------------------------------------------------------------------+  passWhile :: Streamer -> [Identifier] -> (B.ByteString -> Bool) -> Sink+  passWhile s ds f =+    case idsToSocks s ds of+      Left e   -> liftIO $ throwIO $ ProtocolExc e+      Right ss -> do mbX <- C.await +                     case mbX of+                       Nothing -> return ()+                       Just x  | f x       -> go ss x+                               | otherwise -> return ()+    where go ss x = do+            mbY <- C.await+            case mbY of+              Nothing -> liftIO (multiSend ss x [])+              Just y  | f y       -> do liftIO (multiSend ss x [Z.SndMore])+                                        go ss y+                      | otherwise -> do liftIO (multiSend ss x [])+                                        ignoreStream++  ------------------------------------------------------------------------+  -- | Ignore an incoming stream+  ------------------------------------------------------------------------+  ignoreStream :: Sink+  ignoreStream = do mb <- C.await+                    case mb of +                      Nothing -> return ()+                      Just _  -> ignoreStream++  {- $controller+     The controller is passed in to the control action of 'withStreams'.+     It allows the application to control the polling device.+     Through the controller, the device can be stopped, restarted,+     paused and resumed and it is possible to send and receive+     streams through the controler.+     To relay streams to the controller+     (/i.e./ directly to application code) the 'internal' stream,+     which is identified by the string \"_internal\"+     can be used.+  -}+  ------------------------------------------------------------------------+  -- | The internal stream that represents the 'Controller'.+  --   StreamSinks can write to this stream, /e.g./:+  --+  --   > passAll s [internal]+  --+  --   And the streamer may also receive from this stream, /e.g./:+  --+  --   > if getSource s == internal+  ------------------------------------------------------------------------+  internal :: Identifier+  internal = "_internal" ++  ------------------------------------------------------------------------+  -- | Controller+  ------------------------------------------------------------------------+  data Controller = Controller {+                      ctrlCtx  :: Context,+                      ctrlCmd  :: Z.Socket Z.XReq,+                      ctrlOpts :: [Z.SocketOption]}++  ------------------------------------------------------------------------+  -- | Control Action+  ------------------------------------------------------------------------+  type Control a = Controller -> IO a++  ------------------------------------------------------------------------+  -- Get the socket from a controller and send a string to it+  ------------------------------------------------------------------------+  sndCmd :: String -> Controller -> IO ()+  sndCmd cmd ctrl = let s = ctrlCmd ctrl+                     in Z.send s (B.pack cmd) []++  ------------------------------------------------------------------------+  -- | Stop streams+  ------------------------------------------------------------------------+  stop :: Controller -> IO ()+  stop = sndCmd "stop"++  ------------------------------------------------------------------------+  -- | Pause streams+  ------------------------------------------------------------------------+  pause :: Controller -> IO ()+  pause = sndCmd "pause"++  ------------------------------------------------------------------------+  -- | Resume streams+  ------------------------------------------------------------------------+  resume :: Controller -> IO ()+  resume = sndCmd "resume"++  ------------------------------------------------------------------------+  -- | Receive a stream through the controller+  --   that was sink\'d to the target 'internal'.+  ------------------------------------------------------------------------+  receive :: Controller -> Timeout -> SinkR (Maybe a) -> IO (Maybe a)+  receive c tmo snk = do+    mb_ <- waitForRecv (ctrlCmd c) tmo+    case mb_ of +      Nothing -> return Nothing+      Just _  -> C.runResourceT $ recv (ctrlCmd c) $$ snk++  ------------------------------------------------------------------------+  -- | Send a stream through the controller+  ------------------------------------------------------------------------+  send :: Controller -> [Identifier] -> Source -> IO ()+  send c is src = +    let s = ctrlCmd c+     in do Z.send s (B.pack "send") [Z.SndMore]+           sendIds s+           C.runResourceT $ src $$ relay s+    where sendIds s  = do mapM_ (sendId s) is+                          Z.send s B.empty [Z.SndMore]+          sendId s i = Z.send s (B.pack i) [Z.SndMore]++  ------------------------------------------------------------------------+  -- | A poll entry describes how to access and identify a socket+  ------------------------------------------------------------------------+  data PollEntry = Poll {+         -- | How to address this particular stream+         pollId   :: Identifier,+         -- | The address to link to+         pollAdd  :: String,+         -- | The zeromq socket type+         pollType :: AccessType,+         -- | How to link (bind or connect)+         pollLink :: LinkType,+         -- | List of 'Service' (or topics)+         --   for subscribers+         pollSub  :: [Service],+         -- | zeromq socket options+         pollOs   :: [Z.SocketOption]+       }+    deriving (Show, Read)++  instance Eq PollEntry where+    x == y = pollId x == pollId y++  ------------------------------------------------------------------------+  -- | Defines the type of a 'PollEntry';+  --   the names of the constructors are similar +  --   to the corresponding ZMQ socket types.+  ------------------------------------------------------------------------+  data AccessType = +         -- | Represents a server and expects connections from clients;+         --   corresponds to ZMQ Socket Type 'Z.Rep'+         ServerT    +         -- | Represents a client and connects to a server;+         --   corresponds to ZMQ Socket Type 'Z.Req'+         | ClientT+         -- | Represents a load balancer, +         --   expecting connections from clients;+         --   corresponds to ZMQ Socket Type 'Z.XRep'+         | RouterT +         -- | Represents a router+         --   expecting connections from servers;+         --   corresponds to ZMQ Socket Type 'Z.XReq'+         | DealerT +         -- | Represents a publisher;+         --   corresponds to ZMQ Socket Type 'Z.Pub'+         | PubT    +         -- | Represents a subscriber;+         --   corresponds to ZMQ Socket Type 'Z.Sub'+         | SubT    +         -- | Represents a Pipe;+         --   corresponds to ZMQ Socket Type 'Z.Push'+         | PipeT+         -- | Represents a Puller;+         --   corresponds to ZMQ Socket Type 'Z.Pull'+         | PullT+         -- | Represents a Peer;+         --   corresponds to ZMQ Socket Type 'Z.Pair'+         | PeerT   +    deriving (Eq, Show, Read)++  ------------------------------------------------------------------------+  -- | Safely read 'AccessType';+  --   ignores the case of the input string+  --   (/e.g./ \"servert\" -> 'ServerT')+  ------------------------------------------------------------------------+  parseAccess :: String -> Maybe AccessType+  parseAccess s = case map toLower s of+                    "servert" -> Just ServerT    +                    "clientt" -> Just ClientT+                    "routert" -> Just RouterT +                    "dealert" -> Just DealerT +                    "pubt"    -> Just PubT    +                    "subt"    -> Just SubT    +                    "pipet"   -> Just PipeT+                    "pullt"   -> Just PullT+                    "peert"   -> Just PeerT+                    _         -> Nothing++  -------------------------------------------------------------------------+  -- | Safely read 'LinkType';+  --   ignores the case of the input string+  --   and, besides \"bind\" and \"connect\", +  --   also accepts \"bin\", \"con\" and \"conn\";+  --   intended for use with command line parameters+  -------------------------------------------------------------------------+  parseLink :: String -> Maybe LinkType +  parseLink s = case map toLower s of+                  "bind"    -> Just Bind+                  "bin"     -> Just Bind+                  "b"       -> Just Bind+                  "c"       -> Just Connect+                  "con"     -> Just Connect+                  "conn"    -> Just Connect+                  "connect" -> Just Connect+                  _         -> Nothing++  -------------------------------------------------------------------------+  -- | Binds or connects a socket to an address+  -------------------------------------------------------------------------+  link :: LinkType -> Z.Socket a -> String -> [Z.SocketOption] -> IO ()+  link t s add os = do setSockOs s os+                       case t of+                         Bind    -> Z.bind s add +                         Connect -> trycon s add 10++  ------------------------------------------------------------------------+  -- some helpers+  ------------------------------------------------------------------------+  retries :: Int+  retries = 100++  ------------------------------------------------------------------------+  -- try n times to connect+  -- this is particularly useful for "inproc" sockets:+  -- the socket, in this case, must be bound before we can connect to it.+  ------------------------------------------------------------------------+  trycon :: Z.Socket a -> String -> Int -> IO ()+  trycon sock add i = catch (Z.connect sock add) +                            (\e -> if i <= 0 +                                     then throwIO (e::SomeException)+                                     else do+                                       threadDelay 1000+                                       trycon sock add (i-1))++  -------------------------------------------------------------------------+  -- Sets Socket Options+  -------------------------------------------------------------------------+  setSockOs :: Z.Socket a -> [Z.SocketOption] -> IO ()+  setSockOs s = mapM_ (Z.setOption s)++  {- $example+     The following code implements a ping pong communication+     using two streamers.+     The code is somewhat simplistic;+     it does not use timeout,+     ignores errors and+     does not provide means for clean shutdown.+     It focuses instead on demonstrating the core of+     the streamer functionality.+     +     For more examples on how to use streams,+     you may want to refer to the MDP Broker code in+     Network.Mom.Patterns.Broker.Broker.++    > import           Control.Monad.Trans+    > import           Control.Monad (forever)+    > import           Control.Concurrent+    > import qualified Data.Conduit          as C+    > import qualified Data.ByteString.Char8 as B+    > import           Network.Mom.Patterns.Streams +    > import qualified System.ZMQ as Z++    > main :: IO ()+    > main = Z.withContext 1 $ \ctx -> do+    >          ready <- newEmptyMVar+    >          _ <- forkIO (ping ctx ready)+    >          _ <- forkIO (pong ctx ready)+    >          forever $ threadDelay 100000++    > ping :: Z.Context -> MVar () -> IO ()+    > ping ctx ready = withStreams ctx "pong" (-1)+    >                        [Poll "ping" "inproc://ping" PeerT Bind [] []]+    >                        (\_ -> return ())      -- no timeout+    >                        (\_ _ _ -> return ())  -- ignore errors+    >                        pinger $ \c -> do+    >                    putMVar ready () -- ping is ready+    >                    putStrLn "starting game!"+    >                    send c ["ping"] startPing -- send through controller+    >                                              -- to initialise ping pong+    >                    putStrLn "game started!"+    >                    forever $ threadDelay 100000+    >   where startPing = C.yield $ B.pack "ping"++    > pong :: Z.Context -> MVar () -> IO ()+    > pong ctx ready = do +    >   _ <- takeMVar ready -- wait for ping getting ready+    >   withStreams ctx "ping" (-1)+    >               [Poll "pong" "inproc://ping" PeerT Connect [] []]+    >               (\_ -> return ())+    >               (\_ _ _ -> return ())+    >               pinger $ \_ -> forever $ threadDelay 100000++    > pinger :: StreamSink+    > pinger s = C.awaitForever $ \i -> +    >              let x = B.unpack i +    >               in do liftIO $ putStrLn x+    >                     liftIO $ threadDelay 500000+    >                     case x of+    >                       "ping" -> stream s ["pong"] [B.pack "pong"]+    >                       "pong" -> stream s ["ping"] [B.pack "ping"]+    >                       _      -> return ()+  -}+

@@ -0,0 +1,165 @@+{-# LANGUAGE DeriveDataTypeable,RankNTypes #-}+-------------------------------------------------------------------------------+-- |+-- Module     : Network/Mom/Patterns/Types.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+--+-- Fundamental streaming types +-------------------------------------------------------------------------------+module Network.Mom.Patterns.Types+where++  import           Prelude hiding (catch)+  import           Control.Exception (Exception, SomeException) +  import           Data.Typeable (Typeable)+  import qualified Data.Conduit          as C+  import qualified Data.ByteString.Char8 as B+  import qualified System.ZMQ            as Z++  ------------------------------------------------------------------------+  -- * Conduits+  ------------------------------------------------------------------------+  -- | The IO Resource transformer.+  --   See the conduit package for details+  type RIO     = C.ResourceT IO++  -- | A stream source+  type Source  = C.Source RIO B.ByteString  ++  -- | A stream sink without return type+  type Sink    = C.Sink B.ByteString RIO () ++  -- | A stream sink wit return type+  type SinkR r = C.Sink B.ByteString RIO r ++  -- | A conduit that links source and sink+  --   applying some transformation to the stream.+  --   Input is always 'B.ByteString',+  --   output and return type may vary.+  type Conduit o r = C.ConduitM B.ByteString o RIO r++  -- | Simplified Conduit where output+  --   is always 'B.ByteString' +  --   and no final value is returned.+  type Conduit_ = Conduit B.ByteString ()++  -- | Streaming the elements of a list+  streamList :: [B.ByteString] -> C.Producer RIO B.ByteString +  streamList = mapM_ C.yield ++  -- | Pass the stream through without applying+  --   any transformation to it+  passThrough :: Conduit B.ByteString ()+  passThrough = C.awaitForever C.yield ++  ------------------------------------------------------------------------+  -- * Link Type+  ------------------------------------------------------------------------+  -------------------------------------------------------------------------+  -- | A /zeromq/ 'AccessPoint'+  --   can be bound or connected to its address.+  --   Only one peer can bind the address,+  --   all other parties have to connect.+  -------------------------------------------------------------------------+  data LinkType = +         -- | Bind the address+         Bind +         -- | Connect to the address+         | Connect+    deriving (Show, Read)++  ------------------------------------------------------------------------+  -- * Error Handling+  ------------------------------------------------------------------------+  ------------------------------------------------------------------------+  -- | Error handler for all services +  --   that are implemented as background services,+  --   /e.g./ servers and brokers.+  --   The handler receives the 'Criticality' of the error event,+  --   the exception and an additional descriptive string.+  --   +  --   A good policy is+  --   to terminate or restart the service+  --   when a 'Fatal' or 'Critical' error occurs+  --   and to continue, if possible,+  --   on a plain 'Error'.+  --   The error handler may perform additional, user-defined actions, +  --   such as logging the incident or +  --   sending an SMS.+  ------------------------------------------------------------------------+  type OnError_  = Criticality         -> +                   SomeException       -> +                   String              -> IO ()++  -------------------------------------------------------------------------+  -- | Indicates criticality of the error event+  -------------------------------------------------------------------------+  data Criticality = +                     -- | The current operation +                     --   (/e.g./ processing a request)+                     --   has not terminated properly,+                     --   but the service is able to continue;+                     --   the error may have been caused by a faulty+                     --   request or other temporal conditions.+                     Error +                     -- | The event has impact on the process,+                     --   leaving it in an unkown state.+                   | Critical +                     -- | The service cannot recover and will terminate+                   | Fatal+    deriving (Eq, Ord, Show, Read)++  -------------------------------------------------------------------------+  -- | Stream Exception+  -------------------------------------------------------------------------+  data StreamException = +              -- | low-level error+              SocketExc   String+              -- | IO error+              | IOExc       String+              -- | Protocol error+              | ProtocolExc String +              -- | Application-defined error+              | AppExc      String+              -- | Internal error, indicating a code error in library+              | Ouch        String+    deriving (Show, Read, Typeable, Eq)++  instance Exception StreamException++  ------------------------------------------------------------------------+  -- * Some convenient definitions+  ------------------------------------------------------------------------+  -- | String identifying a stream in the streams device+  type Identifier = String++  -- | String identifying a service provided, /e.g./ by a /server/+  type Service    = String++  -- | Identity of a communication peer,+  --   needed for complex patterns (/e.g./ broker)+  type Identity    = B.ByteString++  -- | Message body,+  --   needed for complex patterns (/e.g./ broker)+  type Body        = [B.ByteString]++  ------------------------------------------------------------------------+  -- | Milliseconds +  ------------------------------------------------------------------------+  type Msec = Int++  -- | Reexport from zeromq (timeout in microseconds)+  type Timeout    = Z.Timeout+  -- | Reexport from zeromq+  type Context    = Z.Context+  -- | Reexport from zeromq+  type Size       = Z.Size++  -- | Reexport from zeromq+  withContext :: Size -> (Context -> IO a) -> IO a+  withContext = Z.withContext+

@@ -1,378 +0,0 @@-module Service (-          Service, srvContext, srvName, srvId,-          stop, pause, resume, changeParam, changeOption,-          addDevice, remDevice, changeTimeout,-          withService, poll, xpoll, XPoll(..),-          periodic, periodicSend-          , Command(..), DevCmd(..) -- if test-          )-where--  import           Factory-  import           Types--  import qualified Data.ByteString.Char8 as B-  import           Data.Time.Clock-  import           Data.List (isPrefixOf)-  import           Data.Map (Map)-  import qualified Data.Map as Map--  import           Control.Concurrent -  import           Control.Applicative ((<$>))-  import           Control.Monad-  import           Prelude hiding (catch)-  import           Control.Exception (bracket, finally)-  import qualified System.ZMQ as Z--  -------------------------------------------------------------------------  -- | Generic Service data type;-  --   'Service' is passed to application-defined actions-  --   used with background services, namely-  --   withServer, withPeriodicPub, withSub, withPuller and withDevice. -  -------------------------------------------------------------------------  data Service = Service {-                   srvCtx    :: Z.Context,-                   -- | Obtains the service name-                   srvName   :: String,-                   srvCmd    :: Z.Socket Z.Pub,-                   srvId     :: ThreadId-                 }--  -------------------------------------------------------------------------  -- | Obtains the 'Z.Context' from 'Service'-  -------------------------------------------------------------------------  srvContext :: Service -> Z.Context-  srvContext = srvCtx--  -------------------------------------------------------------------------  -- Stops a service - used internally only-  -------------------------------------------------------------------------  stop  :: Service -> IO ()-  stop = sendCmd STOP--  -------------------------------------------------------------------------  -- | Pauses the 'Service'-  -------------------------------------------------------------------------  pause :: Service -> IO ()-  pause = sendCmd PAUSE--  -------------------------------------------------------------------------  -- | Resumes the 'Service'-  -------------------------------------------------------------------------  resume :: Service -> IO ()-  resume = sendCmd RESUME--  -------------------------------------------------------------------------  -- | Changes the 'Service' control parameter-  -------------------------------------------------------------------------  changeParam :: Service -> Parameter -> IO ()-  changeParam s c = sendCmd (APP c) s--  -------------------------------------------------------------------------  -- | Changes SocketOption-  -------------------------------------------------------------------------  changeOption :: Service -> Z.SocketOption -> IO ()-  changeOption s o = sendCmd (OPT o) s--  -------------------------------------------------------------------------  -- | Adds a 'PollEntry' to a device;-  --   the 'Service', of course, must be a device, -  --   the command is otherwise ignored.-  -------------------------------------------------------------------------  addDevice  :: Service -> PollEntry -> IO ()-  addDevice s p = sendDevCmd (ADD p) s--  -------------------------------------------------------------------------  -- | Removes a 'PollEntry' from a device;-  --   the 'Service', of course, must be a device, -  --   the command is otherwise ignored.-  -------------------------------------------------------------------------  remDevice :: Service -> Identifier -> IO ()-  remDevice s i = sendDevCmd (REM i) s--  -------------------------------------------------------------------------  -- | Changes the timeout of a device;-  --   the 'Service', of course, must be a device,-  --   the command is otherwise ignored.-  -------------------------------------------------------------------------  changeTimeout :: Service -> Timeout -> IO ()-  changeTimeout s t = sendDevCmd (TMO t) s--  -------------------------------------------------------------------------  -- Service commands-  -------------------------------------------------------------------------  data Command = STOP | PAUSE  | RESUME -               | DEVICE DevCmd      -- device specific commands-               | APP String         -- change control parameter -               | OPT Z.SocketOption -- change socket option-    deriving (Eq, Show, Read)--  -------------------------------------------------------------------------  -- Device-specific commands-  -------------------------------------------------------------------------  data DevCmd = ADD PollEntry | REM Identifier | TMO Z.Timeout-    deriving (Eq, Show, Read)--  -------------------------------------------------------------------------  -- Send a command-  -------------------------------------------------------------------------  sendCmd :: Command -> Service -> IO ()-  sendCmd c s = Z.send (srvCmd s) (B.pack $ show c) []--  -------------------------------------------------------------------------  -- Send device-specific command-  -------------------------------------------------------------------------  sendDevCmd :: DevCmd -> Service -> IO ()-  sendDevCmd d = sendCmd (DEVICE d)--  -------------------------------------------------------------------------  -- Parse a command string-  -------------------------------------------------------------------------  readCmd :: String -> Either String Command-  readCmd s = case s of-               "STOP"   -> Right STOP-               "PAUSE"  -> Right PAUSE-               "RESUME" -> Right RESUME-               x        -> -                 if "APP"    `isPrefixOf` x ||-                    "OPT"    `isPrefixOf` x ||-                    "DEVICE" `isPrefixOf` x-                   then Right $ read x-                   else Left  $ "No Command: " ++ x--  -------------------------------------------------------------------------  -- The work horse behind "with*" services-  -- - starts the service in a separate thread and-  --   waits until it is ready-  -- - executes the control action-  -- - stops the service and waits for its termination-  -------------------------------------------------------------------------  withService :: Z.Context -> String -> String -> -                (Z.Context -> String -> String -> String -> IO () -> IO ()) ->-                (Service -> IO a) -> IO a-  withService ctx name param service action = do-    running <- newEmptyMVar-    ready   <- newEmptyMVar-    Z.withSocket ctx Z.Pub $ \cmd -> do-      sn <- ("inproc://srv_" ++) <$> show <$> mkUniqueId-      Z.bind cmd sn-      bracket (start sn cmd ready running) -              (\s -> stop s >> takeMVar running)-              (doAction ready)-    where start sn cmd ready m = do-            let imReady = putMVar ready ()-            tid <- forkIO $ finally (service ctx name sn param imReady) -                                    (putMVar m ())-            return $ Service ctx name cmd tid-          doAction ready srv = takeMVar ready >>= \_ -> action srv--  -------------------------------------------------------------------------  -- Poll on a command socket and the service socket-  -------------------------------------------------------------------------  poll :: Bool -> [Z.Poll] -> (String -> IO ()) -> String -> IO ()-  poll paused poller rcv param -    | paused    = handleCmd paused poller rcv param-    | otherwise = do-        [c, s] <- Z.poll poller (-1)-        case c of -          Z.S _ Z.In -> handleCmd paused poller rcv param-          _          -> -            case s of-              Z.S _ Z.In -> rcv param >> poll paused poller rcv param-              _          ->              poll paused poller rcv param--  -------------------------------------------------------------------------  -- Handle a message received on the command socket-  -------------------------------------------------------------------------  handleCmd :: Bool -> [Z.Poll] -> (String -> IO ()) -> String -> IO ()-  handleCmd paused poller@[Z.S sock _, _] rcv param = do-    x <- Z.receive sock []-    case readCmd $ B.unpack x of-      Left  _   -> poll paused poller rcv param -- ignore-      Right cmd -> case cmd of-                     STOP   -> return ()-                     PAUSE  -> poll True   poller rcv param-                     RESUME -> poll False  poller rcv param-                     APP p  -> poll paused poller rcv p-                     OPT o  -> changeOpt poller o >> -                               poll paused poller rcv param-                     _      -> poll paused poller rcv param -- ignore-  handleCmd _ _ _ _ = ouch "invalid poller in 'handleCmd'!"--  -------------------------------------------------------------------------  -- Change a socket option-  -------------------------------------------------------------------------  changeOpt :: [Z.Poll] -> Z.SocketOption -> IO ()-  changeOpt (_:Z.S s _:_) o = Z.setOption s o-  changeOpt _             _ = return ()--  -------------------------------------------------------------------------  -- XPoll descriptor for device services-  -------------------------------------------------------------------------  data XPoll = XPoll {-                 xpCtx   :: Z.Context,-                 xpTmo   :: Z.Timeout,-                 xpMap   :: Map Identifier Z.Poll,-                 xpIds   :: [Identifier],-                 xpPoll  :: [Z.Poll]-               }--  -------------------------------------------------------------------------  -- Remove a poll entry-  -------------------------------------------------------------------------  xpDelete :: Identifier -> XPoll -> XPoll-  xpDelete i xp = let (p:pp)     = xpPoll xp-                      (is, ps)   = go (xpIds xp) pp-                   in xp {xpMap  = Map.delete i $ xpMap xp,-                          xpIds  = is,-                          xpPoll = p:ps}-    where  go _        []     = ([], [])-           go []       _      = ([], [])-           go (d:ds) (p:ps) = -             if i == d then              go ds ps-               else let (  ds',   ps') = go ds ps-                     in (d:ds', p:ps')--  -------------------------------------------------------------------------  -- Polling for device-based services:-  -- - a command socket-  -- - and a set of device sockets-  -------------------------------------------------------------------------  xpoll :: Bool -> MVar XPoll -> -           (String -> IO ()) ->-           (Identifier -> Z.Poll -> String -> IO ()) -> String -> IO ()-  xpoll paused mxp ontmo rcv param -    | paused    = handleCmdX paused mxp ontmo rcv param-    | otherwise = do-        xp     <- readMVar mxp-        (c:ss) <- Z.poll (xpPoll xp) (xpTmo xp)-        case c of -          Z.S _ Z.In -> handleCmdX paused mxp ontmo rcv param-          _          -> go (xpIds xp) ss-    where go _      []     = ontmo param >>-                             xpoll paused mxp ontmo rcv param-          go (i:is) (s:ss) =-            case s of-              Z.S _ Z.In -> rcv i s param >>-                            xpoll paused mxp ontmo rcv param-              _          -> go is ss-          go _      _     = ouch "Invalid xpoll entries"--  -------------------------------------------------------------------------  -- Handle messages received through the command socket-  -- of a device service-  -------------------------------------------------------------------------  handleCmdX :: Bool -> MVar XPoll    -> -                (String -> IO ())     -> -                (Identifier -> Z.Poll -> String -> IO ()) -> String -> IO ()-  handleCmdX paused mxp ontmo rcv param = do-    xp <- readMVar mxp-    case xpPoll xp of-      (Z.S sock _ : _) -> do-        x <- Z.receive sock []-        case readCmd $ B.unpack x of-          Left e    -> do putStrLn $ e ++ ": " ++ B.unpack x-                          xpoll paused mxp ontmo rcv param-          Right cmd -> case cmd of-                         STOP     -> return ()-                         PAUSE    -> xpoll True   mxp ontmo rcv param-                         RESUME   -> xpoll False  mxp ontmo rcv param-                         APP p    -> xpoll paused mxp ontmo rcv p-                         OPT _    -> xpoll paused mxp ontmo rcv param -- opt!-                         DEVICE d -> do modifyMVar_ mxp -                                          (\_ -> handleDevCmd d xp)-                                        xpoll False mxp ontmo rcv param-      _ -> ouch "invalid poller in 'handleCmdX'!"--  -------------------------------------------------------------------------  -- Handle a device command-  -------------------------------------------------------------------------  handleDevCmd :: DevCmd -> XPoll -> IO XPoll-  handleDevCmd d xp = -    case d of-      TMO t -> return   xp {xpTmo = t}-      REM i -> case Map.lookup i (xpMap xp) of-                 Just (Z.S s _) -> safeClose s >> return (xpDelete i xp)-                 _              -> return xp-      ADD p -> do-        s <- access (xpCtx   xp)-                    (pollType p) -                    (pollLink p) -                    (pollOs   p) -                    (pollAdd  p) -                    (pollSub  p)-        case xpPoll xp of-          (c:ss) -> do let i = pollId p-                       return xp {xpPoll = c:s:ss,-                                  xpIds  = i:xpIds xp,-                                  xpMap  = Map.insert i s $ xpMap xp}-          _      -> return xp--  -------------------------------------------------------------------------  -- Publish periodically-  -------------------------------------------------------------------------  periodicSend :: Bool -> Z.Timeout -> Z.Socket Z.Sub -> (String -> IO ()) -> String -> IO ()-  periodicSend paused period cmd send param = do-    release <- getCurrentTime-    periodicSend_ paused period release cmd send param--  periodicSend_ :: Bool -> Z.Timeout -> UTCTime -> Z.Socket Z.Sub -> (String -> IO ()) -> String -> IO ()-  periodicSend_ paused period release cmd send param-    | paused    = handleCmdSnd True period release cmd send param -    | otherwise = send param >> handleCmdSnd paused period release cmd send param --  -------------------------------------------------------------------------  -- Poll on a publisher's command socket-  -------------------------------------------------------------------------  handleCmdSnd :: Bool -> Z.Timeout -> UTCTime -> Z.Socket Z.Sub -> (String -> IO ()) -> String -> IO ()-  handleCmdSnd paused period release sock send param = do-    [Z.S _ evt] <- Z.poll [Z.S sock Z.In] 0-    case evt of -      Z.In   -> do-        x        <- Z.receive sock []-        release' <- waitNext period release-        case readCmd $ B.unpack x of-          Left  _   -> periodicSend_ paused period release' sock send param-          Right cmd -> -            case cmd of-              STOP   -> return ()-              PAUSE  -> periodicSend_ True   period release' sock send param-              RESUME -> periodicSend_ False  period release' sock send param-              APP p  -> periodicSend_ paused period release' sock send p-              _      -> periodicSend_ paused period release' sock send param-      _ -> do-        release' <- waitNext period release-        periodicSend_ paused period release' sock send param--  -------------------------------------------------------------------------  -- Doing something periodically-  -------------------------------------------------------------------------  periodic :: Z.Timeout -> IO () -> IO ()-  periodic period act = getCurrentTime                 >>= go-    where go release  = act >> waitNext period release >>= go --  -------------------------------------------------------------------------  -- Wait for the next release point-  -------------------------------------------------------------------------  waitNext :: Z.Timeout -> UTCTime -> IO UTCTime-  waitNext period release = do-     now <- getCurrentTime-     let next = release `timeAdd` period-     if (now `timeAdd` 1) >= next-       then return now-       else do-         let sleepTime = next `diffUTCTime` now-         threadDelay (nominal2mu sleepTime)-         getCurrentTime--  -------------------------------------------------------------------------  -- Some time processing helpers -  -------------------------------------------------------------------------  timeAdd :: UTCTime -> Z.Timeout -> UTCTime-  timeAdd t p = mu2nominal p `addUTCTime` t--  mu2nominal :: Z.Timeout -> NominalDiffTime-  mu2nominal m = (fromIntegral m / 1000000)::NominalDiffTime--  nominal2mu :: NominalDiffTime -> Int-  nominal2mu n = ceiling (n * fromIntegral (1000000::Int))

@@ -1,564 +0,0 @@-module Types (-          -- * Service Access Point-          AccessPoint(..), LinkType(..), parseLink, link,-          AccessType(..), access, safeClose, setSockOs,-          -- * PollEntry-          PollEntry(..), pollEntry,-          -- * Enumerators-          Fetch, Fetch_, -          FetchHelper, FetchHelper', FetchHelper_, FetchHelper_',-          Dump,-          rcvEnum, itSend,-          -- * Converters-          InBound, OutBound,-          idIn, idOut, inString, outString, inUTF8, outUTF8,-          -- * Error Handlers-          Criticality(..),-          OnError, OnError_,-          chainIO, chainIOe, tryIO, tryIOe,-          -- * ZMQ Context-          Z.Context, Z.withContext,-          Z.SocketOption(..),-          -- * Helpers-          retries, trycon, ifLeft, (?>), ouch,-          Timeout, OnTimeout,-          Topic, alltopics, notopic,-          Identifier, Parameter, noparam)--where--  import qualified Data.ByteString.Char8  as B-  import qualified Data.ByteString.UTF8   as U -- standard converters-  import           Data.Char (toLower)-  import qualified Data.Enumerator        as E-  import           Data.Enumerator (($$))-  import qualified Data.Enumerator.List   as EL (head)--  import           Control.Concurrent-  import           Control.Monad-  import           Control.Monad.Trans-  import           Prelude hiding (catch)-  import           Control.Exception (SomeException, try, catch, throwIO)-  import           System.ZMQ as Z--  -------------------------------------------------------------------------  -- | Defines the type of a 'PollEntry';-  --   the names of the constructors are similar to ZMQ socket types-  --   but with some differences to keep the terminology in line-  --   with basic patterns.-  --   The leading \"X\" stands for \"Access\" -  --   (not for \"eXtended\" as in XRep and XReq).-  -------------------------------------------------------------------------  data AccessType = -         -- | Represents a server and expects connections from clients;-         --   should be used with 'Bind';-         --   corresponds to ZMQ Socket Type 'Z.Rep'-         XServer    -         -- | Represents a client and connects to a server;-         --   should be used with 'Connect';-         --   corresponds to ZMQ Socket Type 'Z.Req'-         | XClient-         -- | Represents a load balancer, -         --   expecting connections from clients;-         --   should be used with 'Bind';-         --   corresponds to ZMQ Socket Type 'Z.XRep'-         | XDealer -         -- | Represents a router-         --   expecting connections from servers;-         --   should be used with 'Bind';-         --   corresponds to ZMQ Socket Type 'Z.XReq'-         | XRouter -         -- | Represents a publisher;-         --   should be used with 'Bind';-         --   corresponds to ZMQ Socket Type 'Z.Pub'-         | XPub    -         -- | Represents a subscriber;-         --   should be used with 'Connect';-         --   corresponds to ZMQ Socket Type 'Z.Sub'-         | XSub    -         -- | Represents a Pipe;-         --   should be used with 'Bind';-         --   corresponds to ZMQ Socket Type 'Z.Push'-         | XPipe-         -- | Represents a Puller;-         --   should be used with 'Connect';-         --   corresponds to ZMQ Socket Type 'Z.Pull'-         | XPull-         -- | Represents a Peer;-         --   corresponding peers must use complementing 'LinkType';-         --   corresponds to ZMQ Socket Type 'Z.Pair'-         | XPeer   -    deriving (Eq, Show, Read)--  -------------------------------------------------------------------------  -- Creates a socket, binds or links it and sets the socket options-  -------------------------------------------------------------------------  access :: Z.Context -> AccessType -> LinkType -> [Z.SocketOption] ->-               String -> [Topic] -> IO Z.Poll-  access ctx a l os u ts = -    case a of -      XServer -> Z.socket ctx Z.Rep  >>= go-      XClient -> Z.socket ctx Z.Req  >>= go-      XDealer -> Z.socket ctx Z.XRep >>= go-      XRouter -> Z.socket ctx Z.XReq >>= go-      XPub    -> Z.socket ctx Z.Pub  >>= go-      XPipe   -> Z.socket ctx Z.Push >>= go-      XPull   -> Z.socket ctx Z.Pull >>= go-      XPeer   -> Z.socket ctx Z.Pair >>= go-      XSub    -> Z.socket ctx Z.Sub  >>= \s -> -                  mapM_ (Z.subscribe s) ts >> go s-   where go s = do setSockOs s os-                   case l of-                     Bind    -> Z.bind s u-                     Connect -> trycon s u retries-                   return $ Z.S s Z.In--  -------------------------------------------------------------------------  -- Close without throughing an exception-  -------------------------------------------------------------------------  safeClose :: Z.Socket a -> IO ()-  safeClose s = catch (Z.close s)-                      (\e -> let _ = (e::SomeException)-                              in return ())--  -------------------------------------------------------------------------  -- | Describes how to access a service;-  --   an 'AccessPoint' usually consists of an address -  --   and a list of 'Z.SocketOption'.-  --   Addresses are passed in as strings of the form:-  ---  --   * \"tcp:\/\/*:5555\": for binding the port /5555/ via TCP\/IP-  --                         on all network interfaces;-  --                         an IPv4 address or the operating system-  --                         interface name could be given instead. -  ---  --   * \"tcp:\/\/localhost:5555\": for connecting to the port /5555/ -  --                                 on /localhost/ via TCP\/IP;-  --                                 the endpoint may given as /DNS/ name-  --                                 or as an IPv4 address.-  ---  --   * \"ipc:\/\/tmp\/queues/0\": for binding and connecting to-  --                                  a local inter-process communication-  --                                  endpoint, in this case created under-  --                                  \/tmp\/queues\/0;-  --                                  only available on UNIX.-  -- -  --   * \"inproc:\/\/worker\": for binding and connecting to-  --                            the process internal address /worker/-  ---  --   For more options, please refer to the zeromq documentation.-  -------------------------------------------------------------------------  data AccessPoint = Address {-                       -- | Address string-                       acAdd :: String,-                       -- | Socket options-                       acOs  :: [Z.SocketOption]}-  -  instance Show AccessPoint where-    show (Address s _) = s-  -  -------------------------------------------------------------------------  -- | A poll entry describes how to handle an 'AccessPoint'-  -------------------------------------------------------------------------  data PollEntry = Poll {-                     pollId   :: Identifier,-                     pollAdd  :: String,-                     pollType :: AccessType,-                     pollLink :: LinkType,-                     pollSub  :: [Topic],-                     pollOs   :: [Z.SocketOption]-                   }-    deriving (Show, Read)--  instance Eq PollEntry where-    x == y = pollId x == pollId y--  -------------------------------------------------------------------------  -- | Creates a 'PollEntry';-  ---  --   Parameters:-  ---  --   * 'Identifier': identifies an 'AccessPoint'; -  --                   the identifier shall be unique within the device.-  ---  --   * 'AccessType': the 'AccessType' of this 'AccessPoint'-  ---  --   * 'AccessPoint': the 'AccessPoint'-  ---  --   * 'LinkType': how to link to this 'AccessPoint'-  ---  --   * ['Topic']: The subscription topics - -  --                ignored for all poll entries, but those-  --                with 'AccessType' 'XSub' -  -------------------------------------------------------------------------  pollEntry :: Identifier -> -               AccessType -> AccessPoint -> LinkType ->-               [Topic]    -> PollEntry-  pollEntry i at ac lt sub = Poll {-                               pollId   = i,-                               pollAdd  = acAdd ac,-                               pollType = at,-                               pollLink = lt,-                               pollSub  = sub,-                               pollOs   = acOs ac}--  -------------------------------------------------------------------------  -- | 'E.Enumerator' to process data segments of type /o/;-  --   receives the 'Z.Context', the control parameter -  --   and an input of type /i/;-  --   'Fetch' is used by 'Server's that receive requests of type /i/-  --   and produce an outgoing stream with segments of type /o/.-  -------------------------------------------------------------------------  type Fetch       i o = Z.Context -> Parameter -> i -> E.Enumerator o IO ()--  -------------------------------------------------------------------------  -- | A variant of 'Fetch' without input-  -------------------------------------------------------------------------  type Fetch_        o = Fetch () o--  -------------------------------------------------------------------------  -- | A function that may be used with some of the fetchers;-  --   The helper returns 'Nothing' to signal -  --   that no more data are available-  --   and 'Just' /o/ to continue the stream.-  --   FetchHelpers are used with 'Server's -  --   that receive requests of type /i/.-  --   The function -  --   receives the 'Z.Context', the conrol parameter-  --   and an input of type /i/;-  -------------------------------------------------------------------------  type FetchHelper i o = Z.Context -> Parameter -> i -> IO (Maybe o)--  -------------------------------------------------------------------------  -- | A variant of 'FetchHelper' that returns type /o/ instead of-  --   'Maybe' /o/.-  --   Please note that /'/ does not mean /strict/, here;-  --   it just means that the result is not a 'Maybe'.-  -------------------------------------------------------------------------  type FetchHelper' i o = Z.Context -> Parameter -> i -> IO o--  -------------------------------------------------------------------------  -- | A variant of 'FetchHelper' without input-  -------------------------------------------------------------------------  type FetchHelper_  o = FetchHelper () o--  -------------------------------------------------------------------------  -- | A variant of 'FetchHelper_' that returns type /o/ instead of-  --   'Maybe' /o/.-  --   Please note that /'/ does not mean /strict/, here;-  --   it just means that the result is not a 'Maybe'.-  -------------------------------------------------------------------------  type FetchHelper_' o = FetchHelper' () o--  -------------------------------------------------------------------------  -- | 'E.Iteratee' to process data segments of type /i/;-  --   receives the 'Z.Context' and the control parameter-  -------------------------------------------------------------------------  type Dump i = Z.Context -> Parameter -> E.Iteratee i IO ()--  -------------------------------------------------------------------------  -- | Error handler for servers;-  --   receives the 'Criticality' of the error event,-  --   the exception, the server name and the service control parameter.-  --   If the error handler returns 'Just' a 'B.ByteString'-  --   this 'B.ByteString' is sent to the client as error message.-  --   -  --   A good policy for implementing servers is-  --   to terminate or restart the 'Server'-  --   when a 'Fatal' or 'Critical' error occurs-  --   and to send an error message to the client-  --   on a plain 'Error'.-  --   The error handler, additionally, may log the incident-  --   or inform an administrator.-  -------------------------------------------------------------------------  type OnError   = Criticality         -> -                   SomeException       -> -                   String -> Parameter -> IO (Maybe B.ByteString)--  -------------------------------------------------------------------------  -- | Error handler for all services but servers;-  --   receives the 'Criticality' of the error event,-  --   the exception, the service name -  --   and the service control parameter.-  --   -  --   A good policy is-  --   to terminate or restart the service-  --   when a 'Fatal' error occurs-  --   and to continue, if possible,-  --   on a plain 'Error'.-  --   The error handler, additionally, may log the incident-  --   or inform an administrator.-  -------------------------------------------------------------------------  type OnError_  = Criticality         -> -                   SomeException       -> -                   String -> Parameter -> IO ()--  --------------------------------------------------------------------------  -- | Indicates criticality of the error event-  --------------------------------------------------------------------------  data Criticality = -                     -- | The current operation -                     --   (/e.g./ processing a request)-                     --   has not terminated properly,-                     --   but the service is able to continue;-                     --   the error may have been caused by a faulty-                     --   request or other temporal conditions.-                     --   Note that if an application-defined 'E.Iteratee' or-                     --   'E.Enumerator' results in 'SomeException'-                     --   (by means of 'E.throwError'),-                     --   the incident is classified as 'Error';-                     --   if it throws an IO Error, however,-                     --   the incident is classified as 'Fatal'.-                     Error -                     -- | One worker thread is lost ('Server' only)-                   | Critical -                     -- | The service cannot recover and will terminate-                   | Fatal-    deriving (Eq, Ord, Show, Read)--  --------------------------------------------------------------------------  -- | How to link to an 'AccessPoint'-  --------------------------------------------------------------------------  data LinkType = -         -- | Bind the address-         Bind -         -- | Connect to the address-         | Connect-    deriving (Show, Read)--  --------------------------------------------------------------------------  -- | Safely read 'LinkType';-  --   ignores the case of the input string-  --   and, besides \"bind\" and \"connect\", -  --   also accepts \"bin\", \"con\" and \"conn\";-  --   intended for use with command line parameters-  --------------------------------------------------------------------------  parseLink :: String -> Maybe LinkType -  parseLink s = case map toLower s of-                  "bind"    -> Just Bind-                  "bin"     -> Just Bind-                  "con"     -> Just Connect-                  "conn"    -> Just Connect-                  "connect" -> Just Connect-                  _         -> Nothing--  --------------------------------------------------------------------------  -- binds or connects to the address-  --------------------------------------------------------------------------  link :: LinkType -> AccessPoint -> Z.Socket a -> IO ()-  link t ac s = do setSockOs s (acOs ac) -                   case t of-                     Bind    -> Z.bind s (acAdd ac)-                     Connect -> trycon s (acAdd ac) 10--  --------------------------------------------------------------------------  -- Sets Socket Options-  --------------------------------------------------------------------------  setSockOs :: Z.Socket a -> [Z.SocketOption] -> IO ()-  setSockOs s = mapM_ (Z.setOption s)--  -------------------------------------------------------------------------  -- | Converters are user-defined functions-  --   that convert a 'B.ByteString' to a value of type /a/ ('InBound') or-  --                a value of type /a/ to 'B.ByteString'   ('OutBound'). -  --   Converters are, hence, similar to /put/ and /get/ in the /Binary/-  --   monad. -  --   The reason for using explicit, user-defined converters -  --   instead of /Binary/ /encode/ and /decode/-  --   is that the conversion -  --   may be more complex, involving reading configurations -  --   or other 'IO' actions.-  ---  --   The simplest possible in-bound converter for plain strings is:-  ---  --   > let iconv = return . toString-  -------------------------------------------------------------------------  type InBound  a = B.ByteString -> IO a-  -------------------------------------------------------------------------  -- | A simple string 'OutBound' converter may be:-  ---  --   > let oconv = return . fromString-  -------------------------------------------------------------------------  type OutBound a = a -> IO B.ByteString--  -------------------------------------------------------------------------  -- | 'InBound' 'B.ByteString' -> 'B.ByteString' -  -------------------------------------------------------------------------  idIn :: InBound B.ByteString-  idIn = return--  -------------------------------------------------------------------------  -- | 'OutBound' 'B.ByteString' -> 'B.ByteString' -  -------------------------------------------------------------------------  idOut :: OutBound B.ByteString-  idOut = return--  -------------------------------------------------------------------------  -- | 'OutBound' UTF8 String -> 'B.ByteString' -  -------------------------------------------------------------------------  outUTF8 :: OutBound String-  outUTF8 = return . U.fromString--  -------------------------------------------------------------------------  -- | 'InBound' 'B.ByteString' -> UTF8 String-  -------------------------------------------------------------------------  inUTF8 :: InBound String-  inUTF8 = return . U.toString--  -------------------------------------------------------------------------  -- | 'OutBound' String -> 'B.ByteString'-  -------------------------------------------------------------------------  outString :: OutBound String-  outString = return . B.pack--  -------------------------------------------------------------------------  -- | 'InBound' 'B.ByteString' -> String-  -------------------------------------------------------------------------  inString :: InBound String-  inString = return . B.unpack--  -------------------------------------------------------------------------  -- enumerator-  -------------------------------------------------------------------------  rcvEnum :: Z.Socket a -> InBound i -> E.Enumerator i IO b-  rcvEnum s iconv = go True-    where go more step = -            case step of -              E.Continue k ->-                if more then do-                    x <- liftIO $ Z.receive s []-                    m <- liftIO $ Z.moreToReceive s-                    i <- tryIO  $ iconv x-                    go m $$ k (E.Chunks [i])-                  else E.continue k-              _ -> E.returnI step--  -------------------------------------------------------------------------  -- iteratee -  -------------------------------------------------------------------------  itSend :: Z.Socket a -> OutBound o -> E.Iteratee o IO ()-  itSend s oconv = EL.head >>= go-    where go mbO =-            case mbO of-              Nothing -> return ()-              Just o  -> do-                x    <- tryIO $ oconv o           -                mbO' <- EL.head-                let opt = case mbO' of-                            Nothing -> []-                            Just _  -> [Z.SndMore]-                liftIO $ Z.send s x opt-                go mbO'--  -------------------------------------------------------------------------  -- some helpers-  -------------------------------------------------------------------------  retries :: Int-  retries = 100--  -------------------------------------------------------------------------  -- try n times to connect-  -- this is particularly useful for "inproc" sockets:-  -- the socket, in this case, must be bound before we can connect to it.-  -------------------------------------------------------------------------  trycon :: Z.Socket a -> String -> Int -> IO ()-  trycon sock add i = catch (Z.connect sock add) -                            (\e -> if i <= 0 -                                     then throwIO (e::SomeException)-                                     else do-                                       threadDelay 1000-                                       trycon sock add (i-1))--  -------------------------------------------------------------------------  -- either with the arguments flipped -  -------------------------------------------------------------------------  ifLeft :: Either a b -> (a -> c) -> (b -> c) -> c-  ifLeft e l r = either l r e--  -------------------------------------------------------------------------  -- | Chains IO Actions in an 'E.Enumerator' together;-  --   returns 'E.Error' when an error occurs-  -------------------------------------------------------------------------  chainIOe :: IO a -> (a -> E.Iteratee b IO c) -> E.Iteratee b IO c-  chainIOe x f = liftIO (try x) >>= \ei ->-                   case ei of-                     Left  e -> E.returnI (E.Error e)-                     Right y -> f y--  -------------------------------------------------------------------------  -- | Chains IO Actions in an 'E.Enumerator' together;-  --   throws 'SomeException' -  --   using 'E.throwError'-  --   when an error occurs-  -------------------------------------------------------------------------  chainIO :: IO a -> (a -> E.Iteratee b IO c) -> E.Iteratee b IO c-  chainIO x f = liftIO (try x) >>= \ei ->-                  case ei of-                    Left  e -> E.throwError (e::SomeException)-                    Right y -> f y--  -------------------------------------------------------------------------  -- | Executes an IO Actions in an 'E.Iteratee';-  --   throws 'SomeException'-  --   using 'E.throwError' when an error occurs-  -------------------------------------------------------------------------  tryIO :: IO a -> E.Iteratee i IO a-  tryIO act = -    liftIO (try act) >>= \ei ->-      case ei of-        Left  e -> E.throwError (e::SomeException)-        Right x -> return x--  -------------------------------------------------------------------------  -- | Executes an IO Actions in an 'E.Iteratee';-  --   returns 'E.Error' when an error occurs-  -------------------------------------------------------------------------  tryIOe :: IO a -> E.Iteratee i IO a-  tryIOe act = -    liftIO (try act) >>= \ei ->-      case ei of-        Left  e -> E.returnI (E.Error e)-        Right x -> return x--  -------------------------------------------------------------------------  -- Ease working with either-  -------------------------------------------------------------------------  eiCombine :: IO (Either a b) -> (b -> IO (Either a c)) -> IO (Either a c)-  eiCombine x f = x >>= \mbx ->-                  case mbx of-                    Left  e -> return $ Left e-                    Right y -> f y--  infixl 9 ?>-  (?>) :: IO (Either a b) -> (b -> IO (Either a c)) -> IO (Either a c)-  (?>) = eiCombine--  -------------------------------------------------------------------------  -- Ouch message-  -------------------------------------------------------------------------  ouch :: String -> a-  ouch s = error $ "Ouch! You hit a bug, please report: " ++ s--  -- | A device identifier is just a plain 'String'-  type Identifier  = String--  -- | A timeout action is just an IO action without arguments-  type OnTimeout   = IO ()--  -- | Control Parameter-  type Parameter = String--  -- | Ignore parameter-  noparam :: Parameter-  noparam = ""--  -- | Subscription Topic-  type Topic = String--  -- | Subscribe to all topics-  alltopics :: [Topic]-  alltopics = [""]--  -- | Subscribe to no topic-  notopic :: [Topic]-  notopic = []

@@ -0,0 +1,91 @@+-------------------------------------------------------------------------------+-- |+-- Module     : Heartbeat.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Heartbeat+-------------------------------------------------------------------------------+module Heartbeat+where++  import Network.Mom.Patterns.Types (Msec)++  import Data.Time.Clock+  import Control.Concurrent.MVar++  -------------------------------------------------------------------------+  -- Tolerance for foreign heartbeat -+  -- Period until disconnect = tolerance * heartbeat period+  -------------------------------------------------------------------------+  tolerance :: Int+  tolerance = 10++  -------------------------------------------------------------------------+  -- Helper to check for expiration of a period +  -------------------------------------------------------------------------+  hbPeriodReached :: MVar UTCTime -> Msec -> IO Bool+  hbPeriodReached m tmo = modifyMVar m $ \t -> do+                            now <- getCurrentTime+                            if t `timeAdd` tmo <= now+                              then return (now `timeAdd` tmo, True )+                              else return (                t, False)++  ------------------------------------------------------------------------+  -- Heartbeat descriptor+  ------------------------------------------------------------------------+  data Heartbeat = Heart {+                     hbNextMe     :: UTCTime,+                     hbNextHe     :: UTCTime,+                     hbPeriod     :: Msec+                   }++  ------------------------------------------------------------------------+  -- Create a new heartbeat descriptor+  ------------------------------------------------------------------------+  newHeartbeat :: Msec -> IO Heartbeat+  newHeartbeat period = do+    now <- getCurrentTime+    return Heart {+               hbNextHe = timeAdd now (tolerance * period),+               hbNextMe = timeAdd now period,+               hbPeriod = period}++  ------------------------------------------------------------------------+  -- Update heartbeat descriptor+  -- - my next heartbeat+  ------------------------------------------------------------------------+  updMe :: UTCTime -> Heartbeat -> Heartbeat+  updMe now hb = hb {hbNextMe = now `timeAdd` (hbPeriod hb)}++  ------------------------------------------------------------------------+  -- - his next heartbeat+  ------------------------------------------------------------------------+  updHim :: UTCTime -> Heartbeat -> Heartbeat+  updHim now hb = hb {hbNextHe = timeAdd now $ tolerance * (hbPeriod hb)}++  ------------------------------------------------------------------------+  -- Check me, him+  ------------------------------------------------------------------------+  checkMe :: UTCTime -> Heartbeat -> Bool+  checkMe now hb | now <= hbNextMe hb = False+                 | otherwise          = True++  alive :: UTCTime -> Heartbeat -> Bool+  alive now hb | now <= hbNextHe hb = True+               | otherwise          = False++  -----------------------------------------------------------------------+  -- Adding period to time+  -----------------------------------------------------------------------+  timeAdd :: UTCTime -> Msec -> UTCTime+  timeAdd t p = ms2nominal p `addUTCTime` t++  -----------------------------------------------------------------------+  -- Convert milliseconds to seconds+  -----------------------------------------------------------------------+  ms2nominal :: Msec -> NominalDiffTime+  ms2nominal m = fromIntegral m / (1000::NominalDiffTime)+    

@@ -0,0 +1,582 @@+{-# LANGUAGE CPP #-}+-------------------------------------------------------------------------------+-- |+-- Module     : Registry.hs+-- Copyright  : (c) Tobias Schoofs+-- License    : LGPL +-- Stability  : experimental+-- Portability: non-portable+-- +-- Backoffice for majordomo broker+-------------------------------------------------------------------------------+module Registry +#ifndef TEST+       (insert, remove, +        checkWorker, getWorker, freeWorker, +        getServiceName,+        updWorkerHb, lookupService, +        clean, setHbPeriod, +        size, stat, statPerService,+        printQ, getQ)+#endif+where++  import           Heartbeat+  import           Network.Mom.Patterns.Types (Msec, Identity)++  import           Control.Concurrent+  import           Control.Monad (when)+  import           Control.Applicative ((<$>))++  import           System.IO.Unsafe++  import qualified Data.ByteString.Char8 as B+  import           Data.Map   (Map)+  import qualified Data.Map   as   M -- filter, keys+  import qualified Data.Sequence as  S+  import           Data.Sequence (Seq, (|>), (<|), (><), ViewL(..), ViewR(..)) +  import           Data.Foldable (toList)+  import           Data.Time.Clock++  ------------------------------------------------------------------------+  -- Worker may be free or busy+  ------------------------------------------------------------------------+  data State = Free | Busy +    deriving (Show, Eq)+                 +  ------------------------------------------------------------------------+  -- Worker +  ------------------------------------------------------------------------+  data Worker = Worker {+                  wrkId    :: Identity,+                  wrkState :: State,+                  wrkHB    :: Heartbeat,+                  wrkQ     :: MVar Queue+                }++  ------------------------------------------------------------------------+  -- Base for Heartbeat calculation:+  -- expected heartbeat is _hb * tolerance,+  -- heartbeat to be sent is _hb +  ------------------------------------------------------------------------+  {-# NOINLINE _hb #-}+  _hb :: MVar Msec+  _hb = unsafePerformIO $ newMVar 2000 -- default: 2 seconds++  -------------------------------------------------------------------------+  -- Set common heartbeat period+  -------------------------------------------------------------------------+  setHbPeriod :: Msec -> IO ()+  setHbPeriod hb = modifyMVar_ _hb $ \_ -> return hb++  -------------------------------------------------------------------------+  -- Update worker's heartbeat descriptor according to function+  -------------------------------------------------------------------------+  updHb :: (UTCTime -> Heartbeat -> Heartbeat) -> +            UTCTime -> WrkNode -> WrkNode+  updHb upd t (i, w) = (i, w {wrkHB = upd t $ wrkHB w})++  ------------------------------------------------------------------------+  -- A Worker Node is an Identity paired with a worker (lookup)+  -- A Worker Tree is a map of identities and services,+  -- where service is a queue of free and busy workers;+  -- the worker tree, hence, does not point directly to the worker,+  -- but to the queue where the worker can be found:+  --+  -- tree+  --  |+  --  ----> (i,srv)+  --            |+  --            ----> (sn,srvQ)+  --                       |+  --                       -----> Free Seq (i,wrk)+  --                       -----> Busy Seq (i,wrk)+  -------------------------------------------------------------------------+  type WrkNode = (Identity, Worker)+  type WrkTree = Map Identity Service++  -------------------------------------------------------------------------+  -- A service is a queue of free and busy workers+  -------------------------------------------------------------------------+  data Service = Service {+                   srvName :: B.ByteString,+                   srvQ    :: MVar Queue+                 }++  -------------------------------------------------------------------------+  -- A service node is a pair of (ServiceName, Service)+  -- A service tree is a map  of ServiceName and Service+  -------------------------------------------------------------------------+  type SrvNode = (B.ByteString, Service)+  type SrvTree = Map B.ByteString Service++  -------------------------------------------------------------------------+  -- service tree+  -------------------------------------------------------------------------+  {-# NOINLINE _srv #-}+  _srv :: MVar SrvTree+  _srv = unsafePerformIO $ newMVar M.empty ++  -------------------------------------------------------------------------+  -- A sequence of services, used for removing unresponsive workers:+  -- We clean up one service at a time, so there are not too long+  -- work intervals due to clean-up.+  -------------------------------------------------------------------------+  {-# NOINLINE _s #-}+  _s :: MVar (Seq B.ByteString)+  _s = unsafePerformIO $ newMVar S.empty ++  -------------------------------------------------------------------------+  -- worker tree+  -------------------------------------------------------------------------+  {-# NOINLINE _wrk #-}+  _wrk :: MVar WrkTree+  _wrk = unsafePerformIO $ newMVar M.empty ++  -------------------------------------------------------------------------+  -- Register new worker for service+  -------------------------------------------------------------------------+  insert :: Identity -> B.ByteString -> IO ()+  insert i s = do+    mbW <- lookupW i+    case mbW of+      Just _  -> return () -- identity already known+      Nothing -> do+        sn <- lookupS s >>= getSn+        initW   i sn+        insertW i sn+    where getSn mbS = +            case mbS of+              Just sn -> return sn+              Nothing -> do+                 q <- newMVar $ Q S.empty S.empty+                 let sn = Service {+                            srvName = s,+                            srvQ    = q}+                 insertS s sn+                 return sn++  -------------------------------------------------------------------------+  -- Remove worker +  -------------------------------------------------------------------------+  remove :: Identity -> IO ()+  remove i = do+    mbW <- lookupW i+    case mbW of+      Nothing -> return ()+      Just sn -> do+        deleteW i+        e <- modifyMVar (srvQ sn) $ \q -> +               let q' = removeQ i q+                   e  = emptyQ q'+                in return (q', e)+        when e $ deleteS (srvName sn)++  -------------------------------------------------------------------------+  -- Free a worker that has been busy+  -- updating his heatbeat+  -------------------------------------------------------------------------+  freeWorker :: Identity -> IO ()+  freeWorker i = getCurrentTime >>= \now ->+                   freeWorkerWithUpd i (updHb updHim now)++  -------------------------------------------------------------------------+  -- Free a worker that has been busy+  -- updating heartbeat (because we apparently have feedback )+  -------------------------------------------------------------------------+  freeWorkerWithUpd :: Identity -> (WrkNode -> WrkNode) -> IO ()+  freeWorkerWithUpd i upd = do+    mbW <- lookupW i+    case mbW of+      Nothing -> return () -- silent error+      Just sn -> modifyMVar_ (srvQ sn) $ \q -> return $ setStateQ i Free upd q ++  -------------------------------------------------------------------------+  -- Update heartbeat+  -------------------------------------------------------------------------+  updWorkerHb :: Identity -> IO ()+  updWorkerHb i = getCurrentTime >>= \now -> updWorker i (updHb updHim now)++  -------------------------------------------------------------------------+  -- Generic worker update+  -------------------------------------------------------------------------+  updWorker :: Identity -> (WrkNode -> WrkNode) -> IO ()+  updWorker i f = do+    mbW <- lookupW i+    case mbW of+      Nothing -> return ()+      Just sn -> modifyMVar_ (srvQ sn) $ \q -> +                   return $ updateQ i f q++  -------------------------------------------------------------------------+  -- Get service by worker identity+  -------------------------------------------------------------------------+  getServiceName :: Identity -> IO (Maybe B.ByteString)+  getServiceName i = do+    mbW <- lookupW i+    case mbW of+      Nothing -> return Nothing+      Just s  -> return $ Just (srvName s)++  -------------------------------------------------------------------------+  -- Check whether service exists+  -------------------------------------------------------------------------+  lookupService :: B.ByteString -> IO Bool+  lookupService s = do+    mbS <- lookupS s+    case mbS of+      Nothing -> return False+      Just  _ -> return True++  -------------------------------------------------------------------------+  -- Get worker for service:+  --   - removing unresponsive workers on the way +  --   - get first of free q, put it to the busy queue +  --                          and update its heartbeat+  --   - if free q is emtpy, take the first of the busy queue+  -------------------------------------------------------------------------+  getWorker :: B.ByteString -> IO (Maybe Identity)+  getWorker s = do+    now <- getCurrentTime+    mbS <- lookupS s+    case mbS of+      Nothing -> return Nothing+      Just sn -> do+        bury now sn -- remove non-responsive workers+        modifyMVar (srvQ sn) $ \q ->+          case firstFreeQ q of+            Just (i,_) ->+              return (setStateQ i Busy (updHb updMe now) q, Just i)+            Nothing    -> +              case firstBusyQ q of+                Nothing    -> return (q, Nothing)+                Just (i,_) -> return (+                      setStateQ i Busy (updHb updMe now) q, Just i)+          +  -------------------------------------------------------------------------+  -- Remove non-responsive workers+  -- and heartbeat those that have been inactive +  --     for at least one heartbeat period+  -------------------------------------------------------------------------+  checkWorker :: IO [Identity]+  checkWorker = do+    now <- getCurrentTime+    checkService >>= \mbS -> case mbS of+                               Nothing -> return ()+                               Just s  -> bury now s+    concat <$> withMVar _srv (mapM (getWorkers now) . M.elems)++    where getWorkers now s = do+            is <- withMVar (srvQ s) ( +                    return . map fst . takeIfQ Free tst)+            mapM_ (`updWorker` updSt) is+            return is++            where updSt (i,w) = let hb  = updMe now $ wrkHB w+                                 in (i, w{wrkHB = hb}) +                  tst = checkMe now . wrkHB . snd++  -------------------------------------------------------------------------+  -- Get next service to check+  -- check on the way if service still exists+  -- if not, remove service and try next.+  -- Note: there is no upper bound;+  --       if a lot of services have been removed since our last visit+  --          we will go through this many times+  -------------------------------------------------------------------------+  checkService :: IO (Maybe Service)+  checkService = do+    mbS <- modifyMVar _s getS+    case mbS of +      Nothing -> return Nothing+      Just s  -> do+        mbSrv <- lookupS s+        case mbSrv of+          Nothing  -> rm >> checkService+          Just srv -> return $ Just srv+    where getS s = +            case S.viewl s of+              EmptyL  -> return (s, Nothing)+              x :< xs -> +                return (xs |> x, Just x)+          rmTail s = +            case S.viewr s of+              EmptyR  -> return s+              xs :> _ -> return xs+          rm = modifyMVar_ _s rmTail++  -------------------------------------------------------------------------+  -- Remove unresponsive workers+  -------------------------------------------------------------------------+  bury :: UTCTime -> Service -> IO ()+  bury now sn = do+    q <- readMVar $ srvQ sn+    mapM_ (remove . fst) $ findDeads $ qFree q+    mapM_ (remove . fst) $ findDeads $ qBusy q+    where findDeads = filter (not . alive now . wrkHB . snd) . toList++  -------------------------------------------------------------------------+  -- Clean up global variables+  -------------------------------------------------------------------------+  clean :: IO ()+  clean = do modifyMVar_ _wrk $ \_ -> return M.empty+             modifyMVar_ _srv $ \_ -> return M.empty+             modifyMVar_ _s   $ \_ -> return S.empty++  -------------------------------------------------------------------------+  -- Number of workers+  -------------------------------------------------------------------------+  size :: IO Int+  size = withMVar _wrk $ \t -> return $ M.size t++  -------------------------------------------------------------------------+  -- Per service:+  --     number of free workers+  --     number of busy workers+  -------------------------------------------------------------------------+  statPerService :: B.ByteString -> IO (B.ByteString, Int, Int)+  statPerService s = do+    mbS <- lookupS s+    case mbS of+      Nothing -> return (s, 0, 0)+      Just sn -> perService (s, sn)++  -------------------------------------------------------------------------+  -- For all services:+  --     number of free workers+  --     number of busy workers+  -------------------------------------------------------------------------+  stat :: IO [(B.ByteString, Int, Int)]+  stat = withMVar _srv $ \t -> mapM perService $ M.assocs t++  -------------------------------------------------------------------------+  -- Get stats per service+  -------------------------------------------------------------------------+  perService :: SrvNode -> IO (B.ByteString, Int, Int)+  perService (s, sn) = withMVar (srvQ sn) $ \q ->+                         return (s, S.length $ qFree q,+                                    S.length $ qBusy q)++  -------------------------------------------------------------------------+  -- Debug: print q+  -------------------------------------------------------------------------+  printQ :: B.ByteString -> IO ()+  printQ s = do+    mbS <- lookupS s+    case mbS of+      Nothing -> return ()+      Just x  -> do q <- readMVar (srvQ x)+                    putStr "Free: "+                    print $ map fst $ toList $ qFree q+                    putStrLn ""+                    putStr "Busy: "+                    print $ map fst $ toList $ qBusy q ++  -------------------------------------------------------------------------+  -- Create worker+  --        initialising heartbeat+  --        inserting    into service queue+  -------------------------------------------------------------------------+  initW :: Identity -> Service -> IO ()+  initW i sn = modifyMVar_ (srvQ sn) $ \q -> do+                 hb <- withMVar _hb newHeartbeat +                 let w = Worker {+                           wrkId    = i,+                           wrkState = Free,+                           wrkHB    = hb,+                           wrkQ     = srvQ sn}+                 return $ insertQ (i,w) q+                    +  -------------------------------------------------------------------------+  -- Lookup worker, returning service+  -------------------------------------------------------------------------+  lookupW :: Identity -> IO (Maybe Service)+  lookupW i = withMVar _wrk $ \t -> return $ M.lookup i t++  -------------------------------------------------------------------------+  -- Lookup service+  -------------------------------------------------------------------------+  lookupS :: B.ByteString -> IO (Maybe Service)+  lookupS sn = withMVar _srv $ \t -> return $ M.lookup sn t++  -------------------------------------------------------------------------+  -- Insert worker+  -------------------------------------------------------------------------+  insertW :: Identity -> Service -> IO ()+  insertW i sn = modifyMVar_ _wrk $ \t -> return $ M.insert i sn t++  -------------------------------------------------------------------------+  -- Insert service+  -------------------------------------------------------------------------+  insertS :: B.ByteString -> Service -> IO ()+  insertS s sn = do+    modifyMVar_ _srv $ \t  -> return $ M.insert s sn t+    modifyMVar_ _s   $ \ss -> return $ ss |> s+  +  -------------------------------------------------------------------------+  -- Delete worker+  -------------------------------------------------------------------------+  deleteW :: Identity -> IO ()+  deleteW i = modifyMVar_ _wrk $ \t -> return $ M.delete i t+  +  -------------------------------------------------------------------------+  -- Delete service+  -------------------------------------------------------------------------+  deleteS :: B.ByteString -> IO ()+  deleteS s = modifyMVar_ _srv $ \t -> return $ M.delete s t++  -------------------------------------------------------------------------+  -- Queue:+  --    Seq free workers+  --    Seq busy workers+  -------------------------------------------------------------------------+  data Queue = Q {+                 qFree :: Seq WrkNode,+                 qBusy :: Seq WrkNode +               }++  -------------------------------------------------------------------------+  -- Get a worker (either from free or busy)+  -------------------------------------------------------------------------+  getQ :: Identity -> Queue -> Maybe WrkNode+  getQ i q = lookupQ i Free q ~> lookupQ i Busy q++  -------------------------------------------------------------------------+  -- lookup one seq, either+  --    free or+  --    busy+  -------------------------------------------------------------------------+  lookupQ :: Identity -> State -> Queue -> Maybe WrkNode+  lookupQ i s q = +      let r = toList $ snd $ S.breakl (eq i) l+       in if null r then Nothing else Just $ head r+    where l  = getList q s+          +  -------------------------------------------------------------------------+  -- Insert worker into q (always free)+  -------------------------------------------------------------------------+  insertQ :: WrkNode -> Queue -> Queue+  insertQ w q = q{qFree = qFree q |> w}++  -------------------------------------------------------------------------+  -- Remove worker from q (either free or busy)+  -------------------------------------------------------------------------+  removeQ :: Identity -> Queue -> Queue+  removeQ i = removeWithStateQ i Free . removeWithStateQ i Busy++  -------------------------------------------------------------------------+  -- Remove worker from q depending on state+  -------------------------------------------------------------------------+  removeWithStateQ :: Identity -> State -> Queue -> Queue+  removeWithStateQ i s q = +    case S.viewl t of+      EmptyL    -> q+      (_ :< xs) -> case s of+                     Free -> q{qFree = h >< xs}+                     Busy -> q{qBusy = h >< xs}+    where (h,t) = getWithStateQ i s q++  -------------------------------------------------------------------------+  -- Generic update of a worker in a queue+  -------------------------------------------------------------------------+  updateQ :: Identity -> (WrkNode -> WrkNode) -> Queue -> Queue+  updateQ i f = updateWithStateQ i Free f . updateWithStateQ i Busy f ++  -------------------------------------------------------------------------+  -- Generic update of a worker in a queue depending on state+  -------------------------------------------------------------------------+  updateWithStateQ :: Identity -> State    -> +                      (WrkNode -> WrkNode) -> Queue -> Queue+  updateWithStateQ i s f q =+    case S.viewl t of+      EmptyL    -> q+      (x :< xs) -> case s of+                     Free -> q{qFree = h >< f x <| xs}+                     Busy -> q{qBusy = h >< f x <| xs}+    where (h,t) = getWithStateQ i s q++  -------------------------------------------------------------------------+  -- Remove worker from Seq with state x and+  -- Add    it     to   Seq with state y+  -------------------------------------------------------------------------+  setStateQ :: Identity -> State -> (WrkNode -> WrkNode) -> Queue -> Queue+  setStateQ i s f q = case S.viewl t of+                        EmptyL    -> q+                        (x :< xs) -> case s of +                                       Free -> q{qFree = qFree q |> f x,+                                                 qBusy = h >< xs}+                                       Busy -> q{qBusy = qBusy q |> f x,+                                                 qFree = h >< xs}+    where (h,t) = let s' = case s of+                             Free -> Busy+                             Busy -> Free+                   in getWithStateQ i s' q ++  -------------------------------------------------------------------------+  -- Get view of Seq, breaking on worker+  -- either free or busy+  -------------------------------------------------------------------------+  getWithStateQ :: Identity -> State -> Queue -> (Seq WrkNode, Seq WrkNode)+  getWithStateQ i s q = case s of+                          Free -> S.breakl (eq i) $ qFree q+                          Busy -> S.breakl (eq i) $ qBusy q++  -------------------------------------------------------------------------+  -- Queue is empty+  -------------------------------------------------------------------------+  emptyQ :: Queue -> Bool+  emptyQ q = S.null (qFree q) && S.null (qBusy q)++  -------------------------------------------------------------------------+  -- Head of Seq free+  -------------------------------------------------------------------------+  firstFreeQ :: Queue -> Maybe WrkNode+  firstFreeQ q = firstQ (qFree q)++  -------------------------------------------------------------------------+  -- Head of Seq busy+  -------------------------------------------------------------------------+  firstBusyQ :: Queue -> Maybe WrkNode+  firstBusyQ q = firstQ (qBusy q)++  -------------------------------------------------------------------------+  -- Head+  -------------------------------------------------------------------------+  firstQ :: Seq WrkNode -> Maybe WrkNode+  firstQ f = case S.viewl f of+               EmptyL   -> Nothing+               (w :< _) -> Just w++  -------------------------------------------------------------------------+  -- Take n that fulfil f with state s+  -------------------------------------------------------------------------+  takeIfQ :: State -> (WrkNode -> Bool) -> Queue -> [WrkNode]+  takeIfQ s f q = case s of+                      Free -> takeIf f $ qFree q+                      Busy -> takeIf f $ qBusy q++  -------------------------------------------------------------------------+  -- Take n that fulfil f from Seq+  -------------------------------------------------------------------------+  takeIf :: (WrkNode -> Bool) -> Seq WrkNode -> [WrkNode]+  takeIf f s = case S.viewl s of+                 EmptyL    -> []+                 (w :< ws) | f w       -> w : takeIf f ws+                           | otherwise ->     takeIf f ws++  eq :: Identity -> WrkNode -> Bool+  eq i = (== i) . fst ++  getList :: Queue -> State -> Seq WrkNode+  getList q s = case s of+                  Free -> qFree q+                  Busy -> qBusy q++  -------------------------------------------------------------------------+  -- Simple Maybe combinator+  -------------------------------------------------------------------------+  infixl 9 ~>+  (~>) :: Maybe a -> Maybe a -> Maybe a+  (~>) f g = case f of+               Nothing -> g+               Just x  -> Just x