stomp-queue 0.0.3 → 0.0.5
raw patch · 6 files changed
+288/−115 lines, 6 filesdep ~attoparsecdep ~stompl
Dependency ranges changed: attoparsec, stompl
Files
- Network/Mom/Stompl/Client/Factory.hs +7/−0
- Network/Mom/Stompl/Client/Protocol.hs +37/−27
- Network/Mom/Stompl/Client/Queue.hs +170/−63
- Network/Mom/Stompl/Client/Socket.hs +20/−15
- Network/Mom/Stompl/Client/State.hs +22/−6
- stomp-queue.cabal +32/−4
Network/Mom/Stompl/Client/Factory.hs view
@@ -5,6 +5,13 @@ Rec(..), mkUniqueRecc, parseRec) where + ------------------------------------------------------------------------+ -- To-do:+ -- - Currently, we use running numbers to uniquely identify + -- subscription ids, receipts, transactions, etc.+ -- - A better approach is to use random numbers+ ------------------------------------------------------------------------+ import System.IO.Unsafe import Control.Concurrent import Data.Char (isDigit)
Network/Mom/Stompl/Client/Protocol.hs view
@@ -10,8 +10,6 @@ Message(..), mkMessage, MsgId(..), send, ack, nack)- - where import qualified Socket as S@@ -36,7 +34,7 @@ -- Default version, when broker does not send a version --------------------------------------------------------------------- defVersion :: F.Version- defVersion = (1,0)+ defVersion = (1,1) --------------------------------------------------------------------- -- Connection @@ -48,10 +46,11 @@ -- the agreed version -- after connect conBeat :: F.Heart, -- the heart beat- conSrv :: String, -- server descritpion+ conSrv :: String, -- server description conSes :: String, -- session identifier (from broker) conUsr :: String, -- user conPwd :: String, -- passcode+ conCli :: String, -- client-id conMax :: Int, -- max receive conSock :: Maybe Socket, -- if connected: socket conRcv :: Maybe S.Receiver, -- if connected: receiver@@ -134,8 +133,10 @@ --------------------------------------------------------------------- -- Make a connection ---------------------------------------------------------------------- mkConnection :: String -> Int -> Int -> String -> String -> [F.Version] -> F.Heart -> Connection- mkConnection host port mx usr pwd vers beat = + mkConnection :: String -> Int -> Int -> + String -> String -> String -> + [F.Version] -> F.Heart -> Connection+ mkConnection host port mx usr pwd cli vers beat = Connection { conAddr = host, conPort = port,@@ -145,6 +146,7 @@ conSes = "", conUsr = usr, conPwd = pwd,+ conCli = cli, conMax = mx, conSock = Nothing, conRcv = Nothing,@@ -185,14 +187,16 @@ --------------------------------------------------------------------- -- connect ---------------------------------------------------------------------- connect :: String -> Int -> Int -> String -> String -> [F.Version] -> F.Heart -> IO Connection- connect host port mx usr pwd vers beat = + connect :: String -> Int -> Int -> + String -> String -> String -> + [F.Version] -> F.Heart -> [F.Header] -> IO Connection+ connect host port mx usr pwd cli vers beat hs = bracketOnError (do s <- S.connect host port let c = mkConnection host port mx - usr pwd vers beat+ usr pwd cli vers beat return c {conSock = Just s, conTcp = True}) disc- (connectBroker mx vers beat) + (connectBroker mx vers beat hs) --------------------------------------------------------------------- -- disconnect either on broker level or on tcp/ip level@@ -299,15 +303,18 @@ --------------------------------------------------------------------- -- 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+ connectBroker :: Int -> [F.Version] -> F.Heart -> [F.Header] ->+ Connection -> IO Connection+ connectBroker mx vers beat hs c = + case mkConF (conAddr c) + (conUsr c) (conPwd c) + (conCli c) vers beat hs 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)+ eiC <- catch (S.receive rc (getSock c) mx) -- no timeout? (\e -> return $ Left $ show (e::SomeException)) case eiC of Left e -> return c {conErrM = e}@@ -321,30 +328,30 @@ where period = snd . conBeat ---------------------------------------------------------------------- -- handle broker responds connect frame+ -- handle broker response to 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 ++ ")",+ 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)}+ _ -> 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+ errToMsg f = F.getMsg f ++ if B.length (F.getBody f) == 0 + then "."+ else ": " ++ U.toString (F.getBody f) --------------------------------------------------------------------- -- frame constructors@@ -353,14 +360,17 @@ 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 = ++ mkConF :: String -> String -> String -> String -> + [F.Version] -> F.Heart -> [F.Header] -> Either String F.Frame+ mkConF host usr pwd cli vers beat hs = let uHdr = if null usr then [] else [F.mkLogHdr usr] pHdr = if null pwd then [] else [F.mkPassHdr pwd]+ cHdr = if null cli then [] else [F.mkCliIdHdr cli] in F.mkConFrame $ [F.mkHostHdr host, F.mkAcVerHdr $ F.versToVal vers, F.mkBeatHdr $ F.beatToVal beat] ++- uHdr ++ pHdr+ uHdr ++ pHdr ++ cHdr ++ hs mkDiscF :: String -> Either String F.Frame mkDiscF receipt =@@ -382,7 +392,7 @@ 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 + (msgType msg) (msgLen msg) hs -- escape headers! (msgRaw msg) mkAckF :: Bool -> Message a -> String -> [F.Header] -> Either String F.Frame
Network/Mom/Stompl/Client/Queue.hs view
@@ -23,7 +23,6 @@ -- * Connections -- $stomp_con withConnection, - withConnection_, Factory.Con, F.Heart, Copt(..),@@ -31,11 +30,12 @@ -- $stomp_queues Reader, Writer, newReader, newWriter, - withReader, withReader_,+ withReader, withWriter, withPair, ReaderDesc, WriterDesc, Qopt(..), F.AckMode(..), InBound, OutBound, readQ, writeQ, writeQWith,+ writeAdHoc, writeAdHocWith, -- * Messages P.Message, msgContent, P.msgRaw, @@ -46,9 +46,9 @@ waitReceipt, -- * Transactions -- $stomp_trans+ Tx, withTransaction,- withTransaction_,- Topt(..), abort,+ Topt(..), abort, -- * Acknowledgments -- $stomp_acks ack, ackWith, nack, nackWith,@@ -188,17 +188,17 @@ {- $stomp_receipts Receipts are identifiers unique during the life time- of an application that can be added to all kinds of+ of an application; receipts 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.+ The broker, in its turn, uses receipts to acknowledge received messages.+ Receipts, hence, are 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.+ to the subscriber(s), depends on the broker. - Receipts are usually handled internally by the library.+ Receipts are 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@@ -206,10 +206,12 @@ On sending messages, receipt handling can be made explict. The function 'writeQWith'- adds a receipt to the message+ requests a receipt to the message and returns it to the caller. The application can then, later, explicitly wait for the receipt, using 'waitReceipt'.+ Otherwise, receipt handling remains+ inivisible in the application code. -} {- $stomp_trans@@ -296,7 +298,7 @@ > > ping :: String -> IO () > ping qn = - > withConnection_ "localhost" 61613 [] $ \c -> do+ > withConnection "localhost" 61613 [] [] $ \c -> do > let iconv _ _ _ = strToPing . U.toString > let oconv = return . U.fromString . show > inQ <- newReader c "Q-IN" qn [] [] iconv@@ -325,14 +327,6 @@ 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@@ -342,6 +336,8 @@ -- not to terminate the action before all other threads -- working on the connection have finished. --+ -- There should be only one connection to the same broker per process.+ -- -- Paramter: -- -- * 'String': The broker's hostname or IP-address@@ -349,7 +345,10 @@ -- * 'Int': The broker's port -- -- * 'Copt': Control options passed to the connection+ -- (including user/password) --+ -- * 'Header': List of additional, broker-specific headers+ -- -- * ('Con' -> 'IO' a): The action to execute. -- The action receives the connection handle -- and returns a value of type /a/ @@ -362,7 +361,7 @@ -- -- Example: --- -- > withConnection "localhost" 61613 [] $ \c -> do+ -- > withConnection "localhost" 61613 [] [] $ \c -> do -- -- This would connect to a broker listening to the loopback interface, -- port number 61613.@@ -379,15 +378,16 @@ -- -- Example: --- -- > t <- forkIO $ withConnection_ "127.0.0.1" 61613 [] $ \c -> do+ -- > t <- forkIO $ withConnection "127.0.0.1" 61613 [] [] $ \c -> do ------------------------------------------------------------------------- withConnection :: String -> Int -> [Copt] -> + withConnection :: String -> Int -> [Copt] -> [F.Header] -> (Con -> IO a) -> IO a- withConnection host port os act = do+ withConnection host port os hs 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)+ let (ci) = oCliId os+ bracket (P.connect host port mx u p ci vers beat hs) P.disc -- important: At the end, we close the socket! (whenConnected os act) @@ -481,10 +481,13 @@ 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 + -- The handling of receipts is usually transparent to applications, + -- but, in the case of sending message, may be made visible + -- by using 'writeQWith' instead of 'writeQ'.+ -- 'writeQWith' return the receipt identifier+ -- and the application can later invoke 'waitReceipt'+ -- to wait for the broker confirming this receipt.+ -- Note that 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@@ -497,14 +500,14 @@ -- 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+ -- method. You may want to fire and forget - and check + -- for the confirmation of the receipt only later.+ -- 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+ -- If you want to request a receipt with a message -- and wait for the broker to confirm it, you have to use -- both options. --@@ -525,8 +528,9 @@ -- 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+ -- Furthermore, if 'OMode' is either+ -- 'F.Client' or 'F.ClientIndi', then + -- this option forces 'readQ' to send an acknowledgment -- automatically when a message has been read from the queue. | OAck -- | A queue created with 'OForceTx' will throw @@ -696,24 +700,100 @@ -- -- > 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 :: Con -> String -> + String -> [Qopt] -> [F.Header] -> + InBound i -> (Reader i -> IO r) -> IO r 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.+ -- | Creates a 'Writer' 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 writers+ -- that are used only temporarly, /e.g./ during initialisation.+ --+ -- 'withWriter' 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 writer created by 'withWriter'+ -- outside the action will raise a 'QueueException'. ------------------------------------------------------------------------- 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 ())+ withWriter :: Con -> String -> + String -> [Qopt] -> [F.Header] -> + OutBound o -> (Writer o -> IO r) -> IO r+ withWriter cid qn dst os hs conv act = + newWriter cid qn dst os hs conv >>= act ------------------------------------------------------------------------- -- Creating a SendQ is plain an simple.+ -- | Creates a pair of ('Reader' i, 'Writer' o) with limited lifetime. + -- The pair will live only in the scope of the action+ -- that is passed as last parameter. + -- The function is useful for readers\/writers+ -- used in combination, /e.g./ to emulate a client\/server+ -- kind of communication.+ --+ -- 'withPair' returns the result of the action passed in.+ --+ -- The parameters are:+ --+ -- * The connection handle 'Con'+ --+ -- * The name of the pair; + -- the reader will be identified by a string+ -- with \"_r\" added to this name,+ -- the writer by a string with \"_w\" added to this name.+ --+ -- * The description of the 'Reader', 'ReaderDesc'+ --+ -- * The description of the 'Writer', 'WriterDesc'+ --+ -- * The application-defined action+ --+ -- The reason for introducing the reader and writer description+ -- is to provide error detection at compile time:+ -- It is this way much more difficult to accidently confuse+ -- the writer's and the reader's parameters (/e.g./ + -- passing the writer's 'Qopt's to the reader). ------------------------------------------------------------------------+ withPair :: Con -> String -> ReaderDesc i -> + WriterDesc o ->+ ((Reader i, Writer o) -> IO r) -> IO r+ withPair cid n (rq,ro,rh,iconv)+ (wq,wo,wh,oconv) act = + withReader cid (n ++ "_r") rq ro rh iconv $ \r ->+ withWriter cid (n ++ "_w") wq wo wh oconv $ \w -> act (r,w)++ ------------------------------------------------------------------------+ -- | The 'Reader' parameters of 'withPair':+ --+ -- * The reader's queue name + --+ -- * The reader's 'Qopt's+ --+ -- * The reader's 'Header's+ --+ -- * The reader's (inbound) converter+ ------------------------------------------------------------------------+ type ReaderDesc i = (String, [Qopt], [F.Header], InBound i)++ ------------------------------------------------------------------------+ -- | The 'Writer' parameters of 'withPair'+ --+ -- * The writer's queue name+ --+ -- * The writer's 'Qopt's+ --+ -- * The writer's 'Header's+ --+ -- * The writer's (outbound) converter+ ------------------------------------------------------------------------+ type WriterDesc o = (String, [Qopt], [F.Header], OutBound o)++ ------------------------------------------------------------------------+ -- Creating a SendQ is plain and simple.+ ------------------------------------------------------------------------ newSendQ :: Con -> String -> String -> [Qopt] -> OutBound a -> IO (Writer a) newSendQ cid qn dst os conv = @@ -866,6 +946,21 @@ writeQWith q mime hs x >>= (\_ -> return ()) ------------------------------------------------------------------------+ -- | This is a variant of 'writeQ'+ -- that overwrites the destination queue defined in the writer queue.+ -- It can be used for /ad hoc/ communication and+ -- for emulations of client/server-like protocols:+ -- the client would pass the name of the queue+ -- where it expects the server response in a header;+ -- the server would send the resply to the queue+ -- indicated in the header using 'writeAdHoc'.+ -- The additional 'String' parameter contains the destination.+ ------------------------------------------------------------------------+ writeAdHoc :: Writer a -> String -> Mime.Type -> [F.Header] -> a -> IO ()+ writeAdHoc q dest mime hs x =+ writeGeneric q dest mime hs x >>= (\_ -> return ())++ ------------------------------------------------------------------------ -- | This is a variant of 'writeQ' -- that is particularly useful for queues -- created with 'OWithReceipt', but without 'OWaitReceipt'.@@ -884,7 +979,25 @@ -- > r <- writeQWith q nullType [] "hello world!" ------------------------------------------------------------------------ writeQWith :: Writer a -> Mime.Type -> [F.Header] -> a -> IO Receipt- writeQWith q mime hs x = do+ writeQWith q = writeGeneric q (wDest q) ++ ------------------------------------------------------------------------+ -- | This is a variant of 'writeAdHoc' + -- 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'.+ -- Please refer to 'writeQWith' for more details.+ ------------------------------------------------------------------------+ writeAdHocWith :: Writer a -> String -> Mime.Type -> [F.Header] -> a -> IO Receipt+ writeAdHocWith = writeGeneric + + ------------------------------------------------------------------------+ -- internal work horse+ ------------------------------------------------------------------------+ writeGeneric :: Writer a -> String -> + Mime.Type -> [F.Header] -> a -> IO Receipt+ writeGeneric q dest mime hs x = do c <- getCon (wCon q) if not $ P.connected (conCon c) then throwIO $ ConnectException $@@ -902,7 +1015,7 @@ 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) + let m = P.mkMessage P.NoMsg NoSub dest mime (B.length s) tx s x when (wRec q) $ addRec (wCon q) rc logSend $ wCon q@@ -988,26 +1101,19 @@ 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.+ -- even if an exception has been 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.+ -- access from any thread but the one+ -- that has actually started the transaction. -- -- It is /not/ advisable to use 'withTransaction' with /timeout/. -- It is preferred to use /timeout/ on the @@ -1035,7 +1141,7 @@ -- -- Note that 'try' is used to catch any 'StomplException'. ------------------------------------------------------------------------- withTransaction :: Con -> [Topt] -> (Con -> IO a) -> IO a+ withTransaction :: Con -> [Topt] -> (Tx -> IO a) -> IO a withTransaction cid os op = do tx <- mkUniqueTxId let t = mkTrn tx os@@ -1045,20 +1151,20 @@ "Not connected (" ++ show cid ++ ")" else finally (do addTx t cid startTx cid c t - x <- op cid+ x <- op tx 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)+ (terminateTx cid tx `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+ -- > mb_ <- timeout tmo $ waitReceipt c r -- > case mb_ of -- > Nothing -> -- error handling -- > Just _ -> do -- ...@@ -1087,13 +1193,14 @@ -- Terminate the transaction appropriately -- either committing or aborting ------------------------------------------------------------------------- terminateTx :: Tx -> Con -> IO ()- terminateTx tx cid = do+ terminateTx :: Con -> Tx -> IO ()+ terminateTx cid tx = do c <- getCon cid mbT <- getTx tx c case mbT of- Nothing -> throwIO $ OuchException $ - "Transaction disappeared: " ++ show tx+ Nothing -> putStrLn "Transaction terminated!" -- return ()+ -- 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
Network/Mom/Stompl/Client/Socket.hs view
@@ -17,13 +17,15 @@ import Control.Concurrent.MVar import Control.Applicative ((<$>))- import Control.Monad (unless)+ import Control.Monad (unless, when) import Control.Exception (throwIO, finally, SomeException) import qualified Control.Exception as Ex (try) import qualified Data.Attoparsec.ByteString as A ( Result, IResult(..), feed, parse) + import System.IO (stderr, hPutStrLn)+ type Result = A.Result F.Frame maxStep :: Int@@ -62,13 +64,11 @@ getBuffer = get buf, putBuffer = put buf} - lockSock :: MVar Bool -> IO ()- lockSock l = putMVar l True+ lockSock :: MVar () -> IO ()+ lockSock l = putMVar l () - releaseSock :: MVar Bool -> IO ()- releaseSock l = do- _ <- takeMVar l- return ()+ releaseSock :: MVar () -> IO ()+ releaseSock = takeMVar initWriter :: IO Writer initWriter = do@@ -80,7 +80,7 @@ connect :: String -> Int -> IO S.Socket connect host port = do- let p = fromIntegral port :: S.PortNumber+ let p = fromIntegral port :: S.PortNumber prot <- getProtocolNumber "tcp" let hints = S.defaultHints {S.addrSocketType = S.Stream, S.addrProtocol = prot}@@ -104,8 +104,8 @@ eiS <- Ex.try $ S.connect sock a case eiS of Left e -> do- putStrLn $ "Network.Mom.Stompl.Socket - " ++- "Warning: " ++ show (e::SomeException)+ hPutStrLn stderr $ "Network.Mom.Stompl.Socket - " +++ "Warning: " ++ show (e::SomeException) tryConnect p is Right _ -> return sock @@ -119,11 +119,16 @@ #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+ finally (resend sock s) (release wr)++ resend :: S.Socket -> B.ByteString -> IO ()+ resend s b | B.length b == 0 = return ()+ | otherwise = do+ let t = B.length b+ n <- BS.send s b+ if n == 0+ then throwIO $ SocketException "Could not send buffer -- unknown error"+ else when (n < t) $ resend s $ B.drop n b receive :: Receiver -> S.Socket -> Int -> IO (Either String F.Frame) receive rec sock mx = handlePartial rec sock mx Nothing 0
Network/Mom/Stompl/Client/State.hs view
@@ -1,9 +1,10 @@+{-# Language BangPatterns #-} module State ( msgContent, numeric, ms, Connection(..), Copt(..), oHeartBeat, oMaxRecv,- oAuth,+ oAuth, oCliId, Transaction(..), Topt(..), hasTopt, tmo, TxState(..),@@ -117,7 +118,11 @@ OHeartBeat (F.Heart) | -- | Authentication: user and password- OAuth String String+ OAuth String String |++ -- | Identification: specifies the JMS Client ID for persistant connections+ OClientId String+ deriving (Eq, Show) ------------------------------------------------------------------------@@ -128,6 +133,7 @@ is (OMaxRecv _) (OMaxRecv _) = True is (OHeartBeat _) (OHeartBeat _) = True is (OAuth _ _) (OAuth _ _) = True+ is (OClientId _) (OClientId _) = True is _ _ = False noWait :: Int@@ -142,6 +148,9 @@ noAuth :: (String, String) noAuth = ("","") + noCliId :: String+ noCliId = ""+ oWaitBroker :: [Copt] -> Int oWaitBroker os = case find (is $ OWaitBroker 0) os of Just (OWaitBroker d) -> d@@ -162,6 +171,11 @@ Just (OAuth u p) -> (u, p) _ -> noAuth + oCliId :: [Copt] -> String+ oCliId os = case find (is $ OClientId "") os of+ Just (OClientId i) -> i+ _ -> noCliId+ findCon :: Con -> [Connection] -> Maybe Connection findCon cid = find (\c -> conId c == cid) @@ -220,7 +234,7 @@ setMyTime t c = c {conMyBeat = t} updCon :: Connection -> [Connection] -> [Connection]- updCon c cs = c : delete c cs+ updCon c cs = let !cs' = delete c cs in c:cs' ------------------------------------------------------------------------ -- Transaction @@ -451,12 +465,14 @@ case findTx tx ts of Nothing -> return (c, ()) Just t -> - let t' = f 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+ let !trns' = delete t trns+ !ts' = deleteBy eq (tid, trns) ts+ in (tid, t : trns') : ts' ------------------------------------------------------------------------ -- update transaction state@@ -539,7 +555,7 @@ ------------------------------------------------------------------------ -- search for a receipt either in connection or transactions.- -- this is used by the listener that is not in the thread list+ -- this is used by the listener (which is not in the thread list) ------------------------------------------------------------------------ forceRmRec :: Con -> Receipt -> IO () forceRmRec cid r = withCon cid doRmRec
stomp-queue.cabal view
@@ -1,5 +1,5 @@ Name: stomp-queue-Version: 0.0.3+Version: 0.0.5 Cabal-Version: >= 1.8 Copyright: Copyright (c) Tobias Schoofs, 2011 - 2013 License: LGPL@@ -28,19 +28,47 @@ The Stomp specification can be found at <http://stomp.github.com>. + .++ Release History:++ .++ [0.0.5] Underline functions removed, + with* functions: withWriter, withPair,+ new option for connection (ClientId),+ Headers for broker-specific options can be passed to connection,+ memory leaks solved.++ .++ [0.0.3] New interface writeAdHoc++ .++ [0.0.2] Minor corrections+ + .++ [0.0.1] Initial release++ Library Build-Depends: base >= 4.0 && <= 5.0, bytestring >= 0.9 && < 0.10, utf8-string >= 0.3.6,- attoparsec >= 0.10.4.0,+ attoparsec >= 0.9.1.1, split >= 0.1.4.1, network >= 2.3.0.4,- stompl >= 0.0.2,+ stompl >= 0.0.3, 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+ Exposed-Modules: Network.Mom.Stompl.Client.Queue, + Network.Mom.Stompl.Client.Exception+ -- Network.Mom.Stompl.Client.Client,+ -- Network.Mom.Stompl.Client.Server other-modules: Socket, Protocol, State, Factory