stomp-queue (empty) → 0.0.1
raw patch · 9 files changed
+2933/−0 lines, 9 filesdep +attoparsecdep +basedep +bytestringsetup-changed
Dependencies added: attoparsec, base, bytestring, mime, network, split, stompl, time, utf8-string
Files
- Network/Mom/Stompl/Client/Exception.hs +73/−0
- Network/Mom/Stompl/Client/Factory.hs +133/−0
- Network/Mom/Stompl/Client/Protocol.hs +409/−0
- Network/Mom/Stompl/Client/Queue.hs +1314/−0
- Network/Mom/Stompl/Client/Socket.hs +172/−0
- Network/Mom/Stompl/Client/State.hs +619/−0
- Setup.hs +2/−0
- license/lgpl-3.0.txt +165/−0
- stomp-queue.cabal +46/−0
+ Network/Mom/Stompl/Client/Exception.hs view
@@ -0,0 +1,73 @@+{-# OPTIONS -fglasgow-exts #-}+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Stompl/Client/Exception.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- Stability : experimental+-- Portability: portable+--+-- Exceptions for the Stompl Client+-------------------------------------------------------------------------------+module Network.Mom.Stompl.Client.Exception (+ StomplException(..), try, convertError)+where++ import Control.Exception hiding (try)+ import Prelude hiding (catch)+ import Control.Applicative ((<$>))+ import Data.Typeable (Typeable)++ ------------------------------------------------------------------------+ -- | The Stompl Client uses exceptions to communicate errors+ -- to the user application.+ ------------------------------------------------------------------------+ data StomplException =+ -- | Thrown+ -- on problems with the socket, /e.g./+ -- when a message cannot be sent+ SocketException String+ -- | Thrown when something + -- against the protocol happens, /e.g./+ -- an unexpected frame is received+ -- or a message from a queue+ -- that was not subscribed+ | ProtocolException String+ -- | Thrown on wrong uses of queues, /e.g./+ -- use of a queue outside its scope+ | QueueException String+ -- | Thrown on transaction errors, /e.g./+ -- pending acks+ | TxException String+ -- | Thrown on connection errors, /e.g./+ -- connection was disconnected+ | ConnectException String+ -- | Should be thrown + -- by user-defined converters+ | ConvertException String+ -- | Thrown when an error frame is received+ | BrokerException String+ -- | Thrown by /abort/+ | AppException String+ -- | You hit a bug!+ -- This exception is only thrown+ -- when something really strange happened+ | OuchException String+ deriving (Show, Read, Typeable, Eq)++ instance Exception StomplException++ ------------------------------------------------------------------------+ -- | Catches any 'StomplException',+ -- including asynchronous exceptions coming from internal threads+ ------------------------------------------------------------------------+ try :: IO a -> IO (Either StomplException a)+ try act = (Right <$> act) `catch` (return . Left)++ ------------------------------------------------------------------------+ -- | Throws 'ConvertException'+ -- to signal a conversion error.+ ------------------------------------------------------------------------+ convertError :: String -> IO a+ convertError e = throwIO $ ConvertException e+
+ Network/Mom/Stompl/Client/Factory.hs view
@@ -0,0 +1,133 @@+{-# OPTIONS -fglasgow-exts -fno-cse #-}+module Factory (+ Con(..), mkUniqueConId,+ Sub(..), mkUniqueSubId,+ Tx (..), mkUniqueTxId,+ Rec(..), mkUniqueRecc, parseRec)+where++ import System.IO.Unsafe+ import Control.Concurrent+ import Data.Char (isDigit)++ ------------------------------------------------------------------------+ -- | Opaque Connection handle.+ -- Only valid within the action passed to /withConnection/. + ------------------------------------------------------------------------+ newtype Con = Con Int+ deriving (Eq)++ instance Show Con where+ show (Con i) = show i++ ------------------------------------------------------------------------+ -- Subscription Identifier+ ------------------------------------------------------------------------+ data Sub = Sub Int | NoSub+ deriving (Eq)++ instance Show Sub where+ show (Sub i) = show i+ show (NoSub) = ""++ ------------------------------------------------------------------------+ -- Transaction Identifier+ ------------------------------------------------------------------------+ data Tx = Tx Int | NoTx+ deriving (Eq)++ instance Show Tx where+ show (Tx i) = show i+ show (NoTx) = ""++ ------------------------------------------------------------------------+ -- | This is a receipt.+ ------------------------------------------------------------------------+ data Rec = + -- | A valid receipt+ Rec Int + -- | No receipt was sent with this interaction.+ -- Receiving a 'NoRec' is not an error,+ -- but the result of an inconsistent - but harmless -+ -- use of /writeQWith/ on a queue that does not+ -- send receipts. An application should, of course,+ -- not try to wait for a 'NoRec'. It will never be confirmed.+ | NoRec+ deriving (Eq)++ instance Show Rec where+ show (Rec i) = show i+ show NoRec = ""++ parseRec :: String -> Maybe Rec+ parseRec s = + if numeric s then Just (Rec $ read s) else Nothing++ numeric :: String -> Bool+ numeric = all isDigit++ ------------------------------------------------------------------------+ -- Source for unique connection identifiers+ ------------------------------------------------------------------------+ {-# NOINLINE conid #-}+ conid :: MVar Con+ conid = unsafePerformIO $ newMVar (Con 1)++ ------------------------------------------------------------------------+ -- Source for unique subscription identifiers+ ------------------------------------------------------------------------+ {-# NOINLINE subid #-}+ subid :: MVar Sub+ subid = unsafePerformIO $ newMVar (Sub 1)++ ------------------------------------------------------------------------+ -- Source for unique transaction identifiers+ ------------------------------------------------------------------------+ {-# NOINLINE txid #-}+ txid :: MVar Tx+ txid = unsafePerformIO $ newMVar (Tx 1)++ ------------------------------------------------------------------------+ -- Source for unique receipts+ ------------------------------------------------------------------------+ {-# NOINLINE recc #-}+ recc :: MVar Rec+ recc = unsafePerformIO $ newMVar (Rec 1)++ ------------------------------------------------------------------------+ -- Interfaces+ ------------------------------------------------------------------------+ mkUniqueConId :: IO Con+ mkUniqueConId = mkUniqueId conid incCon++ mkUniqueSubId :: IO Sub+ mkUniqueSubId = mkUniqueId subid incSub++ mkUniqueTxId :: IO Tx+ mkUniqueTxId = mkUniqueId txid incTx++ mkUniqueRecc :: IO Rec+ mkUniqueRecc = mkUniqueId recc incRecc++ mkUniqueId :: MVar a -> (a -> a) -> IO a+ mkUniqueId v f = modifyMVar v $ \x -> + let x' = f x in return (x', x')++ incCon :: Con -> Con+ incCon (Con n) = Con (incX n)++ incSub :: Sub -> Sub+ incSub (Sub n) = Sub (incX n)+ incSub NoSub = NoSub++ incTx :: Tx -> Tx+ incTx (Tx n) = Tx (incX n)+ incTx NoTx = NoTx++ incRecc :: Rec -> Rec+ incRecc (Rec n) = Rec (incX n)+ incRecc (NoRec) = NoRec++ incX :: Int -> Int+ incX i = if i == 99999999 then 1 else i+1+
+ Network/Mom/Stompl/Client/Protocol.hs view
@@ -0,0 +1,409 @@+{-# Language CPP #-}+module Protocol (Connection, mkConnection, + conBeat, getVersion,+ getSock, getWr, getRc,+ connected, getErr, conMax,+ connect, disconnect, disc,+ Subscription, mkSub,+ subscribe, unsubscribe,+ begin, commit, abort, sendBeat,+ Message(..), mkMessage,+ MsgId(..),+ send, ack, nack)+ + +where++ import qualified Socket as S+ import qualified Factory as Fac+ import qualified Network.Mom.Stompl.Frame as F+ import Network.Mom.Stompl.Client.Exception+ import Network.Socket (Socket)++ import qualified Data.ByteString.Char8 as B+ import qualified Data.ByteString.UTF8 as U+ import Data.Maybe (fromMaybe)++ import Prelude hiding (catch)+ import Control.Exception (throwIO, catch, + SomeException, bracketOnError)+#ifdef _DEBUG+ import Control.Monad (when)+#endif+ import Codec.MIME.Type as Mime (Type) ++ ---------------------------------------------------------------------+ -- Default version, when broker does not send a version+ ---------------------------------------------------------------------+ defVersion :: F.Version+ defVersion = (1,0)++ ---------------------------------------------------------------------+ -- Connection + ---------------------------------------------------------------------+ data Connection = Connection {+ conAddr :: String, -- the broker's IP address+ conPort :: Int, -- the broker's port+ conVers :: [F.Version], -- the accepted versions+ -- the agreed version + -- after connect+ conBeat :: F.Heart, -- the heart beat+ conSrv :: String, -- server descritpion+ conSes :: String, -- session identifier (from broker)+ conUsr :: String, -- user+ conPwd :: String, -- passcode+ conMax :: Int, -- max receive+ conSock :: Maybe Socket, -- if connected: socket+ conRcv :: Maybe S.Receiver, -- if connected: receiver+ conWrt :: Maybe S.Writer, -- if connected: writer+ conErrM :: String, -- connection error+ conTcp :: Bool, -- flag for tcp/ip connect+ conBrk :: Bool} -- flag fror broker connect++ ---------------------------------------------------------------------+ -- Subscribe abstraction+ ---------------------------------------------------------------------+ data Subscription = Sub {+ subId :: Fac.Sub, -- subscribe identifier+ subName :: String, -- queue name+ subMode :: F.AckMode -- ack mode+ }+ deriving (Show)++ mkSub :: Fac.Sub -> String -> F.AckMode -> Subscription+ mkSub sid qn am = Sub {+ subId = sid,+ subName = qn,+ subMode = am}++ ---------------------------------------------------------------------+ -- | Message Identifier+ ---------------------------------------------------------------------+ data MsgId = MsgId String | NoMsg+ deriving (Eq)++ instance Show MsgId where+ show (MsgId s) = s+ show (NoMsg) = ""++ ------------------------------------------------------------------------+ -- | Any content received from a queue+ -- is wrapped in a message.+ -- It is, in particular, the return value of /readQ/.+ ------------------------------------------------------------------------+ data Message a = Msg {+ -- | The message Identifier+ msgId :: MsgId,+ -- | The subscription+ msgSub :: Fac.Sub,+ -- | The destination+ msgDest :: String,+ -- | The Stomp headers+ -- that came with the message+ msgHdrs :: [F.Header],+ -- | The /MIME/ type of the content+ msgType :: Mime.Type,+ -- | The length of the + -- encoded content+ msgLen :: Int,+ -- | The transaction, in which + -- the message was received+ msgTx :: Fac.Tx,+ -- | The encoded content + msgRaw :: B.ByteString,+ -- | The content + msgCont :: a}+ + ---------------------------------------------------------------------+ -- Create a message+ ---------------------------------------------------------------------+ mkMessage :: MsgId -> Fac.Sub -> String -> + Mime.Type -> Int -> Fac.Tx -> + B.ByteString -> a -> Message a+ mkMessage mid sub dst typ len tx raw cont = Msg {+ msgId = mid,+ msgSub = sub,+ msgDest = dst,+ msgHdrs = [], + msgType = typ, + msgLen = len, + msgTx = tx,+ msgRaw = raw,+ msgCont = cont}++ ---------------------------------------------------------------------+ -- Make a connection+ ---------------------------------------------------------------------+ mkConnection :: String -> Int -> Int -> String -> String -> [F.Version] -> F.Heart -> Connection+ mkConnection host port mx usr pwd vers beat = + Connection {+ conAddr = host,+ conPort = port,+ conVers = vers,+ conBeat = beat,+ conSrv = "",+ conSes = "",+ conUsr = usr, + conPwd = pwd,+ conMax = mx,+ conSock = Nothing,+ conRcv = Nothing,+ conWrt = Nothing,+ conErrM = "",+ conTcp = False,+ conBrk = False}++ ---------------------------------------------------------------------+ -- Error: we are not connected!+ ---------------------------------------------------------------------+ incompleteErr :: String+ incompleteErr = "incomplete Connection touched!"++ ---------------------------------------------------------------------+ -- Connection interfaces+ ---------------------------------------------------------------------+ getSock :: Connection -> Socket+ getSock = fromMaybe (error incompleteErr) . conSock ++ getWr :: Connection -> S.Writer+ getWr = fromMaybe (error incompleteErr) . conWrt++ getRc :: Connection -> S.Receiver+ getRc = fromMaybe (error incompleteErr) . conRcv++ connected :: Connection -> Bool+ connected c = conTcp c && conBrk c++ getErr :: Connection -> String+ getErr = conErrM++ getVersion :: Connection -> F.Version+ getVersion c = if null (conVers c) + then defVersion+ else head $ conVers c++ ---------------------------------------------------------------------+ -- connect+ ---------------------------------------------------------------------+ connect :: String -> Int -> Int -> String -> String -> [F.Version] -> F.Heart -> IO Connection+ connect host port mx usr pwd vers beat = + bracketOnError (do s <- S.connect host port+ let c = mkConnection host port mx + usr pwd vers beat+ return c {conSock = Just s, conTcp = True}) + disc+ (connectBroker mx vers beat) ++ ---------------------------------------------------------------------+ -- disconnect either on broker level or on tcp/ip level+ ---------------------------------------------------------------------+ disconnect :: Connection -> String -> IO Connection+ disconnect c r+ | conBrk c = case mkDiscF r of+ Left e -> return c {+ conErrM = + "Cannot create Frame: " ++ e}+ Right f -> do+ S.send (getWr c) (getSock c) f + return c {conBrk = False}+ | conTcp c = disc c+ | otherwise = return c {conErrM = "Not connected!"}++ ---------------------------------------------------------------------+ -- begin transaction+ ---------------------------------------------------------------------+ begin :: Connection -> String -> String -> IO ()+ begin c tx receipt = sendFrame c tx receipt [] mkBeginF++ ---------------------------------------------------------------------+ -- commit transaction+ ---------------------------------------------------------------------+ commit :: Connection -> String -> String -> IO ()+ commit c tx receipt = sendFrame c tx receipt [] mkCommitF++ ---------------------------------------------------------------------+ -- abort transaction+ ---------------------------------------------------------------------+ abort :: Connection -> String -> String -> IO ()+ abort c tx receipt = sendFrame c tx receipt [] mkAbortF++ ---------------------------------------------------------------------+ -- ack+ ---------------------------------------------------------------------+ ack :: Connection -> Message a -> String -> IO ()+ ack c m receipt = sendFrame c m receipt [] (mkAckF True)++ ---------------------------------------------------------------------+ -- nack+ ---------------------------------------------------------------------+ nack :: Connection -> Message a -> String -> IO ()+ nack c m receipt = sendFrame c m receipt [] (mkAckF False)++ ---------------------------------------------------------------------+ -- subscribe+ ---------------------------------------------------------------------+ subscribe :: Connection -> Subscription -> String -> [F.Header] -> IO ()+ subscribe c sub receipt hs = sendFrame c sub receipt hs mkSubF++ ---------------------------------------------------------------------+ -- unsubscribe+ ---------------------------------------------------------------------+ unsubscribe :: Connection -> Subscription -> String -> [F.Header] -> IO ()+ unsubscribe c sub receipt hs = sendFrame c sub receipt hs mkUnSubF++ ---------------------------------------------------------------------+ -- send+ ---------------------------------------------------------------------+ send :: Connection -> Message a -> String -> [F.Header] -> IO ()+ send c msg receipt hs = sendFrame c msg receipt hs mkSendF++ ---------------------------------------------------------------------+ -- heart beat+ ---------------------------------------------------------------------+ sendBeat :: Connection -> IO ()+ sendBeat c = sendFrame c () "" [] (\_ _ _ -> Right F.mkBeat)++ ---------------------------------------------------------------------+ -- generic sendFrame:+ -- takes a connection some data (like subscribe, message, etc.)+ -- some headers, a function that creates a frame or returns an error+ -- creates the frame and sends it+ ---------------------------------------------------------------------+ sendFrame :: Connection -> a -> String -> [F.Header] -> + (a -> String -> [F.Header] -> Either String F.Frame) -> IO ()+ sendFrame c m receipt hs mkF = + if not (connected c) then throwIO $ ConnectException "Not connected!"+ else case mkF m receipt hs of+ Left e -> throwIO $ ProtocolException $+ "Cannot create Frame: " ++ e+ Right f -> +#ifdef _DEBUG+ do when (not $ F.complies (1,1) f) $+ putStrLn $ "Frame does not comply with 1.1: " ++ show f +#endif+ S.send (getWr c) (getSock c) f++ ---------------------------------------------------------------------+ -- hard disconnect (on tcp/ip) level+ ---------------------------------------------------------------------+ disc :: Connection -> IO Connection+ disc c = do+ let c' = c {conTcp = False,+ conBrk = False,+ conWrt = Nothing,+ conRcv = Nothing,+ conSock = Nothing}+ S.disconnect (getSock c)+ return c'++ ---------------------------------------------------------------------+ -- the hard work on connecting to a broker+ ---------------------------------------------------------------------+ connectBroker :: Int -> [F.Version] -> F.Heart -> Connection -> IO Connection+ connectBroker mx vers beat c = + case mkConF (conAddr c) (conUsr c) (conPwd c) vers beat of+ Left e -> return c {conErrM = e}+ Right f -> do+ rc <- S.initReceiver+ wr <- S.initWriter+ S.send wr (getSock c) f + eiC <- catch (S.receive rc (getSock c) mx)+ (\e -> return $ Left $ show (e::SomeException)) + case eiC of+ Left e -> return c {conErrM = e}+ Right r -> do+ let c' = handleConnected r c+ if period c' > 0 && period c' < fst beat+ then return c {conErrM = "Beat frequency too high"}+ else return c' {conBrk = True,+ conRcv = Just rc,+ conWrt = Just wr}+ where period = snd . conBeat++ ---------------------------------------------------------------------+ -- handle broker responds connect frame+ ---------------------------------------------------------------------+ handleConnected :: F.Frame -> Connection -> Connection+ handleConnected f c = + case F.typeOf f of+ F.Connected -> c {+ conSrv = let srv = F.getServer f+ in F.getSrvName srv ++ "/" +++ F.getSrvVer srv ++ " (" +++ F.getSrvCmts srv ++ ")",+ conBeat = F.getBeat f,+ conVers = [F.getVersion f],+ conSes = F.getSession f}+ F.Error -> c {conErrM = errToMsg f}+ _ -> c {conErrM = "Unexpected Frame: " ++ U.toString (F.putCommand f)}++ ---------------------------------------------------------------------+ -- transform an error frame into a string+ ---------------------------------------------------------------------+ errToMsg :: F.Frame -> String+ errToMsg f = let msg = if B.length (F.getBody f) == 0 + then "."+ else ": " ++ U.toString (F.getBody f)+ in F.getMsg f ++ msg++ ---------------------------------------------------------------------+ -- frame constructors+ -- this needs review...+ ---------------------------------------------------------------------+ mkReceipt :: String -> [F.Header]+ mkReceipt receipt = if null receipt then [] else [F.mkRecHdr receipt]++ mkConF :: String -> String -> String -> [F.Version] -> F.Heart -> Either String F.Frame+ mkConF host usr pwd vers beat = + let uHdr = if null usr then [] else [F.mkLogHdr usr]+ pHdr = if null pwd then [] else [F.mkPassHdr pwd]+ in F.mkConFrame $ [F.mkHostHdr host,+ F.mkAcVerHdr $ F.versToVal vers, + F.mkBeatHdr $ F.beatToVal beat] +++ uHdr ++ pHdr++ mkDiscF :: String -> Either String F.Frame+ mkDiscF receipt =+ F.mkDisFrame $ mkReceipt receipt++ mkSubF :: Subscription -> String -> [F.Header] -> Either String F.Frame+ mkSubF sub receipt hs = + F.mkSubFrame $ [F.mkIdHdr $ show $ subId sub,+ F.mkDestHdr $ subName sub,+ F.mkAckHdr $ show $ subMode sub] ++ + mkReceipt receipt ++ hs++ mkUnSubF :: Subscription -> String -> [F.Header] -> Either String F.Frame+ mkUnSubF sub receipt hs =+ let dh = if null (subName sub) then [] else [F.mkDestHdr $ subName sub]+ in F.mkUSubFrame $ [F.mkIdHdr $ show $ subId sub] ++ dh ++ + mkReceipt receipt ++ hs++ mkSendF :: Message a -> String -> [F.Header] -> Either String F.Frame+ mkSendF msg receipt hs = + Right $ F.mkSend (msgDest msg) (show $ msgTx msg) receipt + (msgType msg) (msgLen msg) hs + (msgRaw msg) ++ mkAckF :: Bool -> Message a -> String -> [F.Header] -> Either String F.Frame+ mkAckF ok msg receipt _ =+ let sh = if null $ show $ msgSub msg then [] + else [F.mkSubHdr $ show $ msgSub msg]+ th = if null $ show $ msgTx msg + then [] else [F.mkTrnHdr $ show $ msgTx msg]+ rh = mkReceipt receipt+ mk = if ok then F.mkAckFrame else F.mkNackFrame+ in mk $ F.mkMIdHdr (show $ msgId msg) : (sh ++ rh ++ th)++ mkBeginF :: String -> String -> [F.Header] -> Either String F.Frame+ mkBeginF tx receipt _ = + F.mkBgnFrame $ F.mkTrnHdr tx : mkReceipt receipt++ mkCommitF :: String -> String -> [F.Header] -> Either String F.Frame+ mkCommitF tx receipt _ =+ F.mkCmtFrame $ F.mkTrnHdr tx : mkReceipt receipt++ mkAbortF :: String -> String -> [F.Header] -> Either String F.Frame+ mkAbortF tx receipt _ =+ F.mkAbrtFrame $ F.mkTrnHdr tx : mkReceipt receipt+
+ Network/Mom/Stompl/Client/Queue.hs view
@@ -0,0 +1,1314 @@+-------------------------------------------------------------------------------+-- |+-- Module : Network/Mom/Stompl/Client/Queue.hs+-- Copyright : (c) Tobias Schoofs+-- License : LGPL +-- Stability : experimental+-- Portability: portable+--+-- The Stomp Protocol specifies message-oriented interoperability.+-- Applications connect to a message broker to send (publish)+-- or receive (subscribe) messages through queues. +-- Interoperating applications do not know +-- the location or internal structure of each other.+-- They only see interfaces, /i.e./ the messages+-- published and subscribed through the broker.+-- +-- The Stompl Client library implements+-- a Stomp client using abstractions like+-- 'Connection', 'Transaction' and+-- queues in terms of 'Reader' and 'Writer'.+-------------------------------------------------------------------------------+module Network.Mom.Stompl.Client.Queue (+ -- * Connections+ -- $stomp_con+ withConnection, + withConnection_, + Factory.Con, + F.Heart,+ Copt(..),+ -- * Queues+ -- $stomp_queues+ Reader, Writer, + newReader, newWriter, + withReader, withReader_,+ Qopt(..), F.AckMode(..), + InBound, OutBound, + readQ, + writeQ, writeQWith,+ -- * Messages+ P.Message, + msgContent, P.msgRaw, + P.msgType, P.msgLen, P.msgHdrs,+ -- * Receipts+ -- $stomp_receipts+ Factory.Rec(..), Receipt,+ waitReceipt,+ -- * Transactions+ -- $stomp_trans+ withTransaction,+ withTransaction_,+ Topt(..), abort,+ -- * Acknowledgments+ -- $stomp_acks+ ack, ackWith, nack, nackWith,+ -- * Exceptions+ module Network.Mom.Stompl.Client.Exception+ -- * Complete Example+ -- $stomp_sample+ )++where+ ----------------------------------------------------------------+ -- todo+ -- -- pass a logger (name) to withConnection+ -- - test/check for deadlocks+ ----------------------------------------------------------------++ import qualified Socket as S+ import qualified Protocol as P+ import Factory + import State++ import qualified Network.Mom.Stompl.Frame as F+ import Network.Mom.Stompl.Client.Exception++ import qualified Data.ByteString.Char8 as B+ import qualified Data.ByteString.UTF8 as U+ import Data.List (find)+ import Data.Time.Clock+ import Data.Maybe (isJust, fromJust)++ import Control.Concurrent + import Control.Applicative ((<$>))+ import Control.Monad+ import Control.Exception (bracket, finally, catches, onException,+ AsyncException(..), Handler(..),+ throwIO, SomeException)++ import Codec.MIME.Type as Mime (Type)+ import System.Timeout (timeout)++ {- $stomp_con++ The Stomp protocol is connection-oriented and+ usually implemented on top of /TCP/\//IP/.+ The client initialises the connection+ by sending a /connect/ message which is answered+ by the broker by confirming or rejecting the connection.+ The connection is authenticated by user and passcode.+ The authentication mechanism, however, + varies among brokers.++ During the connection phase,+ the protocol version and a heartbeat + that defines the frequency of /alive/ messages+ exchanged between broker and client+ are negotiated.++ The connection remains active until either the client+ disconnects voluntarily or the broker disconnects+ in consequence of a protocol error.++ The details of the connection, including + protocol version and heartbeats are handled+ internally by the Stompl Client library.+ -}++ {- $stomp_queues++ Stomp program interoperability is based on queues.+ Queues are communication channels of arbitrary size+ that may be written by any client + currently connected to the broker.+ Messages in the queue are stored in /FIFO/ order.+ The process of adding messages to the queue is called /send/.+ In order to read from a queue, a client has to /subscribe/ to it.+ After having subscribed to a queue, + a client will receive all message sent to it.+ Brokers may implement additional selection criteria+ by means of /selectors/ that are expressed in some+ query language, such as /SQL/ or /XPath/.++ There are two different flavours of queues+ distinguished by their communication pattern:+ Queues are either one-to-one channels,+ this is, a message published in this queue+ is sent to exactly one subscriber+ and then removed from it; + or queues may be one-to-many,+ /i.e./, a message published in this queue+ is sent to all current subscribers of the queue.+ This type of queues is sometimes called /topic/.+ Which pattern is supported + and how patterns are controlled, depends on the broker. ++ From the perspective of the Stomp protocol,+ the content of messages in a queue+ has no format.+ The Protocol describes only those aspects of messages+ that are related to their handling;+ this can be seen as a /syntactic/ level of interoperability.+ Introducing meaning to message contents+ is entirely left to applications.+ Message- or service-oriented frameworks,+ usually, define formats and encodings + to describe messages and higher-level+ communication patterns built on top of them,+ to add more /syntactic/ formalism or+ to raise interoperability+ to a /semantic/ or even /pragmatic/ level.++ The Stompl library stresses the importance+ of adding meaning to the message content+ by adding types to queues. + From the perspective of the client Haskell program,+ a queue is a communication channel+ that allows sending and receiving messages of a given type.+ This adds type-safety to Stompl queues, which,+ otherwise, would just return plain bytes.+ It is, on the other hand, always possible+ to ignore this feature by declaring queues+ as '()' or 'B.ByteString'.+ In the first case, the /raw/ bytestring+ may be read from the 'P.Message';+ in the second case, the contents of the 'P.Message'+ will be a 'B.ByteString'.++ In the Stompl library, queues + are unidirectional communication channels+ either for reading or writing.+ This is captured by implementing queues+ with two different data types,+ a 'Reader' and a 'Writer'.+ On creating a queue a set of parameters can be defined+ to control the behaviour of the queue.+ -}++ {- $stomp_receipts++ Receipts are identifiers unique during the life time+ of an application that can be added to all kinds of+ messages sent to the broker.+ The broker uses receipts to acknowledge received messages.+ Receipts are, hence, useful to make a session more reliable.+ When the broker has confirmed the receipt of a frame sent to it,+ the client application can be sure that it has arrived.+ What kind of additional guarantees are made,+ /e.g./ that the frame is saved to disk or has already been sent+ to the subscribers, depends on the broker.++ Receipts are usually handled internally by the library.+ The application, however, decides where receipts should+ be requested, /i.e./ on subcribing to a queue,+ on sending a message, on sending /acks/ and on+ starting and ending transactions.+ On sending messages, + receipt handling can be made explict.+ The function 'writeQWith'+ adds a receipt to the message+ and returns it to the caller.+ The application can then, later,+ explicitly wait for the receipt, using 'waitReceipt'.+ -}++ {- $stomp_trans++ Transactions are units of interactions with+ a Stomp broker, including sending messages to queues+ and acknowledging the receipt of messages.+ All messages sent during a transaction+ are buffered in the broker.+ Only when the application terminates the transaction+ with /commit/ the messages will be eventually processed.+ If an error occurs during the transaction,+ it can be /aborted/ by the client. + Transactions, in consequence, can be used+ to ensure atomicity,+ /i.e./ either all single steps are performed or + no step is performed.++ In the Stompl Client library, transactions+ are sequences of Stompl actions, + queue operations as well as nested transactions,+ that are committed at the end or aborted,+ whenever an error condition becomes true.+ Error conditions are uncaught exceptions+ and conditions defined by options passed+ to the transaction, for example+ that all receipts requested during the transaction,+ have been confirmed by the broker.++ To enforce atomicity, + threads are not allowed to share transactions.+ -}++ {- $stomp_acks++ 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.+ A subscription may use + /auto mode/, /i.e./ a message is considered acknowledged+ when it has been sent to the subscriber;+ /client mode/, /i.e./ a message is considered acknowledged+ only when an /ack/ message has been sent back from the client.+ Note that client mode is cumulative, that means, + the broker will consider all messages acknowledged + that have been sent+ from the previous ack up to the acknowledged message;+ or /client-individual mode/, /i.e./ non-cumulative+ client mode.+ + A message may also be /negatively acknowledged/ (/nack/). + How the broker handles a /nack/, however,+ is not further specified by the Stomp protocol.+ -}++ {- $stomp_sample++ > import Network.Mom.Stompl.Client.Queue+ >+ > import System.Environment (getArgs)+ > import Network.Socket (withSocketsDo)+ > import Control.Monad (forever)+ > import Control.Concurrent (threadDelay)+ > import qualified Data.ByteString.UTF8 as U+ > import Data.Char(toUpper)+ > import Codec.MIME.Type (nullType)+ >+ > main :: IO ()+ > main = do+ > os <- getArgs+ > case os of+ > [q] -> withSocketsDo $ ping q+ > _ -> putStrLn "I need a queue name!"+ > -- error handling...+ > + > data Ping = Ping | Pong+ > deriving (Show)+ >+ > strToPing :: String -> IO Ping+ > strToPing s = case map toUpper s of+ > "PING" -> return Ping+ > "PONG" -> return Pong+ > _ -> convertError $ "Not a Ping: '" ++ s ++ "'"+ >+ > ping :: String -> IO ()+ > ping qn = + > withConnection_ "localhost" 61613 [] $ \c -> do+ > let iconv _ _ _ = strToPing . U.toString+ > let oconv = return . U.fromString . show+ > inQ <- newReader c "Q-IN" qn [] [] iconv+ > outQ <- newWriter c "Q-OUT" qn [] [] oconv+ > writeQ outQ nullType [] Pong+ > listen inQ outQ+ >+ > listen :: Reader Ping -> Writer Ping -> IO ()+ > listen iQ oQ = forever $ do+ > eiM <- try $ readQ iQ + > case eiM of+ > Left e -> do+ > putStrLn $ "Error: " ++ show e+ > -- error handling ...+ > Right m -> do+ > let p = case msgContent m of+ > Ping -> Pong+ > Pong -> Ping+ > putStrLn $ show p+ > writeQ oQ nullType [] p+ > threadDelay 10000+ -}++ -- The versions, we support+ vers :: [F.Version]+ vers = [(1,0), (1,1)]++ ------------------------------------------------------------------------+ -- | A variant of 'withConnection' that returns nothing+ ------------------------------------------------------------------------+ withConnection_ :: String -> Int -> + [Copt] -> (Con -> IO ()) -> IO ()+ withConnection_ host port os act = + withConnection host port os act >>= (\_ -> return ())++ ------------------------------------------------------------------------+ -- | Initialises a connection and executes an 'IO' action.+ -- The connection life time is the scope of this action.+ -- The connection handle, 'Con', that is passed to the action+ -- should not be returned from 'withConnection'.+ -- Connections, however, can be shared among threads.+ -- In this case, the programmer has to take care+ -- not to terminate the action before all other threads+ -- working on the connection have finished.+ --+ -- Paramter:+ --+ -- * 'String': The broker's hostname or IP-address+ --+ -- * 'Int': The broker's port+ --+ -- * 'Copt': Control options passed to the connection+ --+ -- * ('Con' -> 'IO' a): The action to execute.+ -- The action receives the connection handle+ -- and returns a value of type /a/ + -- in the 'IO' monad.+ --+ -- 'withConnection' returns the result of the action passed into it.+ --+ -- 'withConnection' will always disconnect from the broker + -- when the action has terminated, even if an exception is raised.+ --+ -- Example:+ --+ -- > withConnection "localhost" 61613 [] $ \c -> do+ --+ -- This would connect to a broker listening to the loopback interface,+ -- port number 61613.+ -- The action is defined after the /hanging do/.+ --+ -- Internally, connections use concurrent threads;+ -- errors are communicated by throwing exceptions+ -- to the owner of the connection, where+ -- the owner is the thread that created the connection+ -- by calling 'withConnection'.+ -- It is therefore advisable to start different connections+ -- in different threads, so that each thread will receive+ -- only exceptions related to the connection it has opened.+ -- + -- Example:+ --+ -- > t <- forkIO $ withConnection_ "127.0.0.1" 61613 [] $ \c -> do+ ------------------------------------------------------------------------+ withConnection :: String -> Int -> [Copt] -> + (Con -> IO a) -> IO a+ withConnection host port os act = do+ let beat = oHeartBeat os+ let mx = oMaxRecv os+ let (u,p) = oAuth os+ bracket (P.connect host port mx u p vers beat)+ P.disc -- important: At the end, we close the socket!+ (whenConnected os act)++ whenConnected :: [Copt] -> (Con -> IO a) -> P.Connection -> IO a+ whenConnected os act c = + if not $ P.connected c + then throwIO $ ConnectException $ P.getErr c+ else do+ cid <- mkUniqueConId -- connection id+ me <- myThreadId -- connection owner+ now <- getCurrentTime -- heartbeat+ finally (do addCon $ mkConnection cid c me now os+ whenListening cid $ whenBeating cid)+ (rmCon cid)+ where period = snd . P.conBeat + whenListening cid = bracket (forkIO $ listen cid) killThread+ whenBeating cid _ = bracket + (if period c > 0 + then Just <$> (forkIO $ heartBeat cid $ period c)+ else return Nothing)+ (\b -> when (isThrd b) (killThread $ thrd b))+ (\_ -> do r <- act cid+ c' <- getCon cid+ -- wait for receipt on disconnect ----+ if (conWait c' <= 0) then return r+ else do+ rc <- mkUniqueRecc + _ <- P.disconnect c (show rc)+ addRec cid rc+ waitCon cid rc (conWait c') `onException` rmRec cid rc+ return r)++ isThrd :: Maybe a -> Bool+ isThrd = isJust++ thrd :: Maybe a -> a+ thrd = fromJust++ ------------------------------------------------------------------------+ -- wait for receipt+ ------------------------------------------------------------------------+ waitCon :: Con -> Receipt -> Int -> IO ()+ waitCon cid rc delay = do+ c <- getCon cid+ case find (==rc) $ conRecs c of+ Nothing -> return ()+ Just _ -> + if delay <= 0 + then throwIO $ ConnectException $ + "No receipt on disconnect (" ++ + show cid ++ ")."+ else do+ threadDelay $ ms 1+ waitCon cid rc (delay - 1)+ + ------------------------------------------------------------------------+ -- | A Queue for sending messages.+ ------------------------------------------------------------------------+ data Writer a = SendQ {+ wCon :: Con,+ wDest :: String,+ wName :: String,+ wRec :: Bool,+ wWait :: Bool,+ wTx :: Bool,+ wTo :: OutBound a}++ ------------------------------------------------------------------------+ -- | A Queue for receiving messages+ ------------------------------------------------------------------------+ data Reader a = RecvQ {+ rCon :: Con,+ rSub :: Sub,+ rDest :: String,+ rName :: String,+ rMode :: F.AckMode,+ rAuto :: Bool, -- library creates Ack+ rRec :: Bool,+ rFrom :: InBound a}++ instance Eq (Reader a) where+ q1 == q2 = rName q1 == rName q2++ instance Eq (Writer a) where+ q1 == q2 = wName q1 == wName q2++ ------------------------------------------------------------------------+ -- | Options that may be passed + -- to 'newReader' and 'newWriter' and their variants.+ ------------------------------------------------------------------------+ data Qopt = + -- | A queue created with 'OWithReceipt' will request a receipt+ -- on all interactions with the broker.+ -- The handling of receipts may be transparent to applications or+ -- may be made visible by using 'writeQWith'.+ -- Note that the option has effect right from the beginning, /i.e./+ -- a 'Reader' created with 'OWithReceipt' will + -- issue a request for receipt when subscribing to a Stomp queue.+ OWithReceipt + -- | A queue created with 'OWaitReceipt' will wait for the receipt+ -- before returning from a call that has issued a request for receipt.+ -- This implies that the current thread will yield the processor.+ -- 'writeQ' will internally create a request for receipt and + -- wait for the broker to confirm the receipt before returning.+ -- Note that, for 'newReader', there is no difference between+ -- 'OWaitReceipt' and 'OWithReceipt'. Either option will cause+ -- the thread to preempt until the receipt is confirmed.+ --+ -- On writing a message, this is not always the preferred+ -- method. You may want to fire and forget - for a while,+ -- before you check that your message has actually been + -- handled by the broker. In this case, you will create the+ -- 'Writer' with 'OWithReceipt' only and, later, after having+ -- sent a message with 'writeQWith', wait for the receipt using+ -- 'waitReceipt'. Note that 'OWaitReceipt' without 'OWithReceipt'+ -- has no meaning with 'writeQ' and 'writeQWith'. + -- If you want to send a receipt+ -- and wait for the broker to confirm it, you have to use + -- both options.+ --+ -- It is good practice to use /timeout/ with all calls+ -- that may wait for receipts, + -- /ie/ 'newReader' and 'withReader' + -- with options 'OWithReceipt' or 'OWaitReceipt',+ -- or 'writeQ' and 'writeQWith' with options 'OWaitReceipt',+ -- or 'ackWith' and 'nackWith'.+ | OWaitReceipt + -- | The option defines the 'F.AckMode' of the queue,+ -- which is relevant for 'Reader' only.+ -- 'F.AckMode' is one of: + -- 'F.Auto', 'F.Client', 'F.ClientIndi'.+ --+ -- If 'OMode' is not given, 'F.Auto' is assumed as default.+ --+ -- For more details, see 'F.AckMode'.+ | OMode F.AckMode + -- | Expression often used by René Artois.+ -- What he tries to say is: If 'OMode' is either+ -- 'F.Client' or 'F.ClientIndi', send an acknowledgment+ -- automatically when a message has been read from the queue. + | OAck+ -- | A queue created with 'OForceTx' will throw + -- 'QueueException' when used outside a 'Transaction'.+ | OForceTx+ deriving (Show, Read, Eq) ++ hasQopt :: Qopt -> [Qopt] -> Bool+ hasQopt o os = o `elem` os++ ackMode :: [Qopt] -> F.AckMode+ ackMode os = case find isMode os of+ Just (OMode x) -> x+ _ -> F.Auto+ where isMode x = case x of+ OMode _ -> True+ _ -> False++ ------------------------------------------------------------------------+ -- | Converters are user-defined actions passed to + -- 'newReader' ('InBound') and+ -- 'newWriter' ('OutBound')+ -- that convert a 'B.ByteString' to a value of type /a/ ('InBound') or+ -- a value of type /a/ to 'B.ByteString' ('OutBound'). + -- Converters are, hence, similar to /put/ and /get/ in the /Binary/+ -- monad. + --+ -- The reason for using explicit, user-defined converters + -- instead of /Binary/ /encode/ and /decode/+ -- is that the conversion with queues+ -- may be much more complex, involving reading configurations + -- or other 'IO' actions.+ -- Furthermore, we have to distinguish between data types and + -- there binary encoding when sent over the network.+ -- This distinction is made by /MIME/ types.+ -- Two applications may send the same data type,+ -- but one encodes this type as \"text/plain\",+ -- the other as \"text/xml\".+ -- 'InBound' conversions have to consider the /MIME/ type+ -- and, hence, need more input parameters than provided by /decode/.+ -- /encode/ and /decode/, however,+ -- can be used internally by user-defined converters.+ --+ -- The parameters expected by an 'InBound' converter are:+ --+ -- * the /MIME/ type of the content+ --+ -- * the content size + --+ -- * the list of 'F.Header' coming with the message+ --+ -- * the contents encoded as 'B.ByteString'.+ --+ -- The simplest possible in-bound converter for plain strings+ -- may be created like this:+ --+ -- > let iconv _ _ _ = return . toString+ ------------------------------------------------------------------------+ type InBound a = Mime.Type -> Int -> [F.Header] -> B.ByteString -> IO a+ ------------------------------------------------------------------------+ -- | Out-bound converters are much simpler.+ -- Since the application developer knows,+ -- which encoding to use, the /MIME/ type is not needed.+ -- The converter receives only the value of type /a/+ -- and converts it into a 'B.ByteString'.+ -- A simple example to create an out-bound converter + -- for plain strings could be:+ --+ -- > let oconv = return . fromString+ ------------------------------------------------------------------------+ type OutBound a = a -> IO B.ByteString++ ------------------------------------------------------------------------+ -- | Creates a 'Reader' with the life time of the connection 'Con'.+ -- Creating a receiving queue involves interaction with the broker;+ -- this may result in preempting the calling thread, + -- depending on the options ['Qopt'].+ -- + -- Parameters:+ --+ -- * The connection handle 'Con'+ --+ -- * A queue name that should be unique in your application.+ -- The queue name is useful for debugging, since it appears+ -- in error messages.+ --+ -- * The Stomp destination, /i.e./ the name of the queue+ -- as it is known to the broker and other applications.+ --+ -- * A list of options ('Qopt').+ --+ -- * A list of headers ('F.Header'), + -- which will be passed to the broker.+ -- the 'F.Header' parameter is actually a breach in the abstraction+ -- from the Stomp protocol. A header may be, for instance,+ -- a selector that restricts the subscription to this queue,+ -- such that only messages with certain attributes + -- (/i.e./ specific headers) are sent to the subscribing client.+ -- Selectors are broker-specific and typically expressed+ -- as SQL or XPath.+ --+ -- * An in-bound converter.+ --+ -- A usage example to create a 'Reader'+ -- with 'Connection' /c/ and the in-bound converter+ -- /iconv/ would be:+ --+ -- > q <- newReader c "TestQ" "/queue/test" [] [] iconv+ --+ -- A call to 'newReader' may result in preemption when+ -- one of the options 'OWaitReceipt' or 'OWithReceipt' are given;+ -- an example for such a call + -- with /tmo/ an 'Int' value representing a /timeout/+ -- in microseconds and + -- the result /mbQ/ of type 'Maybe' is:+ --+ -- > mbQ <- timeout tmo $ newReader c "TestQ" "/queue/test" [OWaitReceipt] [] oconv+ -- > case mbQ of+ -- > Nothing -> -- handle error+ -- > Just q -> do -- ...+ ------------------------------------------------------------------------+ newReader :: Con -> String -> String -> [Qopt] -> [F.Header] -> + InBound a -> IO (Reader a)+ newReader cid qn dst os hs conv = do+ c <- getCon cid+ if not $ P.connected (conCon c)+ then throwIO $ ConnectException $ + "Not connected (" ++ show cid ++ ")"+ else newRecvQ cid c qn dst os hs conv+ + ------------------------------------------------------------------------+ -- | Creates a 'Writer' with the life time of the connection 'Con'.+ -- Creating a sending queue does not involve interaction with the broker+ -- and will not preempt the calling thread.+ -- + -- A sending queue may be created like in the following+ -- code fragment, where /oconv/ is + -- an already defined out-bound converter:+ --+ -- > q <- newWriter c "TestQ" "/queue/test" [] [] oconv+ ------------------------------------------------------------------------+ newWriter :: Con -> String -> String -> [Qopt] -> [F.Header] -> + OutBound a -> IO (Writer a)+ newWriter cid qn dst os _ conv = do+ c <- getCon cid+ if not $ P.connected (conCon c)+ then throwIO $ ConnectException $ + "Not connected (" ++ show cid ++ ")"+ else newSendQ cid qn dst os conv++ ------------------------------------------------------------------------+ -- | Creates a 'Reader' with limited life time. + -- The queue will live only in the scope of the action+ -- that is passed as last parameter. + -- The function is useful for readers+ -- that are used only temporarly, /e.g./ during initialisation.+ -- When the action terminates, the client unsubscribes from + -- the broker queue - even if an exception is raised.+ --+ -- 'withReader' returns the result of the action.+ -- Since the life time of the queue is limited to the action,+ -- it should not be returned.+ -- Any operation on a reader created by 'withReader'+ -- outside the action will raise 'QueueException'.+ --+ -- A usage example is: + --+ -- > x <- withReader c "TestQ" "/queue/test" [] [] iconv $ \q -> do+ ------------------------------------------------------------------------+ withReader :: Con -> String -> String -> [Qopt] -> [F.Header] -> + InBound a -> (Reader a -> IO b) -> IO b+ withReader cid qn dst os hs conv act = do+ q <- newReader cid qn dst os hs conv+ act q `finally` unsub q++ ------------------------------------------------------------------------+ -- | A variant of 'withReader' + -- for actions that do not return anything.+ ------------------------------------------------------------------------+ withReader_ :: Con -> String -> String -> [Qopt] -> [F.Header] -> + InBound a -> (Reader a -> IO ()) -> IO ()+ withReader_ cid qn dst os hs conv act = + withReader cid qn dst os hs conv act >>= (\_ -> return ())++ ------------------------------------------------------------------------+ -- Creating a SendQ is plain an simple.+ ------------------------------------------------------------------------+ newSendQ :: Con -> String -> String -> [Qopt] -> + OutBound a -> IO (Writer a)+ newSendQ cid qn dst os conv = + return SendQ {+ wCon = cid,+ wDest = dst,+ wName = qn,+ wRec = hasQopt OWithReceipt os, + wWait = hasQopt OWaitReceipt os, + wTx = hasQopt OForceTx os, + wTo = conv}++ ------------------------------------------------------------------------+ -- Creating a ReceivQ, however, involves some more hassle,+ -- in particular 'IO'.+ ------------------------------------------------------------------------+ newRecvQ :: Con -> Connection -> String -> String -> + [Qopt] -> [F.Header] ->+ InBound a -> IO (Reader a)+ newRecvQ cid c qn dst os hs conv = do+ let am = ackMode os+ let au = hasQopt OAck os+ let with = hasQopt OWithReceipt os || hasQopt OWaitReceipt os+ sid <- mkUniqueSubId+ rc <- if with then mkUniqueRecc else return NoRec+ logSend cid+ P.subscribe (conCon c) (P.mkSub sid dst am) (show rc) hs+ ch <- newChan + addSub cid (sid, ch) + addDest cid (dst, ch) + let q = RecvQ {+ rCon = cid,+ rSub = sid,+ rDest = dst,+ rName = qn,+ rMode = am,+ rAuto = au,+ rRec = with,+ rFrom = conv}+ when with $ waitReceipt cid rc+ return q++ ------------------------------------------------------------------------+ -- Unsubscribe a queue+ ------------------------------------------------------------------------+ unsub :: Reader a -> IO ()+ unsub q = do+ let cid = rCon q+ let sid = rSub q+ let dst = rDest q+ rc <- if rRec q then mkUniqueRecc else return NoRec+ c <- getCon cid+ logSend cid+ finally (P.unsubscribe (conCon c) + (P.mkSub sid dst F.Client)+ (show rc) [])+ (do rmSub cid sid+ rmDest cid dst)+ when (rRec q) $ waitReceipt cid rc++ ------------------------------------------------------------------------+ -- | Removes the oldest message from the queue+ -- and returns it as 'P.Message'.+ -- The message cannot be read from the queue+ -- by another call to 'readQ' within the same connection.+ -- Wether other connections will receive the message as well+ -- depends on the broker and the queue patterns it implements.+ -- If the queue is currently empty,+ -- the thread will preempt until a message arrives.+ --+ -- If the queue was created with + -- 'OMode' other than 'F.Auto' + -- and with 'OAck', then an /ack/ + -- will be automatically sent to the broker;+ -- if 'OAck' was not set,+ -- the message will be registered as pending /ack/.+ --+ -- Note that, when 'readQ' sends an /ack/ internally,+ -- it will not request a receipt from the broker.+ -- The rationale for this design is simplicity.+ -- If the function expected a receipt, + -- it would have to either wait for the receipt+ -- or return it.+ -- In the first case, it would be difficult+ -- for the programmer to distinguish, on a timeout, between+ -- /no message available/ and+ -- /no receipt arrived/.+ -- In the second case, the receipt+ -- would need to be returned.+ -- This would unnecessarily blow up the interface.+ -- If you need the reliability of receipts,+ -- you should create the queue without 'OAck'+ -- and use 'ackWith' to acknowledge + -- the message explicitly.+ ------------------------------------------------------------------------+ readQ :: Reader a -> IO (P.Message a)+ readQ q = do+ c <- getCon (rCon q)+ if not $ P.connected (conCon c)+ then throwIO $ QueueException $ "Not connected: " ++ show (rCon q)+ else case getSub (rSub q) c of+ Nothing -> throwIO $ QueueException $ + "Unknown queue " ++ rName q+ Just ch -> do+ m <- readChan ch >>= frmToMsg q+ when (rMode q /= F.Auto) $+ if rAuto q+ then ack (rCon q) m+ else addAck (rCon q) (P.msgId m)+ return m++ ------------------------------------------------------------------------+ -- We do not support this, because of GHC ticket 4154:+ -- deadlock on isEmptyChan with concurrent read+ ------------------------------------------------------------------------+ -- isEmptyQ :: Reader a -> IO Bool++ ------------------------------------------------------------------------+ -- | Adds the value /a/ as message at the end of the queue.+ -- The Mime type as well as the headers + -- are added to the message.+ --+ -- If the queue was created with the option+ -- 'OWithReceipt',+ -- 'writeQ' will request a receipt from the broker.+ -- If the queue was additionally created with+ -- 'OWaitReceipt',+ -- 'writeQ' will preempt until the receipt is confirmed.+ --+ -- The Stomp headers are useful for brokers+ -- that provide selectors on /subscribe/,+ -- see 'newReader' for details.+ --+ -- A usage example for a 'Writer' /q/ of type 'String'+ -- may be (/nullType/ is defined as /text/\//plain/ in Codec.MIME):+ --+ -- > writeQ q nullType [] "hello world!"+ --+ -- For a 'Writer' that was created + -- with 'OWithReceipt' and 'OWaitReceipt',+ -- the function should be called with /timeout/:+ --+ -- > mbR <- timeout tmo $ writeQ q nullType [] "hello world!"+ -- > case mbR of+ -- > Nothing -> -- error handling+ -- > Just r -> do -- ...+ ------------------------------------------------------------------------+ writeQ :: Writer a -> Mime.Type -> [F.Header] -> a -> IO ()+ writeQ q mime hs x =+ writeQWith q mime hs x >>= (\_ -> return ())++ ------------------------------------------------------------------------+ -- | This is a variant of 'writeQ' + -- that is particularly useful for queues + -- created with 'OWithReceipt', but without 'OWaitReceipt'.+ -- It returns the 'Receipt', so that it can be waited for+ -- later, using 'waitReceipt'.+ --+ -- Note that the behaviour of 'writeQWith', + -- besides of returning the receipt, is the same as 'writeQ',+ -- /i.e./, on a queue with 'OWithReceipt' and 'OWaitReceipt'+ -- 'writeQWith' will wait for the receipt being confirmed.+ -- In this case, the returned receipt is, in fact, + -- of no further use for the application.+ --+ -- The function is used like:+ --+ -- > r <- writeQWith q nullType [] "hello world!"+ ------------------------------------------------------------------------+ writeQWith :: Writer a -> Mime.Type -> [F.Header] -> a -> IO Receipt+ writeQWith q mime hs x = do+ c <- getCon (wCon q)+ if not $ P.connected (conCon c)+ then throwIO $ ConnectException $+ "Not connected (" ++ show (wCon q) ++ ")"+ else do+ tx <- getCurTx c >>= (\mbT -> + case mbT of+ Nothing -> return NoTx+ Just t -> return t)+ if tx == NoTx && wTx q+ then throwIO $ QueueException $+ "Queue '" ++ wName q ++ + "' with OForceTx used outside Transaction"+ else do+ let conv = wTo q+ s <- conv x+ rc <- if wRec q then mkUniqueRecc else return NoRec+ let m = P.mkMessage P.NoMsg NoSub (wDest q) + mime (B.length s) tx s x+ when (wRec q) $ addRec (wCon q) rc + logSend $ wCon q+ P.send (conCon c) m (show rc) hs + when (wRec q && wWait q) $ waitReceipt (wCon q) rc + return rc++ ------------------------------------------------------------------------+ -- | Acknowledges the arrival of 'P.Message' to the broker.+ -- It is used with a 'Connection' /c/ and a 'P.Message' /x/ like:+ --+ -- > ack c x+ ------------------------------------------------------------------------+ ack :: Con -> P.Message a -> IO ()+ ack cid msg = do+ ack' cid True False msg+ rmAck cid $ P.msgId msg++ ------------------------------------------------------------------------+ -- | Acknowledges the arrival of 'P.Message' to the broker,+ -- requests a receipt and waits until it is confirmed.+ -- Since it preempts the calling thread,+ -- it is usually used with /timeout/,+ -- for a 'Connection' /c/, a 'P.Message' /x/ + -- and a /timeout/ in microseconds /tmo/ like:+ --+ -- > mbR <- timeout tmo $ ackWith c x + -- > case mbR of+ -- > Nothing -> -- error handling+ -- > Just _ -> do -- ...+ ------------------------------------------------------------------------+ ackWith :: Con -> P.Message a -> IO ()+ ackWith cid msg = do+ ack' cid True True msg + rmAck cid $ P.msgId msg++ ------------------------------------------------------------------------+ -- | Negatively acknowledges the arrival of 'P.Message' to the broker.+ -- For more details see 'ack'.+ ------------------------------------------------------------------------+ nack :: Con -> P.Message a -> IO ()+ nack cid msg = do+ ack' cid False False msg+ rmAck cid $ P.msgId msg++ ------------------------------------------------------------------------+ -- | Negatively acknowledges the arrival of 'P.Message' to the broker,+ -- requests a receipt and waits until it is confirmed.+ -- For more details see 'ackWith'.+ ------------------------------------------------------------------------+ nackWith :: Con -> P.Message a -> IO ()+ nackWith cid msg = do+ ack' cid False True msg+ rmAck cid $ P.msgId msg++ ------------------------------------------------------------------------+ -- Checks for Transaction,+ -- if a transaction is ongoing,+ -- the TxId is added to the message+ -- and calls P.ack on the message.+ -- If called with True for "with receipt"+ -- the function creates a receipt and waits for its confirmation. + ------------------------------------------------------------------------+ ack' :: Con -> Bool -> Bool -> P.Message a -> IO ()+ ack' cid ok with msg = do+ c <- getCon cid+ if not $ P.connected (conCon c) + then throwIO $ ConnectException $ + "Not connected (" ++ show cid ++ ")"+ else if null (show $ P.msgId msg)+ then throwIO $ ProtocolException "No message id in message!"+ else do+ tx <- getCurTx c >>= (\mbT -> + case mbT of+ Nothing -> return NoTx+ Just x -> return x)+ let msg' = msg {P.msgTx = tx}+ rc <- if with then mkUniqueRecc else return NoRec+ when with $ addRec cid rc+ logSend cid+ if ok then P.ack (conCon c) msg' $ show rc+ else P.nack (conCon c) msg' $ show rc+ when with $ waitReceipt cid rc ++ ------------------------------------------------------------------------+ -- | Variant of 'withTransaction' that does not return anything.+ ------------------------------------------------------------------------+ withTransaction_ :: Con -> [Topt] -> (Con -> IO ()) -> IO ()+ withTransaction_ cid os op = do+ _ <- withTransaction cid os op+ return ()++ ------------------------------------------------------------------------+ -- | Starts a transaction and executes the action+ -- in the last parameter.+ -- After the action has finished, + -- the transaction will be either committed or aborted+ -- even if an exception was raised.+ -- Note that, depending on the options,+ -- the way a transaction is terminated may vary,+ -- refer to 'Topt' for details.+ --+ -- Transactions cannot be shared among threads.+ -- Transactions are internally protected against+ -- access from any thread that has not started the transaction.+ --+ -- It is /not/ advisable to use 'withTransaction' with /timeout/.+ -- It is preferred to use /timeout/ on the + -- the actions executed within this transaction.+ -- Whether and how much time the transaction itself+ -- shall wait for the completion of on-going interactions with the broker,+ -- in particular pending receipts,+ -- shall be controlled+ -- by the 'OTimeout' option.+ --+ -- 'withTransaction' returns the result of the action.+ --+ -- The simplest usage example with a 'Connection' /c/ is:+ --+ -- > r <- withTransaction c [] $ \_ -> do+ --+ -- If the transaction shall use receipts and, before terminating, wait 100/ms/+ -- for all receipts to be confirmed by the broker+ -- 'withTransaction' is called like:+ --+ -- > eiR <- try $ withTransaction c [OTimeout 100, OWithReceipts] \_ -> do+ -- > case eiR of+ -- > Left e -> -- error handling+ -- > Right x -> do -- ..+ --+ -- Note that 'try' is used to catch any 'StomplException'.+ ------------------------------------------------------------------------+ withTransaction :: Con -> [Topt] -> (Con -> IO a) -> IO a+ withTransaction cid os op = do+ tx <- mkUniqueTxId+ let t = mkTrn tx os+ c <- getCon cid+ if not $ P.connected (conCon c)+ then throwIO $ ConnectException $+ "Not connected (" ++ show cid ++ ")"+ else finally (do addTx t cid+ startTx cid c t + x <- op cid+ updTxState tx cid TxEnded+ return x)+ -- if an exception is raised in terminate+ -- we at least will remove the transaction+ -- from our state and then reraise + (terminateTx tx cid `onException` rmThisTx tx cid)+ + ------------------------------------------------------------------------+ -- | Waits for the 'Receipt' to be confirmed by the broker.+ -- Since the thread will preempt, the call should be protected+ -- with /timeout/, /e.g./:+ --+ -- > mb_ <- waitReceipt c r+ -- > case mb_ of+ -- > Nothing -> -- error handling+ -- > Just _ -> do -- ...+ ------------------------------------------------------------------------+ waitReceipt :: Con -> Receipt -> IO ()+ waitReceipt cid r =+ case r of+ NoRec -> return ()+ _ -> waitForMe + where waitForMe = do+ ok <- checkReceipt cid r+ unless ok $ do+ threadDelay $ ms 1+ waitForMe ++ ------------------------------------------------------------------------+ -- | Aborts the transaction immediately by raising 'AppException'.+ -- The string passed in to 'abort' will be added to the + -- exception message.+ ------------------------------------------------------------------------+ abort :: String -> IO ()+ abort e = throwIO $ AppException $+ "Tx aborted by application: " ++ e++ ------------------------------------------------------------------------+ -- Terminate the transaction appropriately+ -- either committing or aborting+ ------------------------------------------------------------------------+ terminateTx :: Tx -> Con -> IO ()+ terminateTx tx cid = do+ c <- getCon cid+ mbT <- getTx tx c+ case mbT of+ Nothing -> throwIO $ OuchException $ + "Transaction disappeared: " ++ show tx+ Just t | txState t /= TxEnded -> endTx False cid c tx t+ | txReceipts t || txPendingAck t -> do + ok <- waitTx tx cid $ txTmo t+ if ok+ then endTx True cid c tx t+ else do+ endTx False cid c tx t+ let m = if txPendingAck t then "Acks" else "Receipts"+ throwIO $ TxException $+ "Transaction aborted: Missing " ++ m+ | otherwise -> endTx True cid c tx t++ -----------------------------------------------------------------------+ -- Send begin frame+ -- We don't wait for the receipt now+ -- we will wait for receipts + -- on terminating the transaction anyway+ -----------------------------------------------------------------------+ startTx :: Con -> Connection -> Transaction -> IO ()+ startTx cid c t = do+ rc <- if txAbrtRc t then mkUniqueRecc else return NoRec+ when (txAbrtRc t) $ addRec cid rc + logSend cid+ P.begin (conCon c) (show $ txId t) (show rc)++ -----------------------------------------------------------------------+ -- Send commit or abort frame+ -- and, if we work with receipts,+ -- wait for the receipt+ -----------------------------------------------------------------------+ endTx :: Bool -> Con -> Connection -> Tx -> Transaction -> IO ()+ endTx x cid c tx t = do+ let w = txAbrtRc t && txTmo t > 0 + rc <- if w then mkUniqueRecc else return NoRec+ when w $ addRec cid rc + logSend cid+ if x then P.commit (conCon c) (show tx) (show rc)+ else P.abort (conCon c) (show tx) (show rc)+ mbR <- if w then timeout (ms $ txTmo t) $ waitReceipt cid rc+ else return $ Just ()+ rmTx cid+ case mbR of+ Just _ -> return ()+ Nothing -> throwIO $ TxException $+ "Transaction in unknown State: " +++ "missing receipt for " ++ (if x then "commit!" + else "abort!")++ -----------------------------------------------------------------------+ -- Check if there are pending acks,+ -- if so, already wrong,+ -- otherwise wait for receipts+ -----------------------------------------------------------------------+ waitTx :: Tx -> Con -> Int -> IO Bool+ waitTx tx cid delay = do+ c <- getCon cid+ mbT <- getTx tx c+ case mbT of+ Nothing -> return True+ Just t | txPendingAck t -> return False+ | txReceipts t -> + if delay <= 0 then return False+ else do+ threadDelay $ ms 1+ waitTx tx cid (delay - 1)+ | otherwise -> return True++ -----------------------------------------------------------------------+ -- Transform a frame into a message+ -- using the queue's application callback+ -----------------------------------------------------------------------+ frmToMsg :: Reader a -> F.Frame -> IO (P.Message a)+ frmToMsg q f = do+ let b = F.getBody f+ let conv = rFrom q+ let sid = if null (F.getSub f) || not (numeric $ F.getSub f)+ then NoSub else Sub $ read $ F.getSub f+ x <- conv (F.getMime f) (F.getLength f) (F.getHeaders f) b+ let m = P.mkMessage (P.MsgId $ F.getId f) sid+ (F.getDest f) + (F.getMime f)+ (F.getLength f)+ NoTx + b -- raw bytestring+ x -- converted context+ return m {P.msgHdrs = F.getHeaders f}++ -----------------------------------------------------------------------+ -- Connection listener+ -----------------------------------------------------------------------+ listen :: Con -> IO ()+ listen cid = forever $ do+ c <- getCon cid+ let cc = conCon c+ eiF <- catches (S.receive (P.getRc cc) (P.getSock cc) (P.conMax cc))+ [Handler (\e -> case e of+ ThreadKilled -> throwIO e+ _ -> return $ Left $ show e),+ Handler (\e -> return $ Left $ + show (e::SomeException))]+ case eiF of+ Left e -> throwToOwner c $ + ProtocolException $ "Receive Error: " ++ e+ Right f -> do+ logReceive cid+ case F.typeOf f of+ F.Message -> handleMessage cid f+ F.Error -> handleError cid f+ F.Receipt -> handleReceipt cid f+ F.HeartBeat -> handleBeat cid f+ _ -> throwToOwner c $ + ProtocolException $ "Unexpected Frame: " +++ show (F.typeOf f)++ -----------------------------------------------------------------------+ -- Handle Message Frame+ -----------------------------------------------------------------------+ handleMessage :: Con -> F.Frame -> IO ()+ handleMessage cid f = do+ c <- getCon cid+ case getCh c of+ Nothing -> throwToOwner c $ + ProtocolException $ "Unknown Queue: " ++ show f+ Just ch -> writeChan ch f+ where getCh c = let dst = F.getDest f+ sid = F.getSub f+ in if null sid+ then getDest dst c+ else if numeric sid+ then getSub (Sub $ read sid) c+ else Nothing ++ -----------------------------------------------------------------------+ -- Handle Error Frame+ -----------------------------------------------------------------------+ handleError :: Con -> F.Frame -> IO ()+ handleError cid f = do+ c <- getCon cid+ let r = if null (F.getReceipt f) then ""+ else " (" ++ F.getReceipt f ++ ")" + let e = F.getMsg f ++ r ++ ": " ++ U.toString (F.getBody f)+ throwToOwner c (BrokerException e) ++ -----------------------------------------------------------------------+ -- Throw to owner+ -- important: give the owner some time to react+ -- before continuing and - probably - + -- causing another exception+ -----------------------------------------------------------------------+ throwToOwner :: Connection -> StomplException -> IO ()+ throwToOwner c e = do+ throwTo (conOwner c) e+ threadDelay 10000++ -----------------------------------------------------------------------+ -- Handle Receipt Frame+ -----------------------------------------------------------------------+ handleReceipt :: Con -> F.Frame -> IO ()+ handleReceipt cid f = do+ c <- getCon cid+ case parseRec $ F.getReceipt f of+ Just r -> forceRmRec cid r + Nothing -> throwToOwner c $ + ProtocolException $ "Invalid Receipt: " ++ show f++ -----------------------------------------------------------------------+ -- Handle Beat Frame, i.e. ignore them+ -----------------------------------------------------------------------+ handleBeat :: Con -> F.Frame -> IO ()+ handleBeat _ _ = return () -- putStrLn "Beat!"++ -----------------------------------------------------------------------+ -- My Beat + -----------------------------------------------------------------------+ heartBeat :: Con -> Int -> IO ()+ heartBeat cid period = forever $ do+ now <- getCurrentTime+ c <- getCon cid+ let me = myMust c + let he = hisMust c + when (now > he) $ throwToOwner c $ + ProtocolException $ + "Missing HeartBeat, last was " +++ show (now `diffUTCTime` he) ++ + " seconds ago!"+ when (now >= me) $ P.sendBeat $ conCon c+ threadDelay $ ms period++ -----------------------------------------------------------------------+ -- When we should have sent last heartbeat+ -----------------------------------------------------------------------+ myMust :: Connection -> UTCTime+ myMust c = let t = conMyBeat c+ p = snd $ P.conBeat $ conCon c+ in timeAdd t p++ -----------------------------------------------------------------------+ -- When he should have sent last heartbeat+ -----------------------------------------------------------------------+ hisMust :: Connection -> UTCTime+ hisMust c = let t = conHisBeat c+ tol = 4+ b = fst $ P.conBeat $ conCon c+ p = tol * b+ in timeAdd t 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)+
+ Network/Mom/Stompl/Client/Socket.hs view
@@ -0,0 +1,172 @@+{-# Language CPP #-}+module Socket (Writer, Receiver, + initWriter, initReceiver,+ connect, disconnect, send, receive)+where++ import qualified Network.Socket as S+ import qualified Network.Socket.ByteString as BS+ import Network.BSD (getProtocolNumber)++ import qualified Data.ByteString.Char8 as B+ import qualified Data.ByteString.UTF8 as U++ import Network.Mom.Stompl.Parser (stompParser)+ import qualified Network.Mom.Stompl.Frame as F+ import Network.Mom.Stompl.Client.Exception++ import Control.Concurrent.MVar+ import Control.Applicative ((<$>))+ import Control.Monad (unless)+ import Control.Exception (throwIO, finally, SomeException)+ import qualified Control.Exception as Ex (try)++ import qualified Data.Attoparsec as A (Result(..), feed, parse)++ type Result = A.Result F.Frame++ maxStep :: Int+ maxStep = 100++ data Receiver = Receiver {+ getBuffer :: IO (Maybe B.ByteString),+ putBuffer :: B.ByteString -> IO ()+ }++ data Writer = Writer {+ lock :: IO (),+ release :: IO ()+ }++ get :: MVar B.ByteString -> IO (Maybe B.ByteString)+ get buf = do+ e <- isEmptyMVar buf+ if e+ then return Nothing+ else Just <$> takeMVar buf++ put :: MVar B.ByteString -> B.ByteString -> IO ()+ put buf s = do+ e <- isEmptyMVar buf+ if e + then putMVar buf s+ else do + _ <- takeMVar buf+ putMVar buf s++ initReceiver :: IO Receiver+ initReceiver = do+ buf <- newEmptyMVar+ return Receiver {+ getBuffer = get buf,+ putBuffer = put buf}++ lockSock :: MVar Bool -> IO ()+ lockSock l = putMVar l True++ releaseSock :: MVar Bool -> IO ()+ releaseSock l = do+ _ <- takeMVar l+ return ()++ initWriter :: IO Writer+ initWriter = do+ l <- newEmptyMVar+ return Writer {+ lock = lockSock l,+ release = releaseSock l+ }+ + connect :: String -> Int -> IO S.Socket+ connect host port = do+ let p = fromIntegral port :: S.PortNumber+ prot <- getProtocolNumber "tcp" + let hints = S.defaultHints {S.addrSocketType = S.Stream,+ S.addrProtocol = prot}+ adds <- S.getAddrInfo (Just hints) (Just host) Nothing+ tryConnect p adds++ tryConnect :: S.PortNumber -> [S.AddrInfo] -> IO S.Socket+ tryConnect _ [] = throwIO $ SocketException $ + "Give up: no more address info!"+ tryConnect p (i:is) = + case i of+ S.AddrInfo _ f t prot addr _ -> + let mbA = case addr of+ S.SockAddrInet _ a -> Just $ S.SockAddrInet p a+ S.SockAddrInet6 _ fl a s -> Just $ S.SockAddrInet6 p fl a s+ _ -> Nothing+ in case mbA of+ Nothing -> tryConnect p is+ Just a -> do+ sock <- S.socket f t prot+ eiS <- Ex.try $ S.connect sock a+ case eiS of+ Left e -> do+ putStrLn $ "Network.Mom.Stompl.Socket - " +++ "Warning: " ++ show (e::SomeException)+ tryConnect p is+ Right _ -> return sock++ disconnect :: S.Socket -> IO ()+ disconnect = S.sClose ++ send :: Writer -> S.Socket -> F.Frame -> IO ()+ send wr sock f = do+ let s = F.putFrame f+ lock wr+#ifdef _DEBUG+ putStrLn $ "Sending: " ++ U.toString s+#endif+ n <- finally (BS.send sock s)+ (release wr)+ unless (n == B.length s) $ throwIO $ SocketException $+ "Could not send complete buffer. " +++ "Bytes sent: " ++ show n++ receive :: Receiver -> S.Socket -> Int -> IO (Either String F.Frame)+ receive rec sock mx = handlePartial rec sock mx Nothing 0++ handlePartial :: Receiver -> S.Socket -> Int -> + Maybe Result -> Int -> IO (Either String F.Frame)+ handlePartial rec sock mx mbR step = do+ eiS <- getInput rec sock mx+ case eiS of + Left e -> return $ Left e+ Right s -> do+ let prs = case mbR of + Just r -> A.feed r+ _ -> A.parse stompParser+ case prs s of+ A.Fail _ _ e -> return $ Left $+ U.toString s ++ + ": " ++ e+ r@(A.Partial _) -> + if step > maxStep + then return $ Left "Message too long!" + else handlePartial rec sock mx (Just r) (step + 1)+ + A.Done str f -> do+ unless (B.null str) $ putBuffer rec str+ return $ Right f+ + getInput :: Receiver -> S.Socket -> Int -> IO (Either String B.ByteString)+ getInput rec sock mx = do+ mbB <- getBuffer rec+ case mbB of+ Just s -> +#ifdef _DEBUG+ do putStrLn $ "In Buffer: " ++ U.toString s+#endif+ return $ Right s+ Nothing -> do+ s <- BS.recv sock mx+ if B.null s + then return $ Left "Peer disconnected"+ else +#ifdef _DEBUG+ do putStrLn $ "Receiving: " ++ U.toString s+#endif+ if B.null s+ then getInput rec sock mx+ else return $ Right s
+ Network/Mom/Stompl/Client/State.hs view
@@ -0,0 +1,619 @@+{-# OPTIONS -fglasgow-exts -fno-cse #-}+module State (+ msgContent, numeric, ms,+ Connection(..),+ Copt(..),+ oHeartBeat, oMaxRecv,+ oAuth,+ Transaction(..),+ Topt(..), hasTopt, tmo,+ TxState(..),+ Receipt,+ mkConnection,+ logSend, logReceive,+ addCon, rmCon, getCon,+ addSub, addDest, getSub, getDest,+ rmSub, rmDest,+ mkTrn,+ addTx, getTx, rmTx, rmThisTx,+ getCurTx,+ updTxState,+ txPendingAck, txReceipts,+ addAck, rmAck, addRec, rmRec,+ forceRmRec,+ checkReceipt)+where++ import qualified Protocol as P+ import Factory ++ import qualified Network.Mom.Stompl.Frame as F+ import Network.Mom.Stompl.Client.Exception++ import Control.Concurrent + import Control.Exception (throwIO)++ import System.IO.Unsafe++ import Data.List (find, deleteBy, delete)+ import Data.Char (isDigit)+ import Data.Time.Clock++ ------------------------------------------------------------------------+ -- | Returns the content of the message in the format + -- produced by an in-bound converter+ ------------------------------------------------------------------------+ msgContent :: P.Message a -> a+ msgContent = P.msgCont++ ------------------------------------------------------------------------+ -- some helpers+ ------------------------------------------------------------------------+ numeric :: String -> Bool+ numeric = all isDigit++ ------------------------------------------------------------------------+ -- convert milliseconds to microseconds + ------------------------------------------------------------------------+ ms :: Int -> Int+ ms u = 1000 * u++ ------------------------------------------------------------------------+ -- compare two tuples by the fst+ ------------------------------------------------------------------------+ eq :: Eq a => (a, b) -> (a, b) -> Bool+ eq x y = fst x == fst y++ -- | Just a nicer word for 'Rec'+ type Receipt = Rec++ ------------------------------------------------------------------------+ -- Connection+ ------------------------------------------------------------------------+ data Connection = Connection {+ conId :: Con,+ conCon :: P.Connection,+ conOwner :: ThreadId,+ conHisBeat :: UTCTime,+ conMyBeat :: UTCTime, + conWait :: Int,+ conSubs :: [SubEntry],+ conDests :: [DestEntry], + conThrds :: [ThreadEntry],+ conErrs :: [F.Frame],+ conRecs :: [Receipt],+ conAcks :: [P.MsgId]}++ instance Eq Connection where+ c1 == c2 = conId c1 == conId c2++ -------------------------------------------------------------------------+ -- | Options passed to a connection+ -------------------------------------------------------------------------+ data Copt = + -- | Tells the connection to wait /n/ milliseconds for the 'Receipt' + -- sent with 'F.Disconnect' at the end of the session.+ -- The /Stomp/ protocol advises to request a receipt + -- and to wait for it before actually closing the + -- socket. Many brokers, however, do not + -- implement this feature (or implement it inappropriately,+ -- closing the connection immediately after having sent+ -- the receipt).+ -- 'withConnection', for this reason, ignores + -- the receipt by default and simply closes the socket+ -- after having sent the 'F.Disconnect' frame.+ -- If your broker shows a correct behaviour, + -- it is advisable to use this option.+ OWaitBroker Int |++ -- | The maximum size of TCP/IP packets.+ -- Indirectly, this options also defines the+ -- maximum message size which is /10 * maxReceive/.+ -- By default, the maximum packet size is 1024 bytes.+ OMaxRecv Int |++ -- | This option defines the client\'s bid+ -- for negotiating heart beats (see 'F.HeartBeat'). + -- By default, no heart beats are sent or accepted+ OHeartBeat (F.Heart) |++ -- | Authentication: user and password+ OAuth String String+ deriving (Eq, Show)++ ------------------------------------------------------------------------+ -- Same constructor+ ------------------------------------------------------------------------+ is :: Copt -> Copt -> Bool+ is (OWaitBroker _) (OWaitBroker _) = True+ is (OMaxRecv _) (OMaxRecv _) = True+ is (OHeartBeat _) (OHeartBeat _) = True+ is (OAuth _ _) (OAuth _ _) = True+ is _ _ = False++ noWait :: Int+ noWait = 0++ stdRecv :: Int+ stdRecv = 1024++ noBeat :: F.Heart+ noBeat = (0,0)++ noAuth :: (String, String)+ noAuth = ("","")++ oWaitBroker :: [Copt] -> Int+ oWaitBroker os = case find (is $ OWaitBroker 0) os of+ Just (OWaitBroker d) -> d+ _ -> noWait++ oMaxRecv :: [Copt] -> Int+ oMaxRecv os = case find (is $ OMaxRecv 0) os of+ Just (OMaxRecv i) -> i+ _ -> stdRecv++ oHeartBeat :: [Copt] -> F.Heart+ oHeartBeat os = case find (is $ OHeartBeat (0,0)) os of+ Just (OHeartBeat b) -> b+ _ -> noBeat++ oAuth :: [Copt] -> (String, String)+ oAuth os = case find (is $ OAuth "" "") os of+ Just (OAuth u p) -> (u, p)+ _ -> noAuth++ findCon :: Con -> [Connection] -> Maybe Connection+ findCon cid = find (\c -> conId c == cid)++ mkConnection :: Con -> P.Connection -> ThreadId -> UTCTime -> [Copt] -> Connection+ mkConnection cid c myself t os = Connection cid c myself t t (oWaitBroker os)+ [] [] [] [] [] []++ addAckToCon :: P.MsgId -> Connection -> Connection+ addAckToCon mid c = c {conAcks = mid : conAcks c} ++ rmAckFromCon :: P.MsgId -> Connection -> Connection+ rmAckFromCon mid c = c {conAcks = delete mid $ conAcks c}++ addRecToCon :: Receipt -> Connection -> Connection+ addRecToCon r c = c {conRecs = r : conRecs c}++ rmRecFromCon :: Receipt -> Connection -> Connection+ rmRecFromCon r c = c {conRecs = delete r $ conRecs c}++ checkReceiptCon :: Receipt -> Connection -> Bool+ checkReceiptCon r c = case find (== r) $ conRecs c of+ Nothing -> True+ Just _ -> False++ ------------------------------------------------------------------------+ -- Sub, Dest and Thread Entry+ ------------------------------------------------------------------------+ type SubEntry = (Sub, Chan F.Frame)+ type DestEntry = (String, Chan F.Frame)+ type ThreadEntry = (ThreadId, [Transaction])++ addSubToCon :: SubEntry -> Connection -> Connection+ addSubToCon s c = c {conSubs = s : conSubs c}++ getSub :: Sub -> Connection -> Maybe (Chan F.Frame)+ getSub sid c = lookup sid (conSubs c)++ rmSubFromCon :: SubEntry -> Connection -> Connection+ rmSubFromCon s c = c {conSubs = ss} + where ss = deleteBy eq s (conSubs c)++ addDestToCon :: DestEntry -> Connection -> Connection+ addDestToCon d c = c {conDests = d : conDests c}++ getDest :: String -> Connection -> Maybe (Chan F.Frame)+ getDest dst c = lookup dst (conDests c)++ rmDestFromCon :: DestEntry -> Connection -> Connection+ rmDestFromCon d c = c {conDests = ds}+ where ds = deleteBy eq d (conDests c)++ setHisTime :: UTCTime -> Connection -> Connection+ setHisTime t c = c {conHisBeat = t}++ setMyTime :: UTCTime -> Connection -> Connection+ setMyTime t c = c {conMyBeat = t}++ updCon :: Connection -> [Connection] -> [Connection]+ updCon c cs = c : delete c cs+ + ------------------------------------------------------------------------+ -- Transaction + ------------------------------------------------------------------------+ data Transaction = Trn {+ txId :: Tx,+ txState :: TxState,+ txTmo :: Int,+ txAbrtAck :: Bool,+ txAbrtRc :: Bool,+ txAcks :: [P.MsgId],+ txRecs :: [Receipt]+ }++ instance Eq Transaction where+ t1 == t2 = txId t1 == txId t2++ findTx :: Tx -> [Transaction] -> Maybe Transaction+ findTx tx = find (\x -> txId x == tx) ++ mkTrn :: Tx -> [Topt] -> Transaction+ mkTrn tx os = Trn {+ txId = tx,+ txState = TxStarted,+ txTmo = tmo os,+ txAbrtAck = hasTopt OAbortMissingAcks os,+ txAbrtRc = hasTopt OWithReceipts os,+ txAcks = [],+ txRecs = []}++ ------------------------------------------------------------------------+ -- | Options passed to a transaction.+ ------------------------------------------------------------------------+ data Topt = + -- | The timeout in milliseconds (not microseconds!)+ -- to wait for /pending receipts/.+ -- If receipts are pending, when the transaction+ -- is ready to terminate,+ -- and no timeout or a timeout /<= 0/ is given, + -- and the option 'OWithReceipts' + -- was passed to 'withTransaction',+ -- the transaction will be aborted with 'TxException';+ -- otherwise it will wait until all pending+ -- ineractions with the broker have terminated+ -- or the timeout has expired - whatever comes first.+ -- If the timeout expires first, 'TxException' is raised. + OTimeout Int + -- | This option has two effects:+ -- 1) Internal interactions of the transaction+ -- with the broker will request receipts;+ -- 2) before ending the transaction,+ -- the library will check for receipts+ -- that have not yet been confirmed by the broker+ -- (including receipts requested by user calls+ -- such as /writeQ/ or /ackWith/).+ --+ -- If receipts are pending, when the transaction+ -- is ready to terminate and 'OTimeout' with+ -- a value /> 0/ is given, the transaction will+ -- wait for pending receipts; otherwise+ -- the transaction will be aborted with 'TxException'.+ -- Note that it, usually, does not make sense to use+ -- this options without 'OTimeout',+ -- since it is in all probability that a receipt + -- has not yet been confirmed when the transaction terminates.+ | OWithReceipts + -- | If a message has been received from a + -- queue with 'OMode' option other + -- than 'F.Auto' and this message has not yet been+ -- acknowledged when the transaction is ready+ -- to terminate, the /ack/ is /missing/.+ -- With this option, the transaction + -- will not commit with missing /acks/,+ -- but abort and raise 'TxException'.+ | OAbortMissingAcks + deriving (Eq, Show)++ hasTopt :: Topt -> [Topt] -> Bool+ hasTopt o os = o `elem` os ++ tmo :: [Topt] -> Int+ tmo os = case find isTimeout os of+ Just (OTimeout i) -> i+ _ -> 0+ where isTimeout o = case o of+ OTimeout _ -> True+ _ -> False++ data TxState = TxStarted | TxEnded+ deriving (Eq, Show)++ setTxState :: TxState -> Transaction -> Transaction+ setTxState st t = t {txState = st}++ addAckToTx :: P.MsgId -> Transaction -> Transaction+ addAckToTx mid t = t {txAcks = mid : txAcks t}++ rmAckFromTx :: P.MsgId -> Transaction -> Transaction+ rmAckFromTx mid t = t {txAcks = delete mid $ txAcks t}++ addRecToTx :: Receipt -> Transaction -> Transaction+ addRecToTx r t = t {txRecs = r : txRecs t}++ rmRecFromTx :: Receipt -> Transaction -> Transaction+ rmRecFromTx r t = t {txRecs = delete r $ txRecs t}++ checkReceiptTx :: Receipt -> Transaction -> Bool+ checkReceiptTx r = not . (elem r) . txRecs ++ txPendingAck :: Transaction -> Bool+ txPendingAck t = txAbrtAck t && not (null $ txAcks t)++ txReceipts :: Transaction -> Bool+ txReceipts t = txAbrtRc t && not (null $ txRecs t) ++ ------------------------------------------------------------------------+ -- State + ------------------------------------------------------------------------+ {-# NOINLINE con #-}+ con :: MVar [Connection]+ con = unsafePerformIO $ newMVar []+ + ------------------------------------------------------------------------+ -- Add connection to state+ ------------------------------------------------------------------------+ addCon :: Connection -> IO ()+ addCon c = modifyMVar_ con $ \cs -> return (c:cs)++ ------------------------------------------------------------------------+ -- get connection from state+ ------------------------------------------------------------------------+ getCon :: Con -> IO Connection+ getCon cid = withCon cid $ \c -> return (c, c) ++ ------------------------------------------------------------------------+ -- remove connection from state+ ------------------------------------------------------------------------+ rmCon :: Con -> IO ()+ rmCon cid = modifyMVar_ con $ \cs -> + case findCon cid cs of+ Nothing -> return cs+ Just c -> + return $ delete c cs++ ------------------------------------------------------------------------+ -- Apply an action that may change a connection to the state+ ------------------------------------------------------------------------+ withCon :: Con -> (Connection -> IO (Connection, a)) -> IO a+ withCon cid op = modifyMVar con (\cs -> + case findCon cid cs of+ Nothing -> + throwIO $ ConnectException $+ "No such Connection: " ++ show cid+ Just c -> do+ (c', x) <- op c+ let cs' = updCon c' cs+ return (cs', x))++ ------------------------------------------------------------------------+ -- Log heart-beats+ ------------------------------------------------------------------------+ logTime :: Con -> (UTCTime -> Connection -> Connection) -> IO ()+ logTime cid f = + getCurrentTime >>= \t -> withCon cid (\c -> return (f t c, ()))++ logSend :: Con -> IO ()+ logSend cid = logTime cid setMyTime++ logReceive :: Con -> IO ()+ logReceive cid = logTime cid setHisTime++ ------------------------------------------------------------------------+ -- add and remove sub and dest+ ------------------------------------------------------------------------+ addSub :: Con -> SubEntry -> IO ()+ addSub cid s = withCon cid $ \c -> return (addSubToCon s c, ())++ addDest :: Con -> DestEntry -> IO ()+ addDest cid d = withCon cid $ \c -> return (addDestToCon d c, ())++ rmSub :: Con -> Sub -> IO ()+ rmSub cid sid = withCon cid rm+ where rm c = case getSub sid c of + Nothing -> return (c, ())+ Just ch -> return (rmSubFromCon (sid, ch) c, ())++ rmDest :: Con -> String -> IO ()+ rmDest cid dst = withCon cid rm+ where rm c = case getDest dst c of + Nothing -> return (c, ())+ Just ch -> return (rmDestFromCon (dst, ch) c, ())++ ------------------------------------------------------------------------+ -- add transaction to connection+ -- Note: transactions are kept per threadId+ ------------------------------------------------------------------------+ addTx :: Transaction -> Con -> IO ()+ addTx t cid = withCon cid $ \c -> do+ tid <- myThreadId+ case lookup tid (conThrds c) of+ Nothing -> + return (c {conThrds = [(tid, [t])]}, ())+ Just ts -> + return (c {conThrds = addTx2Thrds t tid (conThrds c) ts}, ())+ where addTx2Thrds tx tid ts trns = + (tid, tx : trns) : deleteBy eq (tid, trns) ts++ ------------------------------------------------------------------------+ -- get transaction from connection+ -- Note: transactions are kept per threadId+ ------------------------------------------------------------------------+ getTx :: Tx -> Connection -> IO (Maybe Transaction)+ getTx tx c = do+ tid <- myThreadId+ case lookup tid (conThrds c) of+ Nothing -> return Nothing+ Just ts -> return $ findTx tx ts++ ------------------------------------------------------------------------+ -- apply an action that may change a transaction to the state+ ------------------------------------------------------------------------+ updTx :: Tx -> Con -> (Transaction -> Transaction) -> IO ()+ updTx tx cid f = withCon cid $ \c -> do+ tid <- myThreadId+ case lookup tid (conThrds c) of+ Nothing -> return (c, ())+ Just ts -> + case findTx tx ts of+ Nothing -> return (c, ())+ Just t -> + let t' = f t+ in return (c {conThrds = + updTxInThrds t' tid (conThrds c) ts}, + ())+ where updTxInThrds t tid ts trns =+ (tid, t : delete t trns) : deleteBy eq (tid, trns) ts++ ------------------------------------------------------------------------+ -- update transaction state+ ------------------------------------------------------------------------+ updTxState :: Tx -> Con -> TxState -> IO ()+ updTxState tx cid st = updTx tx cid (setTxState st)++ ------------------------------------------------------------------------+ -- get current transaction for thread+ ------------------------------------------------------------------------+ getCurTx :: Connection -> IO (Maybe Tx)+ getCurTx c = do+ tid <- myThreadId+ case lookup tid (conThrds c) of+ Nothing -> return Nothing+ Just ts -> if null ts then return Nothing+ else return $ Just $ (txId . head) ts++ ------------------------------------------------------------------------+ -- apply a change of the current transaction + -- or the connection (if there is no transaction) to the state+ ------------------------------------------------------------------------+ updCurTx :: (Transaction -> Transaction) ->+ (Connection -> Connection) ->+ Connection -> IO (Connection, ())+ updCurTx onTx onCon c = do+ tid <- myThreadId+ case lookup tid $ conThrds c of+ Nothing -> return (onCon c, ())+ Just ts -> if null ts + then return (onCon c, ())+ else do+ let t = head ts+ let t' = onTx t + let ts' = t' : tail ts+ let c' = c {conThrds = + (tid, ts') : + deleteBy eq (tid, ts) (conThrds c)}+ return (c', ())++ ------------------------------------------------------------------------+ -- add a pending ack either to the current transaction+ -- or - if there is no transaction - to the connection+ ------------------------------------------------------------------------+ addAck :: Con -> P.MsgId -> IO ()+ addAck cid mid = do+ let toTx = addAckToTx mid+ let toCon = addAckToCon mid+ withCon cid $ updCurTx toTx toCon++ ------------------------------------------------------------------------+ -- remove a pending ack either from the current transaction+ -- or - if there is no transaction - from the connection+ ------------------------------------------------------------------------+ rmAck :: Con -> P.MsgId -> IO ()+ rmAck cid mid = do+ let fromTx = rmAckFromTx mid+ let fromCon = rmAckFromCon mid+ withCon cid $ updCurTx fromTx fromCon++ ------------------------------------------------------------------------+ -- add a pending receipt either to the current transaction+ -- or - if there is no transaction - to the connection+ ------------------------------------------------------------------------+ addRec :: Con -> Receipt -> IO ()+ addRec cid r = do+ let toTx = addRecToTx r+ let toCon = addRecToCon r+ withCon cid $ updCurTx toTx toCon++ ------------------------------------------------------------------------+ -- remove a pending receipt either from the current transaction+ -- or - if there is no transaction - from the connection+ ------------------------------------------------------------------------+ rmRec :: Con -> Receipt -> IO ()+ rmRec cid r = do+ let fromTx = rmRecFromTx r+ let fromCon = rmRecFromCon r+ withCon cid $ updCurTx fromTx fromCon++ ------------------------------------------------------------------------+ -- search for a receipt either in connection or transactions.+ -- this is used by the listener that is not in the thread list+ ------------------------------------------------------------------------+ forceRmRec :: Con -> Receipt -> IO ()+ forceRmRec cid r = withCon cid doRmRec + where doRmRec c = + case find (== r) $ conRecs c of+ Just _ -> return (rmRecFromCon r c, ())+ Nothing -> + let thrds = map rmRecFromThrd $ conThrds c+ in return (c {conThrds = thrds}, ()) + rmRecFromThrd (thrd, ts) = (thrd, map (rmRecFromTx r) ts)++ ------------------------------------------------------------------------+ -- check a condition either on the current transaction or+ -- - if there is not transaction - on the connection+ ------------------------------------------------------------------------+ checkCurTx :: (Transaction -> Bool) ->+ (Connection -> Bool) -> + Con -> IO Bool+ checkCurTx onTx onCon cid = do+ c <- getCon cid+ tid <- myThreadId+ case lookup tid $ conThrds c of+ Nothing -> return $ onCon c+ Just ts -> if null ts then return $ onCon c+ else return $ onTx $ head ts++ ------------------------------------------------------------------------+ -- check a receipt+ ------------------------------------------------------------------------+ checkReceipt :: Con -> Receipt -> IO Bool+ checkReceipt cid r = do+ let onTx = checkReceiptTx r+ let onCon = checkReceiptCon r+ checkCurTx onTx onCon cid++ ------------------------------------------------------------------------+ -- remove a specific transaction+ ------------------------------------------------------------------------+ rmThisTx :: Tx -> Con -> IO ()+ rmThisTx tx cid = withCon cid $ \c -> do+ tid <- myThreadId+ case lookup tid (conThrds c) of+ Nothing -> return (c, ())+ Just ts -> + if null ts + then return (c {conThrds = deleteBy eq (tid, []) (conThrds c)}, ())+ else + case findTx tx ts of+ Nothing -> return (c, ())+ Just t -> do+ let ts' = delete t ts+ if null ts' + then return (c {conThrds = + deleteBy eq (tid, ts) (conThrds c)}, ())+ else return (c {conThrds = (tid, ts') : + deleteBy eq (tid, ts) (conThrds c)}, ())++ ------------------------------------------------------------------------+ -- remove the current transaction+ ------------------------------------------------------------------------+ rmTx :: Con -> IO ()+ rmTx cid = withCon cid $ \c -> do+ tid <- myThreadId+ case lookup tid (conThrds c) of+ Nothing -> return (c, ())+ Just ts -> + if null ts + then return (c {conThrds = deleteBy eq (tid, []) (conThrds c)}, ())+ else do+ let ts' = tail ts+ if null ts' + then return (c {conThrds = + deleteBy eq (tid, ts) (conThrds c)}, ())+ else return (c {conThrds = (tid, ts') : + deleteBy eq (tid, ts) (conThrds c)}, ())+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ license/lgpl-3.0.txt view
@@ -0,0 +1,165 @@+ GNU LESSER GENERAL PUBLIC LICENSE+ Version 3, 29 June 2007++ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>+ Everyone is permitted to copy and distribute verbatim copies+ of this license document, but changing it is not allowed.+++ This version of the GNU Lesser General Public License incorporates+the terms and conditions of version 3 of the GNU General Public+License, supplemented by the additional permissions listed below.++ 0. Additional Definitions.++ As used herein, "this License" refers to version 3 of the GNU Lesser+General Public License, and the "GNU GPL" refers to version 3 of the GNU+General Public License.++ "The Library" refers to a covered work governed by this License,+other than an Application or a Combined Work as defined below.++ An "Application" is any work that makes use of an interface provided+by the Library, but which is not otherwise based on the Library.+Defining a subclass of a class defined by the Library is deemed a mode+of using an interface provided by the Library.++ A "Combined Work" is a work produced by combining or linking an+Application with the Library. The particular version of the Library+with which the Combined Work was made is also called the "Linked+Version".++ The "Minimal Corresponding Source" for a Combined Work means the+Corresponding Source for the Combined Work, excluding any source code+for portions of the Combined Work that, considered in isolation, are+based on the Application, and not on the Linked Version.++ The "Corresponding Application Code" for a Combined Work means the+object code and/or source code for the Application, including any data+and utility programs needed for reproducing the Combined Work from the+Application, but excluding the System Libraries of the Combined Work.++ 1. Exception to Section 3 of the GNU GPL.++ You may convey a covered work under sections 3 and 4 of this License+without being bound by section 3 of the GNU GPL.++ 2. Conveying Modified Versions.++ If you modify a copy of the Library, and, in your modifications, a+facility refers to a function or data to be supplied by an Application+that uses the facility (other than as an argument passed when the+facility is invoked), then you may convey a copy of the modified+version:++ a) under this License, provided that you make a good faith effort to+ ensure that, in the event an Application does not supply the+ function or data, the facility still operates, and performs+ whatever part of its purpose remains meaningful, or++ b) under the GNU GPL, with none of the additional permissions of+ this License applicable to that copy.++ 3. Object Code Incorporating Material from Library Header Files.++ The object code form of an Application may incorporate material from+a header file that is part of the Library. You may convey such object+code under terms of your choice, provided that, if the incorporated+material is not limited to numerical parameters, data structure+layouts and accessors, or small macros, inline functions and templates+(ten or fewer lines in length), you do both of the following:++ a) Give prominent notice with each copy of the object code that the+ Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the object code with a copy of the GNU GPL and this license+ document.++ 4. Combined Works.++ You may convey a Combined Work under terms of your choice that,+taken together, effectively do not restrict modification of the+portions of the Library contained in the Combined Work and reverse+engineering for debugging such modifications, if you also do each of+the following:++ a) Give prominent notice with each copy of the Combined Work that+ the Library is used in it and that the Library and its use are+ covered by this License.++ b) Accompany the Combined Work with a copy of the GNU GPL and this license+ document.++ c) For a Combined Work that displays copyright notices during+ execution, include the copyright notice for the Library among+ these notices, as well as a reference directing the user to the+ copies of the GNU GPL and this license document.++ d) Do one of the following:++ 0) Convey the Minimal Corresponding Source under the terms of this+ License, and the Corresponding Application Code in a form+ suitable for, and under terms that permit, the user to+ recombine or relink the Application with a modified version of+ the Linked Version to produce a modified Combined Work, in the+ manner specified by section 6 of the GNU GPL for conveying+ Corresponding Source.++ 1) Use a suitable shared library mechanism for linking with the+ Library. A suitable mechanism is one that (a) uses at run time+ a copy of the Library already present on the user's computer+ system, and (b) will operate properly with a modified version+ of the Library that is interface-compatible with the Linked+ Version.++ e) Provide Installation Information, but only if you would otherwise+ be required to provide such information under section 6 of the+ GNU GPL, and only to the extent that such information is+ necessary to install and execute a modified version of the+ Combined Work produced by recombining or relinking the+ Application with a modified version of the Linked Version. (If+ you use option 4d0, the Installation Information must accompany+ the Minimal Corresponding Source and Corresponding Application+ Code. If you use option 4d1, you must provide the Installation+ Information in the manner specified by section 6 of the GNU GPL+ for conveying Corresponding Source.)++ 5. Combined Libraries.++ You may place library facilities that are a work based on the+Library side by side in a single library together with other library+facilities that are not Applications and are not covered by this+License, and convey such a combined library under terms of your+choice, if you do both of the following:++ a) Accompany the combined library with a copy of the same work based+ on the Library, uncombined with any other library facilities,+ conveyed under the terms of this License.++ b) Give prominent notice with the combined library that part of it+ is a work based on the Library, and explaining where to find the+ accompanying uncombined form of the same work.++ 6. Revised Versions of the GNU Lesser General Public License.++ The Free Software Foundation may publish revised and/or new versions+of the GNU Lesser General Public License from time to time. Such new+versions will be similar in spirit to the present version, but may+differ in detail to address new problems or concerns.++ Each version is given a distinguishing version number. If the+Library as you received it specifies that a certain numbered version+of the GNU Lesser General Public License "or any later version"+applies to it, you have the option of following the terms and+conditions either of that published version or of any later version+published by the Free Software Foundation. If the Library as you+received it does not specify a version number of the GNU Lesser+General Public License, you may choose any version of the GNU Lesser+General Public License ever published by the Free Software Foundation.++ If the Library as you received it specifies that a proxy can decide+whether future versions of the GNU Lesser General Public License shall+apply, that proxy's public statement of acceptance of any version is+permanent authorization for you to choose that version for the+Library.
+ stomp-queue.cabal view
@@ -0,0 +1,46 @@+Name: stomp-queue+Version: 0.0.1+Cabal-Version: >= 1.8+Copyright: Copyright (c) Tobias Schoofs, 2011+License: LGPL+license-file: license/lgpl-3.0.txt+Author: Tobias Schoofs+Maintainer: tobias.schoofs@gmx.net+Homepage: http://github.com/toschoo/stompl+Category: Network, Message-oriented Middleware, Stomp, Client+Build-Type: Simple+Synopsis: Stompl Client Library +Description:+ The Stomp Protocol specifies message-oriented interoperability.+ Applications connect to a message broker to send (publish)+ or receive (subscribe) messages through queues. + Interoperating applications do not know + 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 <http://github.com/toschoo/stompl>.+ The Stomp specification can be found at+ <http://stomp.github.com>.++Library+ Build-Depends: base >= 4.0 && <= 5.0,+ bytestring >= 0.9 && < 0.10,+ utf8-string >= 0.3.6,+ attoparsec >= 0.9.1.1,+ split >= 0.1.4.1,+ network >= 2.3.0.4,+ stompl >= 0.0.1,+ mime >= 0.3.3,+ time >= 1.1.4++ hs-source-dirs: Network/Mom/Stompl/Client, .+ + Exposed-Modules: Network.Mom.Stompl.Client.Queue, Network.Mom.Stompl.Client.Exception+ other-modules: Socket, Protocol, State, Factory+