diff --git a/Network/Mom/Stompl/Client/Factory.hs b/Network/Mom/Stompl/Client/Factory.hs --- a/Network/Mom/Stompl/Client/Factory.hs +++ b/Network/Mom/Stompl/Client/Factory.hs @@ -1,4 +1,4 @@ -module Factory ( +module Network.Mom.Stompl.Client.Factory ( Con(..), mkUniqueConId, Sub(..), mkUniqueSubId, Tx (..), mkUniqueTxId, diff --git a/Network/Mom/Stompl/Client/Patterns.hs b/Network/Mom/Stompl/Client/Patterns.hs deleted file mode 100644 --- a/Network/Mom/Stompl/Client/Patterns.hs +++ /dev/null @@ -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 - 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 @@ -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 diff --git a/Network/Mom/Stompl/Client/State.hs b/Network/Mom/Stompl/Client/State.hs --- a/Network/Mom/Stompl/Client/State.hs +++ b/Network/Mom/Stompl/Client/State.hs @@ -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) diff --git a/Network/Mom/Stompl/Client/Stream.hs b/Network/Mom/Stompl/Client/Stream.hs --- a/Network/Mom/Stompl/Client/Stream.hs +++ b/Network/Mom/Stompl/Client/Stream.hs @@ -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 diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,7 @@ +__0.5.0__ + Changes: + - new conduit version (> 1.0) + __0.3.1__ Changes: diff --git a/stomp-queue.cabal b/stomp-queue.cabal --- a/stomp-queue.cabal +++ b/stomp-queue.cabal @@ -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