------------------------------------------------------------------------------- -- | -- 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)