diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/license/lgpl-3.0.txt b/license/lgpl-3.0.txt
new file mode 100644
--- /dev/null
+++ b/license/lgpl-3.0.txt
@@ -0,0 +1,165 @@
+ GNU LESSER GENERAL PUBLIC LICENSE
+ Version 3, 29 June 2007
+
+ Copyright (C) 2007 Free Software Foundation, Inc.
+ 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.
diff --git a/patterns.cabal b/patterns.cabal
new file mode 100644
--- /dev/null
+++ b/patterns.cabal
@@ -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
+ .
+
+ More examples and a test suite are available
+ on .
+
+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
+
diff --git a/src/Factory.hs b/src/Factory.hs
new file mode 100644
--- /dev/null
+++ b/src/Factory.hs
@@ -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
+
diff --git a/src/Network/Mom/Patterns.hs b/src/Network/Mom/Patterns.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Mom/Patterns.hs
@@ -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
+-- .
+-------------------------------------------------------------------------------
+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./
+
+ -}
+
diff --git a/src/Network/Mom/Patterns/Basic.hs b/src/Network/Mom/Patterns/Basic.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Mom/Patterns/Basic.hs
@@ -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)
diff --git a/src/Network/Mom/Patterns/Device.hs b/src/Network/Mom/Patterns/Device.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Mom/Patterns/Device.hs
@@ -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 ())
+
diff --git a/src/Network/Mom/Patterns/Enumerator.hs b/src/Network/Mom/Patterns/Enumerator.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Mom/Patterns/Enumerator.hs
@@ -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)
+
diff --git a/src/Service.hs b/src/Service.hs
new file mode 100644
--- /dev/null
+++ b/src/Service.hs
@@ -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)))
diff --git a/src/Types.hs b/src/Types.hs
new file mode 100644
--- /dev/null
+++ b/src/Types.hs
@@ -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 = []