diff --git a/Network/Mom/Stompl/Client/Factory.hs b/Network/Mom/Stompl/Client/Factory.hs --- a/Network/Mom/Stompl/Client/Factory.hs +++ b/Network/Mom/Stompl/Client/Factory.hs @@ -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) diff --git a/Network/Mom/Stompl/Client/Protocol.hs b/Network/Mom/Stompl/Client/Protocol.hs --- a/Network/Mom/Stompl/Client/Protocol.hs +++ b/Network/Mom/Stompl/Client/Protocol.hs @@ -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 diff --git a/Network/Mom/Stompl/Client/Queue.hs b/Network/Mom/Stompl/Client/Queue.hs --- a/Network/Mom/Stompl/Client/Queue.hs +++ b/Network/Mom/Stompl/Client/Queue.hs @@ -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 diff --git a/Network/Mom/Stompl/Client/Socket.hs b/Network/Mom/Stompl/Client/Socket.hs --- a/Network/Mom/Stompl/Client/Socket.hs +++ b/Network/Mom/Stompl/Client/Socket.hs @@ -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 diff --git a/Network/Mom/Stompl/Client/State.hs b/Network/Mom/Stompl/Client/State.hs --- a/Network/Mom/Stompl/Client/State.hs +++ b/Network/Mom/Stompl/Client/State.hs @@ -1,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 diff --git a/stomp-queue.cabal b/stomp-queue.cabal --- a/stomp-queue.cabal +++ b/stomp-queue.cabal @@ -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 . + . + + 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