stomp-patterns (empty) → 0.0.1
raw patch · 9 files changed
+2633/−0 lines, 9 filesdep +basedep +bytestringdep +containerssetup-changed
Dependencies added: base, bytestring, containers, mime, mtl, split, stomp-queue, stompl, time
Files
- Network/Mom/Stompl/Patterns/Balancer.hs +142/−0
- Network/Mom/Stompl/Patterns/Basic.hs +1053/−0
- Network/Mom/Stompl/Patterns/Bridge.hs +178/−0
- Network/Mom/Stompl/Patterns/Desk.hs +179/−0
- Network/Mom/Stompl/Patterns/Registry.hs +609/−0
- Network/Mom/Stompl/Patterns/Types.hs +196/−0
- Setup.hs +2/−0
- license/lgpl-3.0.txt +165/−0
- stomp-patterns.cabal +109/−0
+ Network/Mom/Stompl/Patterns/Balancer.hs view
@@ -0,0 +1,142 @@+{-# Language BangPatterns #-}+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Stompl/Patterns/Balancer.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- Stability : experimental+-- Portability: portable+--+-- This module provides a balancer for services and tasks+-- and a topic router.+-- Balancers for services and tasks improve scalability and reliability+-- of servers and workers. Workers should always be used with a balancer+-- (since balancing workload is the main idea of workers);+-- servers can very well be used without a balancer, but won't scale+-- with increasing numbers of clients.+--+-- A balancer consists of a registry to which +-- servers and workers connect;+-- servers and workers are maintained in lists +-- according to the job they provide.+-- Clients and pushers send requests to the balancer,+-- which then forwards the request to a server or worker.+-- The client will receive the reply not through the balancer,+-- but directly from the server (to which the reply queue+-- was forwarded as part of the request message -- +-- see 'ClientA' for details).+-- +-- With servers and workers sending heartbeats,+-- a balancer also improves reliability+-- in contrast to a topology+-- where a task is pushed to a single worker or +-- a request is sent to only one server.+-- +-- A router is a forwarder of a topic.+-- A router is very similar to a publisher ('PubA')+-- with the difference that the router+-- does not create new topic data, +-- but uses topic data received from a publisher+-- (a router, hence, is a subscriber and a publisher).+-- Routers can be used to balance the workload of publishers:+-- Instead of one publisher serving thousands of subscribers,+-- the initial publisher would serve thousands of routers,+-- which, in their turn, serve thousands of subscribers +-- (or even other routers).+-------------------------------------------------------------------------------+module Network.Mom.Stompl.Patterns.Balancer (+ -- * Balancer+ withBalancer,+ -- * Router + withRouter)+where++ import Registry+ import Types+ import Network.Mom.Stompl.Client.Queue + import Network.Mom.Stompl.Patterns.Basic+ import Codec.MIME.Type (nullType)+ import Prelude hiding (catch)+ import Control.Exception (throwIO, catches)+ import Control.Monad (forever, unless)++ -----------------------------------------------------------------------+ -- | Create a Service and Task Balancer with the lifetime+ -- of the application-defined action passed in+ -- and start it in a background thread:+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the balancer, used for error handling;+ --+ -- * 'QName': Registration queue -- this queue is used+ -- by providers to connect to the registry,+ -- it is not used for consumer requests;+ --+ -- * (Int, Int): Heartbeat range of the 'Registry' + -- (see 'withRegistry' for details);+ -- + -- * 'QName': Request queue -- this queue is used+ -- for consumer requests;+ --+ -- * 'OnError': Error handling;+ --+ -- * IO r: Action that defines the lifetime of the balancer;+ -- the result /r/ is also the result of /withBalancer/.+ -----------------------------------------------------------------------+ withBalancer :: Con -> String -> QName -> (Int, Int) -> + QName -> OnError -> IO r -> IO r+ withBalancer c n qn (mn,mx) rq onErr action =+ withRegistry c n qn (mn,mx) onErr $ \reg -> + withPair c n (rq, [], [], bytesIn)+ ("unknown", [], [], bytesOut) $ \(r,w) -> + withThread (balance reg r w) action+ where balance reg r w = + forever $ catches (do+ m <- readQ r+ jn <- getJobName m+ t <- mapR reg jn (send2Prov w m)+ unless t $ throwIO $ NoProviderX jn)+ (ignoreHandler n onErr)+ send2Prov w m p = writeAdHoc w (prvQ p) nullType + (msgHdrs m) $ msgContent m++ -----------------------------------------------------------------------+ -- | Create a router with the lifetime of the + -- application-defined action passed in+ -- and start it in a background thread:+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the router, used for error handling;+ --+ -- * 'JobName': Routed topic;+ --+ -- * 'QName': Registration queue of the source publisher;+ --+ -- * 'QName': Queue through which the internal subscriber+ -- will receive the topic data from the source publisher;+ --+ -- * 'QName': Registration queue of the target publisher+ -- to which subscribers will connect;+ --+ -- * Int: Registration timeout + -- (timeout to register at the source publisher);+ -- + -- * 'QName': Request queue -- this queue is used+ -- for consumer requests;+ --+ -- * 'OnError': Error handling;+ --+ -- * IO r: Action that defines the lifetime of the router;+ -- the result /r/ is also the result of /withRouter/.+ -----------------------------------------------------------------------+ withRouter :: Con -> String -> JobName -> + QName -> QName -> QName -> + Int -> OnError -> IO r -> IO r+ withRouter c n jn srq ssq trq tmo onErr action = + withPub c n jn trq onErr + ("unknown", [], [], bytesOut) $ \p ->+ withSubThread c n jn srq tmo + (ssq, [], [], bytesIn) (pub p) onErr action+ where pub p m = publish p nullType (msgHdrs m) $ msgContent m
+ Network/Mom/Stompl/Patterns/Basic.hs view
@@ -0,0 +1,1053 @@+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Stompl/Patterns/Basic.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- Stability : experimental+-- Portability: portable+--+-- This module provides the /basic/ patterns+-- client\/server, publish and subscribe and pusher\/worker+-- (a.k.a. pipeline) as well as a /registry/,+-- a means to support patterns+-- where one application uses or serves a set of other applications. +-- +-- Basic patterns can be distinguished by the data exchange+-- defined by their protocol:+--+-- * client\/server: The client, first, sends data to the server+-- (the request) and, then, +-- the server sends data to this client+-- (the reply);+--+-- * publish and subscribe: The publisher sends data to all subscribers+-- (the topic);+--+-- * pusher\/worker: The pusher sends data to the worker (the request)+-- without receiving a reply.+--+-- We call the processing performed by an application on behalf of +-- another application a /job/.+-- There are three different job types:+--+-- * service: must be requested explicitly and includes a message+-- sent from the application that perfoms the service (server)+-- to the application that requests the service (client).+--+-- * task: must be requested explicitly, but does not include a reply.+--+-- * topic: is sent to registered subscribers without being requested+-- explicitly.+--+-- Applications providing a job are generically called providers, +-- applications requesting a job are called consumers.+-- Providers, hence, are servers, workers and publishers, and+-- consumers are clients, pushers and subscribers.+-- Note that this is somewhat different from the +-- data-centric terminology \"producer\" and \"consumer\".+-- It is not very useful+-- to distinugish servers from clients or+-- pushers from workers by referring to the distinction+-- of producing or not producing data.+-- It is in fact the pusher that produces data,+-- not the worker. The pusher, however, is the one+-- that requests something from the worker. The task, in this case,+-- is the \"good\" that is provided by one side and consumed+-- by the other.+--+-- This distinction is relevant when we start to think+-- about /reliability/.+-- Reliability is a relation between a provider and a consumer:+-- The consumer relies on the producer,+-- not the other way round, /e.g./+-- a pusher relies on a woker+-- and a client on a server to get the job done.+--+-- The interfaces in this library +-- give some guarantees related to reliability, +-- but there are also some pitfalls:+-- +-- * A client using timeouts can be sure that +-- the requested service has been performed+-- when the reply arrives before the timeout expires.+-- If no reply arrives before timeout expiration,+-- no such claim can be made (in particular not+-- that the service has not been performed).+-- The service may have been performed, but +-- it may have taken more time than expected or+-- it may have been performed, +-- but the reply message has failed to arrive.+-- If the service is /idempotent/ -+-- /i.e./ calling the service twice has +-- the same effect as calling it once -+-- the client, when the timeout has expired,+-- can just send the request once again;+-- otherwise, it has to use other means to recover +-- from this situation.+--+-- * A pusher will never know if the task has been performed+-- correctly, since there is no response from the worker.+-- This is one of the reasons that the pipeline pattern +-- should usually not be used alone, but in the context of+-- a balancer. (You usually want to push +-- a request to a worker through a balancer --+-- one of the ideas behind pusher/woker is work balancing.)+-- A balancer may request providers to send /heartbeats/+-- and, this way, minimise the risk of failure.+-- The worker still may fail between a heartbeat+-- and a request and even the fact that it does send hearbeats+-- does not necessarily mean that it is operating correctly.+-- If it is essential for the client to know+-- that all tasks have been performed correctly,+-- other verification means are required.+-- +-- * Finally, a subscriber will never know+-- whether a publisher is still working correctly+-- or not, if the publisher does not send data+-- periodically. A reliable design+-- would use periodic publishers, /i.e./+-- publishers that send data at a constant rate,+-- even if no new data are available.+-- The data update, in this case,+-- would have the effect of a heartbeat.+--+-- The library uses a set of headers that must not be used by applications.+-- All internal header keys start and end with two underscores.+-- By avoiding this naming of header keys, application code easily avoids+-- naming conflicts. The headers used in basic patterns are:+--+-- * __channel__: Reply queue (client\/server)+--+-- * __job__: The requested /job/ (client\/server, pusher\/worker and registry)+--+-- * __type__: Request type (registry)+--+-- * __job-type__: Type of job (registry)+--+-- * __queue__: Queue to register (registry)+--+-- * __hb__: Heartbeat specification (registry)+--+-- * __sc__: Status Code (registry) +-------------------------------------------------------------------------------+module Network.Mom.Stompl.Patterns.Basic (+ -- * Client+ ClientA, clName, withClient, request, checkRequest,++ -- * Server+ ServerA, srvName, withServer, reply,++ -- * Registry+ -- $registry_intro++ register, unRegister, + heartbeat, HB, mkHB,++ -- $registry_howto++ -- * withServerThread+ withServerThread, RegistryDesc,++ -- * Pusher+ PusherA, pushName, withPusher, push,++ -- * Worker+ withTaskThread,++ -- * More about Registries+ -- $registry_core++ Registry, withRegistry, + mapR, getProvider, showRegistry,+ Provider, prvQ, JobType(..),++ -- $registry_usage++ -- * Publisher+ PubA, pubName, withPub, publish, withPubThread,++ -- * Subscriber+ SubA, subName, withSub, checkIssue,+ withSubThread, withSubMVar,++ -- * Heartbeats for Pub+ withPubProxy,++ -- * Exceptions and Error Handling+ PatternsException(..), OnError,+ StatusCode(..), readStatusCode,++ -- * Useful Types and Helpers+ JobName, QName,+ nobody, ignorebody,+ bytesOut, bytesIn,+ getJobName, getJobType, getQueue, getChannel, + getHB,+ getSC,+ getHeader)+where++ import Registry+ import Types++ import Network.Mom.Stompl.Client.Queue + import qualified Network.Mom.Stompl.Frame as F+ import System.Timeout+ import Codec.MIME.Type (Type)+ import Prelude hiding (catch)+ import Control.Exception (throwIO, catches, finally)+ import Control.Concurrent + import Control.Monad (forever, unless, when, void)+ import Data.Time+ import qualified Data.ByteString.Char8 as B++ {- $registry_intro+ Before we continue the survey of basic patterns,+ we have to introduce registries.+ Registries are used by some patterns, advanced patterns,+ but also publishers,+ to use a set of providers that register beforehand+ or, in the case of publishers, + to serve a set of consumers that have registered to+ the publisher.+ A typical example is balancers (/majordomo pattern/):+ Servers or tasks register to a balancer, which + on receiving a request from a client for a certain job,+ forwards the request to one of the registered providers+ of this job. Internally, the register balances the request,+ such that, with more than one provider currently registered,+ two consecutive requests will not be served+ by the same provider.+ Note that registers provide different modes of using providers:+ a load balancer will send a request to only one of its providers,+ whereas publishers will send the data they produce to all+ currently registered consumers.+ The difference is defined by the 'JobType' of a given 'Job'.+ Of course, only providers of the same type may register+ for the same job. ++ Registers provide a queue through which services can register;+ patterns using registers would provide + another queue through which they receive requests. + Registers allow for some level of reliability,+ /i.e./ registers can ensure with certain probability + that providers are available at the time,+ when a request is made.+ Therefore, registries may request heartbeats from providers.+ Heartbeats are negotiated on registration.+ Note that registries do not send heartbeats back to providers.+ Providers have to use other strategies to make sure+ that the registry to which they have registered is actually+ available. + -}++ {- $registry_howto+ The following example shows+ how to use the registration functions and heartbeats+ together with a server:++ > -- The definition of the variables+ > -- reg, jn, rn, tmo+ > -- is out of the scope of this listing;+ > -- their data type and meaning + > -- can be inferred from the context.+ >+ > withConnection "127.0.0.1" 61613 [] [] $ \c -> + > withServer c "Test" (q, [], [], iconv)+ > ("unknown", [], [], oconv) $ \s -> do+ > (sc,me) <- if null reg -- if parameter reg is null+ > then return (OK, 0)+ > else register c jn Service reg rn tmo 500+ > case sc of+ > -- ok ------------------------------+ > OK -> + > if me < 0 || me > 5000 -- accept heartbeat from + > -- 0 (no heartbeats) to+ > -- 5 seconds+ > then do void $ unRegister c jn wn rn tmo+ > throwIO $ UnacceptableHbX me+ > else do hb <- mkHB me+ > m <- newMVar hb + > let p = if me <= 0 then (-1) else 1000 * me + > withWriter c "HB" reg [] [] nobody $ \w -> + > finally (forever $+ > reply s p t hs transform >> heartbeat m w jn rn) (do+ > -- "don't forget to unregister!" (Frank Zappa)+ > sc <- unRegister c jn wn rn tmo+ > unless (sc == OK) $ + > throwIO $ NotOKX sc "on unregister")+ > -- not ok ---------------------------+ > e -> throwIO $ NotOKX e "on register"++ There is, however, a function that does all of this internally: + 'withServerThread'.+ -}++ {- $registry_core+ Until now, we have only looked at how to connect to a registry,+ not at how to use it and what a registry actually is in terms of+ data types.+ Well, answering the second question is simple:+ a registry, from the perspective of the user application,+ is an opaque data type with a set of functions:+ -}++ {- $registry_usage+ A typical example of how to use a registry in practice + is the balancer pattern, which is shown (without error handling)+ below:++ > -- The definition of the variables+ > -- c, n qn, mn, mx, onErr, rq+ > -- is out of the scope of this listing;+ > -- their data type and meaning + > -- can be inferred from the context.+ >+ > withRegistry c n qn (mn, mx) onErr $ \reg ->+ > withPair c n (rq, [], [], bytesIn) + > ("unknown", [], [], bytesOut) $ \(r,w) -> + > forever $ do+ > m <- readQ r -- receive a request+ > jn <- getJobName m -- get the job name from the request + > t <- mapR reg jn (send2Prov w m) -- apply job+ > unless t $ throwIO $ NoProviderX jn -- throw exception+ > -- when job is not provided+ > where send2Prov w m p = writeAdHoc w (prvQ p) nullType + > (msgHdrs m) $ msgContent m++ User applications, usually, do not need to use registries directly.+ Registries are used in patterns, namely in Desks, Balancers+ and in /Pub/s.+ -}++ ------------------------------------------------------------------------+ -- | The client data type, which implements the client side+ -- of the client\/server protocol.+ ------------------------------------------------------------------------+ data ClientA i o = Cl {+ -- | Access to the client name+ clName :: String,+ clChn :: QName,+ clJob :: JobName,+ clIn :: Reader i,+ clOut :: Writer o}+ + ------------------------------------------------------------------------+ -- | The function creates a client that lives within its scope.+ --+ -- Parameters:+ --+ -- * 'Con': Connection to a Stomp broker+ --+ -- * 'String': Name of the Client, which can be used for error reporting.+ --+ -- * 'JobName': Name of the 'Service' the client will request+ --+ -- * 'ReaderDesc' i: Description of a reader queue;+ -- this is the queue through which the server+ -- will send its response.+ --+ -- * 'WriterDesc' o: Description of a writer queue;+ -- this is the queue through which the server+ -- is expecting requests.+ --+ -- * 'ClientA' i o -> IO r: An application-defined action+ -- whose scope defines the client's lifetime+ ------------------------------------------------------------------------+ withClient :: Con -> String ->+ JobName ->+ ReaderDesc i ->+ WriterDesc o ->+ (ClientA i o -> IO r) -> IO r+ withClient c n jn rd@(rn, _, _, _) wd act =+ withPair c n rd wd $ \(r,w) -> act $ Cl n rn jn r w++ ------------------------------------------------------------------------+ -- | The client will send the request of type /o/+ -- and wait for the reply until the timeout exprires.+ -- The reply is of type /i/ and is returned as 'Message' /i/.+ -- If the timeout expires before the reply has been received,+ -- the function returns 'Nothing'.+ --+ -- Since servers do not know the clients they are serving,+ -- 'request' sends the name of its reader queue (the /reply queue/)+ -- as message header to the server.+ --+ -- Parameters:+ --+ -- * 'ClientA' i o: The client; note that i is the type of the reply,+ -- o is the type of the request.+ --+ -- * 'Int': The timeout in microseconds.+ --+ -- * 'Type': The /MIME/ type of the request.+ --+ -- * ['F.Header']: List of additional headers + -- to be sent with the request.+ --+ -- * /o/: The request + ------------------------------------------------------------------------+ request :: ClientA i o -> + Int -> Type -> [F.Header] -> o -> IO (Maybe (Message i))+ request c tmo t hs r = + let hs' = [("__channel__", clChn c),+ ("__job__", clJob c)] ++ hs+ in writeQ (clOut c) t hs' r >> timeout tmo (readQ (clIn c))++ ------------------------------------------------------------------------+ -- | This function serves as a \"delayed\" receiver for the case+ -- that the timeout of a request has expired.+ -- When using this function, it is assumed+ -- that a request has been made, but no response has been received.+ -- It can be used in time-critical applications,+ -- where the client may use the time between request and reply+ -- productively, instead of passively blocking on the reply queue.+ --+ -- Use this function with care! It can be easily abused+ -- to break the client\/server pattern, when it is called+ -- without a request having been made before.+ -- If, in this case, /timout/ is /-1/,+ -- the application will block forever.+ --+ -- The function receives those parameters from 'request'+ -- that are related to receiving the reply, /i.e./+ -- 'Type', ['F.Header'] and /o/ are not passed to /checkRequest/.+ ------------------------------------------------------------------------+ checkRequest :: ClientA i o -> Int -> IO (Maybe (Message i))+ checkRequest c tmo = timeout tmo $ readQ (clIn c)++ ------------------------------------------------------------------------+ -- | The server data type, which implements the server side+ -- of the client\/server protocol.+ ------------------------------------------------------------------------+ data ServerA i o = Srv {+ -- | Access to the server name+ srvName :: String,+ srvIn :: Reader i,+ srvOut :: Writer o}+ + ------------------------------------------------------------------------+ -- | The function creates a server+ -- that lives within the scope of the application-defined action+ -- passed into it.+ --+ -- Parameters:+ --+ -- * 'Con': Connection to a Stomp broker+ --+ -- * 'String': Name of the Server, which can be used for error reporting.+ --+ -- * 'ReaderDesc' i: Description of a reader queue;+ -- this is the queue through which clients+ -- are expected to send requests.+ --+ -- * 'WriterDesc' o: Description of a writer queue;+ -- this is the queue through which+ -- a specific client will expect the reply.+ -- Note that the server will overwrite+ -- the destination of this queue+ -- using 'writeAdHoc'; + -- the destination of this queue, hence,+ -- is irrelevant.+ --+ -- * 'ServerA' i o -> IO r: An application-defined action+ -- whose scope defines the server's lifetime+ ------------------------------------------------------------------------+ withServer :: Con -> String ->+ ReaderDesc i ->+ WriterDesc o ->+ (ServerA i o -> IO r) -> IO r+ withServer c n rd wd act =+ withPair c n rd wd $ \(r,w) -> act $ Srv n r w++ ------------------------------------------------------------------------+ -- | Waits for a client request, + -- calls the application-defined transformer to generate a reply+ -- and sends this reply through the reply queue+ -- whose name is indicated by a header in the request.+ -- The time a server waits for a request may be restricted+ -- by the timeout. Typically, you would call reply with + -- timeout set to /-1/ (/wait eternally/).+ -- There may be situations, however, where it actually+ -- makes sense to restrict the waiting time,+ -- /i.e./ to perform some housekeeping in between.+ --+ -- Typically, you call reply in a loop like+ --+ -- > forever $ reply srv (-1) nullType [] f+ --+ -- where /f/ is a function of type + --+ -- > Message i -> IO o.+ --+ -- Parameters:+ --+ -- * 'ServerA' i o: The server; note that i is the request queue+ -- and o the reply queue.+ --+ -- * 'Int': The timeout in microseconds.+ --+ -- * 'Type': The /MIME/ type of the reply.+ --+ -- * ['F.Header']: Additional headers to be sent with the reply.+ --+ -- * 'Message' i -> IO o: Transforms the request into a reply -+ -- this defines the service provided by this+ -- application.+ ------------------------------------------------------------------------+ reply :: ServerA i o -> Int -> Type -> [F.Header] -> + (Message i -> IO o) -> IO ()+ reply s tmo t hs transform = do+ mbM <- timeout tmo $ readQ (srvIn s)+ case mbM of+ Nothing -> return ()+ Just m -> do+ c <- getChannel m+ o <- transform m+ writeAdHoc (srvOut s) c t hs o++ ------------------------------------------------------------------------+ -- | Create a server that works in a background thread:+ -- The background thread (and with it the server)+ -- is running until the action passed in to the function (IO r)+ -- terminates; when it terminates, the background thread is+ -- terminated as well.+ -- /withServerThread/ may connect to a registry+ -- (to serve as a provider of a balancer for instance),+ -- which is automatically handled internally+ -- when a RegistryDesc is passed in with a 'QName'+ -- that is not null. + --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * 'String': The name of the server, used for error reporting;+ --+ -- * 'JobName': The job provided by this server+ --+ -- * 'Type': The MIME Type (passed to 'reply')+ --+ -- * ['F.Header']: Additional headers (passed to 'reply')+ --+ -- * 'Message' i -> IO o: The core of the reply function:+ -- transforming a request of type /i/+ -- into a reply of type /o/+ --+ -- * 'ReaderDesc' i: The reader through which requests are expected;+ -- + -- * 'WriterDesc' o: The writer through which replies are sent;+ --+ -- * 'RegistryDesc': Describes whether and how to connect to a registry:+ -- if the queue name of the registry description + -- is null,+ -- the function will not connect to a registry;+ -- otherwise it will connect to the registry+ -- proposing the best value of the 'RegistryDesc'+ -- as its preferred heartbeat rate;+ -- should the heartbeat rate returned by the registry+ -- be outside the scope of min and max,+ -- /withServerThread/ will terminate + -- with 'UnacceptableHbX'.+ --+ -- * 'OnError': Error handler+ --+ -- * IO r: The function starts a new thread on which the + -- the server is working; + -- the thread from which the function was called+ -- continues in this action.+ -- Its return value is also the result of /withServerThread/.+ -- When the action terminates,+ -- the new thread is terminated internally.+ ------------------------------------------------------------------------+ withServerThread :: Con -> String -> JobName ->+ Type -> [F.Header] -> (Message i -> IO o) ->+ ReaderDesc i -> + WriterDesc o ->+ RegistryDesc -> + OnError -> IO r -> IO r+ withServerThread c n jn t hs transform+ rd@(rn, _, _, _)+ wd+ (reg, tmo, (best, mn, mx))+ onErr action =+ withServer c n rd wd $ \s -> do+ (sc,me) <- if null reg + then return (OK, 0)+ else register c jn Service reg rn tmo best+ case sc of+ OK -> + if me < mn || me > mx+ then do finalise c jn reg rn tmo+ throwIO $ UnacceptableHbX me++ else do hb <- mkHB me+ m <- newMVar hb + let p = if me <= 0 then (-1) else 1000 * me + withThread (finally (srv m p s) + (finalise c jn reg rn tmo)) action+ e -> throwIO $ NotOKX e "on register"+ where srv m p s = withWriter c "HB" reg [] [] nobody $ \w -> + forever $ catches (+ reply s p t hs transform >> heartbeat m w jn rn) (+ ignoreHandler (srvName s) onErr)++ ------------------------------------------------------------------------+ -- Finaliser for the registry+ ------------------------------------------------------------------------+ finalise :: Con -> JobName -> QName -> QName -> Int -> IO ()+ finalise c jn wn rn tmo | null wn = return ()+ | otherwise = do + sc <- unRegister c jn wn rn tmo+ unless (sc == OK) $ + throwIO $ NotOKX sc "on unregister"++ ------------------------------------------------------------------------+ -- | The publisher data type+ ------------------------------------------------------------------------+ data PubA o = Pub {+ -- | Access to the name of the publisher+ pubName :: String,+ pubJob :: JobName,+ pubReg :: Registry,+ pubConv :: OutBound o,+ pubOut :: Writer B.ByteString}++ ------------------------------------------------------------------------+ -- | Create a publisher with the lifetime of the scope+ -- of the user action passed in.+ -- The publisher, internally, creates a registry+ -- to which subscribers will connect to obtain the topic data.+ -- The registry will not expect heartbeats from subscribers,+ -- since the dependability relation is the other way round:+ -- the publisher does not depend on subscribers,+ -- but subscribers depend on a publisher.+ -- The publisher, usually, does not send heartbeats either.+ -- For exceptions to this rule, see 'withPubProxy'.+ --+ -- * 'Con': Connect to a Stomp broker;+ --+ -- * String: Name of the publisher used for error reporting;+ --+ -- * 'JobName': The name of the topic;+ --+ -- * 'QName': Name of the registration queue (see 'withRegistry');+ --+ -- * 'OnError': Error Handler passed to the registry;+ --+ -- * 'WriterDesc': Queue through which data are published;+ -- note that the queue name is irrelevant.+ -- The publisher will send data to the queues+ -- of registered subscribers (see 'mapR');+ --+ -- * 'PubA' -> IO r: Action that defines the lifetime+ -- of the publisher; the result (/r/)+ -- is also the result of /withPub/.+ ------------------------------------------------------------------------+ withPub :: Con -> String -> JobName -> QName -> OnError -> + WriterDesc o -> (PubA o -> IO r) -> IO r+ withPub c n jn rn onErr (_, wos, wh, oconv) act = + withRegistry c n rn (0,0) onErr $ \r ->+ withWriter c jn "unknown" wos wh bytesOut $ \w -> + act $ Pub n jn r oconv w++ ------------------------------------------------------------------------+ -- | Publish data of type /o/:+ --+ -- * 'PubA' o: Publisher to use;+ --+ -- * 'Type': MIME Type of the message to be sent;+ --+ -- * ['F.Header']: Additional headers to be sent with the message;+ --+ -- * /o/: The message content.+ ------------------------------------------------------------------------+ publish :: PubA o -> Type -> [F.Header] -> o -> IO ()+ publish p t hs x = let oc = pubConv p+ in oc x >>= \m ->+ void $ mapR (pubReg p) (pubJob p) $ \prv -> + writeAdHoc (pubOut p) (prvQ prv) t hs m++ ------------------------------------------------------------------------+ -- | Create a publisher that works in a background thread+ -- publishing periodically at a monotonic rate,+ -- /i.e./ it creates data and publishes them,+ -- computes the difference + -- of the publication rate minus the time needed+ -- to create and publish the data + -- and will then suspend the thread for this period.+ -- For a publication rate of /p/ microseconds,+ -- the thread will be delayed for /p - x/ microseconds,+ -- if /x/ corresponds to the time that was spent+ -- on creating and publishing the data.+ --+ -- The precision depends of course on your system and+ -- its current workload.+ -- For most cases, this will be equal to just suspending the thread+ -- for the publication rate.+ --+ -- Parameters:+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the publisher used for error reporting;+ --+ -- * 'JobName': Name of the topic;+ --+ -- * 'QName': Registration queue;+ --+ -- * Type: MIME Type of the published message;+ --+ -- * ['F.Header']: Additional headers to be sent+ -- with the message;+ --+ -- * IO o: Action to create the message content;+ --+ -- * 'WriterDesc' o: Queue through which the message+ -- will be published (remember, however,+ -- that the queue name is irrelevant);+ --+ -- * Int: Publication rate in microseconds;+ --+ -- * 'OnError': Error handler for the registry+ -- and the publisher;+ --+ -- * IO r: Action that defines the lifetime of the publisher;+ -- The result /r/ is also the result of /withPubThread/. + ------------------------------------------------------------------------+ withPubThread :: Con -> String -> JobName -> QName ->+ Type -> [F.Header] -> IO o ->+ WriterDesc o -> Int -> + OnError -> IO r -> IO r+ withPubThread c n jn rn t hs create wd period onErr action = + withPub c n jn rn onErr wd $ \p -> withThread (doPub p) action+ where doPub p = forever $ catches (do+ n1 <- getCurrentTime+ create >>= publish p t hs+ n2 <- getCurrentTime+ let d = nominal2us (n2 `diffUTCTime` n1)+ when (d < period) $ threadDelay (period - d)) (+ ignoreHandler (pubName p) onErr)++ ------------------------------------------------------------------------+ -- | Subscriber data type+ ------------------------------------------------------------------------+ data SubA i = Sub {+ -- | Access to the subscriber name+ subName :: String,+ subIn :: Reader i+ }++ ------------------------------------------------------------------------+ -- | Create a subscriber with the lifetime + -- of the user action passed in.+ -- The subscriber will internally connect to a publisher's+ -- registry and receive data as long as it stays connected.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Subscriber name useful for error reporting;+ --+ -- * 'JobName': Subscribed topic;+ --+ -- * 'QName': Queue of a registry to connect to+ -- (the 'Pub's registration queue!)+ --+ -- * Int: Registration timeout in microseconds;+ --+ -- * 'ReaderDesc': This is the queue through which+ -- the subscriber will receive data.+ --+ -- * 'SubA' i -> IO r: Action that defines the lifetime+ -- of the subscriber. Its result /r/+ -- is also the result of /withSub/.+ ------------------------------------------------------------------------+ withSub :: Con -> String -> JobName -> QName -> Int ->+ ReaderDesc i -> (SubA i -> IO r) -> IO r+ withSub c n jn wn tmo (rn, ros, rh, iconv) act = + withReader c n rn ros rh iconv $ \r -> do+ mbR <- if null wn + then return $ Just (OK,0)+ else timeout tmo $ register c jn Topic wn rn tmo 0+ case mbR of+ Nothing -> throwIO $ TimeoutX "on register"+ Just (OK,_) -> finally (act $ Sub n r) + (finalise c jn wn rn tmo)+ Just (sc,_) -> throwIO $ NotOKX sc "on register "++ ------------------------------------------------------------------------+ -- | Check if data have been arrived for this subscriber;+ -- if data are available before the timeout expires,+ -- the function results in 'Just' ('Message' i);+ -- if the timeout expires first, the result is 'Nothing'.+ --+ -- * 'SubA' i: The subscriber to check + --+ -- * Int: Timeout in microseconds+ ------------------------------------------------------------------------+ checkIssue :: SubA i -> Int -> IO (Maybe (Message i))+ checkIssue s tmo = timeout tmo $ readQ (subIn s)++ ------------------------------------------------------------------------+ -- | Create a subscriber that works in a background thread;+ -- Whenever data are available, an application callback passed in+ -- to the function is called with the message that has arrived.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Subscriber name used for error reporting;+ --+ -- * 'JobName': Subscribed topic;+ --+ -- * 'QName': The publisher's registration queue;+ --+ -- * Int: Registration timeout in microseconds;+ --+ -- * 'ReaderDesc' i: Queue through which the subscriber+ -- shall receive data;+ --+ -- * 'Message' i -> IO (): Application callback;+ --+ -- * 'OnError': Error handler; + --+ -- * IO r: Action that defines the lifetime of the subscriber;+ -- the result /r/ is also the result of /withSubThread/.+ ------------------------------------------------------------------------+ withSubThread :: Con -> String -> JobName -> QName -> Int ->+ ReaderDesc i -> (Message i -> IO ()) -> OnError -> + IO r -> IO r+ withSubThread c n jn wn tmo rd job onErr action = + withSub c n jn wn tmo rd $ \s -> withThread (go s) action+ where go s = forever $ catches (chk s >>= job)+ (ignoreHandler (subName s) onErr)+ chk s = checkIssue s (-1) >>= \mbX ->+ case mbX of+ Nothing -> chk s+ Just m -> return m++ ------------------------------------------------------------------------+ -- | Create a subscriber that works in a background thread + -- and updates an MVar, whenever new data are available;+ -- the function is in fact a special case of 'withSubThread',+ -- where the application callback updates an MVar.+ -- Note that the MVar must not be empty when the function+ -- is called, otherwise, it will block on modifying the MVar.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Subscriber name used for error reporting;+ --+ -- * 'JobName': Subscribed topic;+ --+ -- * 'QName': The publisher's registration queue;+ --+ -- * Int: Registration timeout in microseconds;+ --+ -- * 'ReaderDesc' i: Queue through which the subscriber+ -- shall receive data;+ --+ -- * 'MVar' i: MVar to update;+ --+ -- * 'OnError': Error handler; + --+ -- * IO r: Action that defines the lifetime of the subscriber;+ -- the result /r/ is also the result of /withSubMVar/.+ ------------------------------------------------------------------------+ withSubMVar :: Con -> String -> JobName -> QName -> Int ->+ ReaderDesc i -> MVar i -> OnError -> + IO r -> IO r+ withSubMVar c n jn wn tmo rd v = + withSubThread c n jn wn tmo rd job + where job m = modifyMVar_ v $ \_ -> return $ msgContent m ++ ------------------------------------------------------------------------+ -- | The Pusher data type, which implements+ -- the consumer side of the pipeline protocol.+ -- Note that, when we say "consumer" here,+ -- the pusher is actually a data producer,+ -- but consumes the effect of having a task done.+ -- The pusher can be seen as a client+ -- that does not expect a reply.+ ------------------------------------------------------------------------+ data PusherA o = Pusher {+ -- | Access to the pusher's name+ pushName :: String,+ pushJob :: JobName,+ pushQ :: Writer o+ }++ ------------------------------------------------------------------------+ -- | Create a 'Pusher' with the lifetime of the action passed in:+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the pusher, which may be used for error reporting;+ --+ -- * 'JobName': Name of the job requested by this pusher;+ --+ -- * 'WriterDesc' o: 'Writer' queue through which + -- the job request is pushed;+ --+ -- * ('PusherA' o -> IO r): Action that defines the lifetime of+ -- the pusher; the result /r/ is also+ -- the result of 'withPusher'.+ ------------------------------------------------------------------------+ withPusher :: Con -> String -> JobName -> WriterDesc o -> + (PusherA o -> IO r) -> IO r+ withPusher c n jn (wq, wos, wh, oconv) action = + withWriter c n wq wos wh oconv $ \w -> action $ Pusher n jn w++ ------------------------------------------------------------------------+ -- | Push a 'Job':+ --+ -- * 'PusherA' o: The pusher to be used;+ --+ -- * 'Type': The MIME Type of the message to be sent;+ --+ -- * ['F.Header']: The headers to be sent with the message;+ --+ -- * /o/: The message contents.+ ------------------------------------------------------------------------+ push :: PusherA o -> Type -> [F.Header] -> o -> IO ()+ push p t hs m = let hs' = ("__job__", pushJob p) : hs+ in writeQ (pushQ p) t hs' m++ ------------------------------------------------------------------------+ -- | On the other side of the pipeline,+ -- there sits a worker waiting for requests.+ -- Note that no /Worker/ data type is defined.+ -- Instead, there is only a /withTaskThread/ function+ -- that, internally, creates a worker acting in a background thread.+ -- The rationale is that it does not make too much sense+ -- to have a pipeline with only one worker. + -- It is in fact part of the idea of the pipeline pattern + -- that several workers are used through a balancer.+ -- /withTaskThread/ implements the interaction with the registry+ -- internally and frees the programmer from concerns related+ -- to registration. If you really need a single worker,+ -- you can call the function with an empty RegistryDesc, + -- /i.e./ with an empty queue name.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the worker used for error reporting;+ --+ -- * 'JobName': Name of the job, the worker provides;+ --+ -- * ('Message' i -> IO ()): The job provided by the worker.+ -- Note that the function does not+ -- return a value: Since workers do+ -- not produce a reply, no result+ -- is necessary;+ --+ -- * 'ReaderDesc' i: Queue through which the worker receives+ -- requests;+ --+ -- * 'RegistryDesc': The registry to which the worker connects;+ --+ -- * OnError: Error handler;+ --+ -- * IO r: Action that defines the worker's lifetime.+ --+ ------------------------------------------------------------------------+ withTaskThread :: Con -> String -> JobName ->+ (Message i -> IO ()) -> + ReaderDesc i -> RegistryDesc -> + OnError -> IO r -> IO r+ withTaskThread c n jn task+ (rn, ros, rh, iconv)+ (reg, tmo, (best, mn, mx))+ onErr action = do+ (sc,me) <- if null reg+ then return (OK,0)+ else register c jn Task reg rn tmo best+ case sc of+ OK -> + if me < mn || me > mx+ then do finalise c jn reg rn tmo+ throwIO $ UnacceptableHbX me++ else do hb <- mkHB me+ m <- newMVar hb + let p = if me <= 0 then (-1) else 1000 * me + withReader c n rn ros rh iconv $ \r -> + withThread (finally (tsk m r p) + (finalise c jn reg rn tmo)) action+ e -> throwIO $ NotOKX e "on register"+ where tsk m r p = withWriter c "HB" reg [] [] nobody $ \w -> + forever $ catches (do+ mbM <- timeout p $ readQ r + case mbM of+ Nothing -> return ()+ Just x -> task x + heartbeat m w jn rn)+ (ignoreHandler n onErr)++ ------------------------------------------------------------------------+ -- | Unlike servers and workers,+ -- publishers have no interface to connect + -- internally to a registry.+ -- The rationale for this is that+ -- publishers do not need load balancers or similar means+ -- that would require registration.+ -- As a consequence, there is no means to send heartbeats internally.+ -- Sometimes, however, the need to connect to a registry may arise.+ -- The Desk pattern is an example where it makes sense + -- to register a publisher.+ -- But then, there is no means to internally send heartbeats+ -- proving that the publisher is still alive.+ -- For this case, a simple solution + -- for periodic publishers is available:+ -- a heartbeat proxy that is implemented as a subscriber+ -- receiving data from the publisher and + -- sending a heartbeat on every dataset that arrives.+ -- + -- This function provides a proxy that internally+ -- connects to a registry on behalf of a publisher+ -- and sends heartbeats.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the proxy used for error reporting;+ --+ -- * 'JobName': Name of the topic, the publisher provides;+ --+ -- * 'QName': Registration queue of the publisher -+ -- this is the queue + -- to which the internal subscriber connects;+ --+ -- * 'ReaderDesc' i: The queue through which the internal+ -- subscriber receives data;+ --+ -- * 'RegistryDesc': The other registry - + -- it is this registry to which the + -- proxy will send heartbeats;+ --+ -- * 'OnError': Error Handler;+ --+ -- * IO r: Action that definex the proxy's lifetime;+ -- its result /r/ is also the result of /withPubProxy/.+ ------------------------------------------------------------------------+ withPubProxy :: Con -> String -> JobName -> QName ->+ ReaderDesc i -> RegistryDesc -> OnError -> IO r -> IO r+ withPubProxy c n jn pq rd (reg, tmo, (best, mn, mx)) onErr action =+ withSub c n jn pq tmo rd $ \s -> + withWriter c "HB" reg [] [] nobody $ \w -> do+ (sc, h) <- register c jn Topic reg pq tmo best+ if sc /= OK+ then throwIO $ NotOKX sc "on register proxy"+ else if h < mn || h > mx+ then throwIO $ UnacceptableHbX h+ else do hb <- mkHB h >>= newMVar+ withThread (finally (beat hb (h * 1000) s w 0)+ (finalise c jn reg pq tmo)) + action+ where beat :: MVar HB -> Int -> SubA i -> Writer () -> Int -> IO ()+ beat hb h s w i = forever $ do -- forever continues + -- in case of exception+ mbM <- checkIssue s h+ case mbM of+ Nothing -> if i == 10 + then throwIO $ MissingHbX "No input from pub"+ else beat hb h s w (i+1)+ Just _ -> heartbeat hb w jn pq >> beat hb h s w 0 + `catches` (ignoreHandler n onErr)+
+ Network/Mom/Stompl/Patterns/Bridge.hs view
@@ -0,0 +1,178 @@+{-# Language BangPatterns #-}+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Stompl/Patterns/Bridge.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- Stability : experimental+-- Portability: portable+--+-- Bridges link providers connected to one broker+-- to consumers connected to another broker.+--+-- For publishers and workers, this is quite trivial:+-- the bridge implements +-- the corresponding consumer on one broker+-- and the corresponding provider on the other.+--+-- For servers, the task is somewhat more complicated:+-- since servers use the client's reply queue to send the result+-- back to the client and this queue only exists on the broker+-- to which the client is connected, the bridge has to remember +-- the client's reply queue and use its own queue on the server-side broker+-- to finally route the reply back to the original client.+-- With many broker connected by a service bridge,+-- this can result in long chains of clients and servers +-- sending requests and waiting for replies.+-------------------------------------------------------------------------------+module Network.Mom.Stompl.Patterns.Bridge (+ -- * Forwarder+ withForwarder, + -- * TaskBridge+ withTaskBridge,+ -- * ServiceBridge+ withServiceBridge)+where++ import Types+ import Network.Mom.Stompl.Patterns.Basic + import Network.Mom.Stompl.Client.Queue + import Codec.MIME.Type (nullType)+ import Prelude hiding (catch)+ import Control.Exception (throwIO) ++ -----------------------------------------------------------------------+ -- | Create a forwarder with the lifetime of the application-defined+ -- action passed in and start it in a background thread:+ --+ -- * 'Con': Connection to the source broker+ -- (the one where the original publisher is connected);+ --+ -- * 'Con': Connection to the target broker+ -- (the one where the target subscribers are connected);+ --+ -- * String: Name of the forwarder used for error handling;+ --+ -- * 'JobName': Name of the Topic that is bridged;+ --+ -- * 'QName': Registration queue of the source publisher;+ --+ -- * 'QName': Queue through which the internal subscriber+ -- will receive topic data from the source publisher;+ --+ -- * 'QName': Registration queue of the target publisher;+ --+ -- * Int: Timeout on registering to the source publisher+ -- in microseconds;+ --+ -- * 'OnError': Error handler;+ --+ -- * IO r: Action that defines the lifetime of the forwarder;+ -- its result /r/ is also the result of /withForwarder/.+ --+ -- Note the remarkable similarity to the router pattern ('withRouter').+ -- In fact, a router is but a forwarder where source and target broker+ -- are the same.+ -------------------------------------------------------------------------+ withForwarder :: Con -> Con -> String -> JobName -> + QName -> QName -> QName -> Int -> + OnError -> IO r -> IO r+ withForwarder src trg n jn srq ssq trq tmo onErr action = + withPub trg n jn trq onErr + ("unknown", [], [], bytesOut) $ \p ->+ withSubThread src n jn srq tmo + (ssq, [], [], bytesIn) (pub p) onErr action+ where pub p m = publish p nullType (msgHdrs m) $ msgContent m++ -----------------------------------------------------------------------+ -- | Create a TaskBridge with the lifetime of the action passed in+ -- and start it on a background thread:+ --+ -- * 'Con': Connection to the source broker+ -- (the one to which the pusher is connected);+ --+ -- * 'Con': Connection to the target broker+ -- (the one to which the worker is connected);+ --+ -- * String: Name of the bridge used for error handling;+ --+ -- * 'JobName': Name of the Task that is bridged;+ --+ -- * 'QName': Queue of the worker on the source side;+ -- (if the worker is connected to a balancer+ -- on the source side, this is an internal queue+ -- only visible in the bridge and in the balancer);+ --+ -- * 'QName': Queue of the worker on the target side+ -- (which may be a balancer's request queue);+ --+ -- * 'RegistryDesc': 'Registry' (/i.e./ balancer)+ -- to which the bridge is connected+ -- on the source side;+ --+ -- * 'OnError': Error handler;+ --+ -- * IO r: Action that defines the lifetime of the bridge;+ -- its result /r/ is also the result of /withTaskBridge/.+ -----------------------------------------------------------------------+ withTaskBridge :: Con -> Con -> String -> JobName ->+ QName -> QName ->+ RegistryDesc -> OnError -> IO r -> IO r+ withTaskBridge src trg n jn srq twq reg onErr action =+ withPusher trg n jn (twq, [], [], bytesOut) $ \p ->+ withTaskThread src n jn (fwd p) + (srq, [], [], bytesIn) reg onErr action+ where fwd p m = let hs = filter ((/= "__job__") . fst) $ msgHdrs m+ in push p nullType hs $ msgContent m++ -----------------------------------------------------------------------+ -- | Create a ServiceBridge with the lifetime of the action passed in+ -- and start it on a background thread:+ --+ -- * 'Con': Connection to the source broker+ -- (the one to which the client is connected);+ --+ -- * 'Con': Connection to the target broker+ -- (the one to which the server is connected);+ --+ -- * String: Name of the bridge used for error handling;+ --+ -- * 'JobName': Name of the Service that is bridged;+ --+ -- * 'QName': Queue of the server on the source side;+ -- (if the server is connected to a balancer+ -- on the source side, this is an internal queue+ -- only visible in the bridge and in the balancer);+ --+ -- * 'QName': Reader queue of the internal client on the target side;+ --+ -- * 'QName': Queue of the server on the target side+ -- (which may be a balancer's request queue);+ --+ -- * 'RegistryDesc': 'Registry' (/i.e./ balancer)+ -- to which the bridge is connected+ -- on the source side;+ --+ -- * 'OnError': Error handler;+ --+ -- * IO r: Action that defines the lifetime of the bridge;+ -- its result /r/ is also the result of /withServiceBridge/.+ -----------------------------------------------------------------------+ withServiceBridge :: Con -> Con -> String -> JobName -> + QName -> QName -> QName ->+ RegistryDesc -> OnError -> IO r -> IO r+ withServiceBridge src trg n jn srq trq twq+ reg@(_, tmo, _) onErr action =+ withClient trg n jn (trq, [], [], bytesIn)+ (twq, [], [], bytesOut) $ \c ->+ withServerThread src n jn nullType [] (fwd c)+ (srq, [], [], bytesIn) + ("unknown", [], [], bytesOut)+ reg onErr action+ where fwd c m = + let hs = filter (clFilter . fst) (msgHdrs m)+ in do mbR <- request c tmo nullType hs $ msgContent m+ case mbR of+ Nothing -> throwIO $ TimeoutX "on requesting target"+ Just r -> return $ msgContent r+ clFilter = not . (`elem` ["__channel__", "__job__"])
+ Network/Mom/Stompl/Patterns/Desk.hs view
@@ -0,0 +1,179 @@+{-# Language BangPatterns #-}+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Stompl/Patterns/Desk.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- Stability : experimental+-- Portability: portable+--+-- A Desk is a server that supplies information+-- about providers. A client requests providers+-- for a specific job (service, task or topic)+-- and the desk will reply with a list of queue names+-- of providers of the enquired job.+-- +-- The desk is not statically configured,+-- but uses a registry to which providers connect.+-- Providers that cease to work can disconnect or,+-- if heartbeats are required, will be removed from the+-- list of available providers internally when no more heartbeats+-- are sent. This way, the information provided by a desk is +-- always up-to-date.+--+-- Desk balances providers, /i.e./ providers rotate in a list+-- from which always the first /n/ providers are handed out+-- to requesting consumers (where /n/ corresponds to the number of +-- providers requested by the consumer.) +--+-- Since providers are managed dynamically, +-- the result of two consecutive calls is probably not the same.+-- Desk is thus not idempotent in the strict sense.+-- But, since the call itself does only cause +-- a change of the order of providers+-- (and since it should be irrelevant for the consumer+-- which provider is actually used),+-- two consecutive calls will have the same effect+-- -- if not all providers disconnect between the two calls.+--+-- Internally, the Desk protocol uses the following headers:+--+-- * __jobs__: Comma-separated list of providers;+--+-- * __redundancy__: Requested number of providers.+-------------------------------------------------------------------------------+module Network.Mom.Stompl.Patterns.Desk (+ withDesk, requestProvider)+where++ import Registry+ import Types+ import Network.Mom.Stompl.Patterns.Basic + import Network.Mom.Stompl.Client.Queue + import Data.Char (isDigit)+ import Data.List (intercalate)+ import Data.List.Split (endBy)+ import Codec.MIME.Type (nullType)+ import Prelude hiding (catch)+ import Control.Exception (throwIO, catches)+ import Control.Monad (forever)+ import Control.Applicative ((<$>))++ -----------------------------------------------------------------------+ -- | Creates a desk with the lifetime of the application-defined+ -- action:+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the desk, used for error handling;+ --+ -- * 'QName': Registration queue -- this queue is used+ -- by providers to connect to the registry,+ -- it is not used for consumer requests;+ --+ -- * (Int, Int): Heartbeat range of the 'Registry' + -- (see 'withRegistry' for details);+ --+ -- * 'OnError': Error handling;+ --+ -- * 'QName': Request queue -- this queue is used+ -- by consumers to request information about+ -- available providers;+ --+ -- * IO r: Action that defines the lifetime of the desk;+ -- the result is also the result of /withDesk/.+ -----------------------------------------------------------------------+ withDesk :: Con -> String -> QName -> (Int, Int) -> OnError -> + QName -> IO r -> IO r+ withDesk c n qn (mn,mx) onErr rq action =+ withRegistry c n qn (mn,mx) onErr $ \reg -> + withPair c n (rq, [], [], ignorebody)+ ("unknown", [], [], nobody) $ \(r,w) -> + withThread (doDesk reg r w) action+ where doDesk reg r w = + forever $ catches (do+ m <- readQ r+ j <- getJobName m+ q <- getChannel m+ i <- getRedundancy m+ ps <- (intercalate "," . map prvQ) <$> getProvider reg j i+ let hs = case length ps of+ 0 -> [("__sc__", show NotFound),+ ("__jobs__", ""),+ ("__redundancy__", "0")]+ x -> [("__sc__", show OK),+ ("__redundancy__", show x),+ ("__jobs__", ps)]+ in writeAdHoc w q nullType hs ())+ (ignoreHandler n onErr)++ -----------------------------------------------------------------------+ -- | Function used by consumer to request provider information + -- from a desk:+ --+ -- * 'ClientA' () (): The request to the desk is sent+ -- through a client of type () ().+ -- This client must be created by + -- the application beforehand+ -- (/e.g./: the client could be created+ -- once during initialisation+ -- and then be used repeatedly+ -- to obtain or update information + -- on providers according to the+ -- application needs);+ --+ -- * Int: Timeout in microseconds;+ --+ -- * 'JobName': Name of the job for which the consumer+ -- needs providers;+ --+ -- * Int: Number of providers needed by the consumer.+ -- This can be used for redundancy:+ -- if one provider fails, + -- the consumer passes to the next.+ -- Be aware, however, that the information, + -- at the point in time, when a provider fails, + -- may already be outdated.+ -- Therefore, the redundant providers should be used+ -- immediately and, when the main provider fails later,+ -- the information should be updated by requesting+ -- new providers from the desk.+ --+ -- The result is a tuple of ('StatusCode', ['QName']).+ -- If the 'StatusCode' is not 'OK', + -- the list of 'QName' will be empty;+ -- otherwise, it will contain at least one provider+ -- and maximum /n/ providers (where /n/ is the number of providers+ -- requested). If fewer providers than requested are available,+ -- the list will contain less than /n/ providers. + -- But note that this, as long as there is at least one provider,+ -- does not count as an error, /i.e./ the 'StatusCode' is still 'OK'.+ -----------------------------------------------------------------------+ requestProvider :: ClientA () () -> Int -> + JobName -> Int -> IO (StatusCode, [QName])+ requestProvider c tmo jn r = do+ mbR <- request c tmo nullType [("__job__", jn), + ("__redundancy__", show r)] ()+ case mbR of+ Nothing -> return (Timeout, [])+ Just m -> do+ eiSC <- getSC m+ case eiSC of+ Left sc -> throwIO $ BadStatusCodeX sc+ Right OK -> do qs <- getJobs m+ return (OK, qs)+ Right sc -> return (sc, [])+ + getJobs :: Message i -> IO [QName]+ getJobs m = do+ x <- getHeader "__jobs__"+ "no jobs in header" m+ return $ endBy "," x++ getRedundancy :: Message i -> IO Int+ getRedundancy m = do+ x <- getHeader "__redundancy__" + "No redundancy level in headers" m+ if all isDigit x then return $ read x + else throwIO $ HeaderX "__redundancy" $+ "Redundancy level not numberic: " ++ x
+ Network/Mom/Stompl/Patterns/Registry.hs view
@@ -0,0 +1,609 @@+{-# Language BangPatterns #-}+module Registry +where++ import Types+ import Network.Mom.Stompl.Client.Queue + import System.Timeout+ import Data.Time+ import Data.Char (isDigit, toUpper)+ import Data.List (nub)+ import Data.Maybe (fromMaybe)+ import Data.Map (Map)+ import qualified Data.Map as M+ import Data.Sequence (Seq, (|>), (<|), ViewL(..))+ import qualified Data.Sequence as S+ import Data.Foldable (toList)+ import Codec.MIME.Type (nullType)+ import Prelude hiding (catch)+ import Control.Exception (throwIO, catches)+ import Control.Concurrent + import Control.Monad (forever)+ import Control.Applicative ((<$>))++ -----------------------------------------------------------------------+ -- | JobType: Service, Task or Topic+ -----------------------------------------------------------------------+ data JobType = Service | Task | Topic+ deriving (Eq, Show)++ -----------------------------------------------------------------------+ -- | Safe read method for JobType+ -----------------------------------------------------------------------+ readJobType :: String -> Maybe JobType+ readJobType s = + case map toUpper s of+ "SERVICE" -> Just Service+ "TASK" -> Just Task+ "TOPIC" -> Just Topic+ _ -> Nothing++ ------------------------------------------------------------------------+ -- | A helper that shall ease the use of the registers.+ -- A registry to which a call wants to connect is described as+ -- + -- * The 'QName' through which the registry receives requests;+ --+ -- * The 'Timeout' in microseconds, /i.e./ the time the caller+ -- will wait before the request fails;+ --+ -- * A triple of heartbeat specifications:+ -- the /best/ value, /i.e./ + -- the rate at which the caller + -- prefers to send heartbeats,+ -- the /minimum/ rate at which the caller + -- can accept to send heartbeats,+ -- the /maximum/ rate at which the caller + -- can accept to send heartbeats.+ -- Note that all these values are in milliseconds!+ ------------------------------------------------------------------------+ type RegistryDesc = (QName, Int, (Int, Int, Int))++ ------------------------------------------------------------------------+ -- | Connect to a registry:+ -- The caller registers itself at the registry.+ -- The owner of the registry will then+ -- use the caller depending on its purpose.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * 'JobName': The name of the job provided by the caller;+ --+ -- * 'JobType': The type of the job provided by the caller;+ --+ -- * 'QName': The registry's registration queue;+ --+ -- * 'QName': The queue to register;+ -- this is the queue the register will actually+ -- use (for forwarding requests or whatever+ -- it does in this specific case).+ -- The registry, internally,+ -- uses 'JobName' together with this queue+ -- as a /key/ to identify the provider. + --+ -- * Int: Timeout in microseconds;+ --+ -- * Int: Preferred heartbeat in milliseconds+ -- (0 for no heartbeats).+ --+ -- The function returns a tuple of 'StatusCode' + -- and the heartbeat proposed by the registry+ -- (which may differ from the preferred heartbeat of the caller).+ -- Whenever the 'StatusCode' is not 'OK', + -- the heartbeat is 0.+ -- If the 'JobName' is null, the 'StatusCode' will be 'BadRequest'.+ -- If the timeout expires, register throws 'TimeoutX'.+ ------------------------------------------------------------------------+ register :: Con -> JobName -> JobType -> + QName -> QName -> + Int -> Int -> IO (StatusCode, Int)+ register c j t o i to me | null j = return (BadRequest,0)+ | otherwise =+ let i' = o ++ "/" ++ j ++ "/" ++ i+ hs = [("__type__", "register"),+ ("__job-type__", show t),+ ("__job__", j),+ ("__queue__", i),+ ("__hb__", show me),+ ("__channel__", i')]+ in withWriter c "RegistryW" o [] [] nobody $ \w -> + withReader c "RegistryR" i' [] [] ignorebody $ \r -> do+ writeQ w nullType hs ()+ mbF <- timeout to $ readQ r + case mbF of+ Nothing -> throwIO $ TimeoutX+ "No response from registry"+ Just m -> do eiS <- getSC m+ case eiS of + Left s -> throwIO $ BadStatusCodeX s+ Right OK -> do h <- getHB m+ return (OK, h)+ Right sc -> return (sc, 0)++ ------------------------------------------------------------------------+ -- | Disconnect from a registry:+ -- The caller disconnects from a registry+ -- to which it has registered before.+ -- For the case that the registry is not receiving heartbeats+ -- from the caller,+ -- it is essential to unregister, when+ -- the service is no longer provided.+ -- Otherwise, the registry has no way to know+ -- that it should not send requests to this provider anymore.+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * 'JobName': The 'JobName' to unregister;+ --+ -- * 'QName': The registry's registration queue ;+ --+ -- * 'QName': The queue to unregister;+ --+ -- * Int: The timeout in microseconds.+ --+ -- The function returns a 'StatusCode'. + -- If 'JobName' is null, the 'StatusCode' will be 'BadRequest'.+ -- If the timeout expires, the function will throw 'TimeoutX'.+ ------------------------------------------------------------------------+ unRegister :: Con -> JobName -> + QName -> QName -> + Int -> IO StatusCode+ unRegister c j o i tmo | null j = return BadRequest+ | otherwise = + let i' = o ++ "/" ++ j ++ i+ hs = [("__type__", "unreg"),+ ("__job__", j),+ ("__queue__", i),+ ("__channel__", i')]+ in withWriter c "RegistryW" o [] [] nobody $ \w -> + withReader c "RegistryR" i' [] [] ignorebody $ \r -> do+ writeQ w nullType hs ()+ mbF <- timeout tmo $ readQ r + case mbF of+ Nothing -> throwIO $ TimeoutX "No response from register"+ Just m -> do eiS <- getSC m+ case eiS of+ Left s -> throwIO $ BadStatusCodeX s+ Right sc -> return sc++ ------------------------------------------------------------------------+ -- | Send heartbeats:+ --+ -- * 'MVar' 'HB': An MVar of type 'HB', this MVar will be used+ -- to keep track of when the heartbeat has actually to+ -- be sent.+ --+ -- * 'Writer' (): The writer through which to send the heartbeat;+ -- The queue name of the writer is the registration queue+ -- of the registry; note that its type is ():+ -- heartbeats are empty messages.+ --+ -- * 'JobName': The 'JobName' for which to send heartbeats;+ --+ -- * 'QName': The queue for which to send heartbeats.+ ------------------------------------------------------------------------+ heartbeat :: MVar HB -> Writer () -> JobName -> QName -> IO ()+ heartbeat m w j q | null q = return ()+ | otherwise = + let hs = [("__type__", "hb"),+ ("__job__", j),+ ("__queue__", q)]+ in do now <- getCurrentTime + modifyMVar_ m (go now hs)+ where go now hs hb@(HB me nxt)+ | me > 0 && nxt < now = do writeQ w nullType hs ()+ return hb{hbMeNext = timeAdd now me}+ | otherwise = return hb++ ------------------------------------------------------------------------+ -- | A provider is an opaque data type;+ -- most of its attributes are used only internally by the registry.+ -- Interesting for user applications, however, is the queue+ -- that identifies the provider.+ ------------------------------------------------------------------------+ data Provider = Provider {+ -- | Queue through which the job is provided + prvQ :: QName,+ prvHb :: Int,+ prvNxt :: UTCTime+ }+ deriving Show++ ------------------------------------------------------------------------+ -- Two providers are identical if they have the same queue name+ ------------------------------------------------------------------------+ instance Eq Provider where+ x == y = prvQ x == prvQ y++ ------------------------------------------------------------------------+ -- Add provider to seq or, if already in, + -- update according to the values of the new node.+ ------------------------------------------------------------------------+ updOrAddProv :: Bool -> (Provider -> Provider) -> Provider -> + Seq Provider -> Seq Provider+ updOrAddProv add upd p s = + case S.viewl s of+ S.EmptyL -> if add then S.singleton p else S.empty+ x :< ss -> if prvQ x == prvQ p + then upd x <| ss+ else x <| updOrAddProv add upd p ss++ ------------------------------------------------------------------------+ -- Remove one provider from the seq+ ------------------------------------------------------------------------+ remProv :: QName -> Seq Provider -> Seq Provider+ remProv q s =+ case S.viewl s of+ S.EmptyL -> S.empty+ x :< ss -> if prvQ x == q then ss+ else x <| remProv q ss++ ------------------------------------------------------------------------+ -- Get head of seq and add to end of sequence;+ -- remove all "dead" nodes on the way+ ------------------------------------------------------------------------+ getHeads :: UTCTime -> Seq Provider -> ([Provider], Seq Provider)+ getHeads now s = + case S.viewl s of+ S.EmptyL -> ([], S.empty)+ x :< ss -> if prvHb x > 0 &&+ prvNxt x < now then getHeads now ss+ else ([x], ss |> x)++ ------------------------------------------------------------------------+ -- Job: 'JobType' plus 'Sequence' of 'Provider's+ ------------------------------------------------------------------------+ data JobNode = JobNode {+ jobType :: JobType,+ jobProvs :: Seq Provider+ }++ ------------------------------------------------------------------------+ -- The inner heart of the registry: + -- a 'Map' of 'JobName', 'JobNode'+ ------------------------------------------------------------------------+ data Reg = Reg {+ regName :: String,+ regWork :: Map JobName JobNode+ }++ ------------------------------------------------------------------------+ -- | Registry: An opaque data type+ ------------------------------------------------------------------------+ data Registry = Registry {+ regM :: MVar Reg+ }++ ------------------------------------------------------------------------+ -- Use registry (with return value)+ ------------------------------------------------------------------------+ useRegistry :: Registry -> (Reg -> IO (Reg, r)) -> IO r+ useRegistry r = modifyMVar (regM r)++ ------------------------------------------------------------------------+ -- Use registry (without return value)+ ------------------------------------------------------------------------+ useRegistry_ :: Registry -> (Reg -> IO Reg) -> IO ()+ useRegistry_ r = modifyMVar_ (regM r)++ ------------------------------------------------------------------------+ -- Add provider to job+ ------------------------------------------------------------------------+ insertR :: Registry -> JobName -> JobType -> QName -> Int -> IO ()+ insertR r jn w qn i = + useRegistry_ r $ \reg -> do now <- getCurrentTime+ return reg{regWork = ins now $ regWork reg}+ where ins now m = + let j = fromMaybe (JobNode w S.empty) $ M.lookup jn m+ p = Provider qn i $ nextHB now True i+ ps = updOrAddProv True (upd p) p $ jobProvs j+ in M.insert jn j{jobProvs = ps} m+ upd n _ = n++ ------------------------------------------------------------------------+ -- Update heartbeat of provider + ------------------------------------------------------------------------+ updR :: Registry -> JobName -> QName -> IO ()+ updR r jn qn = + useRegistry_ r $ \reg -> do now <- getCurrentTime+ return reg{regWork = ins now $ regWork reg}+ where ins now m = + case M.lookup jn m of+ Nothing -> m+ Just j -> let p = Provider qn 0 now+ ps = updOrAddProv False (upd now) p + (jobProvs j)+ in M.insert jn j{jobProvs = ps} m+ upd now o = o{prvNxt = nextHB now True $ tolerance * prvHb o}+ + ------------------------------------------------------------------------+ -- Remove 'Provider' from the job+ ------------------------------------------------------------------------+ removeR :: Registry -> JobName -> QName -> IO ()+ removeR r jn qn = + useRegistry_ r $ \reg -> return reg{regWork = ins $ regWork reg}+ where ins m = + case M.lookup jn m of+ Nothing -> m+ Just j -> + let ps = remProv qn $ jobProvs j+ in if S.null ps then M.delete jn m+ else M.insert jn j{jobProvs = ps} m++ ------------------------------------------------------------------------+ -- | Map action to 'Provider's of job 'JobName';+ -- mapping means different things for:+ --+ -- * Serice, Task: action is applied to the first+ -- active provider of a list of providers+ -- and this provider+ -- is then sent to the back of the list,+ -- hence, implementing a balancer.+ --+ -- * Topic: action is applied to all providers,+ -- hence, implementing a publisher.+ --+ -- Parameters:+ --+ -- * 'Registry': The registry to use;+ --+ -- * 'JobName': The job to which to apply the action;+ --+ -- * ('Provider' -> IO ()): The action to apply.+ --+ -- The function returns False iff the requested job is not available+ -- and True otherwise. (Note that a job without providers is removed;+ -- when the function returns True, the job, thus, + -- was applied at least once.+ ------------------------------------------------------------------------+ mapR :: Registry -> JobName -> (Provider -> IO ()) -> IO Bool+ mapR r jn f = + useRegistry r $ \reg -> getCurrentTime >>= \now ->+ ins now (regWork reg) >>= \(js,t) ->+ return (reg{regWork = js},t)+ where ins now m = + case M.lookup jn m of+ Nothing -> return (m, False)+ Just j -> + let (xs, ps) = if jobType j `elem` [Service, Task]+ then getHeads now $ jobProvs j+ else (toList $ jobProvs j, + jobProvs j)+ in mapM_ f xs >> + return (M.insert jn j{jobProvs = ps} m, True)++ ------------------------------------------------------------------------+ -- | Map function of type + --+ -- > 'Provider' -> 'Provider'+ --+ -- to all 'Provider's of job 'JobName'+ -- (independent of 'JobType')+ ------------------------------------------------------------------------+ mapAllR :: Registry -> JobName -> (Provider -> Provider) -> IO ()+ mapAllR r jn f = + useRegistry_ r $ \reg -> ins (regWork reg) >>= \m ->+ return reg{regWork = m}+ where ins m = + case M.lookup jn m of+ Nothing -> return m+ Just j -> return (M.insert jn j{jobProvs = go $ jobProvs j} m)+ go s = case S.viewl s of+ S.EmptyL -> S.empty+ x :< ss -> f x <| go ss+ + ------------------------------------------------------------------------+ -- | Retrieves /n/ 'Provider's of a certain job;+ -- getProvider works, for all 'JobType's+ -- according to the work balancer logic, /i.e./:+ -- it returns the first n providers of the list for this job+ -- and moves them to the end of the list.+ -- 'getProvider' is used, for instance, in the Desk pattern. + --+ -- * 'Registry': The registry in use;+ --+ -- * 'JobName': The job for which the caller needs a provider;+ --+ -- * Int: The number /n/ of providers to retrieve; + -- if less than /n/ providers are available for this job,+ -- all available providers will be returned,+ -- but no error event is created.+ ------------------------------------------------------------------------+ getProvider :: Registry -> JobName -> Int -> IO [Provider]+ getProvider r jn n = + useRegistry r $ \reg -> do now <- getCurrentTime+ let (x,m) = ins now $ regWork reg+ return (reg{regWork = m}, x)+ where ins now m = case M.lookup jn m of+ Nothing -> ([], m)+ Just j -> + let (x,ps) = go now (jobProvs j) n+ in (x, M.insert jn j{jobProvs = ps} m)+ go now ps i | i <= 0 = ([],ps)+ | otherwise = let (!x ,ps1) = getHeads now ps+ (!x',ps2) = go now ps1 (i-1)+ in (nub (x++x'), ps2)++ ------------------------------------------------------------------------+ -- | This function shows all jobs with all their providers+ -- in a registry; the function is intended for debugging only.+ ------------------------------------------------------------------------+ showRegistry :: Registry -> IO ()+ showRegistry r = + useRegistry_ r $ \reg -> let l = map fst $ M.toList (regWork reg)+ p = map (getProvs reg) l+ lp = zip l p+ in print lp >> return reg+ where getProvs reg jn = case M.lookup jn $ regWork reg of+ Nothing -> []+ Just ps -> toList $ jobProvs ps++ ------------------------------------------------------------------------+ -- | A registry is used through a function + -- that, internally, creates a registry+ -- and defines its lifetime in terms of the scope of an action+ -- passed in to the function:+ --+ -- * 'Con': Connection to a Stomp broker;+ --+ -- * String: Name of the registry used for error handling;+ --+ -- * 'QName': Name of the registration queue.+ -- It is this queue to which 'register'+ -- sends a registration request;+ --+ -- * (Int, Int): Minimal and maximal accepted heartbeat interval;+ --+ -- * 'OnError': Error handler;+ --+ -- * ('Registry' -> IO r): The action that defines + -- the registry's lifetime;+ -- the result of this action, /r/, + -- is also the result of /withRegistry/.+ ------------------------------------------------------------------------+ withRegistry :: Con -> String -> QName -> (Int, Int)+ -> OnError -> (Registry -> IO r) -> IO r+ withRegistry c n rq (mn, mx) onErr action = + -- always start the reader in the main thread -------------+ -- for if started in the background thread -------------+ -- the action may send a message -------------+ -- without the reader having subscribed to its queue ------+ withReader c (n ++ "Reader") rq [] [] ignorebody $ \r -> do+ let nm = n ++ "Registry"+ reg <- Registry <$> newMVar (Reg nm M.empty)+ withThread (startReg reg r nm) (action reg)+ where startReg reg r nm = + withWriter c (n ++ "Writer") "unknown" [] [] nobody $ \w -> + forever $ catches + (do m <- readQ r+ t <- getMType m+ case t of+ "register" -> handleRegister reg m w (mn,mx)+ "unreg" -> handleUnRegister reg m w+ "hb" -> handleHeartbeat reg m+ x -> throwIO $ HeaderX "__type__" $+ "Unknown type: " ++ x)+ (ignoreHandler nm onErr)++ ------------------------------------------------------------------------+ -- Handle registration request+ ------------------------------------------------------------------------+ handleRegister :: Registry -> Message m -> Writer () -> (Int, Int) -> IO ()+ handleRegister r m w (mn,mx) = do+ (j,q) <- getJobQueue m+ ch <- getChannel m+ t <- getJobType m + hb <- getHB m + let h | hb < mn || hb > mx = if (mn - hb) < (hb - mx) then mn else mx+ | otherwise = hb+ insertR r j t q h+ let hs = [("__sc__", show OK),+ ("__hb__", show h)]+ writeAdHoc w ch nullType hs ()++ ------------------------------------------------------------------------+ -- Handle unRegister request+ ------------------------------------------------------------------------+ handleUnRegister :: Registry -> Message m -> Writer () -> IO ()+ handleUnRegister r m w = do+ (j,q) <- getJobQueue m+ ch <- getChannel m+ removeR r j q + let hs=[("__sc__", show OK)]+ writeAdHoc w ch nullType hs ()++ ------------------------------------------------------------------------+ -- Handle heartbeat+ ------------------------------------------------------------------------+ handleHeartbeat :: Registry -> Message m -> IO ()+ handleHeartbeat r m = do+ (j,q) <- getJobQueue m+ updR r j q+ -- print $ msgHdrs m -- test++ ------------------------------------------------------------------------+ -- | Get JobQueue+ -- (and throw an exception if at least + -- one of the headers does not exist)+ ------------------------------------------------------------------------+ getJobQueue :: Message m -> IO (String, String)+ getJobQueue m = getJobName m >>= \j -> getQueue m >>= \q -> return (j,q)++ ------------------------------------------------------------------------+ -- | Get Message Type from headers+ -- (and throw an exception if the header does not exist)+ ------------------------------------------------------------------------+ getMType :: Message m -> IO String+ getMType = getHeader "__type__" "No message type in headers"++ ------------------------------------------------------------------------+ -- | Get Job name from headers+ -- (and throw an exception if the header does not exist)+ ------------------------------------------------------------------------+ getJobName :: Message m -> IO String+ getJobName = getHeader "__job__" "No job name in headers" ++ ------------------------------------------------------------------------+ -- | Get Reply queue (channel) from headers+ -- (and throw an exception if the header does not exist)+ ------------------------------------------------------------------------+ getChannel :: Message m -> IO String+ getChannel = getHeader "__channel__" "No response q in headers" ++ ------------------------------------------------------------------------+ -- | Get Queue name from headers+ -- (and throw an exception if the header does not exist)+ ------------------------------------------------------------------------+ getQueue :: Message m -> IO String+ getQueue = getHeader "__queue__" "No queue q in headers" ++ ------------------------------------------------------------------------+ -- | Get Job type from headers+ -- (and throw an exception if the header does not exist+ -- or contains an invalid value)+ ------------------------------------------------------------------------+ getJobType :: Message m -> IO JobType+ getJobType m = + getHeader "__job-type__" "No job type in headers" m >>= \x ->+ case readJobType x of+ Nothing -> throwIO $ HeaderX "__job-type__" $+ "unknown type: " ++ x + Just t -> return t ++ ------------------------------------------------------------------------+ -- | Get Heartbeat specification from headers+ -- (and throw an exception if the header does not exist+ -- or if its value is not numeric)+ ------------------------------------------------------------------------+ getHB :: Message m -> IO Int+ getHB m = + case lookup "__hb__" $ msgHdrs m of+ Nothing -> return 0 + Just v -> if all isDigit v + then return $ read v+ else throwIO $ HeaderX "__hb__" $+ "heartbeat not numeric: " ++ show v++ ------------------------------------------------------------------------+ -- | Get Status code from headers+ -- (and throw an exception if the header does not exist)+ ------------------------------------------------------------------------+ getSC :: Message m -> IO (Either String StatusCode)+ getSC m = readStatusCode <$> getHeader "__sc__"+ "No status code in message" m+ + ------------------------------------------------------------------------+ -- | Get Generic function to retrieve a header value+ -- (and throw an exception if the header does not exist):+ --+ -- * String: Key of the wanted header+ --+ -- * String: Error message in case there is no such header+ --+ -- * 'Message' m: The message whose headers we want to search+ ------------------------------------------------------------------------+ getHeader :: String -> String -> Message m -> IO String+ getHeader h e m = case lookup h $ msgHdrs m of+ Nothing -> throwIO $ HeaderX h e+ Just v -> return v+
+ Network/Mom/Stompl/Patterns/Types.hs view
@@ -0,0 +1,196 @@+{-# Language DeriveDataTypeable #-}+module Types+where++ import Network.Mom.Stompl.Client.Queue + import Data.Time.Clock+ import qualified Data.ByteString.Char8 as B+ import Data.Typeable (Typeable)+ import Prelude hiding (catch)+ import Control.Exception (throwIO, + Exception, SomeException, Handler(..),+ AsyncException(..),+ bracket, finally)+ import Control.Concurrent + import Control.Monad (void)+++ ------------------------------------------------------------------------+ -- | Status code to communicate the state of a request+ -- between two applications. + -- The wire format is inspired by HTTP status codes:+ --+ -- * OK (200): Everything is fine+ --+ -- * BadRequest (400): Syntax error in the request message+ --+ -- * Forbidden (403): Not used+ --+ -- * NotFound (404): For the requested job no provider is available+ --+ -- * Timeout (408): Timeout expired+ ------------------------------------------------------------------------+ data StatusCode = OK | BadRequest | Forbidden | NotFound | Timeout+ deriving (Eq)++ instance Show StatusCode where+ show OK = "200"+ show BadRequest = "400"+ show Forbidden = "403"+ show NotFound = "404"+ show Timeout = "408"++ instance Read StatusCode where+ readsPrec _ r = case r of+ "200" -> [(OK,"")]+ "400" -> [(BadRequest,"")]+ "403" -> [(Forbidden,"")]+ "404" -> [(NotFound,"")]+ "408" -> [(Timeout,"")]+ _ -> undefined+ + ------------------------------------------------------------------------+ -- | Safe StatusCode parser + -- ('StatusCode' is instance of 'Read',+ -- but 'read' would cause an error on an invalid StatusCode)+ ------------------------------------------------------------------------+ readStatusCode :: String -> Either String StatusCode+ readStatusCode s = case s of+ "200" -> Right OK+ "400" -> Right BadRequest+ "403" -> Right Forbidden+ "404" -> Right NotFound+ "408" -> Right Timeout+ _ -> Left $ "Unknown status code: " ++ s++ -- | Name of a service, task or topic+ type JobName = String++ -- | Name of a Stomp queue+ type QName = String++ -- | OutBound converter for messages of type ()+ nobody :: OutBound ()+ nobody _ = return B.empty++ -- | InBound converter for messages of type ()+ ignorebody :: InBound ()+ ignorebody _ _ _ _ = return ()++ -- | OutBound converter for messages of type 'B.ByteString'+ bytesOut :: OutBound B.ByteString+ bytesOut = return++ -- | InBound converter for messages of type 'B.ByteString'+ bytesIn :: InBound B.ByteString+ bytesIn _ _ _ = return++ -----------------------------------------------------------------------+ -- Error and Exception+ -----------------------------------------------------------------------+ reThrowHandler :: String -> OnError -> [Handler a] + reThrowHandler s onErr = [+ Handler (\e -> throwIO (e::AsyncException)),+ Handler (\e -> onErr e s >> throwIO e)]++ ignoreHandler :: String -> OnError -> [Handler ()]+ ignoreHandler s onErr = [+ Handler (\e -> throwIO (e::AsyncException)),+ Handler (\e -> onErr e s)]++ killAndWait :: MVar () -> ThreadId -> IO ()+ killAndWait m tid = do killThread tid+ void $ takeMVar m++ withThread :: IO () -> IO r -> IO r+ withThread thrd action = do+ stp <- newEmptyMVar+ bracket (forkIO $ finally thrd (putMVar stp ()))+ (killAndWait stp)+ (\_ -> action)++ ------------------------------------------------------------------------+ -- | Patterns Exception+ ------------------------------------------------------------------------+ data PatternsException = + -- | Timout expired+ TimeoutX String+ -- | Invalid status code + | BadStatusCodeX String+ -- | Status code other than OK+ | NotOKX StatusCode String+ -- | Error on Header identified by the first String+ | HeaderX String String+ -- | Thrown on missing heartbeat + -- (after tolerance of 10 missing heartbeats)+ | MissingHbX String+ -- | Heartbeat proposed by registry+ -- out of acceptable range+ | UnacceptableHbX Int+ -- | No provider for the requested job available+ | NoProviderX String+ -- | Application-defined exception+ | AppX String+ deriving (Show, Read, Typeable, Eq)++ instance Exception PatternsException++ ------------------------------------------------------------------------+ -- | Error Handler:+ --+ -- * 'SomeException': Exception that led the invocation;+ --+ -- * String: Name of the entity (client name, server name, /etc./)+ ------------------------------------------------------------------------+ type OnError = SomeException -> String -> IO ()++ ------------------------------------------------------------------------+ -- | Heartbeat controller type+ ------------------------------------------------------------------------+ data HB = HB {+ hbMe :: Int,+ hbMeNext :: UTCTime+ }+ + ------------------------------------------------------------------------+ -- | Create a heartbeat controller;+ -- receives the heartbeat in milliseconds.+ ------------------------------------------------------------------------+ mkHB :: Int -> IO HB+ mkHB me = do+ now <- getCurrentTime+ return HB {+ hbMe = me,+ hbMeNext = timeAdd now me}++ -- HB tolerance + tolerance :: Int+ tolerance = 10++ ------------------------------------------------------------------------+ -- Comput the next hearbeat from current one+ ------------------------------------------------------------------------+ nextHB :: UTCTime -> Bool -> Int -> UTCTime+ nextHB now t p = let tol = if t then tolerance else 1+ in timeAdd now $ tol * p++ -----------------------------------------------------------------------+ -- Adding a period to a point in time+ -----------------------------------------------------------------------+ timeAdd :: UTCTime -> Int -> UTCTime+ timeAdd t p = ms2nominal p `addUTCTime` t++ -----------------------------------------------------------------------+ -- Convert milliseconds to seconds+ -----------------------------------------------------------------------+ ms2nominal :: Int -> NominalDiffTime+ ms2nominal m = fromIntegral m / (1000::NominalDiffTime)++ -----------------------------------------------------------------------+ -- Convert 'NominalDiffTime' to microseconds+ -----------------------------------------------------------------------+ nominal2us :: NominalDiffTime -> Int+ nominal2us m = round (fact * realToFrac m :: Double)+ where fact = 10.0^(6::Int)++
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ license/lgpl-3.0.txt view
@@ -0,0 +1,165 @@+ GNU LESSER GENERAL PUBLIC LICENSE+ Version 3, 29 June 2007++ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>+ Everyone is permitted to copy and distribute verbatim copies+ of this license document, but changing it is not allowed.+++ This version of the GNU Lesser General Public License incorporates+the terms and conditions of version 3 of the GNU General Public+License, supplemented by the additional permissions listed below.++ 0. Additional Definitions.++ As used herein, "this License" refers to version 3 of the GNU Lesser+General Public License, and the "GNU GPL" refers to version 3 of the GNU+General Public License.++ "The Library" refers to a covered work governed by this License,+other than an Application or a Combined Work as defined below.++ An "Application" is any work that makes use of an interface provided+by the Library, but which is not otherwise based on the Library.+Defining a subclass of a class defined by the Library is deemed a mode+of using an interface provided by the Library.++ A "Combined Work" is a work produced by combining or linking an+Application with the Library. The particular version of the Library+with which the Combined Work was made is also called the "Linked+Version".++ The "Minimal Corresponding Source" for a Combined Work means the+Corresponding Source for the Combined Work, excluding any source code+for portions of the Combined Work that, considered in isolation, are+based on the Application, and not on the Linked Version.++ The "Corresponding Application Code" for a Combined Work means the+object code and/or source code for the Application, including any data+and utility programs needed for reproducing the Combined Work from the+Application, but excluding the System Libraries of the Combined Work.++ 1. Exception to Section 3 of the GNU GPL.++ You may convey a covered work under sections 3 and 4 of this License+without being bound by section 3 of the GNU GPL.++ 2. Conveying Modified Versions.++ If you modify a copy of the Library, and, in your modifications, a+facility refers to a function or data to be supplied by an Application+that uses the facility (other than as an argument passed when the+facility is invoked), then you may convey a copy of the modified+version:++ a) under this License, provided that you make a good faith effort to+ ensure that, in the event an Application does not supply the+ function or data, the facility still operates, and performs+ whatever part of its purpose remains meaningful, or++ b) under the GNU GPL, with none of the additional permissions of+ this License applicable to that copy.++ 3. Object Code Incorporating Material from Library Header Files.++ The object code form of an Application may incorporate material from+a header file that is part of the Library. You may convey such object+code under terms of your choice, provided that, if the incorporated+material is not limited to numerical parameters, data structure+layouts and accessors, or small macros, inline functions and templates+(ten or fewer lines in length), you do both of the following:++ a) Give prominent notice with each copy of the object code that the+ Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the object code with a copy of the GNU GPL and this license+ document.++ 4. Combined Works.++ You may convey a Combined Work under terms of your choice that,+taken together, effectively do not restrict modification of the+portions of the Library contained in the Combined Work and reverse+engineering for debugging such modifications, if you also do each of+the following:++ a) Give prominent notice with each copy of the Combined Work that+ the Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the Combined Work with a copy of the GNU GPL and this license+ document.++ c) For a Combined Work that displays copyright notices during+ execution, include the copyright notice for the Library among+ these notices, as well as a reference directing the user to the+ copies of the GNU GPL and this license document.++ d) Do one of the following:++ 0) Convey the Minimal Corresponding Source under the terms of this+ License, and the Corresponding Application Code in a form+ suitable for, and under terms that permit, the user to+ recombine or relink the Application with a modified version of+ the Linked Version to produce a modified Combined Work, in the+ manner specified by section 6 of the GNU GPL for conveying+ Corresponding Source.++ 1) Use a suitable shared library mechanism for linking with the+ Library. A suitable mechanism is one that (a) uses at run time+ a copy of the Library already present on the user's computer+ system, and (b) will operate properly with a modified version+ of the Library that is interface-compatible with the Linked+ Version.++ e) Provide Installation Information, but only if you would otherwise+ be required to provide such information under section 6 of the+ GNU GPL, and only to the extent that such information is+ necessary to install and execute a modified version of the+ Combined Work produced by recombining or relinking the+ Application with a modified version of the Linked Version. (If+ you use option 4d0, the Installation Information must accompany+ the Minimal Corresponding Source and Corresponding Application+ Code. If you use option 4d1, you must provide the Installation+ Information in the manner specified by section 6 of the GNU GPL+ for conveying Corresponding Source.)++ 5. Combined Libraries.++ You may place library facilities that are a work based on the+Library side by side in a single library together with other library+facilities that are not Applications and are not covered by this+License, and convey such a combined library under terms of your+choice, if you do both of the following:++ a) Accompany the combined library with a copy of the same work based+ on the Library, uncombined with any other library facilities,+ conveyed under the terms of this License.++ b) Give prominent notice with the combined library that part of it+ is a work based on the Library, and explaining where to find the+ accompanying uncombined form of the same work.++ 6. Revised Versions of the GNU Lesser General Public License.++ The Free Software Foundation may publish revised and/or new versions+of the GNU Lesser General Public License from time to time. Such new+versions will be similar in spirit to the present version, but may+differ in detail to address new problems or concerns.++ Each version is given a distinguishing version number. If the+Library as you received it specifies that a certain numbered version+of the GNU Lesser General Public License "or any later version"+applies to it, you have the option of following the terms and+conditions either of that published version or of any later version+published by the Free Software Foundation. If the Library as you+received it does not specify a version number of the GNU Lesser+General Public License, you may choose any version of the GNU Lesser+General Public License ever published by the Free Software Foundation.++ If the Library as you received it specifies that a proxy can decide+whether future versions of the GNU Lesser General Public License shall+apply, that proxy's public statement of acceptance of any version is+permanent authorization for you to choose that version for the+Library.
+ stomp-patterns.cabal view
@@ -0,0 +1,109 @@+Name: stomp-patterns+Version: 0.0.1+Cabal-Version: >= 1.8+Copyright: Copyright (c) Tobias Schoofs, 2013+License: LGPL+license-file: license/lgpl-3.0.txt+Author: Tobias Schoofs+Maintainer: tobias dot schoofs at gmx dot net+Homepage: http://github.com/toschoo/mom+Category: Network, Message-oriented Middleware, Stomp, Client+Build-Type: Simple+Synopsis: Stompl MOM Stomp Patterns+Description:+ The Stomp Protocol specifies a reduced message broker+ with queues usually read by one application and written+ by one or more applications.+ The specification does not include other, more advanced,+ interoperability patterns, where, for example,+ a client requests a job from a server or+ a publisher sends data to many subscribers.+ Such communication patterns, apparantly,+ are left to be implemented by applications.+ Patterns like client-server, publish and subscribe+ and many others, however, are used over and over again+ in message-oriented applications.++ .+ + This library implements a number of communication patterns+ on top of the Stomp specification+ that are often used and often described in the literature.+ There is a set of /basic/ patterns,++ .++ * client-server,++ .++ * publish and subscribe and++ .++ * pusher-worker (pipeline)++ .++ as well as a set of derived patterns, namely:++ .++ * Desk: A service to obtain the access point+ (/i.e./ queue name) of a specified provider;++ .++ * Load balancers: Services to balance requests+ among a group of connected /workers/+ (a.k.a. /Majordomo/ pattern);++ .++ * Bridge: Connections between brokers.+ + .+ + More information, examples and a test suite are available + on <http://github.com/toschoo/mom>.+ The Stomp specification can be found at+ <http://stomp.github.com>.++ .++ The notion of /pattern/ and the related concepts,+ as they are presented here,+ rely heavily on + Pieter Hintjens, \"Code Connected\", O'Reilly, 2013+ (see also <http://hintjens.com/books>).++ .++ Release History:++ .++ [0.0.1] Initial release+++Library+ Build-Depends: base >= 4.0 && <= 5.0,+ stomp-queue >= 0.1.4,+ stompl >= 0.1.0,+ mtl >= 2.1.2,+ mime >= 0.3.3,+ bytestring >= 0.10.4.0,+ time >= 1.4,+ containers >= 0.4.2.1,+ split >= 0.2.2++ hs-source-dirs: Network/Mom/Stompl/Patterns, .+ + Exposed-Modules: Network.Mom.Stompl.Patterns.Basic,+ Network.Mom.Stompl.Patterns.Desk,+ Network.Mom.Stompl.Patterns.Balancer,+ Network.Mom.Stompl.Patterns.Bridge++ other-modules: Types, Registry+ +