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