stomp-queue 0.1.4 → 0.2.0
raw patch · 7 files changed
+840/−868 lines, 7 filesdep +conduitdep +conduit-extradep +mtldep −network
Dependencies added: conduit, conduit-extra, mtl, network-conduit-tls
Dependencies removed: network
Files
- Network/Mom/Stompl/Client/Exception.hs +20/−8
- Network/Mom/Stompl/Client/Protocol.hs +0/−428
- Network/Mom/Stompl/Client/Queue.hs +419/−171
- Network/Mom/Stompl/Client/Socket.hs +0/−178
- Network/Mom/Stompl/Client/State.hs +242/−67
- Network/Mom/Stompl/Client/Stream.hs +119/−0
- stomp-queue.cabal +40/−16
Network/Mom/Stompl/Client/Exception.hs view
@@ -7,26 +7,38 @@ -- Stability : experimental -- Portability: portable ----- Exceptions for the Stompl Client+-- Exceptions for the Stompl Client.+-- Note that exceptions thrown in internal worker threads+-- (sender and listener) will be forwarded to the connection owner,+-- that is the thread that actually initialised the connection.+-- Since, in some circumstances, several exceptions may be thrown+-- in response to one error event (/e.g./ the broker sends an error frame+-- and, immediately afterwards, closes the connection),+-- the connection owner should implement a robust exception handling mechanism. ------------------------------------------------------------------------------- module Network.Mom.Stompl.Client.Exception ( StomplException(..), try, convertError) where import Control.Exception hiding (try)- import Prelude hiding (catch) import Control.Applicative ((<$>)) import Data.Typeable (Typeable) ------------------------------------------------------------------------- -- | The Stompl Client uses exceptions to communicate errors- -- to the user application.+ -- The Stompl Exception ------------------------------------------------------------------------ data StomplException =- -- | Thrown- -- on problems with the socket, /e.g./- -- when a message cannot be sent+ -- | Currently not used SocketException String+ -- | Thrown+ -- when a worker thread terminates+ -- unexpectedly;+ -- usually, this is a consequence + -- of another error (/e.g./ the broker+ -- closed the socket) and you will+ -- probably receive another exception+ -- (/e.g./ a BrokerException)+ | WorkerException String -- | Thrown when something -- against the protocol happens, /e.g./ -- an unexpected frame is received@@ -40,7 +52,7 @@ -- pending acks | TxException String -- | Thrown on connection errors, /e.g./- -- connection was disconnected+ -- connection was closed | ConnectException String -- | Should be thrown -- by user-defined converters
− Network/Mom/Stompl/Client/Protocol.hs
@@ -1,428 +0,0 @@-{-# Language CPP #-}-module Protocol (Connection, mkConnection, - conBeat, getVersion,- getSock, getWr, getRc,- connected, getErr, conMax,- connect, disconnect, disc,- Subscription, mkSub,- subscribe, unsubscribe,- begin, commit, abort, sendBeat,- Message(..), mkMessage,- MsgId(..),- send, ack, nack)-where-- import qualified Socket as S- import qualified Factory as Fac- import qualified Network.Mom.Stompl.Frame as F- import Network.Mom.Stompl.Client.Exception- import Network.Socket (Socket)-- import qualified Data.ByteString.Char8 as B- import qualified Data.ByteString.UTF8 as U- import Data.Maybe (fromMaybe)-- import Prelude hiding (catch)- import Control.Exception (throwIO, catch, - SomeException, bracketOnError)-#ifdef _DEBUG- import Control.Monad (when)-#endif- import Codec.MIME.Type as Mime (Type) -- ---------------------------------------------------------------------- -- Default version, when broker does not send a version- ---------------------------------------------------------------------- defVersion :: F.Version- defVersion = (1,2)-- ---------------------------------------------------------------------- -- Connection - ---------------------------------------------------------------------- data Connection = Connection {- conAddr :: String, -- the broker's IP address- conPort :: Int, -- the broker's port- conVers :: [F.Version], -- the accepted versions- -- the agreed version - -- after connect- conBeat :: F.Heart, -- the heart beat- conSrv :: String, -- server 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- conWrt :: Maybe S.Writer, -- if connected: writer- conErrM :: String, -- connection error- conTcp :: Bool, -- flag for tcp/ip connect- conBrk :: Bool} -- flag fror broker connect-- ---------------------------------------------------------------------- -- Subscribe abstraction- ---------------------------------------------------------------------- data Subscription = Sub {- subId :: Fac.Sub, -- subscribe identifier- subName :: String, -- queue name- subMode :: F.AckMode -- ack mode- }- deriving (Show)-- mkSub :: Fac.Sub -> String -> F.AckMode -> Subscription- mkSub sid qn am = Sub {- subId = sid,- subName = qn,- subMode = am}-- ---------------------------------------------------------------------- -- | Message Identifier- ---------------------------------------------------------------------- data MsgId = MsgId String | NoMsg- deriving (Eq)-- instance Show MsgId where- show (MsgId s) = s- show (NoMsg) = ""-- ------------------------------------------------------------------------- -- | Any content received from a queue- -- is wrapped in a message.- -- It is, in particular, the return value of /readQ/.- ------------------------------------------------------------------------- data Message a = Msg {- -- | The message Identifier- msgId :: MsgId,- -- | The subscription- msgSub :: Fac.Sub,- -- | The destination- msgDest :: String,- -- | The Ack identifier- msgAck :: String,- -- | The Stomp headers- -- that came with the message- msgHdrs :: [F.Header],- -- | The /MIME/ type of the content- msgType :: Mime.Type,- -- | The length of the - -- encoded content- msgLen :: Int,- -- | The transaction, in which - -- the message was received- msgTx :: Fac.Tx,- -- | The encoded content - msgRaw :: B.ByteString,- -- | The content - msgCont :: a}- - ---------------------------------------------------------------------- -- Create a message- ---------------------------------------------------------------------- mkMessage :: MsgId -> Fac.Sub -> String -> String ->- Mime.Type -> Int -> Fac.Tx -> - B.ByteString -> a -> Message a- mkMessage mid sub dst ak typ len tx raw cont = Msg {- msgId = mid,- msgSub = sub,- msgDest = dst,- msgAck = ak,- msgHdrs = [], - msgType = typ, - msgLen = len, - msgTx = tx,- msgRaw = raw,- msgCont = cont}-- ---------------------------------------------------------------------- -- Make a connection- ---------------------------------------------------------------------- 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,- conVers = vers,- conBeat = beat,- conSrv = "",- conSes = "",- conUsr = usr, - conPwd = pwd,- conCli = cli,- conMax = mx,- conSock = Nothing,- conRcv = Nothing,- conWrt = Nothing,- conErrM = "",- conTcp = False,- conBrk = False}-- ---------------------------------------------------------------------- -- Error: we are not connected!- ---------------------------------------------------------------------- incompleteErr :: String- incompleteErr = "incomplete Connection touched!"-- ---------------------------------------------------------------------- -- Connection interfaces- ---------------------------------------------------------------------- getSock :: Connection -> Socket- getSock = fromMaybe (error incompleteErr) . conSock -- getWr :: Connection -> S.Writer- getWr = fromMaybe (error incompleteErr) . conWrt-- getRc :: Connection -> S.Receiver- getRc = fromMaybe (error incompleteErr) . conRcv-- connected :: Connection -> Bool- connected c = conTcp c && conBrk c-- getErr :: Connection -> String- getErr = conErrM-- getVersion :: Connection -> F.Version- getVersion c = if null (conVers c) - then defVersion- else head $ conVers c-- ---------------------------------------------------------------------- -- connect- ---------------------------------------------------------------------- connect :: String -> Int -> Int -> - String -> String -> String -> F.FrameType -> - [F.Version] -> F.Heart -> [F.Header] -> IO Connection- connect host port mx usr pwd cli t vers beat hs = - bracketOnError (do s <- S.connect host port- let c = mkConnection host port mx - usr pwd cli vers beat- return c {conSock = Just s, conTcp = True}) - disc- (connectBroker mx t vers beat hs) -- ---------------------------------------------------------------------- -- disconnect either on broker level or on tcp/ip level- ---------------------------------------------------------------------- disconnect :: Connection -> String -> IO Connection- disconnect c r- | conBrk c = case mkDiscF r of- Left e -> return c {- conErrM = - "Cannot create Frame: " ++ e}- Right f -> do- S.send (getWr c) (getSock c) f - return c {conBrk = False}- | conTcp c = disc c- | otherwise = return c {conErrM = "Not connected!"}-- ---------------------------------------------------------------------- -- begin transaction- ---------------------------------------------------------------------- begin :: Connection -> String -> String -> IO ()- begin c tx receipt = sendFrame c tx receipt [] mkBeginF-- ---------------------------------------------------------------------- -- commit transaction- ---------------------------------------------------------------------- commit :: Connection -> String -> String -> IO ()- commit c tx receipt = sendFrame c tx receipt [] mkCommitF-- ---------------------------------------------------------------------- -- abort transaction- ---------------------------------------------------------------------- abort :: Connection -> String -> String -> IO ()- abort c tx receipt = sendFrame c tx receipt [] mkAbortF-- ---------------------------------------------------------------------- -- ack- ---------------------------------------------------------------------- ack :: Connection -> Message a -> String -> IO ()- ack c m receipt = sendFrame c m receipt [] (mkAckF True)-- ---------------------------------------------------------------------- -- nack- ---------------------------------------------------------------------- nack :: Connection -> Message a -> String -> IO ()- nack c m receipt = sendFrame c m receipt [] (mkAckF False)-- ---------------------------------------------------------------------- -- subscribe- ---------------------------------------------------------------------- subscribe :: Connection -> Subscription -> String -> [F.Header] -> IO ()- subscribe c sub receipt hs = sendFrame c sub receipt hs mkSubF-- ---------------------------------------------------------------------- -- unsubscribe- ---------------------------------------------------------------------- unsubscribe :: Connection -> Subscription -> String -> [F.Header] -> IO ()- unsubscribe c sub receipt hs = sendFrame c sub receipt hs mkUnSubF-- ---------------------------------------------------------------------- -- send- ---------------------------------------------------------------------- send :: Connection -> Message a -> String -> [F.Header] -> IO ()- send c msg receipt hs = sendFrame c msg receipt hs mkSendF-- ---------------------------------------------------------------------- -- heart beat- ---------------------------------------------------------------------- sendBeat :: Connection -> IO ()- sendBeat c = sendFrame c () "" [] (\_ _ _ -> Right F.mkBeat)-- ---------------------------------------------------------------------- -- generic sendFrame:- -- takes a connection some data (like subscribe, message, etc.)- -- some headers, a function that creates a frame or returns an error- -- creates the frame and sends it- ---------------------------------------------------------------------- sendFrame :: Connection -> a -> String -> [F.Header] -> - (a -> String -> [F.Header] -> Either String F.Frame) -> IO ()- sendFrame c m receipt hs mkF = - if not (connected c) then throwIO $ ConnectException "Not connected!"- else case mkF m receipt hs of- Left e -> throwIO $ ProtocolException $- "Cannot create Frame: " ++ e- Right f -> -#ifdef _DEBUG- do when (not $ F.complies (1,2) f) $- putStrLn $ "Frame does not comply with 1.2: " ++ show f -#endif- S.send (getWr c) (getSock c) f-- ---------------------------------------------------------------------- -- hard disconnect (on tcp/ip) level- ---------------------------------------------------------------------- disc :: Connection -> IO Connection- disc c = do- let c' = c {conTcp = False,- conBrk = False,- conWrt = Nothing,- conRcv = Nothing,- conSock = Nothing}- S.disconnect (getSock c)- return c'-- ---------------------------------------------------------------------- -- the hard work on connecting to a broker- ---------------------------------------------------------------------- connectBroker :: Int -> F.FrameType -> - [F.Version] -> F.Heart -> [F.Header] ->- Connection -> IO Connection- connectBroker mx t vers beat hs c = - let mk = case t of- F.Connect -> F.mkConFrame- F.Stomp -> F.mkStmpFrame- _ -> error "Ouch: Unknown Connect-type"- in case mkConF mk- (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) -- no timeout?- (\e -> return $ Left $ show (e::SomeException)) - case eiC of- Left e -> return c {conErrM = e}- Right r -> - let c' = handleConnected r c- in if period c' > 0 && period c' < fst beat- then return c {conErrM = "Beat frequency too high"}- else return c' {conBrk = True,- conRcv = Just rc,- conWrt = Just wr}- where period = snd . conBeat-- ---------------------------------------------------------------------- -- handle broker 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 ++ ")",- conBeat = F.getBeat f,- conVers = [F.getVersion f],- conSes = F.getSession f}- F.Error -> c {conErrM = errToMsg f}- _ -> c {conErrM = "Unexpected Frame: " ++ - U.toString (F.putCommand f)}-- ---------------------------------------------------------------------- -- transform an error frame into a string- ---------------------------------------------------------------------- errToMsg :: F.Frame -> String- errToMsg f = F.getMsg f ++ if B.length (F.getBody f) == 0 - then "."- else ": " ++ U.toString (F.getBody f)-- ---------------------------------------------------------------------- -- frame constructors- -- this needs review...- ---------------------------------------------------------------------- mkReceipt :: String -> [F.Header]- mkReceipt receipt = if null receipt then [] else [F.mkRecHdr receipt]-- mkConF :: ([F.Header] -> Either String F.Frame) ->- String -> String -> String -> String -> - [F.Version] -> F.Heart -> [F.Header] -> Either String F.Frame- mkConF mk 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 mk $ [F.mkHostHdr host,- F.mkAcVerHdr $ F.versToVal vers, - F.mkBeatHdr $ F.beatToVal beat] ++- uHdr ++ pHdr ++ cHdr ++ hs-- mkDiscF :: String -> Either String F.Frame- mkDiscF receipt =- F.mkDisFrame $ mkReceipt receipt-- mkSubF :: Subscription -> String -> [F.Header] -> Either String F.Frame- mkSubF sub receipt hs = - F.mkSubFrame $ [F.mkIdHdr $ show $ subId sub,- F.mkDestHdr $ subName sub,- F.mkAckHdr $ show $ subMode sub] ++ - mkReceipt receipt ++ hs-- mkUnSubF :: Subscription -> String -> [F.Header] -> Either String F.Frame- mkUnSubF sub receipt hs =- let dh = if null (subName sub) then [] else [F.mkDestHdr $ subName sub]- in F.mkUSubFrame $ [F.mkIdHdr $ show $ subId sub] ++ dh ++ - mkReceipt receipt ++ hs-- mkSendF :: Message a -> String -> [F.Header] -> Either String F.Frame- mkSendF msg receipt hs = - Right $ F.mkSend (msgDest msg) (show $ msgTx msg) receipt - (msgType msg) (msgLen msg) hs -- escape headers! - (msgRaw msg) -- mkAckF :: Bool -> Message a -> String -> [F.Header] -> Either String F.Frame- mkAckF ok msg receipt _ =- let sh = if null $ show $ msgSub msg then [] - else [F.mkSubHdr $ show $ msgSub msg]- th = if null $ show $ msgTx msg - then [] else [F.mkTrnHdr $ show $ msgTx msg]- rh = mkReceipt receipt- mk = if ok then F.mkAckFrame else F.mkNackFrame- in mk $ F.mkIdHdr (msgAck msg) : (sh ++ rh ++ th)-- mkBeginF :: String -> String -> [F.Header] -> Either String F.Frame- mkBeginF tx receipt _ = - F.mkBgnFrame $ F.mkTrnHdr tx : mkReceipt receipt-- mkCommitF :: String -> String -> [F.Header] -> Either String F.Frame- mkCommitF tx receipt _ =- F.mkCmtFrame $ F.mkTrnHdr tx : mkReceipt receipt-- mkAbortF :: String -> String -> [F.Header] -> Either String F.Frame- mkAbortF tx receipt _ =- F.mkAbrtFrame $ F.mkTrnHdr tx : mkReceipt receipt-
Network/Mom/Stompl/Client/Queue.hs view
@@ -38,9 +38,9 @@ writeQ, writeQWith, writeAdHoc, writeAdHocWith, -- * Messages- P.Message, - msgContent, P.msgRaw, - P.msgType, P.msgLen, P.msgHdrs,+ Message, + msgContent, msgRaw, + msgType, msgLen, msgHdrs, -- * Receipts -- $stomp_receipts Factory.Rec(..), Receipt,@@ -50,11 +50,11 @@ Tx, withTransaction, Topt(..), abort, - -- * Acknowledgements+ -- * Acknowledgement -- $stomp_acks ack, ackWith, nack, nackWith, #ifdef TEST- frmToMsg, P.msgAck,+ frmToMsg, msgAck, #endif -- * Exceptions module Network.Mom.Stompl.Client.Exception@@ -63,14 +63,8 @@ ) where- ----------------------------------------------------------------- -- todo- -- -- pass a logger (name) to withConnection- -- - test/check for deadlocks- ---------------------------------------------------------------- - import qualified Socket as S- import qualified Protocol as P+ import Stream import Factory import State @@ -81,11 +75,13 @@ import qualified Data.ByteString.UTF8 as U import Data.List (find) import Data.Time.Clock- import Data.Maybe (isJust, fromJust)+ import Data.Maybe (isNothing)+ import Data.Conduit.Network+ import Data.Conduit.Network.TLS import Control.Concurrent import Control.Applicative ((<$>))- import Control.Monad+ import Control.Monad (when, unless, forever) import Control.Exception (bracket, finally, catches, onException, AsyncException(..), Handler(..), throwIO, SomeException)@@ -175,8 +171,8 @@ 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'+ may be read from the 'Message';+ in the second case, the contents of the 'Message' will be a 'B.ByteString'. In the Stompl library, queues @@ -210,7 +206,7 @@ On sending messages, receipt handling can be made explict. The function 'writeQWith'- requests a receipt to the message+ requests a receipt for the message it sends and returns it to the caller. The application can then, later, explicitly wait for the receipt, using 'waitReceipt'.@@ -269,6 +265,9 @@ A message may also be /negatively acknowledged/ (/nack/). How the broker handles a /nack/, however, is not further specified by the Stomp protocol.+ Some brokers respond with an error frame and+ close the connection immediately after.+ This may cause a surprising cascade of exceptions! -} {- $stomp_sample@@ -321,7 +320,7 @@ > let p = case msgContent m of > Ping -> Pong > Pong -> Ping- > putStrLn $ show p+ > print p > writeQ oQ nullType [] p > threadDelay 10000 -}@@ -379,14 +378,10 @@ -- 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'.+ -- 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] -> [F.Header] -> (Con -> IO a) -> IO a@@ -395,49 +390,121 @@ let mx = oMaxRecv os let (u,p) = oAuth os let (ci) = oCliId os+ let tm = oTmo os+ let cfg = oTLS host port os let t | oStomp os = F.Stomp | otherwise = F.Connect- bracket (P.connect host port mx u p ci t vers beat hs)- P.disc -- important: At the end, we close the socket!- (whenConnected os act)+ runTLSClient cfg $ \ad -> do+ cid <- mkUniqueConId -- connection id+ me <- myThreadId -- connection owner+ now <- getCurrentTime -- heartbeat+ ch <- newChan -- sender+ bracket (do addCon $ mkConnection cid host port + mx u p ci + vers beat ch + me now os+ getCon cid)+ (\_ -> rmCon cid) $ \c -> + withSender c ad ch $ do+ w <- newEmptyMVar+ withListener c ad cid w $ do+ connectBroker t vers beat hs c+ whenConnected cid tm beat w 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)+ ---------------------------------------------------------------------+ -- Start the sender threat+ ---------------------------------------------------------------------+ withSender :: Connection -> AppData -> Chan F.Frame -> IO a -> IO a+ withSender c ad ch = withThread c "sender" (sender ad ch) - isThrd :: Maybe a -> Bool- isThrd = isJust+ ---------------------------------------------------------------------+ -- Start the listener threat+ ---------------------------------------------------------------------+ withListener :: Connection -> AppData -> Con -> MVar () -> IO a -> IO a+ withListener c ad cid m = withThread c "listener" (listen ad cid m) - thrd :: Maybe a -> a- thrd = fromJust+ -----------------------------------------------------------------------+ -- Connection listener+ -----------------------------------------------------------------------+ listen :: AppData -> Con -> MVar () -> IO ()+ listen ad cid w = do + c <- getCon cid+ ch <- newChan+ withThread c "receiver" (receiver ad ch (throwToOwner c)) $+ forever $ do+ f <- readChan ch+ logReceive cid+ case F.typeOf f of+ F.Connected -> handleConnected cid f w+ 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) + ---------------------------------------------------------------------+ -- Send the Connect frame+ ---------------------------------------------------------------------+ connectBroker :: F.FrameType -> + [F.Version] -> F.Heart -> [F.Header] ->+ Connection -> IO ()+ connectBroker t vs beat hs c = + let mk = case t of+ F.Connect -> F.mkConFrame+ F.Stomp -> F.mkStmpFrame+ _ -> error "Ouch: Unknown Connect-type"+ in case mkConF mk+ (conAddr c) + (conUsr c) (conPwd c) + (conCli c) vs beat hs of+ Left e -> throwIO (ConnectException e)+ Right f -> writeChan (conChn c) f ++ ---------------------------------------------------------------------+ -- Waiting for the Connected frame,+ -- starting the user application and+ -- disconnecting after+ ---------------------------------------------------------------------+ whenConnected :: Con -> Int -> F.Heart -> MVar () -> (Con -> IO a) -> IO a+ whenConnected cid tm bt m act = do+ mbT <- if tm <= 0 then Just <$> takeMVar m + else timeout (ms tm) (takeMVar m)+ when (isNothing mbT) (throwIO $ ConnectException "Timeout expired!")+ c <- getCon cid+ if period c > 0 + then if period c < fst bt+ then throwIO (ConnectException $ "Beat frequency too high: " ++ + show (period c))+ else withThread c "heartbeat" (heartBeat cid $ period c) $ go c+ else go c+ where go c = do+ putMVar m () + updCon cid c {conBrk = True} + finally (act cid) (do+ c' <- getCon cid+ -- wait for receipt on disconnect ----+ unless (conWait c' <= 0) $ do+ rc <- mkUniqueRecc + disconnect c' (show rc)+ addRec cid rc+ waitCon cid rc (conWait c') `onException` rmRec cid rc)+ period = snd . conBeat++ ---------------------------------------------------------------------+ -- disconnect from the broker+ ---------------------------------------------------------------------+ disconnect :: Connection -> String -> IO ()+ disconnect c r+ | conBrk c = case mkDiscF r of+ Left e -> throwIO (ConnectException $+ "Cannot create Disconnect Frame: " ++ e)+ Right f -> writeChan (conChn c) f + | otherwise = throwIO (ConnectException "Not connected")+ ------------------------------------------------------------------------- -- wait for receipt+ -- wait for receipt on disconnect ------------------------------------------------------------------------ waitCon :: Con -> Receipt -> Int -> IO () waitCon cid rc delay = do@@ -449,9 +516,44 @@ then throwIO $ ConnectException $ "No receipt on disconnect (" ++ show cid ++ ")."- else do- threadDelay $ ms 1- waitCon cid rc (delay - 1)+ else do threadDelay $ ms 1+ waitCon cid rc (delay - 1)++ -----------------------------------------------------------------------+ -- Starting the internal worker threads:+ -- - the new thread lives as long as the user action lives+ -- - if the new thread terminates early, the owner is signalled+ -- - if the new thread receives a synchronous exception,+ -- the owner is signalled+ -----------------------------------------------------------------------+ withThread :: Connection -> String -> IO () -> IO r -> IO r+ withThread c nm th act = do+ m <- newEmptyMVar+ x <- newMVar False+ tid <- forkIO (theThread x `finally` putMVar m ())+ (act >>= signalOk x) `finally` (killThread tid >> takeMVar m)+ where theThread x = (th >> signalEnd x) `catches` hndls+ hndls = [Handler (\e -> case e of+ ThreadKilled -> throwIO e+ _ -> throwIO e),+ Handler (\e -> throwToOwner c $ + WorkerException (+ nm ++ " terminated: " +++ show (e::SomeException)))+ ]+ signalEnd x = do t <- readMVar x+ unless t $ throwToOwner c $ WorkerException (+ nm ++ " terminated")+ signalOk x r = modifyMVar x $ \_ -> return (True,r)++ -----------------------------------------------------------------------+ -- Throw to owner+ -- important: give the owner some time to react+ -- before continuing and - probably - + -- causing another exception+ -----------------------------------------------------------------------+ throwToOwner :: Connection -> StomplException -> IO ()+ throwToOwner c = throwTo (conOwner c) ------------------------------------------------------------------------ -- | A Queue for sending messages.@@ -492,13 +594,14 @@ data Qopt = -- | A queue created with 'OWithReceipt' will request a receipt -- on all interactions with the broker.- -- 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'.+ -- The handling of receipts is usually not visible to applications, + -- but may be made visible in the case of sending messages+ -- using 'writeQWith'. -- '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 + -- to explicitly wait for the broker confirming this receipt.+ --+ -- 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@@ -524,7 +627,7 @@ -- -- It is good practice to use /timeout/ with all calls -- that may wait for receipts, - -- /ie/ 'newReader' and 'withReader' + -- /i.e./ 'newReader' and 'withReader' -- with options 'OWithReceipt' or 'OWaitReceipt', -- or 'writeQ' and 'writeQWith' with options 'OWaitReceipt', -- or 'ackWith' and 'nackWith'.@@ -551,9 +654,15 @@ | ONoContentLen deriving (Show, Read, Eq) + ------------------------------------------------------------------------+ -- Option is element of option list+ ------------------------------------------------------------------------ hasQopt :: Qopt -> [Qopt] -> Bool hasQopt o os = o `elem` os + ------------------------------------------------------------------------+ -- What ackMode ('F.Auto' by default)+ ------------------------------------------------------------------------ ackMode :: [Qopt] -> F.AckMode ackMode os = case find isMode os of Just (OMode x) -> x@@ -595,12 +704,11 @@ -- -- * the list of 'F.Header' coming with the message --- -- * the contents encoded as 'B.ByteString'.+ -- * the content encoded as 'B.ByteString'. --- -- The simplest possible in-bound converter for plain strings- -- may be created like this:+ -- A simple in-bound converter for plain strings is for instance: --- -- > let iconv _ _ _ = return . toString+ -- > let iconv _ _ _ = return . unpack ------------------------------------------------------------------------ type InBound a = Mime.Type -> Int -> [F.Header] -> B.ByteString -> IO a ------------------------------------------------------------------------@@ -612,7 +720,7 @@ -- A simple example to create an out-bound converter -- for plain strings could be: --- -- > let oconv = return . fromString+ -- > let oconv = return . pack ------------------------------------------------------------------------ type OutBound a = a -> IO B.ByteString @@ -627,8 +735,7 @@ -- * 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 queue name is used only for debugging. -- -- * The Stomp destination, /i.e./ the name of the queue -- as it is known to the broker and other applications.@@ -637,13 +744,12 @@ -- -- * 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 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.+ -- in some kind of query language such as SQL or XPath. -- -- * An in-bound converter. --@@ -674,7 +780,7 @@ InBound a -> IO (Reader a) newReader cid qn dst os hs conv = do c <- getCon cid- if not $ P.connected (conCon c)+ if not $ connected c then throwIO $ ConnectException $ "Not connected (" ++ show cid ++ ")" else newRecvQ cid c qn dst os hs conv@@ -711,7 +817,7 @@ OutBound a -> IO (Writer a) newWriter cid qn dst os _ conv = do c <- getCon cid- if not $ P.connected (conCon c)+ if not $ connected c then throwIO $ ConnectException $ "Not connected (" ++ show cid ++ ")" else newSendQ cid qn dst os conv@@ -751,8 +857,8 @@ -- | Creates a 'Writer' with limited lifetime. -- 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.+ -- The function should be used for writers with a lifetime+ -- shorter than that of the connection. -- -- 'withWriter' returns the result of the action. -- Since the lifetime of the queue is limited to the action,@@ -861,7 +967,7 @@ sid <- mkUniqueSubId rc <- if with then mkUniqueRecc else return NoRec logSend cid- P.subscribe (conCon c) (P.mkSub sid dst am) (show rc) hs+ fSubscribe c(mkSub sid dst am) (show rc) hs ch <- newChan addSub cid (sid, ch) addDest cid (dst, ch) @@ -888,19 +994,19 @@ 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) [])+ finally (fUnsubscribe c+ (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'.+ -- and returns it as '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+ -- Wether other connections will receive the message too -- depends on the broker and the queue patterns it implements. -- If the queue is currently empty, -- the thread will preempt until a message arrives.@@ -930,10 +1036,10 @@ -- and use 'ackWith' to acknowledge -- the message explicitly. ------------------------------------------------------------------------- readQ :: Reader a -> IO (P.Message a)+ readQ :: Reader a -> IO (Message a) readQ q = do c <- getCon (rCon q)- if not $ P.connected (conCon c)+ if not $ connected c then throwIO $ QueueException $ "Not connected: " ++ show (rCon q) else case getSub (rSub q) c of Nothing -> throwIO $ QueueException $ @@ -943,7 +1049,7 @@ when (rMode q /= F.Auto) $ if rAuto q then ack (rCon q) m- else addAck (rCon q) (P.msgId m)+ else addAck (rCon q) (msgId m) return m ------------------------------------------------------------------------@@ -993,7 +1099,7 @@ -- 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+ -- the server would send the reply to the queue -- indicated in the header using 'writeAdHoc'. -- The additional 'String' parameter contains the destination. ------------------------------------------------------------------------@@ -1040,7 +1146,7 @@ Mime.Type -> [F.Header] -> a -> IO Receipt writeGeneric q dest mime hs x = do c <- getCon (wCon q)- if not $ P.connected (conCon c)+ if not $ connected c then throwIO $ ConnectException $ "Not connected (" ++ show (wCon q) ++ ")" else do@@ -1057,30 +1163,30 @@ s <- conv x rc <- if wRec q then mkUniqueRecc else return NoRec let l = if wCntl q then -1 else B.length s- let m = P.mkMessage P.NoMsg NoSub dest "" mime l tx s x+ let m = mkMessage NoMsg NoSub dest "" mime l tx s x when (wRec q) $ addRec (wCon q) rc logSend $ wCon q- P.send (conCon c) m (show rc) hs + fSend 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:+ -- | Acknowledges the arrival of 'Message' to the broker.+ -- It is used with a 'Connection' /c/ and a 'Message' /x/ like: -- -- > ack c x ------------------------------------------------------------------------- ack :: Con -> P.Message a -> IO ()+ ack :: Con -> Message a -> IO () ack cid msg = do ack' cid True False msg- rmAck cid $ P.msgId msg+ rmAck cid $ msgId msg ------------------------------------------------------------------------- -- | Acknowledges the arrival of 'P.Message' to the broker,+ -- | Acknowledges the arrival of '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/ + -- for a 'Connection' /c/, a 'Message' /x/ -- and a /timeout/ in microseconds /tmo/ like: -- -- > mbR <- timeout tmo $ ackWith c x @@ -1088,57 +1194,57 @@ -- > Nothing -> -- error handling -- > Just _ -> do -- ... ------------------------------------------------------------------------- ackWith :: Con -> P.Message a -> IO ()+ ackWith :: Con -> Message a -> IO () ackWith cid msg = do ack' cid True True msg - rmAck cid $ P.msgId msg+ rmAck cid $ msgId msg ------------------------------------------------------------------------- -- | Negatively acknowledges the arrival of 'P.Message' to the broker.+ -- | Negatively acknowledges the arrival of 'Message' to the broker. -- For more details see 'ack'. ------------------------------------------------------------------------- nack :: Con -> P.Message a -> IO ()+ nack :: Con -> Message a -> IO () nack cid msg = do ack' cid False False msg- rmAck cid $ P.msgId msg+ rmAck cid $ msgId msg ------------------------------------------------------------------------- -- | Negatively acknowledges the arrival of 'P.Message' to the broker,+ -- | Negatively acknowledges the arrival of '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 :: Con -> Message a -> IO () nackWith cid msg = do ack' cid False True msg- rmAck cid $ P.msgId msg+ rmAck cid $ msgId msg ------------------------------------------------------------------------ -- Checks for Transaction, -- if a transaction is ongoing, -- the TxId is added to the message- -- and calls P.ack on the message.+ -- and calls 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' :: Con -> Bool -> Bool -> Message a -> IO () ack' cid ok with msg = do c <- getCon cid- if not $ P.connected (conCon c) + if not $ connected c then throwIO $ ConnectException $ "Not connected (" ++ show cid ++ ")"- else if null (show $ P.msgAck msg)+ else if null (show $ msgAck msg) then throwIO $ ProtocolException "No ack in message!" else do tx <- getCurTx c >>= (\mbT -> case mbT of Nothing -> return NoTx Just x -> return x)- let msg' = msg {P.msgTx = tx}+ let msg' = msg {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+ if ok then fAck c msg' $ show rc+ else fNack c msg' $ show rc when with $ waitReceipt cid rc ------------------------------------------------------------------------@@ -1187,7 +1293,7 @@ tx <- mkUniqueTxId let t = mkTrn tx os c <- getCon cid- if not $ P.connected (conCon c)+ if not $ connected c then throwIO $ ConnectException $ "Not connected (" ++ show cid ++ ")" else finally (do addTx t cid@@ -1265,7 +1371,7 @@ 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)+ fBegin c (show $ txId t) (show rc) ----------------------------------------------------------------------- -- Send commit or abort frame@@ -1278,8 +1384,8 @@ 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)+ if x then fCommit c (show tx) (show rc)+ else fAbort c (show tx) (show rc) mbR <- if w then timeout (ms $ txTmo t) $ waitReceipt cid rc else return $ Just () rmTx cid@@ -1313,49 +1419,40 @@ -- Transform a frame into a message -- using the queue's application callback ------------------------------------------------------------------------ frmToMsg :: Reader a -> F.Frame -> IO (P.Message a)+ frmToMsg :: Reader a -> F.Frame -> IO (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.getMsgAck f) - (F.getMime f)- (F.getLength f)- NoTx - b -- raw bytestring- x -- converted context- return m {P.msgHdrs = F.getHeaders f}+ let m = mkMessage (MsgId $ F.getId f) sid+ (F.getDest f) + (F.getMsgAck f) + (F.getMime f)+ (F.getLength f)+ NoTx + b -- raw bytestring+ x -- converted context+ return m {msgHdrs = F.getHeaders f} ------------------------------------------------------------------------ -- Connection listener+ -- Handle Connected Frame ------------------------------------------------------------------------ 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)+ handleConnected :: Con -> F.Frame -> MVar () -> IO ()+ handleConnected cid f m = do -- withCon cid $ \c -> + c <- getCon cid+ if connected c + then throwToOwner c $+ ProtocolException "Unexptected Connected frame"+ else updCon cid c {conSrv = let srv = F.getServer f+ in F.getSrvName srv ++ "/" +++ F.getSrvVer srv ++ " (" +++ F.getSrvCmts srv ++ ")",+ conBeat = F.getBeat f,+ conVers = [F.getVersion f],+ conSes = F.getSession f}+ >> putMVar m () ----------------------------------------------------------------------- -- Handle Message Frame@@ -1365,7 +1462,7 @@ c <- getCon cid case getCh c of Nothing -> throwToOwner c $ - ProtocolException $ "Unknown Queue: " ++ show f+ ProtocolException $ "Unknown Channel: " ++ show f Just ch -> writeChan ch f where getCh c = let dst = F.getDest f sid = F.getSub f@@ -1387,17 +1484,6 @@ 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 ()@@ -1418,7 +1504,7 @@ -- My Beat ----------------------------------------------------------------------- heartBeat :: Con -> Int -> IO ()- heartBeat cid period = forever $ do+ heartBeat cid p = forever $ do now <- getCurrentTime c <- getCon cid let me = myMust c @@ -1428,15 +1514,15 @@ "Missing HeartBeat, last was " ++ show (now `diffUTCTime` he) ++ " seconds ago!"- when (now >= me) $ P.sendBeat $ conCon c- threadDelay $ ms period+ when (now >= me) $ fSendBeat c+ threadDelay $ ms p ----------------------------------------------------------------------- -- When we should have sent last heartbeat ----------------------------------------------------------------------- myMust :: Connection -> UTCTime myMust c = let t = conMyBeat c- p = snd $ P.conBeat $ conCon c+ p = snd $ conBeat c in timeAdd t p -----------------------------------------------------------------------@@ -1445,7 +1531,7 @@ hisMust :: Connection -> UTCTime hisMust c = let t = conHisBeat c tol = 4- b = fst $ P.conBeat $ conCon c+ b = fst $ conBeat c p = tol * b in timeAdd t p @@ -1460,4 +1546,166 @@ ----------------------------------------------------------------------- ms2nominal :: Int -> NominalDiffTime ms2nominal m = fromIntegral m / (1000::NominalDiffTime)++ ---------------------------------------------------------------------+ -- begin transaction+ ---------------------------------------------------------------------+ fBegin :: Connection -> String -> String -> IO ()+ fBegin c tx receipt = sendFrame c tx receipt [] mkBeginF++ ---------------------------------------------------------------------+ -- commit transaction+ ---------------------------------------------------------------------+ fCommit :: Connection -> String -> String -> IO ()+ fCommit c tx receipt = sendFrame c tx receipt [] mkCommitF++ ---------------------------------------------------------------------+ -- abort transaction+ ---------------------------------------------------------------------+ fAbort :: Connection -> String -> String -> IO ()+ fAbort c tx receipt = sendFrame c tx receipt [] mkAbortF++ ---------------------------------------------------------------------+ -- ack+ ---------------------------------------------------------------------+ fAck :: Connection -> Message a -> String -> IO ()+ fAck c m receipt = sendFrame c m receipt [] (mkAckF True)++ ---------------------------------------------------------------------+ -- nack+ ---------------------------------------------------------------------+ fNack :: Connection -> Message a -> String -> IO ()+ fNack c m receipt = sendFrame c m receipt [] (mkAckF False)++ ---------------------------------------------------------------------+ -- subscribe+ ---------------------------------------------------------------------+ fSubscribe :: Connection -> Subscription -> String -> [F.Header] -> IO ()+ fSubscribe c sub receipt hs = sendFrame c sub receipt hs mkSubF++ ---------------------------------------------------------------------+ -- unsubscribe+ ---------------------------------------------------------------------+ fUnsubscribe :: Connection -> Subscription -> String -> [F.Header] -> IO ()+ fUnsubscribe c sub receipt hs = sendFrame c sub receipt hs mkUnSubF++ ---------------------------------------------------------------------+ -- send+ ---------------------------------------------------------------------+ fSend :: Connection -> Message a -> String -> [F.Header] -> IO ()+ fSend c msg receipt hs = sendFrame c msg receipt hs mkSendF++ ---------------------------------------------------------------------+ -- heart beat+ ---------------------------------------------------------------------+ fSendBeat :: Connection -> IO ()+ fSendBeat c = sendFrame c () "" [] (\_ _ _ -> Right F.mkBeat)++ ---------------------------------------------------------------------+ -- generic sendFrame:+ -- takes a connection some data (like subscribe, message, etc.)+ -- some headers, a function that creates a frame or returns an error+ -- creates the frame and sends it+ ---------------------------------------------------------------------+ sendFrame :: Connection -> a -> String -> [F.Header] -> + (a -> String -> [F.Header] -> Either String F.Frame) -> IO ()+ sendFrame c m receipt hs mkF = + if not (connected c) then throwIO $ ConnectException "Not connected!"+ else case mkF m receipt hs of+ Left e -> throwIO $ ProtocolException $+ "Cannot create Frame: " ++ e+ Right f -> +#ifdef _DEBUG+ do when (not $ F.complies (1,2) f) $+ putStrLn $ "Frame does not comply with 1.2: " ++ show f +#endif+ writeChan (conChn c) f+ + ---------------------------------------------------------------------+ -- frame constructors+ -- this needs review...+ ---------------------------------------------------------------------+ mkReceipt :: String -> [F.Header]+ mkReceipt receipt = if null receipt then [] else [F.mkRecHdr receipt]++ mkConF :: ([F.Header] -> Either String F.Frame) ->+ String -> String -> String -> String -> + [F.Version] -> F.Heart -> [F.Header] -> Either String F.Frame+ mkConF mk host usr pwd cli vs 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 mk $ [F.mkHostHdr host,+ F.mkAcVerHdr $ F.versToVal vs, + F.mkBeatHdr $ F.beatToVal beat] +++ uHdr ++ pHdr ++ cHdr ++ hs++ ---------------------------------------------------------------------+ -- make Disconnect Frame+ ---------------------------------------------------------------------+ mkDiscF :: String -> Either String F.Frame+ mkDiscF receipt =+ F.mkDisFrame $ mkReceipt receipt++ ---------------------------------------------------------------------+ -- make Subscribe Frame+ ---------------------------------------------------------------------+ mkSubF :: Subscription -> String -> [F.Header] -> Either String F.Frame+ mkSubF sub receipt hs = + F.mkSubFrame $ [F.mkIdHdr $ show $ subId sub,+ F.mkDestHdr $ subName sub,+ F.mkAckHdr $ show $ subMode sub] ++ + mkReceipt receipt ++ hs++ ---------------------------------------------------------------------+ -- make Unsubscribe Frame+ ---------------------------------------------------------------------+ mkUnSubF :: Subscription -> String -> [F.Header] -> Either String F.Frame+ mkUnSubF sub receipt hs =+ let dh = if null (subName sub) then [] else [F.mkDestHdr $ subName sub]+ in F.mkUSubFrame $ [F.mkIdHdr $ show $ subId sub] ++ dh ++ + mkReceipt receipt ++ hs++ ---------------------------------------------------------------------+ -- make Send Frame+ ---------------------------------------------------------------------+ 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 -- escape headers! + (msgRaw msg) ++ ---------------------------------------------------------------------+ -- make Ack Frame+ ---------------------------------------------------------------------+ mkAckF :: Bool -> Message a -> String -> [F.Header] -> Either String F.Frame+ mkAckF ok msg receipt _ =+ let sh = if null $ show $ msgSub msg then [] + else [F.mkSubHdr $ show $ msgSub msg]+ th = if null $ show $ msgTx msg + then [] else [F.mkTrnHdr $ show $ msgTx msg]+ rh = mkReceipt receipt+ mk = if ok then F.mkAckFrame else F.mkNackFrame+ in mk $ F.mkIdHdr (msgAck msg) : (sh ++ rh ++ th)++ ---------------------------------------------------------------------+ -- make Begin Frame+ ---------------------------------------------------------------------+ mkBeginF :: String -> String -> [F.Header] -> Either String F.Frame+ mkBeginF tx receipt _ = + F.mkBgnFrame $ F.mkTrnHdr tx : mkReceipt receipt++ ---------------------------------------------------------------------+ -- make Commit Frame+ ---------------------------------------------------------------------+ mkCommitF :: String -> String -> [F.Header] -> Either String F.Frame+ mkCommitF tx receipt _ =+ F.mkCmtFrame $ F.mkTrnHdr tx : mkReceipt receipt++ ---------------------------------------------------------------------+ -- make Abort Frame+ ---------------------------------------------------------------------+ mkAbortF :: String -> String -> [F.Header] -> Either String F.Frame+ mkAbortF tx receipt _ =+ F.mkAbrtFrame $ F.mkTrnHdr tx : mkReceipt receipt
− Network/Mom/Stompl/Client/Socket.hs
@@ -1,178 +0,0 @@-{-# Language CPP #-}-module Socket (Writer, Receiver, - initWriter, initReceiver,- connect, disconnect, send, receive)-where-- import qualified Network.Socket as S- import qualified Network.Socket.ByteString as BS- import Network.BSD (getProtocolNumber)-- import qualified Data.ByteString.Char8 as B- import qualified Data.ByteString.UTF8 as U-- import Network.Mom.Stompl.Parser (stompParser)- import qualified Network.Mom.Stompl.Frame as F- import Network.Mom.Stompl.Client.Exception-- import Control.Concurrent.MVar- import Control.Applicative ((<$>))- import Control.Monad (unless, 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- maxStep = 100-- data Receiver = Receiver {- getBuffer :: IO (Maybe B.ByteString),- putBuffer :: B.ByteString -> IO ()- }-- data Writer = Writer {- lock :: IO (),- release :: IO ()- }-- get :: MVar B.ByteString -> IO (Maybe B.ByteString)- get buf = do- e <- isEmptyMVar buf- if e- then return Nothing- else Just <$> takeMVar buf-- put :: MVar B.ByteString -> B.ByteString -> IO ()- put buf s = do- e <- isEmptyMVar buf- if e - then putMVar buf s- else do - _ <- takeMVar buf- putMVar buf s-- initReceiver :: IO Receiver- initReceiver = do- buf <- newEmptyMVar- return Receiver {- getBuffer = get buf,- putBuffer = put buf}-- lockSock :: MVar () -> IO ()- lockSock l = putMVar l ()-- releaseSock :: MVar () -> IO ()- releaseSock = takeMVar -- initWriter :: IO Writer- initWriter = do- l <- newEmptyMVar- return Writer {- lock = lockSock l,- release = releaseSock l- }- - connect :: String -> Int -> IO S.Socket- connect host port = do- let p = fromIntegral port :: S.PortNumber- prot <- getProtocolNumber "tcp" - let hints = S.defaultHints {S.addrSocketType = S.Stream,- S.addrProtocol = prot}- adds <- S.getAddrInfo (Just hints) (Just host) Nothing- tryConnect p adds-- tryConnect :: S.PortNumber -> [S.AddrInfo] -> IO S.Socket- tryConnect _ [] = throwIO $ SocketException - "Give up: no more address info!"- tryConnect p (i:is) = - case i of- S.AddrInfo _ f t prot addr _ -> - let mbA = case addr of- S.SockAddrInet _ a -> Just $ S.SockAddrInet p a- S.SockAddrInet6 _ fl a s -> Just $ S.SockAddrInet6 p fl a s- _ -> Nothing- in case mbA of- Nothing -> tryConnect p is- Just a -> do- sock <- S.socket f t prot- eiS <- Ex.try $ S.connect sock a- case eiS of- Left e -> do- hPutStrLn stderr $ "Network.Mom.Stompl.Socket - " ++- "Warning: " ++ show (e::SomeException)- tryConnect p is- Right _ -> return sock-- disconnect :: S.Socket -> IO ()- disconnect = S.sClose -- send :: Writer -> S.Socket -> F.Frame -> IO ()- send wr sock f = do- let s = F.putFrame f- lock wr-#ifdef _DEBUG- putStrLn $ "Sending: " ++ U.toString s-#endif- 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-- handlePartial :: Receiver -> S.Socket -> Int -> - Maybe Result -> Int -> IO (Either String F.Frame)- handlePartial rec sock mx mbR step = do- eiS <- getInput rec sock mx- case eiS of - Left e -> return $ Left e- Right s -> do- let prs = case mbR of - Just r -> A.feed r- _ -> A.parse stompParser- case prs s of- A.Fail _ _ e -> return $ Left $- U.toString s ++ - ": " ++ e- r@(A.Partial _) -> - if step > maxStep - then return $ Left "Message too long!" - else handlePartial rec sock mx (Just r) (step + 1)- - A.Done str f -> do- unless (B.null str) $ putBuffer rec str- return $ Right f- - getInput :: Receiver -> S.Socket -> Int -> IO (Either String B.ByteString)- getInput rec sock mx = do- mbB <- getBuffer rec- case mbB of- Just s -> -#ifdef _DEBUG- do putStrLn $ "In Buffer: " ++ U.toString s-#endif- return $ Right s- Nothing -> do- s <- BS.recv sock mx- if B.null s - then return $ Left "Peer disconnected"- else -#ifdef _DEBUG- do putStrLn $ "Receiving: " ++ U.toString s-#endif- if B.null s- then getInput rec sock mx- else return $ Right s
Network/Mom/Stompl/Client/State.hs view
@@ -1,17 +1,19 @@-{-# Language BangPatterns #-}+{-# Language BangPatterns,CPP #-} module State ( msgContent, numeric, ms,- Connection(..),+ Connection(..), mkConnection,+ connected, getVersion, Copt(..), oHeartBeat, oMaxRecv,- oAuth, oCliId, oStomp,+ oAuth, oCliId, oStomp, oTmo, oTLS, Transaction(..), Topt(..), hasTopt, tmo, TxState(..), Receipt,- mkConnection,+ Message(..), mkMessage, MsgId(..),+ Subscription(..), mkSub, logSend, logReceive,- addCon, rmCon, getCon,+ addCon, rmCon, getCon, withCon, updCon, addSub, addDest, getSub, getDest, rmSub, rmDest, mkTrn,@@ -21,11 +23,10 @@ txPendingAck, txReceipts, addAck, rmAck, addRec, rmRec, forceRmRec,- checkReceipt)+ checkReceipt) where - import qualified Protocol as P- import Factory + import qualified Factory as Fac import qualified Network.Mom.Stompl.Frame as F import Network.Mom.Stompl.Client.Exception@@ -38,13 +39,21 @@ import Data.List (find) import Data.Char (isDigit) import Data.Time.Clock+ + import Data.Conduit.Network.TLS (TLSClientConfig, + tlsClientConfig,+ tlsClientUseTLS) + import qualified Data.ByteString.Char8 as B++ import Codec.MIME.Type as Mime (Type) + ------------------------------------------------------------------------ -- | Returns the content of the message in the format -- produced by an in-bound converter ------------------------------------------------------------------------- msgContent :: P.Message a -> a- msgContent = P.msgCont+ msgContent :: Message a -> a+ msgContent = msgCont ------------------------------------------------------------------------ -- some helpers@@ -77,14 +86,27 @@ eq x y = fst x == fst y -- | Just a nicer word for 'Rec'- type Receipt = Rec+ type Receipt = Fac.Rec ------------------------------------------------------------------------ -- Connection ------------------------------------------------------------------------ data Connection = Connection {- conId :: Con,- conCon :: P.Connection,+ conId :: Fac.Con,+ conAddr :: String, -- the broker's IP address+ conPort :: Int, -- the broker's port+ conMax :: Int, -- max receive+ conUsr :: String, -- user+ conPwd :: String, -- passcode+ conCli :: String, -- client-id+ conSrv :: String, -- server description+ conSes :: String, -- session identifier (from broker)+ conVers :: [F.Version], -- the accepted versions+ -- the agreed version + -- after connect+ conBeat :: F.Heart, -- the heart beat+ conChn :: Chan F.Frame, -- sender channel+ conBrk :: Bool, conOwner :: ThreadId, conHisBeat :: UTCTime, conMyBeat :: UTCTime, @@ -94,11 +116,20 @@ conThrds :: [ThreadEntry], conErrs :: [F.Frame], conRecs :: [Receipt],- conAcks :: [P.MsgId]}+ conAcks :: [MsgId]} instance Eq Connection where c1 == c2 = conId c1 == conId c2 + mkConnection :: Fac.Con -> String -> Int -> + Int -> String -> String -> + String -> [F.Version] -> F.Heart -> + Chan F.Frame -> ThreadId -> + UTCTime -> [Copt] -> Connection+ mkConnection cid host port mx usr pwd ci vs hs chn myself t os = + Connection cid host port mx usr pwd ci "" "" vs hs chn False + myself t t (oWaitBroker os) [] [] [] [] [] []+ ------------------------------------------------------------------------- -- | Options passed to a connection -------------------------------------------------------------------------@@ -119,28 +150,65 @@ OWaitBroker Int | -- | The maximum size of TCP/IP packets.- -- Indirectly, this options also defines the- -- maximum message size which is /10 * maxReceive/.- -- By default, the maximum packet size is 1024 bytes.+ -- This option is currently ignored.+ -- Instead, 'Data.Conduit.Network' defines+ -- the packet size (currently hard-wired 4KB).+ -- The maximum message size is 1024 times this value,+ -- /i.e./ 4MB. OMaxRecv Int | -- | This option defines the client\'s bid- -- for negotiating heart beats (see 'F.HeartBeat'). + -- for negotiating heartbeats providing + -- an accepted lower and upper bound+ -- expessed as milliseconds between heartbeats. -- By default, no heart beats are sent or accepted OHeartBeat (F.Heart) | -- | Authentication: user and password OAuth String String | - -- | Identification: specifies the JMS Client ID for persistant connections+ -- | Identification: specifies the JMS Client ID for persistent connections OClientId String | -- | With this option set, "connect" will use -- a "STOMP" frame instead of a "CONNECT" frame- OStomp + OStomp | - deriving (Eq, Show)+ -- | Connection timeout in milliseconds;+ -- if the broker does not respond to a connect request+ -- within this time frame, a 'ConnectException' is thrown.+ -- If the value is <= 0, the program will wait forever.+ OTmo Int | + -- | 'TLSClientConfig'+ -- (see 'Data.Conduit.Network.TLS' for details)+ -- for TLS connections.+ -- If the parameter is not given, + -- a plain TCP/IP connection is used.+ OTLS TLSClientConfig++ instance Show Copt where+ show (OWaitBroker i) = "OWaitBroker" ++ show i+ show (OMaxRecv i) = "OMaxRecv " ++ show i+ show (OHeartBeat h) = "OHeartBeat " ++ show h+ show (OAuth u p) = "OAuth " ++ u ++ "/" ++ p+ show (OClientId u) = "OClientId " ++ u+ show OStomp = "OStomp" + show (OTmo i) = "OTmo" ++ show i+ show (OTLS c) = "OTLS " ++ show (tlsClientUseTLS c)++ instance Eq Copt where+ (OWaitBroker i1) == (OWaitBroker i2) = i1 == i2+ (OMaxRecv i1) == (OMaxRecv i2) = i1 == i2 + (OHeartBeat h1) == (OHeartBeat h2) = h1 == h2+ (OAuth u1 p1) == (OAuth u2 p2) = u1 == u2 && p1 == p2+ (OClientId u1) == (OClientId u2) = u1 == u2+ (OTmo i1) == (OTmo i2) = i1 == i2+ (OTLS c1) == (OTLS c2) = tlsClientUseTLS c1 && + tlsClientUseTLS c2+ OStomp == OStomp = True+ _ == _ = False+ ------------------------------------------------------------------------ -- Same constructor ------------------------------------------------------------------------@@ -151,6 +219,8 @@ is (OAuth _ _) (OAuth _ _) = True is (OClientId _) (OClientId _) = True is (OStomp ) (OStomp ) = True+ is (OTmo _ ) (OTmo _ ) = True+ is (OTLS _ ) (OTLS _ ) = True is _ _ = False noWait :: Int@@ -198,17 +268,24 @@ Just _ -> True Nothing -> False - findCon :: Con -> [Connection] -> Maybe Connection- findCon cid = find (\c -> conId c == cid)+ oTmo :: [Copt] -> Int+ oTmo os = case find (is $ OTmo 0) os of+ Just (OTmo i) -> i+ _ -> 0 - mkConnection :: Con -> P.Connection -> ThreadId -> UTCTime -> [Copt] -> Connection- mkConnection cid c myself t os = Connection cid c myself t t (oWaitBroker os)- [] [] [] [] [] []+ oTLS :: String -> Int -> [Copt] -> TLSClientConfig+ oTLS h p os = case find (is $ OTLS dcfg) os of+ Just (OTLS cfg) -> cfg+ _ -> dcfg+ where dcfg = (tlsClientConfig p $ B.pack h){tlsClientUseTLS=False} - addAckToCon :: P.MsgId -> Connection -> Connection+ findCon :: Fac.Con -> [Connection] -> Maybe Connection+ findCon cid = find (\c -> conId c == cid)++ addAckToCon :: MsgId -> Connection -> Connection addAckToCon mid c = c {conAcks = mid : conAcks c} - rmAckFromCon :: P.MsgId -> Connection -> Connection+ rmAckFromCon :: MsgId -> Connection -> Connection rmAckFromCon mid c = c {conAcks = delete' mid $ conAcks c} addRecToCon :: Receipt -> Connection -> Connection@@ -222,17 +299,28 @@ Nothing -> True Just _ -> False + ---------------------------------------------------------------------+ -- Connection interfaces+ ---------------------------------------------------------------------+ connected :: Connection -> Bool+ connected = conBrk ++ getVersion :: Connection -> F.Version+ getVersion c = if null (conVers c) + then defVersion+ else head $ conVers c+ ------------------------------------------------------------------------ -- Sub, Dest and Thread Entry ------------------------------------------------------------------------- type SubEntry = (Sub, Chan F.Frame)+ type SubEntry = (Fac.Sub, Chan F.Frame) type DestEntry = (String, Chan F.Frame) type ThreadEntry = (ThreadId, [Transaction]) addSubToCon :: SubEntry -> Connection -> Connection addSubToCon s c = c {conSubs = s : conSubs c} - getSub :: Sub -> Connection -> Maybe (Chan F.Frame)+ getSub :: Fac.Sub -> Connection -> Maybe (Chan F.Frame) getSub sid c = lookup sid (conSubs c) rmSubFromCon :: SubEntry -> Connection -> Connection@@ -255,29 +343,29 @@ setMyTime :: UTCTime -> Connection -> Connection setMyTime t c = c {conMyBeat = t} - updCon :: Connection -> [Connection] -> [Connection]- updCon c cs = let !c' = delete' c cs in c:c'+ _updCon :: Connection -> [Connection] -> [Connection]+ _updCon c cs = let !c' = delete' c cs in c:c' ------------------------------------------------------------------------ -- Transaction ------------------------------------------------------------------------ data Transaction = Trn {- txId :: Tx,+ txId :: Fac.Tx, txState :: TxState, txTmo :: Int, txAbrtAck :: Bool, txAbrtRc :: Bool,- txAcks :: [P.MsgId],+ txAcks :: [MsgId], txRecs :: [Receipt] } instance Eq Transaction where t1 == t2 = txId t1 == txId t2 - findTx :: Tx -> [Transaction] -> Maybe Transaction+ findTx :: Fac.Tx -> [Transaction] -> Maybe Transaction findTx tx = find (\x -> txId x == tx) - mkTrn :: Tx -> [Topt] -> Transaction+ mkTrn :: Fac.Tx -> [Topt] -> Transaction mkTrn tx os = Trn { txId = tx, txState = TxStarted,@@ -318,10 +406,11 @@ -- a value /> 0/ is given, the transaction will -- wait for pending receipts; otherwise -- the transaction will be aborted with 'TxException'.- -- Note that it, usually, does not make sense to use- -- this options without 'OTimeout',- -- since it is in all probability that a receipt - -- has not yet been confirmed when the transaction terminates.+ -- Note that it usually does not make sense to use+ -- this option without 'OTimeout',+ -- since, in all probability, there will be receipts + -- that have not yet been confirmed + -- when the transaction terminates. | OWithReceipts -- | If a message has been received from a -- queue with 'OMode' option other @@ -351,10 +440,10 @@ setTxState :: TxState -> Transaction -> Transaction setTxState st t = t {txState = st} - addAckToTx :: P.MsgId -> Transaction -> Transaction+ addAckToTx :: MsgId -> Transaction -> Transaction addAckToTx mid t = t {txAcks = mid : txAcks t} - rmAckFromTx :: P.MsgId -> Transaction -> Transaction+ rmAckFromTx :: MsgId -> Transaction -> Transaction rmAckFromTx mid t = t {txAcks = delete' mid $ txAcks t} addRecToTx :: Receipt -> Transaction -> Transaction@@ -388,13 +477,19 @@ ------------------------------------------------------------------------ -- get connection from state ------------------------------------------------------------------------- getCon :: Con -> IO Connection+ getCon :: Fac.Con -> IO Connection getCon cid = withCon cid $ \c -> return (c, c) ------------------------------------------------------------------------+ -- update connection + ------------------------------------------------------------------------+ updCon :: Fac.Con -> Connection -> IO ()+ updCon cid c = withCon cid $ \_ -> return (c, ()) ++ ------------------------------------------------------------------------ -- remove connection from state ------------------------------------------------------------------------- rmCon :: Con -> IO ()+ rmCon :: Fac.Con -> IO () rmCon cid = modifyMVar_ con $ \cs -> case findCon cid cs of Nothing -> return cs@@ -403,7 +498,7 @@ ------------------------------------------------------------------------ -- Apply an action that may change a connection to the state ------------------------------------------------------------------------- withCon :: Con -> (Connection -> IO (Connection, a)) -> IO a+ withCon :: Fac.Con -> (Connection -> IO (Connection, a)) -> IO a withCon cid op = modifyMVar con (\cs -> case findCon cid cs of Nothing -> @@ -411,38 +506,38 @@ "No such Connection: " ++ show cid Just c -> do (c', x) <- op c- let cs' = updCon c' cs+ let cs' = _updCon c' cs return (cs', x)) ------------------------------------------------------------------------ -- Log heart-beats ------------------------------------------------------------------------- logTime :: Con -> (UTCTime -> Connection -> Connection) -> IO ()+ logTime :: Fac.Con -> (UTCTime -> Connection -> Connection) -> IO () logTime cid f = getCurrentTime >>= \t -> withCon cid (\c -> return (f t c, ())) - logSend :: Con -> IO ()+ logSend :: Fac.Con -> IO () logSend cid = logTime cid setMyTime - logReceive :: Con -> IO ()+ logReceive :: Fac.Con -> IO () logReceive cid = logTime cid setHisTime ------------------------------------------------------------------------ -- add and remove sub and dest ------------------------------------------------------------------------- addSub :: Con -> SubEntry -> IO ()+ addSub :: Fac.Con -> SubEntry -> IO () addSub cid s = withCon cid $ \c -> return (addSubToCon s c, ()) - addDest :: Con -> DestEntry -> IO ()+ addDest :: Fac.Con -> DestEntry -> IO () addDest cid d = withCon cid $ \c -> return (addDestToCon d c, ()) - rmSub :: Con -> Sub -> IO ()+ rmSub :: Fac.Con -> Fac.Sub -> IO () rmSub cid sid = withCon cid rm where rm c = case getSub sid c of Nothing -> return (c, ()) Just ch -> return (rmSubFromCon (sid, ch) c, ()) - rmDest :: Con -> String -> IO ()+ rmDest :: Fac.Con -> String -> IO () rmDest cid dst = withCon cid rm where rm c = case getDest dst c of Nothing -> return (c, ())@@ -452,7 +547,7 @@ -- add transaction to connection -- Note: transactions are kept per threadId ------------------------------------------------------------------------- addTx :: Transaction -> Con -> IO ()+ addTx :: Transaction -> Fac.Con -> IO () addTx t cid = withCon cid $ \c -> do tid <- myThreadId case lookup tid (conThrds c) of@@ -467,7 +562,7 @@ -- get transaction from connection -- Note: transactions are kept per threadId ------------------------------------------------------------------------- getTx :: Tx -> Connection -> IO (Maybe Transaction)+ getTx :: Fac.Tx -> Connection -> IO (Maybe Transaction) getTx tx c = do tid <- myThreadId case lookup tid (conThrds c) of@@ -477,7 +572,7 @@ ------------------------------------------------------------------------ -- apply an action that may change a transaction to the state ------------------------------------------------------------------------- updTx :: Tx -> Con -> (Transaction -> Transaction) -> IO ()+ updTx :: Fac.Tx -> Fac.Con -> (Transaction -> Transaction) -> IO () updTx tx cid f = withCon cid $ \c -> do tid <- myThreadId case lookup tid (conThrds c) of@@ -498,13 +593,13 @@ ------------------------------------------------------------------------ -- update transaction state ------------------------------------------------------------------------- updTxState :: Tx -> Con -> TxState -> IO ()+ updTxState :: Fac.Tx -> Fac.Con -> TxState -> IO () updTxState tx cid st = updTx tx cid (setTxState st) ------------------------------------------------------------------------ -- get current transaction for thread ------------------------------------------------------------------------- getCurTx :: Connection -> IO (Maybe Tx)+ getCurTx :: Connection -> IO (Maybe Fac.Tx) getCurTx c = do tid <- myThreadId case lookup tid (conThrds c) of@@ -538,7 +633,7 @@ -- add a pending ack either to the current transaction -- or - if there is no transaction - to the connection ------------------------------------------------------------------------- addAck :: Con -> P.MsgId -> IO ()+ addAck :: Fac.Con -> MsgId -> IO () addAck cid mid = do let toTx = addAckToTx mid let toCon = addAckToCon mid@@ -548,7 +643,7 @@ -- remove a pending ack either from the current transaction -- or - if there is no transaction - from the connection ------------------------------------------------------------------------- rmAck :: Con -> P.MsgId -> IO ()+ rmAck :: Fac.Con -> MsgId -> IO () rmAck cid mid = do let fromTx = rmAckFromTx mid let fromCon = rmAckFromCon mid@@ -558,7 +653,7 @@ -- add a pending receipt either to the current transaction -- or - if there is no transaction - to the connection ------------------------------------------------------------------------- addRec :: Con -> Receipt -> IO ()+ addRec :: Fac.Con -> Receipt -> IO () addRec cid r = do let toTx = addRecToTx r let toCon = addRecToCon r@@ -568,7 +663,7 @@ -- remove a pending receipt either from the current transaction -- or - if there is no transaction - from the connection ------------------------------------------------------------------------- rmRec :: Con -> Receipt -> IO ()+ rmRec :: Fac.Con -> Receipt -> IO () rmRec cid r = do let fromTx = rmRecFromTx r let fromCon = rmRecFromCon r@@ -578,7 +673,7 @@ -- search for a receipt either in connection or transactions. -- this is used by the listener (which is not in the thread list) ------------------------------------------------------------------------- forceRmRec :: Con -> Receipt -> IO ()+ forceRmRec :: Fac.Con -> Receipt -> IO () forceRmRec cid r = withCon cid doRmRec where doRmRec c = case find (== r) $ conRecs c of@@ -594,7 +689,7 @@ ------------------------------------------------------------------------ checkCurTx :: (Transaction -> Bool) -> (Connection -> Bool) -> - Con -> IO Bool+ Fac.Con -> IO Bool checkCurTx onTx onCon cid = do c <- getCon cid tid <- myThreadId@@ -606,7 +701,7 @@ ------------------------------------------------------------------------ -- check a receipt ------------------------------------------------------------------------- checkReceipt :: Con -> Receipt -> IO Bool+ checkReceipt :: Fac.Con -> Receipt -> IO Bool checkReceipt cid r = do let onTx = checkReceiptTx r let onCon = checkReceiptCon r@@ -615,7 +710,7 @@ ------------------------------------------------------------------------ -- remove a specific transaction ------------------------------------------------------------------------- rmThisTx :: Tx -> Con -> IO ()+ rmThisTx :: Fac.Tx -> Fac.Con -> IO () rmThisTx tx cid = withCon cid $ \c -> do tid <- myThreadId case lookup tid (conThrds c) of@@ -637,7 +732,7 @@ ------------------------------------------------------------------------ -- remove the current transaction ------------------------------------------------------------------------- rmTx :: Con -> IO ()+ rmTx :: Fac.Con -> IO () rmTx cid = withCon cid $ \c -> do tid <- myThreadId case lookup tid (conThrds c) of@@ -652,4 +747,84 @@ deleteBy' eq (tid, ts) (conThrds c)}, ()) else return (c {conThrds = (tid, ts') : deleteBy' eq (tid, ts) (conThrds c)}, ())++ ---------------------------------------------------------------------+ -- Default version, when broker does not send a version+ ---------------------------------------------------------------------+ defVersion :: F.Version+ defVersion = (1,2)++ ---------------------------------------------------------------------+ -- Subscribe abstraction+ ---------------------------------------------------------------------+ data Subscription = Subscription {+ subId :: Fac.Sub, -- subscribe identifier+ subName :: String, -- queue name+ subMode :: F.AckMode -- ack mode+ }+ deriving (Show)++ mkSub :: Fac.Sub -> String -> F.AckMode -> Subscription+ mkSub sid qn am = Subscription {+ subId = sid,+ subName = qn,+ subMode = am}++ ---------------------------------------------------------------------+ -- | Message Identifier+ ---------------------------------------------------------------------+ data MsgId = MsgId String | NoMsg+ deriving (Eq)++ instance Show MsgId where+ show (MsgId s) = s+ show (NoMsg) = ""++ ------------------------------------------------------------------------+ -- | Any content received from a queue+ -- is wrapped in a message.+ -- It is, in particular, the return value of /readQ/.+ ------------------------------------------------------------------------+ data Message a = Msg {+ -- | The message Identifier+ msgId :: MsgId,+ -- | The subscription+ msgSub :: Fac.Sub,+ -- | The destination+ msgDest :: String,+ -- | The Ack identifier+ msgAck :: String,+ -- | The Stomp headers+ -- that came with the message+ msgHdrs :: [F.Header],+ -- | The /MIME/ type of the content+ msgType :: Mime.Type,+ -- | The length of the + -- encoded content+ msgLen :: Int,+ -- | The transaction, in which + -- the message was received+ msgTx :: Fac.Tx,+ -- | The encoded content + msgRaw :: B.ByteString,+ -- | The content + msgCont :: a}+ + ---------------------------------------------------------------------+ -- Create a message+ ---------------------------------------------------------------------+ mkMessage :: MsgId -> Fac.Sub -> String -> String ->+ Mime.Type -> Int -> Fac.Tx -> + B.ByteString -> a -> Message a+ mkMessage mid sub dst ak typ len tx raw cont = Msg {+ msgId = mid,+ msgSub = sub,+ msgDest = dst,+ msgAck = ak,+ msgHdrs = [], + msgType = typ, + msgLen = len, + msgTx = tx,+ msgRaw = raw,+ msgCont = cont}
+ Network/Mom/Stompl/Client/Stream.hs view
@@ -0,0 +1,119 @@+module Stream+where++ import qualified Data.Conduit as C+ import Data.Conduit (($$),(=$))+ import Data.Conduit.Network++ import qualified Data.ByteString.Char8 as B+ import qualified Data.ByteString.UTF8 as U++ import Network.Mom.Stompl.Parser (stompParser)+ import qualified Network.Mom.Stompl.Frame as F+ import Network.Mom.Stompl.Client.Exception++ import Control.Monad (forever)+ import Control.Monad.Trans (liftIO)+ import Control.Concurrent++ import qualified Data.Attoparsec.ByteString as A ++ ------------------------------------------------------------------------+ -- Error Handler+ ------------------------------------------------------------------------+ type EH = StomplException -> IO ()++ ------------------------------------------------------------------------+ -- A TCP/IP fragment read by the Conduit Client has 4096 bytes.+ -- We allow 1000 fragments = 1024 * 4096 Bytes = 4MB+ ------------------------------------------------------------------------+ maxStep :: Int+ maxStep = 1024++ ------------------------------------------------------------------------+ -- Sender thread: get a Frame from a pipe, convert it into a ByteString+ -- and send it through a socket + ------------------------------------------------------------------------+ sender :: AppData -> Chan F.Frame -> IO ()+ sender ad ip = pipeSource ip $$ stream =$ appSink ad++ ------------------------------------------------------------------------+ -- Receiver thread: get a ByteStream through a socket,+ -- parse it to a Frame and send it through a pipe+ ------------------------------------------------------------------------+ receiver :: AppData -> Chan F.Frame -> EH -> IO ()+ receiver ad ip eh = appSource ad $$ parseC eh =$ pipeSink ip ++ ------------------------------------------------------------------------+ -- Put a frame into a pipe (a channel)+ ------------------------------------------------------------------------+ pipeSink :: Chan F.Frame -> C.Sink F.Frame IO ()+ pipeSink ch = C.awaitForever (liftIO . writeChan ch)++ ------------------------------------------------------------------------+ -- Read a frame from a pipe (a channel)+ ------------------------------------------------------------------------+ pipeSource :: Chan F.Frame -> C.Source IO F.Frame+ pipeSource ch = forever (liftIO (readChan ch) >>= C.yield)++ ------------------------------------------------------------------------+ -- Convert a frame to a ByteString+ ------------------------------------------------------------------------+ stream :: C.ConduitM F.Frame B.ByteString IO ()+ stream = C.awaitForever (C.yield . F.putFrame)++ ------------------------------------------------------------------------+ -- Parse a Frame from a ByteString+ ------------------------------------------------------------------------+ parseC :: EH -> C.ConduitM B.ByteString F.Frame IO ()+ parseC eh = goOn+ where goOn = go (A.parse stompParser) 0 -- start with a clean parser+ go prs step = do+ mbNew <- C.await+ case mbNew of + Nothing -> return () -- socket was closed+ Just s -> case parseAll prs s of+ -- parse error: call the error handler ---------+ Left e -> liftIO (eh $ ProtocolException e)+ >> goOn+ -- we got a result -----------------------------+ Right (prs', fs) -> do+ -- Do we have (at least) 1 frame to send? ----+ step' <- if null fs then return (step+1) + else mapM_ C.yield fs >>+ return 0+ -- Too many fragments ------------------------+ if step' > maxStep + then liftIO (eh $ ProtocolException + "Message too long!") ++ -- Continue with the current parser ----------+ else go prs' step'++ ------------------------------------------------------------------------+ -- A parser is something that converts a ByteString into a Frame+ ------------------------------------------------------------------------+ type Parser = B.ByteString -> A.Result F.Frame++ ------------------------------------------------------------------------+ -- Continue parsing until we have a complete frame+ ------------------------------------------------------------------------+ parseAll :: Parser -> B.ByteString -> + Either String (Parser, [F.Frame])+ parseAll prs s = case prs s of+ -- We failed ----------------------------------------+ A.Fail _ _ e -> Left $ U.toString s ++ ": " ++ e++ -- We have a partial result and continue -------------+ -- feeding this partial result --------------------+ r@(A.Partial _) -> Right (A.feed r, [])++ -- We are done ---------------------------------------+ A.Done s' f -> + if B.null s' + then Right (A.parse stompParser, [f])+ -- but there may be a leftover -------------------+ else case parseAll (A.parse stompParser) s' of+ Left e -> Left e+ Right (prs', fs) -> Right (prs',f:fs)+
stomp-queue.cabal view
@@ -1,7 +1,7 @@ Name: stomp-queue-Version: 0.1.4+Version: 0.2.0 Cabal-Version: >= 1.8-Copyright: Copyright (c) Tobias Schoofs, 2011 - 2014+Copyright: Copyright (c) Tobias Schoofs, 2011 - 2015 License: LGPL license-file: license/lgpl-3.0.txt Author: Tobias Schoofs@@ -24,10 +24,8 @@ The Stomp Queue library provides a Stomp client, using abstractions like 'Connection', 'Transaction', 'Queue' and 'Message'.- This basic abstractions are implemented in the module /Queue/.- The /Patterns/ module adds an abstraction layer- on top of queues, in particular an implementation- of the /client-server/ pattern.+ The library may use TLS for secure connections + to brokers that provide security over TLS. . @@ -42,6 +40,29 @@ . + [0.2.0] Changes:++ .++ - Low-level sockets were replaced by network-conduit-tls+ (Be aware that this change might introduce some+ subtle changes in behaviour concerning in particular + performance and connection handling)++ .++ - OMaxRecv not used anymore++ .++ - New Option OTLS for TLS connections++ .++ - New Option OTmo to specify a connection timeout++ .+ [0.1.4] Changes: . @@ -157,20 +178,23 @@ Library- Build-Depends: base >= 4.0 && <= 5.0,- bytestring >= 0.10,- utf8-string >= 0.3.6,- attoparsec >= 0.9.1.1,- split >= 0.1.4.1,- network >= 2.4.0.0,- stompl >= 0.1.1,- mime >= 0.3.3,- time >= 1.1.4+ Build-Depends: base >= 4.0 && <= 5.0,+ bytestring >= 0.10,+ utf8-string >= 0.3.6,+ attoparsec >= 0.9.1.1,+ split >= 0.1.4.1,+ mtl >= 2.2.0.1,+ conduit >= 1.2.3.1,+ conduit-extra >= 1.1.6.2,+ network-conduit-tls >= 1.1.0.2,+ stompl >= 0.1.1,+ mime >= 0.3.3,+ time >= 1.1.4 hs-source-dirs: Network/Mom/Stompl/Client, . Exposed-Modules: Network.Mom.Stompl.Client.Queue, Network.Mom.Stompl.Client.Exception, Network.Mom.Stompl.Client.Patterns- other-modules: Socket, Protocol, State, Factory+ other-modules: Stream, State, Factory