patterns (empty) → 0.0.1
raw patch · 10 files changed
+3479/−0 lines, 10 filesdep +basedep +bytestringdep +containerssetup-changed
Dependencies added: base, bytestring, containers, enumerator, mtl, split, time, utf8-string, zeromq-haskell
Files
- Setup.hs +2/−0
- license/lgpl-3.0.txt +165/−0
- patterns.cabal +48/−0
- src/Factory.hs +23/−0
- src/Network/Mom/Patterns.hs +218/−0
- src/Network/Mom/Patterns/Basic.hs +1096/−0
- src/Network/Mom/Patterns/Device.hs +560/−0
- src/Network/Mom/Patterns/Enumerator.hs +420/−0
- src/Service.hs +381/−0
- src/Types.hs +566/−0
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ license/lgpl-3.0.txt view
@@ -0,0 +1,165 @@+ GNU LESSER GENERAL PUBLIC LICENSE+ Version 3, 29 June 2007++ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>+ Everyone is permitted to copy and distribute verbatim copies+ of this license document, but changing it is not allowed.+++ This version of the GNU Lesser General Public License incorporates+the terms and conditions of version 3 of the GNU General Public+License, supplemented by the additional permissions listed below.++ 0. Additional Definitions.++ As used herein, "this License" refers to version 3 of the GNU Lesser+General Public License, and the "GNU GPL" refers to version 3 of the GNU+General Public License.++ "The Library" refers to a covered work governed by this License,+other than an Application or a Combined Work as defined below.++ An "Application" is any work that makes use of an interface provided+by the Library, but which is not otherwise based on the Library.+Defining a subclass of a class defined by the Library is deemed a mode+of using an interface provided by the Library.++ A "Combined Work" is a work produced by combining or linking an+Application with the Library. The particular version of the Library+with which the Combined Work was made is also called the "Linked+Version".++ The "Minimal Corresponding Source" for a Combined Work means the+Corresponding Source for the Combined Work, excluding any source code+for portions of the Combined Work that, considered in isolation, are+based on the Application, and not on the Linked Version.++ The "Corresponding Application Code" for a Combined Work means the+object code and/or source code for the Application, including any data+and utility programs needed for reproducing the Combined Work from the+Application, but excluding the System Libraries of the Combined Work.++ 1. Exception to Section 3 of the GNU GPL.++ You may convey a covered work under sections 3 and 4 of this License+without being bound by section 3 of the GNU GPL.++ 2. Conveying Modified Versions.++ If you modify a copy of the Library, and, in your modifications, a+facility refers to a function or data to be supplied by an Application+that uses the facility (other than as an argument passed when the+facility is invoked), then you may convey a copy of the modified+version:++ a) under this License, provided that you make a good faith effort to+ ensure that, in the event an Application does not supply the+ function or data, the facility still operates, and performs+ whatever part of its purpose remains meaningful, or++ b) under the GNU GPL, with none of the additional permissions of+ this License applicable to that copy.++ 3. Object Code Incorporating Material from Library Header Files.++ The object code form of an Application may incorporate material from+a header file that is part of the Library. You may convey such object+code under terms of your choice, provided that, if the incorporated+material is not limited to numerical parameters, data structure+layouts and accessors, or small macros, inline functions and templates+(ten or fewer lines in length), you do both of the following:++ a) Give prominent notice with each copy of the object code that the+ Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the object code with a copy of the GNU GPL and this license+ document.++ 4. Combined Works.++ You may convey a Combined Work under terms of your choice that,+taken together, effectively do not restrict modification of the+portions of the Library contained in the Combined Work and reverse+engineering for debugging such modifications, if you also do each of+the following:++ a) Give prominent notice with each copy of the Combined Work that+ the Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the Combined Work with a copy of the GNU GPL and this license+ document.++ c) For a Combined Work that displays copyright notices during+ execution, include the copyright notice for the Library among+ these notices, as well as a reference directing the user to the+ copies of the GNU GPL and this license document.++ d) Do one of the following:++ 0) Convey the Minimal Corresponding Source under the terms of this+ License, and the Corresponding Application Code in a form+ suitable for, and under terms that permit, the user to+ recombine or relink the Application with a modified version of+ the Linked Version to produce a modified Combined Work, in the+ manner specified by section 6 of the GNU GPL for conveying+ Corresponding Source.++ 1) Use a suitable shared library mechanism for linking with the+ Library. A suitable mechanism is one that (a) uses at run time+ a copy of the Library already present on the user's computer+ system, and (b) will operate properly with a modified version+ of the Library that is interface-compatible with the Linked+ Version.++ e) Provide Installation Information, but only if you would otherwise+ be required to provide such information under section 6 of the+ GNU GPL, and only to the extent that such information is+ necessary to install and execute a modified version of the+ Combined Work produced by recombining or relinking the+ Application with a modified version of the Linked Version. (If+ you use option 4d0, the Installation Information must accompany+ the Minimal Corresponding Source and Corresponding Application+ Code. If you use option 4d1, you must provide the Installation+ Information in the manner specified by section 6 of the GNU GPL+ for conveying Corresponding Source.)++ 5. Combined Libraries.++ You may place library facilities that are a work based on the+Library side by side in a single library together with other library+facilities that are not Applications and are not covered by this+License, and convey such a combined library under terms of your+choice, if you do both of the following:++ a) Accompany the combined library with a copy of the same work based+ on the Library, uncombined with any other library facilities,+ conveyed under the terms of this License.++ b) Give prominent notice with the combined library that part of it+ is a work based on the Library, and explaining where to find the+ accompanying uncombined form of the same work.++ 6. Revised Versions of the GNU Lesser General Public License.++ The Free Software Foundation may publish revised and/or new versions+of the GNU Lesser General Public License from time to time. Such new+versions will be similar in spirit to the present version, but may+differ in detail to address new problems or concerns.++ Each version is given a distinguishing version number. If the+Library as you received it specifies that a certain numbered version+of the GNU Lesser General Public License "or any later version"+applies to it, you have the option of following the terms and+conditions either of that published version or of any later version+published by the Free Software Foundation. If the Library as you+received it does not specify a version number of the GNU Lesser+General Public License, you may choose any version of the GNU Lesser+General Public License ever published by the Free Software Foundation.++ If the Library as you received it specifies that a proxy can decide+whether future versions of the GNU Lesser General Public License shall+apply, that proxy's public statement of acceptance of any version is+permanent authorization for you to choose that version for the+Library.
+ patterns.cabal view
@@ -0,0 +1,48 @@+Name: patterns+Version: 0.0.1+Cabal-Version: >= 1.8+Copyright: Copyright (c) Tobias Schoofs, 2011 - 2012+License: LGPL+license-file: license/lgpl-3.0.txt+Author: Tobias Schoofs+Maintainer: tobias dot schoofs at gmx dot net+Category: Network, Message-oriented Middleware, zeromq+Homepage: http://github.com/toschoo/mom/src/patterns+Build-Type: Simple+Synopsis: Common patterns in message-oriented applications+Description:++ There are common patterns often reused + - or, in fact, reimplemented,+ in many distributed, message-oriented applications,+ such as Server\/Client (a.k.a Request\/Response),+ Publish\/Subscribe,+ Pipline (a.k.a. Push\/Pull) and+ Exclusive Pair (a.k.a. Peer-to-Peer).+ The Patterns package implements those patterns based on zeromq.++ More information on zeromq can be found at+ <http://www.zeromq.org>.++ More examples and a test suite are available + on <http://github.com/toschoo/mom>.++Library+ Build-Depends: base >= 4.0 && <= 5.0,+ bytestring >= 0.9.1.9,+ utf8-string >= 0.3.6,+ containers >= 0.3.0.0,+ split >= 0.1.4.1,+ zeromq-haskell >= 0.8.3,+ enumerator >= 0.4.11,+ mtl >= 2.0.1.0,+ time >= 1.1.4++ hs-source-dirs: src/Network/Mom, src+ + Exposed-Modules: Network.Mom.Patterns, + Network.Mom.Patterns.Basic,+ Network.Mom.Patterns.Device,+ Network.Mom.Patterns.Enumerator+ other-modules: Factory, Types, Service+
+ src/Factory.hs view
@@ -0,0 +1,23 @@+{-# OPTIONS -fglasgow-exts -fno-cse #-}+module Factory (+ mkUniqueId)+where++ import System.IO.Unsafe+ import Control.Concurrent++ ------------------------------------------------------------------------+ -- Source for unique identifiers+ ------------------------------------------------------------------------+ {-# NOINLINE _addid #-}+ _addid :: MVar Int+ _addid = unsafePerformIO $ newMVar 1++ mkUniqueId :: IO Int+ mkUniqueId = modifyMVar _addid $ \x -> + let x' = incX x+ in return (x', x')++ incX :: Int -> Int+ incX i = if i == 99999999 then 1 else i+1+
+ src/Network/Mom/Patterns.hs view
@@ -0,0 +1,218 @@+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Patterns.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- 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.+-- More information on /zeromq/ can be found at+-- <http://www.zeromq.org>.+-------------------------------------------------------------------------------+module Network.Mom.Patterns (+ -- * Patterns+ -- $patterns++ -- * Basic Patterns+ -- $basic++ module Network.Mom.Patterns.Basic,++ -- * Devices+ -- $device++ module Network.Mom.Patterns.Device,++ -- * Enumerators+ -- $enumerator++ module Network.Mom.Patterns.Enumerator+ )+where++ import Network.Mom.Patterns.Basic+ import Network.Mom.Patterns.Device+ import Network.Mom.Patterns.Enumerator++ {- $patterns+ Instead of a centralised message broker+ as the main back bone of reliable+ message exchange, zeromq implements+ an advanced socket concept.+ Zeromq sockets are thread-local resources + that connect to each other across+ 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 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+ together to request or provide services, + publish or subscribe data or to + allocate work to processing nodes.+ + -}++ {- $basic+ Basic patterns are:++ * Server\/Client (a.k.a Request\/Response)+ consisting of a server process that responds to requests+ and client processes that request a service+ and wait for the server response;+ + * Publish\/Subscribe+ consisting of a publisher process that+ periodically or sporadically publishes data+ and subscribers that receive data+ corresponding to topics, + to which they have actually subscribed;++ * Pipeline (a.k.a. Push\/Pull)+ consisting of a /pusher/ process that sends jobs+ down the pipeline+ 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+ can communicate with each other. + Since communication may - and usually does -+ cross processes and network nodes,+ there is no way to enforce the correct combination+ by means of the type system. + The programmer has to take care+ that programs pair up correctly.++ The patterns hide details of the underlying + communication protocol and, hence,+ guarantee that the protocol is used correctly.+ The implementation, in particular,+ adheres to the following principles: ++ * a client must send a request to a server + before it can receive messages (from this server)+ + * a server must receive a request from a client + before it can send messages back to this client;++ * a publisher cannot receive messages and+ can send messages only to subscribers;++ * a subscriber cannot send messages and+ can receive messages only from a publisher;++ * a pusher cannot receive messages and+ can send message only to a pipe;++ * a puller cannot send messages and+ can receive messages only from a pipe;++ * peers can exchange messages only with other peers.+ -}++ {- $device+ Devices relay messages between compatible services, /i.e./+ Servers and Clients,+ Publishers and Subscribers,+ Pipes and Pullers and+ Peers and Peers.++ 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.+ When data is available,+ the device activates an application-defined stream transformer.+ 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.+ -}++ {- $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.++ 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./+ + -}+
+ src/Network/Mom/Patterns/Basic.hs view
@@ -0,0 +1,1096 @@+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Patterns/Basic.hs+-- 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 action =+ withService ctx name param service action+ 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+ trycon cmd sockname retries+ Z.subscribe cmd ""+ 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 (\(a,z) -> start add a z) (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 action =+ withService ctx name param service action+ 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+ trycon cmd sockname retries+ Z.subscribe cmd ""+ 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 action =+ withService ctx name param service action+ 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+ trycon cmd sockname retries+ Z.subscribe cmd ""+ 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 t = Z.unsubscribe (subSock s) t++ ------------------------------------------------------------------------+ -- | Subscribe another topic+ ------------------------------------------------------------------------+ resubscribe :: Sub i -> Topic -> IO ()+ resubscribe s t = Z.subscribe (subSock s) t++ ------------------------------------------------------------------------+ -- 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 action =+ withService ctx name param service action+ 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+ trycon cmd sockname retries+ Z.subscribe cmd ""+ 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)
+ src/Network/Mom/Patterns/Device.hs view
@@ -0,0 +1,560 @@+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.ByteString.Char8 as B+ 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 action =+ withService ctx name param service action+ 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 act = + withDevice ctx name noparam (-1)+ [pollEntry "clients" XDealer dealer l1 [],+ pollEntry "servers" XRouter router l2 []]+ return return onerr (\_ -> return ()) (\_ -> putThrough) act++ ------------------------------------------------------------------------+ -- | 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 act = + withDevice ctx name noparam (-1)+ [pollEntry "subscriber" XSub sub l1 topics, + pollEntry "publisher" XPub pub l2 []]+ return return onerr (\_ -> return ()) (\_ -> putThrough) act++ ------------------------------------------------------------------------+ -- | 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 act = + withDevice ctx name noparam (-1)+ [pollEntry "pull" XPull puller l1 [], + pollEntry "push" XPipe pusher l2 []]+ return return onerr (\_ -> return ()) (\_ -> putThrough) act++ ------------------------------------------------------------------------+ -- | 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)+ (do withMVar m (\xp' -> mapM_ closeS (xpPoll xp')) >> 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 ())+
+ src/Network/Mom/Patterns/Enumerator.hs view
@@ -0,0 +1,420 @@+{-# 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 Control.Monad.Trans+ import Prelude hiding (catch)+ import Control.Exception (AssertionFailed(..), catch, throwIO)++ 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)+
+ src/Service.hs view
@@ -0,0 +1,381 @@+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.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 take 4 x == "APP " + then Right $ read x+ else + if take 4 x == "OPT "+ then Right $ read x+ else + if take 6 x == "DEVICE"+ 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)))
+ src/Types.hs view
@@ -0,0 +1,566 @@+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 Data.List (intercalate)+ import Data.List.Split (splitOn)+ 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 -> do+ 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 = []