amqp 0.7.0 → 0.8.0
raw patch · 5 files changed
+123/−30 lines, 5 filesdep +connectiondep +monad-control
Dependencies added: connection, monad-control
Files
- Network/AMQP.hs +13/−4
- Network/AMQP/Internal.hs +47/−19
- Network/AMQP/Lifted.hs +42/−0
- amqp.cabal +5/−4
- test/Runner.hs +16/−3
Network/AMQP.hs view
@@ -49,6 +49,7 @@ -- * Connection Connection, ConnectionOpts(..), + TLSSettings(..), defaultConnectionOpts, openConnection, openConnection', @@ -99,6 +100,7 @@ publishMsg', getMsg, rejectMsg, + rejectEnv, recoverMsgs, ackMsg, @@ -156,13 +158,14 @@ exchangePassive :: Bool, -- ^ (default 'False'); If set, the server will not create the exchange. The client can use this to check whether an exchange exists without modifying the server state. exchangeDurable :: Bool, -- ^ (default 'True'); If set when creating a new exchange, the exchange will be marked as durable. Durable exchanges remain active when a server restarts. Non-durable exchanges (transient exchanges) are purged if/when a server restarts. exchangeAutoDelete :: Bool, -- ^ (default 'False'); If set, the exchange is deleted when all queues have finished using it. - exchangeInternal :: Bool -- ^ (default 'False'); If set, the exchange may not be used directly by publishers, but only when bound to other exchanges. Internal exchanges are used to construct wiring that is not visible to applications. + exchangeInternal :: Bool, -- ^ (default 'False'); If set, the exchange may not be used directly by publishers, but only when bound to other exchanges. Internal exchanges are used to construct wiring that is not visible to applications. + exchangeArguments :: FieldTable -- ^ (default empty); A set of arguments for the declaration. The syntax and semantics of these arguments depends on the server implementation. } deriving (Eq, Ord, Read, Show) -- | an 'ExchangeOpts' with defaults set; you must override at least the 'exchangeName' and 'exchangeType' fields. newExchange :: ExchangeOpts -newExchange = ExchangeOpts "" "" False True False False +newExchange = ExchangeOpts "" "" False True False False (FieldTable M.empty) -- | declares a new exchange on the AMQP server. Can be used like this: @declareExchange channel newExchange {exchangeName = \"myExchange\", exchangeType = \"fanout\"}@ declareExchange :: Channel -> ExchangeOpts -> IO () @@ -176,7 +179,7 @@ (exchangeAutoDelete exchg) -- auto_delete (exchangeInternal exchg) -- internal False -- nowait - (FieldTable M.empty))) -- arguments + (exchangeArguments exchg))) -- arguments return () -- | @bindExchange chan destinationName sourceName routingKey@ binds the exchange to the exchange using the provided routing key @@ -453,6 +456,12 @@ requeue -- requeue )) +-- | Reject a message. This is a wrapper for 'rejectMsg' in case you have the 'Envelope' at hand. +rejectEnv :: Envelope + -> Bool -- ^ requeue + -> IO () +rejectEnv env requeue = rejectMsg (envChannel env) (envDeliveryTag env) requeue + -- | @recoverMsgs chan requeue@ asks the broker to redeliver all messages that were received but not acknowledged on the specified channel. --If @requeue==False@, the message will be redelivered to the original recipient. If @requeue==True@, the server will attempt to requeue the message, potentially then delivering it to an alternative subscriber. recoverMsgs :: Channel -> Bool -> IO () @@ -511,7 +520,7 @@ -- * no limit on the number of used channels -- defaultConnectionOpts :: ConnectionOpts -defaultConnectionOpts = ConnectionOpts [("localhost", 5672)] "/" [plain "guest" "guest"] (Just 131072) Nothing Nothing +defaultConnectionOpts = ConnectionOpts [("localhost", 5672)] "/" [plain "guest" "guest"] (Just 131072) Nothing Nothing Nothing -- | @openConnection hostname virtualHost loginName loginPassword@ opens a connection to an AMQP server running on @hostname@. -- @virtualHost@ is used as a namespace for AMQP resources (default is \"/\"), so different applications could use multiple virtual hosts on the same AMQP server.
Network/AMQP/Internal.hs view
@@ -11,7 +11,6 @@ import Data.Text (Text) import Data.Typeable import Network -import System.IO import qualified Control.Exception as CE import qualified Data.ByteString as BS @@ -23,6 +22,7 @@ import qualified Data.Sequence as Seq import qualified Data.Text as T import qualified Data.Text.Encoding as E +import qualified Network.Connection as Conn import Network.AMQP.Protocol import Network.AMQP.Types @@ -126,7 +126,7 @@ -} data Connection = Connection { - connHandle :: Handle, + connHandle :: Conn.Connection, connChannels :: (MVar (IM.IntMap (Channel, ThreadId))), --open channels (channelID => (Channel, ChannelThread)) connMaxFrameSize :: Int, --negotiated maximum frame size connClosed :: MVar (Maybe String), @@ -148,9 +148,20 @@ 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 '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. + coMaxChannel :: !(Maybe Word16), -- ^ The maximum number of channels the client will use. + coTLSSettings :: Maybe TLSSettings -- ^ Whether or not to connect to servers using TLS. See http://www.rabbitmq.com/ssl.html for details. } +-- | Represents the kind of TLS connection to establish. +data TLSSettings = + TLSTrusted -- ^ Require trusted certificates (Recommended). + | TLSUntrusted -- ^ Allow untrusted certificates (Discouraged. Vulnerable to man-in-the-middle attacks) +connectionTLSSettings :: TLSSettings -> Maybe Conn.TLSSettings +connectionTLSSettings tlsSettings = + Just $ case tlsSettings of + TLSTrusted -> Conn.TLSSettingsSimple False False False + TLSUntrusted -> Conn.TLSSettingsSimple True False False + -- | A 'SASLMechanism' is described by its name ('saslName'), its initial response ('saslInitialResponse'), and an optional function ('saslChallengeFunc') that -- transforms a security challenge provided by the server into response, which is then sent back to the server for verification. data SASLMechanism = SASLMechanism { @@ -188,13 +199,13 @@ openConnection'' connOpts = withSocketsDo $ do handle <- connect $ coServers connOpts (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 - BPut.putWord8 1 --TCP/IP - BPut.putWord8 0 --Major Version - BPut.putWord8 9 --Minor Version - hFlush handle + Conn.connectionPut handle $ BS.append (BC.pack "AMQP") + (BS.pack [ + 1 + , 1 --TCP/IP + , 0 --Major Version + , 9 --Minor Version + ]) -- S: connection.start Frame 0 (MethodPayload (Connection_start _ _ _ (LongString serverMechanisms) _)) <- readFrame handle @@ -236,7 +247,7 @@ --spawn the connectionReceiver connThread <- forkIO $ CE.finally (connectionReceiver conn) $ do -- try closing socket - CE.catch (hClose handle) (\(_ :: CE.SomeException) -> return ()) + CE.catch (Conn.connectionClose handle) (\(_ :: CE.SomeException) -> return ()) -- mark as closed modifyMVar_ cClosed $ return . Just . maybe "unknown reason" id @@ -261,7 +272,13 @@ return conn where connect ((host, port) : rest) = do - result <- CE.try (connectTo host $ PortNumber port) + ctx <- Conn.initConnectionContext + result <- CE.try (Conn.connectTo ctx $ Conn.ConnectionParams + { Conn.connectionHostname = host + , Conn.connectionPort = port + , Conn.connectionUseSecure = tlsSettings + , Conn.connectionUseSocks = Nothing + }) either (\(ex :: CE.SomeException) -> do putStrLn $ "Error connecting to "++show (host, port)++": "++show ex @@ -269,6 +286,7 @@ (return) result connect [] = CE.throwIO $ ConnectionClosedException $ "Could not connect to any of the provided brokers: " ++ show (coServers connOpts) + tlsSettings = maybe Nothing connectionTLSSettings (coTLSSettings connOpts) selectSASLMechanism handle serverMechanisms = let serverSaslList = T.split (== ' ') $ E.decodeUtf8 serverMechanisms clientMechanisms = coAuth connOpts @@ -301,7 +319,7 @@ True))) -- insist; deprecated in 0-9-1 abortHandshake handle msg = do - hClose handle + Conn.connectionClose handle CE.throwIO $ ConnectionClosedException msg abortIfNothing m handle msg = case m of @@ -375,13 +393,15 @@ -- otherwise add it to the list _ -> modifyMVar_ (connClosedHandlers conn) $ \old -> return $ handler:old -readFrame :: Handle -> IO Frame +readFrame :: Conn.Connection -> IO Frame readFrame handle = do - dat <- BL.hGet handle 7 + strictDat <- connectionGetExact handle 7 + let dat = toLazy strictDat -- NB: userError returns an IOException so it will be catched in 'connectionReceiver' when (BL.null dat) $ CE.throwIO $ userError "connection not open" let len = fromIntegral $ peekFrameSize dat - dat' <- BL.hGet handle (len+1) -- +1 for the terminating 0xCE + strictDat' <- connectionGetExact handle (len+1) -- +1 for the terminating 0xCE + let dat' = toLazy strictDat' when (BL.null dat') $ CE.throwIO $ userError "connection not open" let ret = runGetOrFail get (BL.append dat dat') case ret of @@ -390,10 +410,18 @@ error $ "readFrame: parser should read " ++ show (len+8) ++ " bytes; but read " ++ show consumedBytes Right (_, _, frame) -> return frame -writeFrame :: Handle -> Frame -> IO () +-- belongs in connection package and will be removed once it lands there +connectionGetExact :: Conn.Connection -> Int -> IO BS.ByteString +connectionGetExact conn x = loop BS.empty 0 + where loop bs y + | y == x = return bs + | otherwise = do + next <- Conn.connectionGet conn (x - y) + loop (BS.append bs next) (y + (BS.length next)) + +writeFrame :: Conn.Connection -> Frame -> IO () writeFrame handle f = do - BL.hPut handle . runPut . put $ f - hFlush handle + Conn.connectionPut handle . toStrict . runPut . put $ f ------------------------ CHANNEL -----------------------------
+ Network/AMQP/Lifted.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE FlexibleContexts #-} + +module Network.AMQP.Lifted + ( consumeMsgs + , consumeMsgs' + ) + where + +import qualified Network.AMQP as A +import Network.AMQP.Types +import Control.Monad.Trans.Control +import Data.Text (Text) +import Control.Monad + +-- | @consumeMsgs chan queueName ack callback@ subscribes to the given queue and returns a consumerTag. For any incoming message, the callback will be run. If @ack == 'Ack'@ you will have to acknowledge all incoming messages (see 'ackMsg' and 'ackEnv') +-- +-- NOTE: The callback will be run on the same thread as the channel thread (every channel spawns its own thread to listen for incoming data) so DO NOT perform any request on @chan@ inside the callback (however, you CAN perform requests on other open channels inside the callback, though I wouldn't recommend it). +-- Functions that can safely be called on @chan@ are 'ackMsg', 'ackEnv', 'rejectMsg', 'recoverMsgs'. If you want to perform anything more complex, it's a good idea to wrap it inside 'forkIO'. +-- +-- In addition, while the callback function @(('Message', 'Envelope') -> m ())@ +-- has access to the captured state, all its side-effects in m are discarded. +consumeMsgs :: MonadBaseControl IO m + => A.Channel + -> Text -- ^ Specifies the name of the queue to consume from. + -> A.Ack + -> ((A.Message, A.Envelope) -> m ()) + -> m A.ConsumerTag +consumeMsgs chan queueName ack callback = + liftBaseWith $ \runInIO -> + A.consumeMsgs chan queueName ack (void . runInIO . callback) + +-- | an extended version of @consumeMsgs@ that allows you to include arbitrary arguments. +consumeMsgs' :: MonadBaseControl IO m + => A.Channel + -> Text -- ^ Specifies the name of the queue to consume from. + -> A.Ack + -> ((A.Message, A.Envelope) -> m ()) + -> FieldTable + -> m A.ConsumerTag +consumeMsgs' chan queueName ack callback args = + liftBaseWith $ \runInIO -> + A.consumeMsgs' chan queueName ack (void . runInIO . callback) args
amqp.cabal view
@@ -1,5 +1,5 @@ Name: amqp -Version: 0.7.0 +Version: 0.8.0 Synopsis: Client library for AMQP servers (currently only RabbitMQ) Description: Client library for AMQP servers (currently only RabbitMQ) . @@ -18,8 +18,8 @@ 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, split>=0.2, clock >= 0.4.0.1 - Exposed-modules: Network.AMQP, Network.AMQP.Types + 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, monad-control >= 0.3, connection == 0.2.* + Exposed-modules: Network.AMQP, Network.AMQP.Types, Network.AMQP.Lifted Other-modules: Network.AMQP.Generated, Network.AMQP.Helpers, Network.AMQP.Protocol, Network.AMQP.Internal GHC-Options: -Wall @@ -43,6 +43,7 @@ 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 + 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 , hspec >= 1.3 , hspec-expectations >= 0.3.3 + , connection == 0.2.*
test/Runner.hs view
@@ -3,18 +3,31 @@ import qualified ConnectionSpec import qualified ChannelSpec import qualified QueueDeclareSpec -import qualified QueueDeleteSpec +import qualified QueueDeleteSpec +import qualified QueuePurgeSpec import qualified ExchangeDeclareSpec +import qualified ExchangeDeleteSpec +import qualified BasicPublishSpec +import qualified BasicRejectSpec + 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 + -- exchange.delete + describe "ExchangeDeleteSpec" ExchangeDeleteSpec.spec + -- queue.declare + describe "QueueDeclareSpec" QueueDeclareSpec.spec -- queue.delete describe "QueueDeleteSpec" QueueDeleteSpec.spec + -- queue.purge + describe "QueuePurgeSpec" QueuePurgeSpec.spec + -- basic.publish + describe "BasicPublishSpec" BasicPublishSpec.spec + -- basic.reject + describe "BasicRejectSpec" BasicRejectSpec.spec