diff --git a/Network/Mom/Stompl/Client/Protocol.hs b/Network/Mom/Stompl/Client/Protocol.hs --- a/Network/Mom/Stompl/Client/Protocol.hs +++ b/Network/Mom/Stompl/Client/Protocol.hs @@ -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 _ = 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 @@ -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 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 @@ -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) diff --git a/stomp-queue.cabal b/stomp-queue.cabal --- a/stomp-queue.cabal +++ b/stomp-queue.cabal @@ -1,5 +1,5 @@ Name: stomp-queue -Version: 0.0.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