diff --git a/Network/Mom/Stompl/Patterns/Balancer.hs b/Network/Mom/Stompl/Patterns/Balancer.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Patterns/Balancer.hs @@ -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 diff --git a/Network/Mom/Stompl/Patterns/Basic.hs b/Network/Mom/Stompl/Patterns/Basic.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Patterns/Basic.hs @@ -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) + diff --git a/Network/Mom/Stompl/Patterns/Bridge.hs b/Network/Mom/Stompl/Patterns/Bridge.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Patterns/Bridge.hs @@ -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__"]) diff --git a/Network/Mom/Stompl/Patterns/Desk.hs b/Network/Mom/Stompl/Patterns/Desk.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Patterns/Desk.hs @@ -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 diff --git a/Network/Mom/Stompl/Patterns/Registry.hs b/Network/Mom/Stompl/Patterns/Registry.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Patterns/Registry.hs @@ -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 + diff --git a/Network/Mom/Stompl/Patterns/Types.hs b/Network/Mom/Stompl/Patterns/Types.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Patterns/Types.hs @@ -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) + + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/license/lgpl-3.0.txt b/license/lgpl-3.0.txt new file mode 100644 --- /dev/null +++ b/license/lgpl-3.0.txt @@ -0,0 +1,165 @@ + GNU LESSER GENERAL PUBLIC LICENSE + Version 3, 29 June 2007 + + Copyright (C) 2007 Free Software Foundation, Inc. + Everyone is permitted to copy and distribute verbatim copies + of this license document, but changing it is not allowed. + + + This version of the GNU Lesser General Public License incorporates +the terms and conditions of version 3 of the GNU General Public +License, supplemented by the additional permissions listed below. + + 0. Additional Definitions. + + As used herein, "this License" refers to version 3 of the GNU Lesser +General Public License, and the "GNU GPL" refers to version 3 of the GNU +General Public License. + + "The Library" refers to a covered work governed by this License, +other than an Application or a Combined Work as defined below. + + An "Application" is any work that makes use of an interface provided +by the Library, but which is not otherwise based on the Library. +Defining a subclass of a class defined by the Library is deemed a mode +of using an interface provided by the Library. + + A "Combined Work" is a work produced by combining or linking an +Application with the Library. The particular version of the Library +with which the Combined Work was made is also called the "Linked +Version". + + The "Minimal Corresponding Source" for a Combined Work means the +Corresponding Source for the Combined Work, excluding any source code +for portions of the Combined Work that, considered in isolation, are +based on the Application, and not on the Linked Version. + + The "Corresponding Application Code" for a Combined Work means the +object code and/or source code for the Application, including any data +and utility programs needed for reproducing the Combined Work from the +Application, but excluding the System Libraries of the Combined Work. + + 1. Exception to Section 3 of the GNU GPL. + + You may convey a covered work under sections 3 and 4 of this License +without being bound by section 3 of the GNU GPL. + + 2. Conveying Modified Versions. + + If you modify a copy of the Library, and, in your modifications, a +facility refers to a function or data to be supplied by an Application +that uses the facility (other than as an argument passed when the +facility is invoked), then you may convey a copy of the modified +version: + + a) under this License, provided that you make a good faith effort to + ensure that, in the event an Application does not supply the + function or data, the facility still operates, and performs + whatever part of its purpose remains meaningful, or + + b) under the GNU GPL, with none of the additional permissions of + this License applicable to that copy. + + 3. Object Code Incorporating Material from Library Header Files. + + The object code form of an Application may incorporate material from +a header file that is part of the Library. You may convey such object +code under terms of your choice, provided that, if the incorporated +material is not limited to numerical parameters, data structure +layouts and accessors, or small macros, inline functions and templates +(ten or fewer lines in length), you do both of the following: + + a) Give prominent notice with each copy of the object code that the + Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the object code with a copy of the GNU GPL and this license + document. + + 4. Combined Works. + + You may convey a Combined Work under terms of your choice that, +taken together, effectively do not restrict modification of the +portions of the Library contained in the Combined Work and reverse +engineering for debugging such modifications, if you also do each of +the following: + + a) Give prominent notice with each copy of the Combined Work that + the Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the Combined Work with a copy of the GNU GPL and this license + document. + + c) For a Combined Work that displays copyright notices during + execution, include the copyright notice for the Library among + these notices, as well as a reference directing the user to the + copies of the GNU GPL and this license document. + + d) Do one of the following: + + 0) Convey the Minimal Corresponding Source under the terms of this + License, and the Corresponding Application Code in a form + suitable for, and under terms that permit, the user to + recombine or relink the Application with a modified version of + the Linked Version to produce a modified Combined Work, in the + manner specified by section 6 of the GNU GPL for conveying + Corresponding Source. + + 1) Use a suitable shared library mechanism for linking with the + Library. A suitable mechanism is one that (a) uses at run time + a copy of the Library already present on the user's computer + system, and (b) will operate properly with a modified version + of the Library that is interface-compatible with the Linked + Version. + + e) Provide Installation Information, but only if you would otherwise + be required to provide such information under section 6 of the + GNU GPL, and only to the extent that such information is + necessary to install and execute a modified version of the + Combined Work produced by recombining or relinking the + Application with a modified version of the Linked Version. (If + you use option 4d0, the Installation Information must accompany + the Minimal Corresponding Source and Corresponding Application + Code. If you use option 4d1, you must provide the Installation + Information in the manner specified by section 6 of the GNU GPL + for conveying Corresponding Source.) + + 5. Combined Libraries. + + You may place library facilities that are a work based on the +Library side by side in a single library together with other library +facilities that are not Applications and are not covered by this +License, and convey such a combined library under terms of your +choice, if you do both of the following: + + a) Accompany the combined library with a copy of the same work based + on the Library, uncombined with any other library facilities, + conveyed under the terms of this License. + + b) Give prominent notice with the combined library that part of it + is a work based on the Library, and explaining where to find the + accompanying uncombined form of the same work. + + 6. Revised Versions of the GNU Lesser General Public License. + + The Free Software Foundation may publish revised and/or new versions +of the GNU Lesser General Public License from time to time. Such new +versions will be similar in spirit to the present version, but may +differ in detail to address new problems or concerns. + + Each version is given a distinguishing version number. If the +Library as you received it specifies that a certain numbered version +of the GNU Lesser General Public License "or any later version" +applies to it, you have the option of following the terms and +conditions either of that published version or of any later version +published by the Free Software Foundation. If the Library as you +received it does not specify a version number of the GNU Lesser +General Public License, you may choose any version of the GNU Lesser +General Public License ever published by the Free Software Foundation. + + If the Library as you received it specifies that a proxy can decide +whether future versions of the GNU Lesser General Public License shall +apply, that proxy's public statement of acceptance of any version is +permanent authorization for you to choose that version for the +Library. diff --git a/stomp-patterns.cabal b/stomp-patterns.cabal new file mode 100644 --- /dev/null +++ b/stomp-patterns.cabal @@ -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 . + The Stomp specification can be found at + . + + . + + 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 ). + + . + + 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 + +