diff --git a/Network/Mom/Stompl/Client/Exception.hs b/Network/Mom/Stompl/Client/Exception.hs --- a/Network/Mom/Stompl/Client/Exception.hs +++ b/Network/Mom/Stompl/Client/Exception.hs @@ -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 diff --git a/Network/Mom/Stompl/Client/Protocol.hs b/Network/Mom/Stompl/Client/Protocol.hs deleted file mode 100644 --- a/Network/Mom/Stompl/Client/Protocol.hs +++ /dev/null @@ -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 - 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 @@ -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 diff --git a/Network/Mom/Stompl/Client/Socket.hs b/Network/Mom/Stompl/Client/Socket.hs deleted file mode 100644 --- a/Network/Mom/Stompl/Client/Socket.hs +++ /dev/null @@ -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 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,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} diff --git a/Network/Mom/Stompl/Client/Stream.hs b/Network/Mom/Stompl/Client/Stream.hs new file mode 100644 --- /dev/null +++ b/Network/Mom/Stompl/Client/Stream.hs @@ -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) + diff --git a/stomp-queue.cabal b/stomp-queue.cabal --- a/stomp-queue.cabal +++ b/stomp-queue.cabal @@ -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