stomp-queue 0.0.8 → 0.1.0
raw patch · 4 files changed
+136/−59 lines, 4 filesdep ~stompl
Dependency ranges changed: stompl
Files
- Network/Mom/Stompl/Client/Protocol.hs +55/−46
- Network/Mom/Stompl/Client/Queue.hs +12/−5
- Network/Mom/Stompl/Client/State.hs +12/−2
- stomp-queue.cabal +57/−6
Network/Mom/Stompl/Client/Protocol.hs view
@@ -34,7 +34,7 @@ -- Default version, when broker does not send a version --------------------------------------------------------------------- defVersion :: F.Version- defVersion = (1,1)+ defVersion = (1,2) --------------------------------------------------------------------- -- Connection @@ -97,6 +97,8 @@ msgSub :: Fac.Sub, -- | The destination msgDest :: String,+ -- | The Ack identifier+ msgAck :: String, -- | The Stomp headers -- that came with the message msgHdrs :: [F.Header],@@ -116,19 +118,20 @@ --------------------------------------------------------------------- -- Create a message ---------------------------------------------------------------------- mkMessage :: MsgId -> Fac.Sub -> String -> + mkMessage :: MsgId -> Fac.Sub -> String -> String -> Mime.Type -> Int -> Fac.Tx -> B.ByteString -> a -> Message a- mkMessage mid sub dst typ len tx raw cont = Msg {- msgId = mid,- msgSub = sub,- msgDest = dst,- msgHdrs = [], - msgType = typ, - msgLen = len, - msgTx = tx,- msgRaw = raw,- msgCont = cont}+ 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@@ -188,15 +191,15 @@ -- connect --------------------------------------------------------------------- connect :: String -> Int -> Int -> - String -> String -> String -> + String -> String -> String -> F.FrameType -> [F.Version] -> F.Heart -> [F.Header] -> IO Connection- connect host port mx usr pwd cli vers beat hs = + 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 vers beat hs) + (connectBroker mx t vers beat hs) --------------------------------------------------------------------- -- disconnect either on broker level or on tcp/ip level@@ -282,10 +285,10 @@ "Cannot create Frame: " ++ e Right f -> #ifdef _DEBUG- do when (not $ F.complies (1,1) f) $- putStrLn $ "Frame does not comply with 1.1: " ++ show f + 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+ S.send (getWr c) (getSock c) f --------------------------------------------------------------------- -- hard disconnect (on tcp/ip) level@@ -303,28 +306,34 @@ --------------------------------------------------------------------- -- the hard work on connecting to a broker ---------------------------------------------------------------------- connectBroker :: Int -> [F.Version] -> F.Heart -> [F.Header] ->+ connectBroker :: Int -> F.FrameType -> + [F.Version] -> F.Heart -> [F.Header] -> Connection -> IO Connection- connectBroker mx vers beat hs c = - case mkConF (conAddr c) + 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 -> do- let c' = handleConnected r c- 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}+ 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 ---------------------------------------------------------------------@@ -360,17 +369,17 @@ mkReceipt :: String -> [F.Header] mkReceipt receipt = if null receipt then [] else [F.mkRecHdr receipt] -- mkConF :: String -> String -> String -> String -> - [F.Version] -> F.Heart -> [F.Header] -> Either String F.Frame- mkConF host usr pwd cli vers beat hs = + 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 F.mkConFrame $ [F.mkHostHdr host,- F.mkAcVerHdr $ F.versToVal vers, - F.mkBeatHdr $ F.beatToVal beat] ++- uHdr ++ pHdr ++ cHdr ++ hs+ 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 =@@ -403,7 +412,7 @@ then [] else [F.mkTrnHdr $ show $ msgTx msg] rh = mkReceipt receipt mk = if ok then F.mkAckFrame else F.mkNackFrame- in mk $ F.mkMIdHdr (show $ msgId msg) : (sh ++ rh ++ th)+ in mk $ F.mkIdHdr (msgAck msg) : (sh ++ rh ++ th) mkBeginF :: String -> String -> [F.Header] -> Either String F.Frame mkBeginF tx receipt _ =
Network/Mom/Stompl/Client/Queue.hs view
@@ -1,3 +1,4 @@+{-# Language CPP #-} ------------------------------------------------------------------------------- -- | -- Module : Network/Mom/Stompl/Client/Queue.hs@@ -52,6 +53,9 @@ -- * Acknowledgements -- $stomp_acks ack, ackWith, nack, nackWith,+#ifdef TEST+ frmToMsg, P.msgAck,+#endif -- * Exceptions module Network.Mom.Stompl.Client.Exception -- * Complete Example@@ -324,7 +328,7 @@ -- The versions, we support vers :: [F.Version]- vers = [(1,0), (1,1)]+ vers = [(1,0), (1,1), (1,2)] ------------------------------------------------------------------------ -- | Initialises a connection and executes an 'IO' action.@@ -391,7 +395,9 @@ let mx = oMaxRecv os let (u,p) = oAuth os let (ci) = oCliId os- bracket (P.connect host port mx u p ci vers beat hs)+ 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) @@ -1019,7 +1025,7 @@ let conv = wTo q s <- conv x rc <- if wRec q then mkUniqueRecc else return NoRec- let m = P.mkMessage P.NoMsg NoSub dest+ let m = P.mkMessage P.NoMsg NoSub dest "" mime (B.length s) tx s x when (wRec q) $ addRec (wCon q) rc logSend $ wCon q@@ -1089,8 +1095,8 @@ if not $ P.connected (conCon c) then throwIO $ ConnectException $ "Not connected (" ++ show cid ++ ")"- else if null (show $ P.msgId msg)- then throwIO $ ProtocolException "No message id in message!"+ else if null (show $ P.msgAck msg)+ then throwIO $ ProtocolException "No ack in message!" else do tx <- getCurTx c >>= (\mbT -> case mbT of@@ -1285,6 +1291,7 @@ 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
Network/Mom/Stompl/Client/State.hs view
@@ -4,7 +4,7 @@ Connection(..), Copt(..), oHeartBeat, oMaxRecv,- oAuth, oCliId,+ oAuth, oCliId, oStomp, Transaction(..), Topt(..), hasTopt, tmo, TxState(..),@@ -133,8 +133,12 @@ OAuth String String | -- | Identification: specifies the JMS Client ID for persistant connections- OClientId String+ OClientId String | + -- | With this option set, "connect" will use + -- a "STOMP" frame instead of a "CONNECT" frame+ OStomp + deriving (Eq, Show) ------------------------------------------------------------------------@@ -146,6 +150,7 @@ is (OHeartBeat _) (OHeartBeat _) = True is (OAuth _ _) (OAuth _ _) = True is (OClientId _) (OClientId _) = True+ is (OStomp ) (OStomp ) = True is _ _ = False noWait :: Int@@ -187,6 +192,11 @@ oCliId os = case find (is $ OClientId "") os of Just (OClientId i) -> i _ -> noCliId++ oStomp :: [Copt] -> Bool+ oStomp os = case find (is OStomp) os of+ Just _ -> True+ Nothing -> False findCon :: Con -> [Connection] -> Maybe Connection findCon cid = find (\c -> conId c == cid)
stomp-queue.cabal view
@@ -1,5 +1,5 @@ Name: stomp-queue-Version: 0.0.8+Version: 0.1.0 Cabal-Version: >= 1.8 Copyright: Copyright (c) Tobias Schoofs, 2011 - 2013 License: LGPL@@ -24,6 +24,10 @@ 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. . @@ -38,6 +42,39 @@ . + [0.1.0] Major changes:++ .++ - Compliance with Stomp 1.2:++ .++ - There are major changes in the frame format,+ please refer to the documentation of the + stompl package, version 0.1.0, there are important changes+ that may impact messages for older versions!++ .++ - When generating an Ack frame,+ the /id/ header is by default taken from the /ack/ header+ in the corresponding Message frame.+ Should there be no /ack/ header or if its value is empty,+ the value of the header /message-id/ is taken.+ This behaviour complies with 1.2 + for brokers supporting this version,+ but also continues to work with 1.1 brokers.++ .++ - It is now possible to send a Stomp frame+ to connect to a broker (the broker, of course,+ has to accept Stomp frames and process them correctly).+ There is a new Copt (/OStomp/) to support this feature.++ .+ [0.0.8] Client/Server on top of Queues. .@@ -54,11 +91,25 @@ . - [0.0.5] Underscore functions removed; - with* functions: withWriter, withPair;- new option for connection (ClientId);- Headers for broker-specific options can be passed to connection.+ [0.0.5] Major changes:+ + . + - Underscore functions (/withConnection_/) removed; ++ .++ - New /with*/ functions: /withWriter/, /withPair/;++ .++ - New option for connection (ClientId);++ .++ - Headers for broker-specific options can be passed to connection+ (this changes the /withConnection/ type signature!)+ . [0.0.3] New interface writeAdHoc@@ -79,7 +130,7 @@ attoparsec >= 0.9.1.1, split >= 0.1.4.1, network >= 2.4.0.0,- stompl >= 0.0.3,+ stompl >= 0.1.0, mime >= 0.3.3, time >= 1.1.4