amqp 0.6.0 → 0.7.0
raw patch · 6 files changed
+162/−31 lines, 6 filesdep +clockdep +hspecdep +hspec-expectations
Dependencies added: clock, hspec, hspec-expectations, split
Files
- Network/AMQP.hs +33/−3
- Network/AMQP/Helpers.hs +17/−2
- Network/AMQP/Internal.hs +65/−20
- Network/AMQP/Protocol.hs +8/−1
- amqp.cabal +19/−5
- test/Runner.hs +20/−0
Network/AMQP.hs view
@@ -119,13 +119,18 @@ rabbitCRdemo, -- * Exceptions - AMQPException(..) + AMQPException(..), + + -- * URI parsing + fromURI ) where import Control.Concurrent +import Data.List.Split (splitOn) import Data.Binary import Data.Binary.Put import Network +import Network.URI (unEscapeString) import Data.Text (Text) import qualified Data.ByteString.Lazy as BL @@ -137,6 +142,7 @@ import Network.AMQP.Types import Network.AMQP.Generated import Network.AMQP.Internal +import Network.AMQP.Helpers ----- EXCHANGE ----- @@ -500,7 +506,7 @@ -- -- * max frame size: @131072@ -- --- * no heartbeat expected from the server +-- * use the heartbeat delay suggested by the broker -- -- * no limit on the number of used channels -- @@ -537,7 +543,7 @@ amqplain :: Text -> Text -> SASLMechanism amqplain loginName loginPassword = SASLMechanism "AMQPLAIN" initialResponse Nothing where - initialResponse = BL.toStrict $ BL.drop 4 $ runPut $ put $ FieldTable $ M.fromList [("LOGIN",FVString loginName), ("PASSWORD", FVString loginPassword)] + initialResponse = toStrict $ BL.drop 4 $ runPut $ put $ FieldTable $ M.fromList [("LOGIN",FVString loginName), ("PASSWORD", FVString loginPassword)] -- | The @RABBIT-CR-DEMO@ SASL mechanism needs to be explicitly enabled on the RabbitMQ server and should only be used for demonstration purposes of the challenge-response cycle. -- See <http://www.rabbitmq.com/authentication.html>. @@ -561,3 +567,27 @@ False )) return () + +-- | Parses amqp standard URI of the form @amqp://user:password@host:port/vhost@ and returns a @ConnectionOpts@ for use with @openConnection''@ +-- | Any of these fields may be empty and will be replaced with defaults from @amqp://guest:guest@localhost:5672/@ +fromURI :: String -> ConnectionOpts +fromURI uri = defaultConnectionOpts { + coServers = [(host,fromIntegral nport)], + coVHost = (T.pack vhost), + coAuth = [plain (T.pack uid) (T.pack pw)] + } + where (host,nport,uid,pw,vhost) = fromURI' uri + +fromURI' :: String -> (String,Int,String,String,String) +fromURI' uri = (unEscapeString host, nport, unEscapeString (dropWhile (=='/') uid), unEscapeString pw, unEscapeString vhost) + where (pre :suf : _) = splitOn "@" (uri ++ "@" ) -- look mom, no regexp dependencies + (pro :uid' :pw':_) = splitOn ":" (pre ++ "::") + (hnp :thost: _) = splitOn "/" (suf ++ "/" ) + (hst':port : _) = splitOn ":" (hnp ++ ":" ) + vhost = if null thost then "/" else thost + dport = if pro == "amqps" then 5671 else 5672 + nport = if null port then dport else read port + uid = if null uid' then "guest" else uid' + pw = if null pw' then "guest" else pw' + host = if null hst' then "localhost" else hst' +
Network/AMQP/Helpers.hs view
@@ -1,7 +1,10 @@+{-# LANGUAGE ScopedTypeVariables #-} module Network.AMQP.Helpers where -import Control.Concurrent.MVar +import Control.Concurrent import Control.Monad +import Data.Int (Int64) +import System.Clock import qualified Data.ByteString.Char8 as BS import qualified Data.ByteString.Lazy.Char8 as BL @@ -36,4 +39,16 @@ chooseMin :: Ord a => a -> Maybe a -> a chooseMin a (Just b) = min a b -chooseMin a Nothing = a+chooseMin a Nothing = a + +getTimestamp :: IO Int64 +getTimestamp = fmap µs $ getTime Monotonic + where + seconds spec = (fromIntegral . sec) spec * 1000 * 1000 + micros spec = (fromIntegral . nsec) spec `div` 1000 + µs spec = (seconds spec) + (micros spec) + +scheduleAtFixedRate :: Int -> IO () -> IO ThreadId +scheduleAtFixedRate interval_µs action = forkIO $ forever $ do + action + threadDelay interval_µs
Network/AMQP/Internal.hs view
@@ -6,6 +6,7 @@ import Data.Binary import Data.Binary.Get import Data.Binary.Put as BPut +import Data.Int (Int64) import Data.Maybe import Data.Text (Text) import Data.Typeable @@ -132,19 +133,21 @@ connClosedLock :: MVar (), -- used by closeConnection to block until connection-close handshake is complete connWriteLock :: MVar (), -- to ensure atomic writes to the socket connClosedHandlers :: MVar [IO ()], - lastChannelID :: MVar Int --for auto-incrementing the channelIDs + lastChannelID :: MVar Int, -- for auto-incrementing the channelIDs + connLastReceived :: MVar Int64, -- the timestamp from a monotonic clock when the last frame was received + connLastSent :: MVar Int64 -- the timestamp from a monotonic clock when the last frame was written } -- | Represents the parameters to connect to a broker or a cluster of brokers. -- See 'defaultConnectionOpts'. -- --- /NOTICE/: The fields 'coHeartbeatDelay' and 'coMaxChannel' were only added for future use, as the respective functionality is not yet implemented. +-- /NOTICE/: The field 'coMaxChannel' was only added for future use, as the respective functionality is not yet implemented. data ConnectionOpts = ConnectionOpts { coServers :: ![(String, PortNumber)], -- ^ A list of host-port pairs. Useful in a clustered setup to connect to the first available host. coVHost :: !Text, -- ^ The VHost to connect to. coAuth :: ![SASLMechanism], -- ^ The 'SASLMechanism's to use for authenticating with the broker. coMaxFrameSize :: !(Maybe Word32), -- ^ The maximum frame size to be used. If not specified, no limit is assumed. - coHeartbeatDelay :: !(Maybe Word16), -- ^ The delay in seconds, after which the client expects a heartbeat frame from the broker. If not specified, no heartbeat is expected. + coHeartbeatDelay :: !(Maybe Word16), -- ^ The delay in seconds, after which the client expects a heartbeat frame from the broker. If 'Nothing', the value suggested by the broker is used. Use @Just 0@ to disable the heartbeat mechnism. coMaxChannel :: !(Maybe Word16) -- ^ The maximum number of channels the client will use. } @@ -161,21 +164,17 @@ connectionReceiver conn = do CE.catch (do Frame chanID payload <- readFrame (connHandle conn) + updateLastReceived conn forwardToChannel chanID payload ) - (\(e :: CE.IOException) -> do - modifyMVar_ (connClosed conn) $ const $ return $ Just $ show e - killThread =<< myThreadId - ) + (\(e :: CE.IOException) -> myThreadId >>= killConnection conn (show e)) connectionReceiver conn where - forwardToChannel 0 (MethodPayload Connection_close_ok) = do - modifyMVar_ (connClosed conn) $ const $ return $ Just "closed by user" - killThread =<< myThreadId + forwardToChannel 0 (MethodPayload Connection_close_ok) = myThreadId >>= killConnection conn "closed by user" forwardToChannel 0 (MethodPayload (Connection_close _ (ShortString errorMsg) _ _)) = do writeFrame (connHandle conn) $ Frame 0 $ MethodPayload Connection_close_ok - modifyMVar_ (connClosed conn) $ const $ return $ Just $ T.unpack errorMsg - killThread =<< myThreadId + myThreadId >>= killConnection conn (T.unpack errorMsg) + forwardToChannel 0 HeartbeatPayload = return () forwardToChannel 0 payload = putStrLn $ "Got unexpected msg on channel zero: " ++ show payload forwardToChannel chanID payload = do --got asynchronous msg => forward to registered channel @@ -188,7 +187,7 @@ openConnection'' :: ConnectionOpts -> IO Connection openConnection'' connOpts = withSocketsDo $ do handle <- connect $ coServers connOpts - maxFrameSize <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information") $ do + (maxFrameSize, heartbeatTimeout) <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information") $ do BL.hPut handle $ BPut.runPut $ do BPut.putByteString $ BC.pack "AMQP" BPut.putWord8 1 @@ -204,13 +203,15 @@ -- C: start_ok writeFrame handle $ start_ok selectedSASL -- S: secure or tune - Frame 0 (MethodPayload (Connection_tune _ frame_max _)) <- handleSecureUntilTune handle selectedSASL + Frame 0 (MethodPayload (Connection_tune _ frame_max sendHeartbeat)) <- handleSecureUntilTune handle selectedSASL -- C: tune_ok let maxFrameSize = chooseMin frame_max $ coMaxFrameSize connOpts + finalHeartbeatSec = fromMaybe sendHeartbeat (coHeartbeatDelay connOpts) + heartbeatTimeout = mfilter (/=0) (Just finalHeartbeatSec) writeFrame handle (Frame 0 (MethodPayload --TODO: handle channel_max - (Connection_tune_ok 0 maxFrameSize 0) + (Connection_tune_ok 0 maxFrameSize finalHeartbeatSec) )) -- C: open writeFrame handle open @@ -219,7 +220,7 @@ Frame 0 (MethodPayload (Connection_open_ok _)) <- readFrame handle -- Connection established! - return maxFrameSize + return (maxFrameSize, heartbeatTimeout) --build Connection object cChannels <- newMVar IM.empty @@ -228,10 +229,12 @@ writeLock <- newMVar () ccl <- newEmptyMVar cClosedHandlers <- newMVar [] - let conn = Connection handle cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers lastChanID + cLastReceived <- getTimestamp >>= newMVar + cLastSent <- getTimestamp >>= newMVar + let conn = Connection handle cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers lastChanID cLastReceived cLastSent --spawn the connectionReceiver - void $ forkIO $ CE.finally (connectionReceiver conn) $ do + connThread <- forkIO $ CE.finally (connectionReceiver conn) $ do -- try closing socket CE.catch (hClose handle) (\(_ :: CE.SomeException) -> return ()) @@ -248,6 +251,13 @@ -- notify connection-close-handlers withMVar cClosedHandlers sequence + + case heartbeatTimeout of + Nothing -> return () + Just timeout -> do + heartbeatThread <- watchHeartbeats conn (fromIntegral timeout) connThread + addConnectionClosedHandler conn True (killThread heartbeatThread) + return conn where connect ((host, port) : rest) = do @@ -298,6 +308,39 @@ Nothing -> abortHandshake handle msg Just a -> return a + +watchHeartbeats :: Connection -> Int -> ThreadId -> IO ThreadId +watchHeartbeats conn timeout connThread = scheduleAtFixedRate rate $ do + checkSendTimeout + checkReceiveTimeout + where + rate = timeout * 1000 * 250 -- timeout / 4 in µs + receiveTimeout = (fromIntegral rate) * 4 * 2 -- 2*timeout in µs + sendTimeout = (fromIntegral rate) * 2 -- timeout/2 in µs + + checkReceiveTimeout = check (connLastReceived conn) receiveTimeout + (killConnection conn "killed connection after missing 2 consecutive heartbeats" connThread) + + checkSendTimeout = check (connLastSent conn) sendTimeout + (writeFrame (connHandle conn) (Frame 0 HeartbeatPayload)) + + check var timeout_µs action = withMVar var $ \lastFrameTime -> do + time <- getTimestamp + when (time >= lastFrameTime + timeout_µs) $ do + action + +updateLastSent :: Connection -> IO () +updateLastSent conn = modifyMVar_ (connLastSent conn) (const getTimestamp) + +updateLastReceived :: Connection -> IO () +updateLastReceived conn = modifyMVar_ (connLastReceived conn) (const getTimestamp) + +-- | kill the connection thread abruptly +killConnection :: Connection -> String -> ThreadId -> IO () +killConnection conn msg connThread = do + modifyMVar_ (connClosed conn) $ const $ return $ Just msg + killThread connThread + -- | closes a connection -- -- Make sure to call this function before your program exits to ensure that all published messages are received by the server. @@ -513,8 +556,10 @@ then CE.catch -- ensure at most one thread is writing to the socket at any time - (withMVar (connWriteLock conn) $ \_ -> - mapM_ (\payload -> writeFrame (connHandle conn) (Frame (channelID chan) payload)) payloads) + (do + withMVar (connWriteLock conn) $ \_ -> + mapM_ (\payload -> writeFrame (connHandle conn) (Frame (channelID chan) payload)) payloads + updateLastSent conn) ( \(_ :: CE.IOException) -> do CE.throwIO $ userError "connection not open" )
Network/AMQP/Protocol.hs view
@@ -50,12 +50,14 @@ MethodPayload MethodPayload | ContentHeaderPayload ShortInt ShortInt LongLongInt ContentHeaderProperties --classID, weight, bodySize, propertyFields | ContentBodyPayload BL.ByteString + | HeartbeatPayload deriving Show frameType :: FramePayload -> Word8 frameType (MethodPayload _) = 1 frameType (ContentHeaderPayload _ _ _ _) = 2 frameType (ContentBodyPayload _) = 3 +frameType HeartbeatPayload = 8 getPayload :: Word8 -> PayloadSize -> Get FramePayload getPayload 1 _ = do --METHOD FRAME @@ -70,6 +72,10 @@ getPayload 3 payloadSize = do --content body frame payload <- getLazyByteString $ fromIntegral payloadSize return (ContentBodyPayload payload) +getPayload 8 payloadSize = do + -- ignoring the actual payload, but still need to read the bytes from the network buffer + _ <- getLazyByteString $ fromIntegral payloadSize + return HeartbeatPayload getPayload n _ = error ("Unknown frame payload: " ++ show n) putPayload :: FramePayload -> Put @@ -79,4 +85,5 @@ put weight put bodySize putContentHeaderProperties p -putPayload (ContentBodyPayload payload) = putLazyByteString payload+putPayload (ContentBodyPayload payload) = putLazyByteString payload +putPayload HeartbeatPayload = putLazyByteString BL.empty
amqp.cabal view
@@ -1,5 +1,5 @@ Name: amqp -Version: 0.6.0 +Version: 0.7.0 Synopsis: Client library for AMQP servers (currently only RabbitMQ) Description: Client library for AMQP servers (currently only RabbitMQ) . @@ -13,22 +13,36 @@ Build-Type: Simple Homepage: https://github.com/hreinhardt/amqp bug-reports: https://github.com/hreinhardt/amqp/issues -Cabal-Version: >=1.6 +Cabal-Version: >=1.8 Extra-source-files: examples/ExampleConsumer.hs, examples/ExampleProducer.hs Library - Build-Depends: base >= 4 && < 5, binary >= 0.7, containers>=0.2, bytestring>=0.9, network>=2.2.3.1, data-binary-ieee754>=0.4.2.1, text>=0.11.2 + Build-Depends: base >= 4 && < 5, binary >= 0.7, containers>=0.2, bytestring>=0.9, network>=2.2.3.1, data-binary-ieee754>=0.4.2.1, text>=0.11.2, split>=0.2, clock >= 0.4.0.1 Exposed-modules: Network.AMQP, Network.AMQP.Types Other-modules: Network.AMQP.Generated, Network.AMQP.Helpers, Network.AMQP.Protocol, Network.AMQP.Internal GHC-Options: -Wall Executable amqp-builder - Build-Depends: xml == 1.3.* + Build-Depends: base >= 4 && < 5, xml == 1.3.*, containers >= 0.2 Hs-Source-Dirs: Tools Main-is: Builder.hs GHC-Options: -Wall source-repository head type: git - location: https://github.com/hreinhardt/amqp+ location: https://github.com/hreinhardt/amqp + +test-suite spec + type: + exitcode-stdio-1.0 + ghc-options: + -Wall -Werror + hs-source-dirs: + ., test + main-is: + Runner.hs + build-depends: + base >= 4 && < 5, binary >= 0.7, containers>=0.2, bytestring>=0.9, network>=2.2.3.1, data-binary-ieee754>=0.4.2.1, text>=0.11.2, split>=0.2 + , hspec >= 1.3 + , hspec-expectations >= 0.3.3
+ test/Runner.hs view
@@ -0,0 +1,20 @@+import Test.Hspec + +import qualified ConnectionSpec +import qualified ChannelSpec +import qualified QueueDeclareSpec +import qualified QueueDeleteSpec +import qualified ExchangeDeclareSpec + +main :: IO () +main = hspec $ do + -- connection.* + describe "ConnectionSpec" ConnectionSpec.spec + -- channel.* + describe "ChannelSpec" ChannelSpec.spec + -- queue.declare + describe "QueueDeclareSpec" QueueDeclareSpec.spec + -- exchange.declare + describe "ExchangeDeclareSpec" ExchangeDeclareSpec.spec + -- queue.delete + describe "QueueDeleteSpec" QueueDeleteSpec.spec