stomp-queue 0.3.1 → 0.5.0
raw patch · 7 files changed
+42/−260 lines, 7 filesdep +resourcetdep ~conduitdep ~conduit-extradep ~mime
Dependencies added: resourcet
Dependency ranges changed: conduit, conduit-extra, mime, network-conduit-tls
Files
- Network/Mom/Stompl/Client/Factory.hs +1/−1
- Network/Mom/Stompl/Client/Patterns.hs +0/−230
- Network/Mom/Stompl/Client/Queue.hs +6/−6
- Network/Mom/Stompl/Client/State.hs +3/−3
- Network/Mom/Stompl/Client/Stream.hs +12/−9
- changelog.md +4/−0
- stomp-queue.cabal +16/−11
Network/Mom/Stompl/Client/Factory.hs view
@@ -1,4 +1,4 @@-module Factory (+module Network.Mom.Stompl.Client.Factory ( Con(..), mkUniqueConId, Sub(..), mkUniqueSubId, Tx (..), mkUniqueTxId,
− Network/Mom/Stompl/Client/Patterns.hs
@@ -1,230 +0,0 @@----------------------------------------------------------------------------------- |--- 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 -{-# DEPRECATED "use Network.Mom.Stompl.Patterns.Basic instead!" #-} (- -- * 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-
Network/Mom/Stompl/Client/Queue.hs view
@@ -24,7 +24,7 @@ -- * Connections -- $stomp_con withConnection, - Factory.Con, + Fac.Con, F.Heart, Copt(..), EHandler,@@ -44,7 +44,7 @@ msgType, msgLen, msgHdrs, -- * Receipts -- $stomp_receipts- Factory.Rec(..), Receipt,+ Fac.Rec(..), Receipt, waitReceipt, -- * Transactions -- $stomp_trans@@ -65,9 +65,9 @@ where - import Stream- import Factory - import State+ import Network.Mom.Stompl.Client.Stream+ import Network.Mom.Stompl.Client.Factory as Fac+ import Network.Mom.Stompl.Client.State import qualified Network.Mom.Stompl.Frame as F import Network.Mom.Stompl.Client.Exception@@ -77,7 +77,7 @@ import Data.List (find) import Data.Time.Clock import Data.Maybe (isNothing)- import Data.Conduit.Network+ import Data.Conduit.Network (AppData) import Data.Conduit.Network.TLS import Control.Concurrent
Network/Mom/Stompl/Client/State.hs view
@@ -1,5 +1,5 @@ {-# Language BangPatterns,CPP #-}-module State (+module Network.Mom.Stompl.Client.State ( msgContent, numeric, ms, Connection(..), mkConnection, connected, getVersion, @@ -28,7 +28,7 @@ checkReceipt) where - import qualified Factory as Fac+ import qualified Network.Mom.Stompl.Client.Factory as Fac import qualified Network.Mom.Stompl.Frame as F import Network.Mom.Stompl.Client.Exception@@ -41,7 +41,7 @@ import Data.List (find) import Data.Char (isDigit) import Data.Time.Clock- + import Data.Conduit.Network.TLS (TLSClientConfig, tlsClientConfig, tlsClientUseTLS)
Network/Mom/Stompl/Client/Stream.hs view
@@ -1,10 +1,12 @@-module Stream+module Network.Mom.Stompl.Client.Stream where import qualified Data.Conduit as C- import Data.Conduit (($$),(=$))- import Data.Conduit.Network+ import Data.Conduit ((.|)) + import Data.Conduit.Network (AppData) + import Data.Conduit.Network as N+ import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.UTF8 as U @@ -14,6 +16,7 @@ import Control.Monad (forever) import Control.Monad.Trans (liftIO)+ import Control.Monad.IO.Class (MonadIO) import Control.Concurrent import qualified Data.Attoparsec.ByteString as A @@ -35,37 +38,37 @@ -- and send it through a socket ------------------------------------------------------------------------ sender :: AppData -> Chan F.Frame -> IO ()- sender ad ip = pipeSource ip $$ stream =$ appSink ad+ sender ad ip = C.runConduitRes (pipeSource ip .| stream .| N.appSink ad) ------------------------------------------------------------------------ -- Receiver thread: get a ByteStream through a socket, -- parse it to a Frame and send it through a pipe ------------------------------------------------------------------------ receiver :: AppData -> Chan F.Frame -> EH -> IO ()- receiver ad ip eh = appSource ad $$ parseC eh =$ pipeSink ip + receiver ad ip eh = C.runConduitRes (appSource ad .| parseC eh .| pipeSink ip) ------------------------------------------------------------------------ -- Put a frame into a pipe (a channel) ------------------------------------------------------------------------- pipeSink :: Chan F.Frame -> C.Sink F.Frame IO ()+ pipeSink :: MonadIO m => Chan F.Frame -> C.ConduitT F.Frame C.Void m () pipeSink ch = C.awaitForever (liftIO . writeChan ch) ------------------------------------------------------------------------ -- Read a frame from a pipe (a channel) ------------------------------------------------------------------------- pipeSource :: Chan F.Frame -> C.Source IO F.Frame+ pipeSource :: MonadIO m => Chan F.Frame -> C.ConduitT () F.Frame m () pipeSource ch = forever (liftIO (readChan ch) >>= C.yield) ------------------------------------------------------------------------ -- Convert a frame to a ByteString ------------------------------------------------------------------------- stream :: C.ConduitM F.Frame B.ByteString IO ()+ stream :: MonadIO m => C.ConduitT F.Frame B.ByteString m () stream = C.awaitForever (C.yield . F.putFrame) ------------------------------------------------------------------------ -- Parse a Frame from a ByteString ------------------------------------------------------------------------- parseC :: EH -> C.ConduitM B.ByteString F.Frame IO ()+ parseC :: MonadIO m => EH -> C.ConduitT B.ByteString F.Frame m () parseC eh = goOn where goOn = go (A.parse stompParser) 0 -- start with a clean parser go prs step = do
changelog.md view
@@ -1,3 +1,7 @@+__0.5.0__+ Changes:+ - new conduit version (> 1.0)+ __0.3.1__ Changes:
stomp-queue.cabal view
@@ -1,7 +1,7 @@ Name: stomp-queue-Version: 0.3.1-Cabal-Version: >= 1.8-Copyright: Copyright (c) Tobias Schoofs, 2011 - 2016+Version: 0.5.0+Cabal-Version: >= 1.24+Copyright: Copyright (c) Tobias Schoofs, 2011 - 2020 License: LGPL license-file: license/lgpl-3.0ex.txt Author: Tobias Schoofs@@ -43,17 +43,22 @@ attoparsec >= 0.9.1.1, split >= 0.1.4.1, mtl >= 2.2.0.1,- conduit >= 1.2.3.1,- conduit-extra >= 1.1.6.2,- network-conduit-tls >= 1.1.0.2, stompl >= 0.5.0,- mime >= 0.3.3,- time >= 1.1.4+ mime >= 0.4.0.2,+ time >= 1.1.4,+ conduit >= 1.3.4,+ conduit-extra >= 1.3.5,+ network-conduit-tls >= 1.3.2,+ resourcet >= 1.2.4 hs-source-dirs: Network/Mom/Stompl/Client, .++ default-language: Haskell98 Exposed-Modules: Network.Mom.Stompl.Client.Queue, - Network.Mom.Stompl.Client.Exception,- Network.Mom.Stompl.Client.Patterns- other-modules: Stream, State, Factory+ Network.Mom.Stompl.Client.Exception++ other-modules: Network.Mom.Stompl.Client.Stream,+ Network.Mom.Stompl.Client.State,+ Network.Mom.Stompl.Client.Factory