diff --git a/Network/Mom/Stompl/Client/Patterns.hs b/Network/Mom/Stompl/Client/Patterns.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Client/Patterns.hs @@ -0,0 +1,229 @@ +------------------------------------------------------------------------------- +-- | +-- Module : Network/Mom/Stompl/Client/Patterns.hs +-- Copyright : (c) Tobias Schoofs +-- License : LGPL +-- Stability : experimental +-- Portability: portable +-- +-- The Stomp specification defines only one queuing pattern: +-- /publish and subscribe/. +-- In some situations, other patterns may be more appropriate +-- such as /peer-to-peer/ or /client server/. +-- Whereas patterns like peer-to-peer +-- are easy to simulate with the means provided by Stomp, +-- client/server needs some more coordination +-- between the involved parties, the clients and the server. +-- +-- This module provides abstractions that implement +-- a simple client/server protocol on top of Stomp. +-- A server is an application that provides a /service/ +-- to others. +-- The service must be explicitly requested by a client +-- and only the requesting client must see the response +-- produced by the server. +-- +-- The module, basically, provides two data types ('ClientA' and 'ServerA') +-- and two functions working on these data types, namely +-- 'request' and 'reply'. +-- With the request function, the client requests a service +-- and waits for a response. +-- With the reply function, a server waits for a request, +-- produces a response and sends it back through a channel +-- indicated by the client. +-- +-- Internally, 'request' and 'reply' use a message header +-- called \"__client__\" to agree on the reply queue +-- +------------------------------------------------------------------------------- +module Network.Mom.Stompl.Client.Patterns ( + -- * Client + ClientA, withClient, request, checkRequest, + -- * Server + ServerA, withServer, reply) +where + + import Network.Mom.Stompl.Client.Queue + import qualified Network.Mom.Stompl.Frame as F + import System.Timeout + import qualified Codec.MIME.Type as M + import Control.Exception (throwIO) + + ------------------------------------------------------------------------ + -- | The client data type + ------------------------------------------------------------------------ + data ClientA i o = Cl { + clChn :: String, + 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, used for debugging. + -- + -- * '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 -> + ReaderDesc i -> + WriterDesc o -> + (ClientA i o -> IO r) -> IO r + withClient c n rd@(rn, _, _, _) wd act = + withPair c n rd wd $ \(r,w) -> act $ Cl rn r w + + -- the reply queue header ---------------------------------------------- + channel :: String + channel = "__client__" + + ------------------------------------------------------------------------ + -- | 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'. + -- + -- 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. + -- + -- * 'M.Type': The /MIME/ type of the request. + -- + -- * ['F.Header']: List of additional headers + -- to be sent with the request; + -- note that the function, internally, + -- uses a header named \"__client__\". + -- This header name, hence, is reserved + -- and must not be used by the application. + -- + -- * /o/: The request + ------------------------------------------------------------------------ + request :: ClientA i o -> + Int -> M.Type -> [F.Header] -> o -> IO (Maybe (Message i)) + request c tmo t hs r = + writeQ (clOut c) t ((channel, clChn c) : 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 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 for ever. + -- + -- For parameters, please refer to 'request'. + ------------------------------------------------------------------------ + checkRequest :: ClientA i o -> Int -> IO (Maybe (Message i)) + checkRequest c tmo = timeout tmo $ readQ (clIn c) + + ------------------------------------------------------------------------ + -- | The server data type + ------------------------------------------------------------------------ + data ServerA i o = Srv { + srvIn :: Reader i, + srvOut :: Writer o} + + ------------------------------------------------------------------------ + -- | The function creates a server + -- that lives within its scope. + -- + -- Parameters: + -- + -- * 'Con': Connection to a Stomp broker + -- + -- * 'String': Name of the Server, used for debugging. + -- + -- * '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 r w + + ------------------------------------------------------------------------ + -- | Waits for a client request, + -- calls the application-defined transformer to generate a reply + -- and sends this reply through the queue + -- whose name is indicated by the value of the \"__client__\" header. + -- The time a server waits for a request may be restricted + -- by the timeout. Typically, you would call reply with + -- timeout set to /-1/ (for /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. + -- + -- * 'M.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 -> M.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 -> + case lookup channel $ msgHdrs m of + Nothing -> throwIO $ + ProtocolException "No reply channel defined!" + Just c -> do x <- transform m + writeAdHoc (srvOut s) c t hs x + diff --git a/Network/Mom/Stompl/Client/Queue.hs b/Network/Mom/Stompl/Client/Queue.hs --- a/Network/Mom/Stompl/Client/Queue.hs +++ b/Network/Mom/Stompl/Client/Queue.hs @@ -49,7 +49,7 @@ Tx, withTransaction, Topt(..), abort, - -- * Acknowledgments + -- * Acknowledgements -- $stomp_acks ack, ackWith, nack, nackWith, -- * Exceptions @@ -249,7 +249,7 @@ Acknowledgements are used by the client to confirm the receipt of a message. The Stomp protocol foresees three different - acknowledgment modes, defined when the client subscribes to a queues. + acknowledgement modes, defined when the client subscribes to a queues. A subscription may use /auto mode/, /i.e./ a message is considered acknowledged when it has been sent to the subscriber; @@ -534,7 +534,7 @@ -- | Expression often used by René Artois. -- Furthermore, if 'OMode' is either -- 'F.Client' or 'F.ClientIndi', then - -- this option forces 'readQ' to send an acknowledgment + -- this option forces 'readQ' to send an acknowledgement -- automatically when a message has been read from the queue. | OAck -- | A queue created with 'OForceTx' will throw diff --git a/stomp-queue.cabal b/stomp-queue.cabal --- a/stomp-queue.cabal +++ b/stomp-queue.cabal @@ -1,5 +1,5 @@ Name: stomp-queue -Version: 0.0.7 +Version: 0.0.8 Cabal-Version: >= 1.8 Copyright: Copyright (c) Tobias Schoofs, 2011 - 2013 License: LGPL @@ -18,11 +18,15 @@ the location or internal structure of each other. They see only each other's interfaces, /i.e./ the messages published and subscribed through the broker. + + . The Stomp Queue library provides a Stomp client, using abstractions like 'Connection', 'Transaction', 'Queue' and 'Message'. + . + More information, examples and a test suite are available on . The Stomp specification can be found at @@ -34,6 +38,10 @@ . + [0.0.8] Client/Server on top of Queues. + + . + [0.0.7] Fighting with hackagedb... . @@ -70,7 +78,7 @@ utf8-string >= 0.3.6, attoparsec >= 0.9.1.1, split >= 0.1.4.1, - network >= 2.3.0.4, + network >= 2.4.0.0, stompl >= 0.0.3, mime >= 0.3.3, time >= 1.1.4 @@ -78,8 +86,7 @@ hs-source-dirs: Network/Mom/Stompl/Client, . Exposed-Modules: Network.Mom.Stompl.Client.Queue, - Network.Mom.Stompl.Client.Exception - -- Network.Mom.Stompl.Client.Client, - -- Network.Mom.Stompl.Client.Server + Network.Mom.Stompl.Client.Exception, + Network.Mom.Stompl.Client.Patterns other-modules: Socket, Protocol, State, Factory