diff --git a/Network/AMQP.hs b/Network/AMQP.hs --- a/Network/AMQP.hs +++ b/Network/AMQP.hs @@ -1,11 +1,11 @@ -{-# LANGUAGE BangPatterns, DeriveDataTypeable, OverloadedStrings, ScopedTypeVariables #-} +{-# LANGUAGE BangPatterns, DeriveDataTypeable, OverloadedStrings, ScopedTypeVariables, TupleSections #-} -- | -- -- A client library for AMQP servers implementing the 0-9-1 spec; currently only supports RabbitMQ (see ) -- -- A good introduction to RabbitMQ and AMQP 0-9-1 (in various languages): , -- --- /Example/: +-- == Example -- -- Connect to a server, declare a queue and an exchange and setup a callback for messages coming in on the queue. Then publish a single message to our new exchange -- @@ -41,10 +41,18 @@ -- > -- acknowledge receiving the message -- > ackEnv env -- --- /Exception handling/: +-- == Exception handling notes -- -- Some function calls can make the AMQP server throw an AMQP exception, which has the side-effect of closing the connection or channel. The AMQP exceptions are raised as Haskell exceptions (see 'AMQPException'). So upon receiving an 'AMQPException' you may have to reopen the channel or connection. - +-- +-- == Debugging tips +-- +-- If you need to debug a problem with e.g. channels being closed unexpectedly, here are some tips: +-- +-- - The RabbitMQ log file often has helpful error-messages. The location of the log-file differs by OS. Look for RABBITMQ_LOGS in this page: +-- - The function 'addChannelExceptionHandler' can be used to figure out when and why a channel was closed. +-- - RabbitMQ has a browser-based management console, which allows you to see connections, channels, queues and more. Setup instructions are here: https://www.rabbitmq.com/management.html +-- module Network.AMQP ( -- * Connection Connection, @@ -57,12 +65,15 @@ closeChannel, closeConnection, addConnectionClosedHandler, + addConnectionBlockedHandler, + getServerProperties, -- * Channel Channel, openChannel, addReturnListener, addChannelExceptionHandler, + isNormalChannelClose, qos, -- * Exchanges @@ -107,6 +118,8 @@ ackMsg, ackEnv, + nackMsg, + nackEnv, -- * Transactions txSelect, @@ -119,6 +132,7 @@ waitForConfirmsUntil, addConfirmationListener, ConfirmationResult(..), + AckType(..), -- * Flow Control flow, @@ -131,6 +145,8 @@ -- * Exceptions AMQPException(..), + ChanThreadKilledException, + CloseType(..), -- * URI parsing fromURI @@ -143,7 +159,7 @@ import Data.List.Split (splitOn) import Data.Binary import Data.Binary.Put -import Network +import Network.Socket (PortNumber) import Network.URI (unEscapeString) import Data.Text (Text) @@ -158,23 +174,22 @@ import Network.AMQP.Generated import Network.AMQP.Internal import Network.AMQP.Helpers +import Text.Read (readMaybe) ----- EXCHANGE ----- -- | A record that contains the fields needed when creating a new exhange using 'declareExchange'. The default values apply when you use 'newExchange'. -data ExchangeOpts = ExchangeOpts - { - exchangeName :: Text, -- ^ (must be set); the name of the exchange - exchangeType :: Text, -- ^ (must be set); the type of the exchange (\"fanout\", \"direct\", \"topic\", \"headers\") +data ExchangeOpts = ExchangeOpts { + exchangeName :: Text, -- ^ (must be set); the name of the exchange + exchangeType :: Text, -- ^ (must be set); the type of the exchange (\"fanout\", \"direct\", \"topic\", \"headers\") - -- optional - 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. - 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) + -- optional + 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. + 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 @@ -244,19 +259,17 @@ ----- QUEUE ----- -- | A record that contains the fields needed when creating a new queue using 'declareQueue'. The default values apply when you use 'newQueue'. -data QueueOpts = QueueOpts - { - --must be set - queueName :: Text, -- ^ (default \"\"); the name of the queue; if left empty, the server will generate a new name and return it from the 'declareQueue' method +data QueueOpts = QueueOpts { + --must be set + queueName :: Text, -- ^ (default \"\"); the name of the queue; if left empty, the server will generate a new name and return it from the 'declareQueue' method - --optional - queuePassive :: Bool, -- ^ (default 'False'); If set, the server will not create the queue. The client can use this to check whether a queue exists without modifying the server state. - queueDurable :: Bool, -- ^ (default 'True'); If set when creating a new queue, the queue will be marked as durable. Durable queues remain active when a server restarts. Non-durable queues (transient queues) are purged if/when a server restarts. Note that durable queues do not necessarily hold persistent messages, although it does not make sense to send persistent messages to a transient queue. - queueExclusive :: Bool, -- ^ (default 'False'); Exclusive queues may only be consumed from by the current connection. Setting the 'exclusive' flag always implies 'auto-delete'. - queueAutoDelete :: Bool, -- ^ (default 'False'); If set, the queue is deleted when all consumers have finished using it. Last consumer can be cancelled either explicitly or because its channel is closed. If there was no consumer ever on the queue, it won't be deleted. - queueHeaders :: FieldTable -- ^ (default empty): Headers to use when creating this queue, such as @x-message-ttl@ or @x-dead-letter-exchange@. - } - deriving (Eq, Ord, Read, Show) + --optional + queuePassive :: Bool, -- ^ (default 'False'); If set, the server will not create the queue. The client can use this to check whether a queue exists without modifying the server state. + queueDurable :: Bool, -- ^ (default 'True'); If set when creating a new queue, the queue will be marked as durable. Durable queues remain active when a server restarts. Non-durable queues (transient queues) are purged if/when a server restarts. Note that durable queues do not necessarily hold persistent messages, although it does not make sense to send persistent messages to a transient queue. + queueExclusive :: Bool, -- ^ (default 'False'); Exclusive queues may only be consumed from by the current connection. Setting the 'exclusive' flag always implies 'auto-delete'. + queueAutoDelete :: Bool, -- ^ (default 'False'); If set, the queue is deleted when all consumers have finished using it. Last consumer can be cancelled either explicitly or because its channel is closed. If there was no consumer ever on the queue, it won't be deleted. + queueHeaders :: FieldTable -- ^ (default empty): Headers to use when creating this queue, such as @x-message-ttl@ or @x-dead-letter-exchange@. +} deriving (Eq, Ord, Read, Show) -- | a 'QueueOpts' with defaults set; you should override at least 'queueName'. newQueue :: QueueOpts @@ -269,15 +282,15 @@ -- @messageCount@ is the number of messages in the queue, which will be zero for newly-created queues. @consumerCount@ is the number of active consumers for the queue. declareQueue :: Channel -> QueueOpts -> IO (Text, Int, Int) declareQueue chan queue = do - (SimpleMethod (Queue_declare_ok (ShortString qName) messageCount consumerCount)) <- request chan $ (SimpleMethod (Queue_declare - 1 -- ticket - (ShortString $ queueName queue) - (queuePassive queue) - (queueDurable queue) - (queueExclusive queue) - (queueAutoDelete queue) - False -- no-wait; true means no answer from server - (queueHeaders queue))) + SimpleMethod (Queue_declare_ok (ShortString qName) messageCount consumerCount) <- request chan $ SimpleMethod $ Queue_declare + 1 -- ticket + (ShortString $ queueName queue) + (queuePassive queue) + (queueDurable queue) + (queueExclusive queue) + (queueAutoDelete queue) + False -- no-wait; true means no answer from server + (queueHeaders queue) return (qName, fromIntegral messageCount, fromIntegral consumerCount) -- | @bindQueue chan queue exchange routingKey@ binds the queue to the exchange using the provided routing key. If @exchange@ is the empty string, the default exchange will be used. @@ -316,32 +329,28 @@ -- | remove all messages from the queue; returns the number of messages that were in the queue purgeQueue :: Channel -> Text -> IO Word32 purgeQueue chan queue = do - (SimpleMethod (Queue_purge_ok msgCount)) <- request chan $ (SimpleMethod (Queue_purge + SimpleMethod (Queue_purge_ok msgCount) <- request chan $ SimpleMethod $ Queue_purge 1 -- ticket (ShortString queue) -- queue False -- nowait - )) return msgCount -- | deletes the queue; returns the number of messages that were in the queue before deletion deleteQueue :: Channel -> Text -> IO Word32 deleteQueue chan queue = do - (SimpleMethod (Queue_delete_ok msgCount)) <- request chan $ (SimpleMethod (Queue_delete + SimpleMethod (Queue_delete_ok msgCount) <- request chan $ SimpleMethod $ Queue_delete 1 -- ticket (ShortString queue) -- queue False -- if_unused False -- if_empty False -- nowait - )) return msgCount ----- MSG (the BASIC class in AMQP) ----- -- | a 'Msg' with defaults set; you should override at least 'msgBody' newMsg :: Message -newMsg = Message (BL.empty) Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing - -type ConsumerTag = Text +newMsg = Message BL.empty Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing -- | specifies whether you have to acknowledge messages that you receive from 'consumeMsgs' or 'getMsg'. If you use 'Ack', you have to call 'ackMsg' or 'ackEnv' after you have processed a message, otherwise it might be delivered again in the future data Ack = Ack | NoAck @@ -353,20 +362,47 @@ -- | @consumeMsgs chan queue 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'. +-- === Exceptions in the callback +-- +-- If an exception occurs in @callback@, it will be caught and printed to @stderr@. But you should not depend on this behaviour (it might change in future versions of this library); +-- instead, it is /strongly/ recommended that you catch any exceptions that your callback may throw and handle them appropriately. +-- But make sure not to catch 'ChanThreadKilledException' (or re-throw it if you did catch it), since it is used internally by the library to close channels. +-- +-- So unless you are confident that your callback won't throw exceptions, you may want to structure your code like this: +-- +-- > consumeMsgs chan name Ack $ \(msg, env) -> (do ...) +-- > `CE.catches` +-- > [ +-- > -- rethrow this exception, since the AMPQ library uses it internally +-- > CE.Handler $ \(e::ChanThreadKilledException) -> CE.throwIO e, +-- > +-- > -- (optional) catch individual exceptions that your code may throw +-- > CE.Handler $ \(e::CE.IOException) -> ..., +-- > CE.Handler $ \(e::SomeOtherException) -> ..., +-- > +-- > -- catch all exceptions that weren't handled above +-- > CE.Handler $ \(e::CE.SomeException) -> ... +-- > ] +-- +-- In practice, it might be useful to encapsulate this exception-handling logic in a custom wrapper-function so that you can reuse it for every callback you pass to @consumeMsgs@. +-- +-- === Blocking requests in the callback +-- +-- The @callback@ will be run on the channel's receiver thread (which is responsible for handling all incoming messages on this channel, including responses to requests from the client) so DO NOT perform any blocking request on @chan@ inside the callback, as this would lead to a dead-lock. However, you CAN perform requests on other open channels inside the callback, though that would keep @chan@ blocked until the requests are done, so it is not recommended. +-- +-- Unless you're using AMQP flow control, the following functions can safely be called on @chan@: 'ackMsg', 'ackEnv', 'rejectMsg', 'publishMsg'. If you use flow-control or want to perform anything more complex, it's recommended that instead of using 'consumeMsgs' you use 'getMsg' to fetch messages in a loop (because then your message-handling code will not run in the channel's receiver thread, so there will be no problems when performing blocking requests). consumeMsgs :: Channel -> Text -> Ack -> ((Message,Envelope) -> IO ()) -> IO ConsumerTag consumeMsgs chan queue ack callback = - consumeMsgs' chan queue ack callback (FieldTable M.empty) + consumeMsgs' chan queue ack callback (\_ -> return ()) (FieldTable M.empty) --- | an extended version of @consumeMsgs@ that allows you to include arbitrary arguments. -consumeMsgs' :: Channel -> Text -> Ack -> ((Message,Envelope) -> IO ()) -> FieldTable -> IO ConsumerTag -consumeMsgs' chan queue ack callback args = do +-- | an extended version of @consumeMsgs@ that allows you to define a consumer cancellation callback and include arbitrary arguments. +consumeMsgs' :: Channel -> Text -> Ack -> ((Message,Envelope) -> IO ()) -> (ConsumerTag -> IO ()) -> FieldTable -> IO ConsumerTag +consumeMsgs' chan queue ack callback cancelCB args = do --generate a new consumer tag - newConsumerTag <- (fmap (T.pack . show)) $ modifyMVar (lastConsumerTag chan) $ \c -> return (c+1,c+1) + newConsumerTag <- fmap (T.pack . show) $ modifyMVar (lastConsumerTag chan) $ \c -> return (c+1,c+1) --register the consumer - modifyMVar_ (consumers chan) $ return . M.insert newConsumerTag callback + modifyMVar_ (consumers chan) $ return . M.insert newConsumerTag (callback, cancelCB) writeAssembly chan (SimpleMethod $ Basic_consume 1 -- ticket @@ -383,10 +419,9 @@ -- | stops a consumer that was started with 'consumeMsgs' cancelConsumer :: Channel -> ConsumerTag -> IO () cancelConsumer chan consumerTag = do - (SimpleMethod (Basic_cancel_ok _)) <- request chan $ (SimpleMethod (Basic_cancel + SimpleMethod (Basic_cancel_ok _) <- request chan $ SimpleMethod $ Basic_cancel (ShortString consumerTag) -- consumer_tag False -- nowait - )) --unregister the consumer modifyMVar_ (consumers chan) $ return . M.delete consumerTag @@ -403,30 +438,31 @@ publishMsg' :: Channel -> Text -> Text -> Bool -> Message -> IO (Maybe Int) publishMsg' chan exchange routingKey mandatory msg = do modifyMVar (nextPublishSeqNum chan) $ \nxtSeqNum -> do - writeAssembly chan (ContentMethod (Basic_publish - 1 -- ticket; ignored by rabbitMQ - (ShortString exchange) - (ShortString routingKey) - mandatory -- mandatory; if true, the server might return the msg, which is currently not handled - False) --immediate; not customizable, as it is currently not supported anymore by RabbitMQ - + writeAssembly chan $ ContentMethod + (Basic_publish + 1 -- ticket; ignored by rabbitMQ + (ShortString exchange) + (ShortString routingKey) + mandatory -- mandatory; if true, the server might return the msg, which is currently not handled + False --immediate; not customizable, as it is currently not supported anymore by RabbitMQ + ) (CHBasic - (fmap ShortString $ msgContentType msg) - (fmap ShortString $ msgContentEncoding msg) - (msgHeaders msg) - (fmap deliveryModeToInt $ msgDeliveryMode msg) - (msgPriority msg) - (fmap ShortString $ msgCorrelationID msg) - (fmap ShortString $ msgReplyTo msg) - (fmap ShortString $ msgExpiration msg) - (fmap ShortString $ msgID msg) - (msgTimestamp msg) - (fmap ShortString $ msgType msg) - (fmap ShortString $ msgUserID msg) - (fmap ShortString $ msgApplicationID msg) - (fmap ShortString $ msgClusterID msg) + (ShortString <$> msgContentType msg) + (ShortString <$> msgContentEncoding msg) + (msgHeaders msg) + (deliveryModeToInt <$> msgDeliveryMode msg) + (msgPriority msg) + (ShortString <$> msgCorrelationID msg) + (ShortString <$> msgReplyTo msg) + (ShortString <$> msgExpiration msg) + (ShortString <$> msgID msg) + (msgTimestamp msg) + (ShortString <$> msgType msg) + (ShortString <$> msgUserID msg) + (ShortString <$> msgApplicationID msg) + (ShortString <$> msgClusterID msg) ) - (msgBody msg)) + (msgBody msg) if nxtSeqNum /= 0 then do @@ -434,19 +470,20 @@ return (succ nxtSeqNum, Just nxtSeqNum) else return (0, Nothing) --- | @getMsg chan ack queue@ gets a message from the specified queue. If @ack=='Ack'@, you have to call 'ackMsg' or 'ackEnv' for any message that you get, otherwise it might be delivered again in the future (by calling 'recoverMsgs') +-- | @getMsg chan ack queue@ gets a message from the specified queue. If @ack=='Ack'@, you have to call 'ackMsg' or 'ackEnv' for any message that you get, otherwise it might be delivered again in the future (by calling 'recoverMsgs'). +-- +-- Will return @Nothing@ when no message is currently available. getMsg :: Channel -> Ack -> Text -> IO (Maybe (Message, Envelope)) getMsg chan ack queue = do - ret <- request chan (SimpleMethod (Basic_get + ret <- request chan $ SimpleMethod $ Basic_get 1 -- ticket (ShortString queue) -- queue (ackToBool ack) -- no_ack - )) case ret of ContentMethod (Basic_get_ok deliveryTag redelivered (ShortString exchange) (ShortString routingKey) _) properties body -> - return $ Just $ (msgFromContentHeaderProperties properties body, - Envelope {envDeliveryTag = deliveryTag, envRedelivered = redelivered, - envExchangeName = exchange, envRoutingKey = routingKey, envChannel = chan}) + return $ Just (msgFromContentHeaderProperties properties body, + Envelope {envDeliveryTag = deliveryTag, envRedelivered = redelivered, + envExchangeName = exchange, envRoutingKey = routingKey, envChannel = chan}) _ -> return Nothing {- | @ackMsg chan deliveryTag multiple@ acknowledges one or more messages. A message MUST not be acknowledged more than once. @@ -457,24 +494,39 @@ -} ackMsg :: Channel -> LongLongInt -> Bool -> IO () ackMsg chan deliveryTag multiple = - writeAssembly chan $ (SimpleMethod (Basic_ack + writeAssembly chan $ SimpleMethod $ Basic_ack deliveryTag -- delivery_tag multiple -- multiple - )) -- | Acknowledges a single message. This is a wrapper for 'ackMsg' in case you have the 'Envelope' at hand. ackEnv :: Envelope -> IO () ackEnv env = ackMsg (envChannel env) (envDeliveryTag env) False +{- | @nackMsg chan deliveryTag multiple requeue@ rejects one or more messages. A message MUST not be rejected more than once. + +if @multiple==True@, the @deliverTag@ is treated as \"up to and including\", so that the client can reject multiple messages with a single method call. If @multiple==False@, @deliveryTag@ refers to a single message. + +If @requeue==True@, the server will try to requeue the message. If @requeue==False@, the message will be dropped by the server. +-} +nackMsg :: Channel -> LongLongInt -> Bool -> Bool -> IO () +nackMsg chan deliveryTag multiple requeue = + writeAssembly chan $ SimpleMethod $ Basic_nack + deliveryTag -- delivery_tag + multiple -- multiple + requeue -- requeue + +-- | Reject a single message. This is a wrapper for 'nackMsg' in case you have the 'Envelope' at hand. +nackEnv :: Envelope -> IO () +nackEnv env = nackMsg (envChannel env) (envDeliveryTag env) False False + -- | @rejectMsg chan deliveryTag requeue@ allows a client to reject a message. It can be used to interrupt and cancel large incoming messages, or return untreatable messages to their original queue. If @requeue==False@, the message will be discarded. If it is 'True', the server will attempt to requeue the message. -- -- NOTE: RabbitMQ 1.7 doesn't implement this command rejectMsg :: Channel -> LongLongInt -> Bool -> IO () rejectMsg chan deliveryTag requeue = - writeAssembly chan $ (SimpleMethod (Basic_reject + writeAssembly chan $ SimpleMethod $ Basic_reject deliveryTag -- delivery_tag requeue -- requeue - )) -- | Reject a message. This is a wrapper for 'rejectMsg' in case you have the 'Envelope' at hand. rejectEnv :: Envelope @@ -486,9 +538,8 @@ --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 () recoverMsgs chan requeue = do - SimpleMethod Basic_recover_ok <- request chan $ (SimpleMethod (Basic_recover + SimpleMethod Basic_recover_ok <- request chan $ SimpleMethod $ Basic_recover requeue -- requeue - )) return () ------------------- TRANSACTIONS (TX) -------------------------- @@ -496,19 +547,19 @@ -- | This method sets the channel to use standard transactions. The client must use this method at least once on a channel before using the Commit or Rollback methods. txSelect :: Channel -> IO () txSelect chan = do - (SimpleMethod Tx_select_ok) <- request chan $ SimpleMethod Tx_select + SimpleMethod Tx_select_ok <- request chan $ SimpleMethod Tx_select return () -- | This method commits all messages published and acknowledged in the current transaction. A new transaction starts immediately after a commit. txCommit :: Channel -> IO () txCommit chan = do - (SimpleMethod Tx_commit_ok) <- request chan $ SimpleMethod Tx_commit + SimpleMethod Tx_commit_ok <- request chan $ SimpleMethod Tx_commit return () -- | This method abandons all messages published and acknowledged in the current transaction. A new transaction starts immediately after a rollback. txRollback :: Channel -> IO () txRollback chan = do - (SimpleMethod Tx_rollback_ok) <- request chan $ SimpleMethod Tx_rollback + SimpleMethod Tx_rollback_ok <- request chan $ SimpleMethod Tx_rollback return () @@ -545,17 +596,17 @@ -} waitForConfirmsUntil :: Channel -> Int -> IO ConfirmationResult waitForConfirmsUntil chan timeout = do - delay <- registerDelay timeout - let partial = do - expired <- readTVar delay - if expired - then return . Partial =<< (,,) - <$> swapTVar (ackedSet chan) IntSet.empty - <*> swapTVar (nackedSet chan) IntSet.empty - <*> readTVar (unconfirmedSet chan) - else retry - complete = return . Complete =<< waitForAllConfirms chan - atomically $ complete `orElse` partial + delay <- registerDelay timeout + let partial = do + expired <- readTVar delay + if expired + then Partial <$> ((,,) + <$> swapTVar (ackedSet chan) IntSet.empty + <*> swapTVar (nackedSet chan) IntSet.empty + <*> readTVar (unconfirmedSet chan)) + else retry + complete = Complete <$> waitForAllConfirms chan + atomically $ complete `orElse` partial {- | Adds a handler which will be invoked each time the @Channel@ receives a confirmation from the broker. The parameters passed to the the handler are the @deliveryTag@ for the message being confirmed, a flag @@ -581,7 +632,7 @@ -} flow :: Channel -> Bool -> IO () flow chan active = do - (SimpleMethod (Channel_flow_ok _)) <- request chan $ SimpleMethod (Channel_flow active) + SimpleMethod (Channel_flow_ok _) <- request chan $ SimpleMethod (Channel_flow active) return () -- | Constructs default connection options with the following settings : @@ -595,7 +646,7 @@ -- * no limit on the number of used channels -- defaultConnectionOpts :: ConnectionOpts -defaultConnectionOpts = ConnectionOpts [("localhost", 5672)] "/" [plain "guest" "guest"] (Just 131072) Nothing Nothing Nothing +defaultConnectionOpts = ConnectionOpts [("localhost", 5672)] "/" [plain "guest" "guest"] (Just 131072) Nothing 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. @@ -621,13 +672,13 @@ plain loginName loginPassword = SASLMechanism "PLAIN" initialResponse Nothing where nul = '\0' - initialResponse = E.encodeUtf8 $ (T.cons nul loginName) `T.append` (T.cons nul loginPassword) + initialResponse = E.encodeUtf8 $ T.cons nul loginName `T.append` T.cons nul loginPassword -- | The @AMQPLAIN@ SASL mechanism. See . amqplain :: Text -> Text -> SASLMechanism amqplain loginName loginPassword = SASLMechanism "AMQPLAIN" initialResponse Nothing where - initialResponse = 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 $ E.encodeUtf8 loginName), ("PASSWORD", FVString $ E.encodeUtf8 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 . @@ -635,7 +686,7 @@ rabbitCRdemo loginName loginPassword = SASLMechanism "RABBIT-CR-DEMO" initialResponse (Just $ const challengeResponse) where initialResponse = E.encodeUtf8 loginName - challengeResponse = return $ (E.encodeUtf8 "My password is ") `BS.append` (E.encodeUtf8 loginPassword) + challengeResponse = return $ E.encodeUtf8 "My password is " `BS.append` E.encodeUtf8 loginPassword -- | @qos chan prefetchSize prefetchCount global@ limits the amount of data the server -- delivers before requiring acknowledgements. @prefetchSize@ specifies the @@ -647,33 +698,57 @@ -- NOTE: RabbitMQ does not implement prefetchSize and will throw an exception if it doesn't equal 0. qos :: Channel -> Word32 -> Word16 -> Bool -> IO () qos chan prefetchSize prefetchCount global = do - (SimpleMethod Basic_qos_ok) <- request chan (SimpleMethod (Basic_qos + SimpleMethod Basic_qos_ok <- request chan $ SimpleMethod $ Basic_qos prefetchSize prefetchCount global - )) 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 +-- | Parses an AMQP standard URI of the form @amqp://user:password\@host:port\/vhost@ and returns a 'ConnectionOpts' for use with 'openConnection'''. +-- +-- To pass multiple servers, separate them by comma, like: @amqp://user:password\@host:port,host2:port2\/vhost@ +-- +-- Any of these fields may be empty and will be replaced with defaults from @amqp://guest:guest\@localhost:5672\/@ +-- +-- When parsing fails, a @Left String@ will be returned with a human-readable error-message. +fromURI :: String -> Either String ConnectionOpts +fromURI uri = + case fromURI' uri of + Right (hostPorts, uid, pw, vhost, tls) -> + Right $ defaultConnectionOpts { + coServers = hostPorts, + coVHost = T.pack vhost, + coAuth = [plain (T.pack uid) (T.pack pw)], + coTLSSettings = if tls then Just TLSTrusted else Nothing + } + Left err -> Left err -fromURI' :: String -> (String,Int,String,String,String) -fromURI' uri = (unEscapeString host, nport, unEscapeString (dropWhile (=='/') uid), unEscapeString pw, unEscapeString vhost) +fromURI' :: String -> Either String ([(String, PortNumber)], String, String, String, Bool) +fromURI' uri = + case sequence (map (fromHostPort dport) hostPorts) of + Right hostPorts' -> Right ( + hostPorts', + unEscapeString (dropWhile (=='/') uid), + unEscapeString pw, + unEscapeString vhost, + tls + ) + Left err -> Left err where (pre :suf : _) = splitOn "@" (uri ++ "@" ) -- look mom, no regexp dependencies (pro :uid' :pw':_) = splitOn ":" (pre ++ "::") (hnp :thost: _) = splitOn "/" (suf ++ "/" ) - (hst':port : _) = splitOn ":" (hnp ++ ":" ) + hostPorts = 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' + tls = pro == "amqps" +fromHostPort :: PortNumber -> String -> Either String (String, PortNumber) +fromHostPort defPort hostPort = (unEscapeString host, ) <$> nport + where + (hst':port : _) = splitOn ":" (hostPort ++ ":" ) + host = if null hst' then "localhost" else hst' + nport = if null port + then Right defPort + else maybe (Left $ "invalid port number: "++port) Right $ readMaybe port diff --git a/Network/AMQP/ChannelAllocator.hs b/Network/AMQP/ChannelAllocator.hs --- a/Network/AMQP/ChannelAllocator.hs +++ b/Network/AMQP/ChannelAllocator.hs @@ -14,7 +14,7 @@ newChannelAllocator :: Int -> IO ChannelAllocator newChannelAllocator maxChannel = - fmap (ChannelAllocator maxChannel) $ V.replicate 1024 0 + ChannelAllocator maxChannel <$> V.replicate 1024 0 allocateChannel :: ChannelAllocator -> IO Int allocateChannel (ChannelAllocator maxChannel c) = do diff --git a/Network/AMQP/Generated.hs b/Network/AMQP/Generated.hs --- a/Network/AMQP/Generated.hs +++ b/Network/AMQP/Generated.hs @@ -110,6 +110,8 @@ put (Connection_open_ok a) = putWord16be 10 >> putWord16be 41 >> put a put (Connection_close a b c d) = putWord16be 10 >> putWord16be 50 >> put a >> put b >> put c >> put d put Connection_close_ok = putWord16be 10 >> putWord16be 51 + put (Connection_blocked a) = putWord16be 10 >> putWord16be 60 >> put a + put Connection_unblocked = putWord16be 10 >> putWord16be 61 put (Channel_open a) = putWord16be 20 >> putWord16be 10 >> put a put (Channel_open_ok a) = putWord16be 20 >> putWord16be 11 >> put a put (Channel_flow a) = putWord16be 20 >> putWord16be 20 >> put a @@ -174,6 +176,8 @@ (10,41) -> get >>= \a -> return (Connection_open_ok a) (10,50) -> get >>= \a -> get >>= \b -> get >>= \c -> get >>= \d -> return (Connection_close a b c d) (10,51) -> return Connection_close_ok + (10,60) -> get >>= \a -> return (Connection_blocked a) + (10,61) -> return Connection_unblocked (20,10) -> get >>= \a -> return (Channel_open a) (20,11) -> get >>= \a -> return (Channel_open_ok a) (20,20) -> get >>= \a -> return (Channel_flow a) @@ -273,6 +277,12 @@ Connection_close_ok | + Connection_blocked + ShortString -- reason + | + Connection_unblocked + + | Channel_open ShortString -- reserved-1 | @@ -502,4 +512,4 @@ Confirm_select_ok - deriving Show + deriving Show diff --git a/Network/AMQP/Helpers.hs b/Network/AMQP/Helpers.hs --- a/Network/AMQP/Helpers.hs +++ b/Network/AMQP/Helpers.hs @@ -43,11 +43,11 @@ chooseMin a Nothing = a getTimestamp :: IO Int64 -getTimestamp = fmap µs $ getTime Monotonic +getTimestamp = µs <$> getTime Monotonic where seconds spec = (fromIntegral . sec) spec * 1000 * 1000 micros spec = (fromIntegral . nsec) spec `div` 1000 - µs spec = (seconds spec) + (micros spec) + µs spec = seconds spec + micros spec scheduleAtFixedRate :: Int -> IO () -> IO ThreadId scheduleAtFixedRate interval_µs action = forkIO $ forever $ do diff --git a/Network/AMQP/Internal.hs b/Network/AMQP/Internal.hs --- a/Network/AMQP/Internal.hs +++ b/Network/AMQP/Internal.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE BangPatterns, CPP, OverloadedStrings, ScopedTypeVariables #-} +{-# LANGUAGE BangPatterns, CPP, OverloadedStrings, ScopedTypeVariables, LambdaCase #-} module Network.AMQP.Internal where import Paths_amqp(version) @@ -11,10 +11,11 @@ import Data.Binary import Data.Binary.Get import Data.Binary.Put as BPut +import Data.Default.Class import Data.Int (Int64) import Data.Maybe import Data.Text (Text) -import Network +import Network.Socket (PortNumber, withSocketsDo) import System.IO (hPutStrLn, stderr) import qualified Control.Exception as CE @@ -53,41 +54,37 @@ -- | An AMQP message data Message = Message { - msgBody :: BL.ByteString, -- ^ the content of your message - msgDeliveryMode :: Maybe DeliveryMode, -- ^ see 'DeliveryMode' - msgTimestamp :: Maybe Timestamp, -- ^ use in any way you like; this doesn't affect the way the message is handled - msgID :: Maybe Text, -- ^ use in any way you like; this doesn't affect the way the message is handled - msgType :: Maybe Text, -- ^ use in any way you like; this doesn't affect the way the message is handled - msgUserID :: Maybe Text, - msgApplicationID :: Maybe Text, - msgClusterID :: Maybe Text, - msgContentType :: Maybe Text, - msgContentEncoding :: Maybe Text, - msgReplyTo :: Maybe Text, - msgPriority :: Maybe Octet, - msgCorrelationID :: Maybe Text, - msgExpiration :: Maybe Text, - msgHeaders :: Maybe FieldTable - } - deriving (Eq, Ord, Read, Show) + msgBody :: BL.ByteString, -- ^ the content of your message + msgDeliveryMode :: Maybe DeliveryMode, -- ^ see 'DeliveryMode' + msgTimestamp :: Maybe Timestamp, -- ^ use in any way you like; this doesn't affect the way the message is handled + msgID :: Maybe Text, -- ^ use in any way you like; this doesn't affect the way the message is handled + msgType :: Maybe Text, -- ^ use in any way you like; this doesn't affect the way the message is handled + msgUserID :: Maybe Text, + msgApplicationID :: Maybe Text, + msgClusterID :: Maybe Text, + msgContentType :: Maybe Text, + msgContentEncoding :: Maybe Text, + msgReplyTo :: Maybe Text, + msgPriority :: Maybe Octet, + msgCorrelationID :: Maybe Text, + msgExpiration :: Maybe Text, + msgHeaders :: Maybe FieldTable +} deriving (Eq, Ord, Read, Show) -- | contains meta-information of a delivered message (through 'getMsg' or 'consumeMsgs') -data Envelope = Envelope - { - envDeliveryTag :: LongLongInt, - envRedelivered :: Bool, - envExchangeName :: Text, - envRoutingKey :: Text, - envChannel :: Channel - } +data Envelope = Envelope { + envDeliveryTag :: LongLongInt, + envRedelivered :: Bool, + envExchangeName :: Text, + envRoutingKey :: Text, + envChannel :: Channel +} -data PublishError = PublishError - { - errReplyCode :: ReturnReplyCode, - errExchange :: Maybe Text, - errRoutingKey :: Text - } - deriving (Eq, Read, Show) +data PublishError = PublishError { + errReplyCode :: ReturnReplyCode, + errExchange :: Maybe Text, + errRoutingKey :: Text +} deriving (Eq, Read, Show) data ReturnReplyCode = Unroutable Text | NoConsumers Text @@ -126,7 +123,7 @@ collect x | x <= 0 = return [] collect x = do (ContentBodyPayload payload) <- readChan chan - r <- collect (x - (BL.length payload)) + r <- collect (x - BL.length payload) return $ payload : r ------------ CONNECTION ------------------- @@ -141,50 +138,58 @@ -} data Connection = Connection { - connHandle :: Conn.Connection, - connChanAllocator :: ChannelAllocator, - connChannels :: MVar (IM.IntMap (Channel, ThreadId)), -- open channels (channelID => (Channel, ChannelThread)) - connMaxFrameSize :: Int, --negotiated maximum frame size - connClosed :: MVar (Maybe String), - 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 ()], - 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 - } + connHandle :: Conn.Connection, + connChanAllocator :: ChannelAllocator, + connChannels :: MVar (IM.IntMap (Channel, ThreadId)), -- open channels (channelID => (Channel, ChannelThread)) + connMaxFrameSize :: Int, --negotiated maximum frame size + connClosed :: MVar (Maybe (CloseType, String)), + 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 ()], + connBlockedHandlers :: MVar [(Text -> IO (), IO ())], + 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 + connServerProperties :: FieldTable, -- the server properties sent in Connection_start + connThread :: MVar (Maybe ThreadId) +} -- | Represents the parameters to connect to a broker or a cluster of brokers. -- See 'defaultConnectionOpts'. 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 '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. - coTLSSettings :: Maybe TLSSettings -- ^ Whether or not to connect to servers using TLS. See http://www.rabbitmq.com/ssl.html for details. - } + 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 '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. + coTLSSettings :: Maybe TLSSettings, -- ^ Whether or not to connect to servers using TLS. See http://www.rabbitmq.com/ssl.html for details. + coName :: !(Maybe Text) -- ^ optional connection name (will be displayed in the RabbitMQ web interface) +} deriving Show + -- | 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) | TLSCustom Conn.TLSSettings -- ^ Provide your own custom TLS settings - + deriving Show connectionTLSSettings :: TLSSettings -> Maybe Conn.TLSSettings connectionTLSSettings tlsSettings = - Just $ case tlsSettings of - TLSTrusted -> Conn.TLSSettingsSimple False False False - TLSUntrusted -> Conn.TLSSettingsSimple True False False - TLSCustom x -> x + Just $ case tlsSettings of + TLSTrusted -> Conn.TLSSettingsSimple False False False def + TLSUntrusted -> Conn.TLSSettingsSimple True False False def + TLSCustom x -> x -- | 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 { - saslName :: !Text, -- ^ mechanism name - saslInitialResponse :: !BS.ByteString, -- ^ initial response - saslChallengeFunc :: !(Maybe (BS.ByteString -> IO BS.ByteString)) -- ^ challenge processing function - } + saslName :: !Text, -- ^ mechanism name + saslInitialResponse :: !BS.ByteString, -- ^ initial response + saslChallengeFunc :: !(Maybe (BS.ByteString -> IO BS.ByteString)) -- ^ challenge processing function +} +instance Show SASLMechanism where + show x = "SASL" + -- | reads incoming frames from socket and forwards them to the opened channels connectionReceiver :: Connection -> IO () connectionReceiver conn = do @@ -193,40 +198,54 @@ updateLastReceived conn forwardToChannel chanID payload ) - (\(e :: CE.IOException) -> myThreadId >>= killConnection conn (CE.toException e)) + (\(e :: CE.IOException) -> myThreadId >>= killConnection conn Abnormal (CE.toException e)) connectionReceiver conn where - - closedByUserEx = ConnectionClosedException "closed by user" + closedByUserEx = ConnectionClosedException Normal "closed by user" - forwardToChannel 0 (MethodPayload Connection_close_ok) = myThreadId >>= killConnection conn (CE.toException closedByUserEx) + forwardToChannel 0 (MethodPayload Connection_close_ok) = + myThreadId >>= killConnection conn Normal (CE.toException closedByUserEx) forwardToChannel 0 (MethodPayload (Connection_close _ (ShortString errorMsg) _ _)) = do writeFrame (connHandle conn) $ Frame 0 $ MethodPayload Connection_close_ok - myThreadId >>= killConnection conn (CE.toException . ConnectionClosedException . T.unpack $ errorMsg) + myThreadId >>= killConnection conn Abnormal (CE.toException . ConnectionClosedException Abnormal . T.unpack $ errorMsg) forwardToChannel 0 HeartbeatPayload = return () + forwardToChannel 0 (MethodPayload (Connection_blocked reason)) = handleBlocked reason + forwardToChannel 0 (MethodPayload Connection_unblocked) = handleUnblocked forwardToChannel 0 payload = hPutStrLn stderr $ "Got unexpected msg on channel zero: " ++ show payload forwardToChannel chanID payload = do --got asynchronous msg => forward to registered channel withMVar (connChannels conn) $ \cs -> do case IM.lookup (fromIntegral chanID) cs of Just c -> writeChan (inQueue $ fst c) payload + -- TODO: It's unclear how to handle this, although it probably never + -- happens in practice. Nothing -> hPutStrLn stderr $ "ERROR: channel not open " ++ show chanID + handleBlocked (ShortString reason) = do + withMVar (connBlockedHandlers conn) $ \listeners -> + forM_ listeners $ \(l, _) -> CE.catch (l reason) $ \(ex :: CE.SomeException) -> + hPutStrLn stderr $ "connection blocked listener threw exception: "++ show ex + + handleUnblocked = do + withMVar (connBlockedHandlers conn) $ \listeners -> + forM_ listeners $ \(_, l) -> CE.catch l $ \(ex :: CE.SomeException) -> + hPutStrLn stderr $ "connection unblocked listener threw exception: "++ show ex + -- | Opens a connection to a broker specified by the given 'ConnectionOpts' parameter. openConnection'' :: ConnectionOpts -> IO Connection openConnection'' connOpts = withSocketsDo $ do handle <- connect [] $ coServers connOpts - (maxFrameSize, maxChannel, heartbeatTimeout) <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information") $ do + (maxFrameSize, maxChannel, heartbeatTimeout, serverProps) <- CE.handle (\(_ :: CE.IOException) -> CE.throwIO $ ConnectionClosedException Abnormal "Handshake failed. Please check the RabbitMQ logs for more information") $ do Conn.connectionPut handle $ BS.append (BC.pack "AMQP") - (BS.pack [ - 1 - , 1 --TCP/IP - , 0 --Major Version - , 9 --Minor Version - ]) + (BS.pack [ + 1 + , 1 --TCP/IP + , 0 --Major Version + , 9 --Minor Version + ]) -- S: connection.start - Frame 0 (MethodPayload (Connection_start _ _ _ (LongString serverMechanisms) _)) <- readFrame handle + Frame 0 (MethodPayload (Connection_start _ _ serverProps (LongString serverMechanisms) _)) <- readFrame handle selectedSASL <- selectSASLMechanism handle serverMechanisms -- C: start_ok @@ -251,7 +270,7 @@ Frame 0 (MethodPayload (Connection_open_ok _)) <- readFrame handle -- Connection established! - return (maxFrameSize, maxChannel, heartbeatTimeout) + return (maxFrameSize, maxChannel, heartbeatTimeout, serverProps) --build Connection object cChannels <- newMVar IM.empty @@ -261,58 +280,60 @@ writeLock <- newMVar () ccl <- newEmptyMVar cClosedHandlers <- newMVar [] + cBlockedHandlers <- newMVar [] cLastReceived <- getTimestamp >>= newMVar cLastSent <- getTimestamp >>= newMVar - let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cLastReceived cLastSent + cThread <- newMVar Nothing + let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cBlockedHandlers cLastReceived cLastSent serverProps cThread + --spawn the connectionReceiver - connThread <- forkFinally' (connectionReceiver conn) $ \res -> do - -- try closing socket - CE.catch (Conn.connectionClose handle) (\(_ :: CE.SomeException) -> return ()) + connThreadId <- forkFinally' (connectionReceiver conn) $ \res -> do + -- try closing socket + CE.catch (Conn.connectionClose handle) (\(_ :: CE.SomeException) -> return ()) - -- mark as closed - modifyMVar_ cClosed $ return . Just . maybe "unknown reason" id + -- mark as closed + modifyMVar_ cClosed $ return . Just . fromMaybe (Abnormal, "unknown reason") - -- kill all channel-threads, making sure the channel threads will - -- be killed taking into account the overall state of the - -- connection: if the thread died for an unexpected exception, - -- inform the channel threads downstream accordingly. Otherwise - -- just use a normal 'killThread' finaliser. - let finaliser = case res of - Left ex -> ex - Right _ -> CE.toException CE.ThreadKilled - modifyMVar_ cChannels $ \x -> do - mapM_ (flip CE.throwTo finaliser . snd) $ IM.elems x - return IM.empty + -- kill all channel-threads, making sure the channel threads will + -- be killed taking into account the overall state of the + -- connection: if the thread died for an unexpected exception, + -- inform the channel threads downstream accordingly. Otherwise + -- just use a normal 'killThread' finaliser. + let finaliser = ChanThreadKilledException $ case res of + Left ex -> ex + Right _ -> CE.toException CE.ThreadKilled + modifyMVar_ cChannels $ \x -> do + mapM_ (flip CE.throwTo finaliser . snd) $ IM.elems x + return IM.empty - -- mark connection as closed, so all pending calls to 'closeConnection' can now return - void $ tryPutMVar ccl () + -- mark connection as closed, so all pending calls to 'closeConnection' can now return + void $ tryPutMVar ccl () - -- notify connection-close-handlers - withMVar cClosedHandlers sequence_ + -- notify connection-close-handlers + withMVar cClosedHandlers sequence_ case heartbeatTimeout of Nothing -> return () - Just timeout -> do - heartbeatThread <- watchHeartbeats conn (fromIntegral timeout) connThread + Just timeout -> do + heartbeatThread <- watchHeartbeats conn (fromIntegral timeout) connThreadId addConnectionClosedHandler conn True (killThread heartbeatThread) + modifyMVar_ cThread $ \_ -> return $ Just connThreadId return conn where connect excs ((host, port) : rest) = do 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 - connect (ex:excs) rest) - (return) - result - connect excs [] = CE.throwIO $ ConnectionClosedException $ "Could not connect to any of the provided brokers: " ++ show (zip (coServers connOpts) (reverse excs)) + result <- CE.try $ Conn.connectTo ctx $ Conn.ConnectionParams + { Conn.connectionHostname = host + , Conn.connectionPort = port + , Conn.connectionUseSecure = tlsSettings + , Conn.connectionUseSocks = Nothing + } + either (\(ex :: CE.SomeException) -> connect (ex:excs) rest) + return + result + connect excs [] = CE.throwIO $ ConnectionClosedException Abnormal $ "Could not connect to any of the provided brokers: " ++ show (zip (coServers connOpts) (reverse excs)) tlsSettings = maybe Nothing connectionTLSSettings (coTLSSettings connOpts) selectSASLMechanism handle serverMechanisms = let serverSaslList = T.split (== ' ') $ E.decodeUtf8 serverMechanisms @@ -322,13 +343,23 @@ in abortIfNothing maybeSasl handle ("None of the provided SASL mechanisms "++show clientSaslList++" is supported by the server "++show serverSaslList++".") - start_ok sasl = (Frame 0 (MethodPayload (Connection_start_ok clientProperties - (ShortString $ saslName sasl) - (LongString $ saslInitialResponse sasl) - (ShortString "en_US")) )) - where clientProperties = FieldTable $ M.fromList [ ("platform", FVString "Haskell") - , ("version" , FVString . T.pack $ showVersion version)] + start_ok sasl = Frame 0 $ MethodPayload $ Connection_start_ok + clientProperties + (ShortString $ saslName sasl) + (LongString $ saslInitialResponse sasl) + (ShortString "en_US") + where + clientProperties = FieldTable $ M.fromList $ [ + ("platform", FVString "Haskell"), + ("version" , FVString . E.encodeUtf8 . T.pack $ showVersion version), + ("capabilities", FVFieldTable clientCapabilities) + ] ++ maybe [] (\x -> [("connection_name", FVString $ E.encodeUtf8 x)]) (coName connOpts) + clientCapabilities = FieldTable $ M.fromList [ + ("consumer_cancel_notify", FVBool True), + ("connection.blocked", FVBool True) + ] + handleSecureUntilTune handle sasl = do tuneOrSecure <- readFrame handle case tuneOrSecure of @@ -339,17 +370,17 @@ writeFrame handle (Frame 0 (MethodPayload (Connection_secure_ok (LongString challengeResponse)))) handleSecureUntilTune handle sasl - tune@(Frame 0 (MethodPayload (Connection_tune _ _ _))) -> return tune + tune@(Frame 0 (MethodPayload Connection_tune{})) -> return tune x -> error $ "handleSecureUntilTune fail. received message: "++show x - open = (Frame 0 (MethodPayload (Connection_open + open = Frame 0 $ MethodPayload $ Connection_open (ShortString $ coVHost connOpts) (ShortString $ T.pack "") -- capabilities; deprecated in 0-9-1 - True))) -- insist; deprecated in 0-9-1 + True -- insist; deprecated in 0-9-1 abortHandshake handle msg = do Conn.connectionClose handle - CE.throwIO $ ConnectionClosedException msg + CE.throwIO $ ConnectionClosedException Abnormal msg abortIfNothing m handle msg = case m of Nothing -> abortHandshake handle msg @@ -362,21 +393,20 @@ 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 + receiveTimeout = fromIntegral rate * 4 * 2 -- 2*timeout in µs + sendTimeout = fromIntegral rate * 2 -- timeout/2 in µs - skippedBeatEx = ConnectionClosedException "killed connection after missing 2 consecutive heartbeats" + skippedBeatEx = ConnectionClosedException Abnormal "killed connection after missing 2 consecutive heartbeats" - checkReceiveTimeout = doCheck (connLastReceived conn) receiveTimeout - (killConnection conn (CE.toException skippedBeatEx) connThread) + checkReceiveTimeout = doCheck (connLastReceived conn) receiveTimeout $ + killConnection conn Abnormal (CE.toException skippedBeatEx) connThread - checkSendTimeout = doCheck (connLastSent conn) sendTimeout - (writeFrame (connHandle conn) (Frame 0 HeartbeatPayload)) + checkSendTimeout = doCheck (connLastSent conn) sendTimeout $ + writeFrame (connHandle conn) (Frame 0 HeartbeatPayload) doCheck var timeout_µs action = withMVar var $ \lastFrameTime -> do time <- getTimestamp - when (time >= lastFrameTime + timeout_µs) $ do - action + when (time >= lastFrameTime + timeout_µs) action updateLastSent :: Connection -> IO () updateLastSent conn = modifyMVar_ (connLastSent conn) (const getTimestamp) @@ -385,10 +415,10 @@ updateLastReceived conn = modifyMVar_ (connLastReceived conn) (const getTimestamp) -- | kill the connection thread abruptly -killConnection :: Connection -> CE.SomeException -> ThreadId -> IO () -killConnection conn ex connThread = do - modifyMVar_ (connClosed conn) $ const $ return $ Just (show ex) - throwTo connThread ex +killConnection :: Connection -> CloseType -> CE.SomeException -> ThreadId -> IO () +killConnection conn closeType ex connThreadId = do + modifyMVar_ (connClosed conn) $ const $ return $ Just (closeType, show ex) + throwTo connThreadId ex -- | closes a connection -- @@ -396,34 +426,47 @@ closeConnection :: Connection -> IO () closeConnection c = do CE.catch ( - withMVar (connWriteLock c) $ \_ -> writeFrame (connHandle c) $ (Frame 0 (MethodPayload (Connection_close + withMVar (connWriteLock c) $ \_ -> writeFrame (connHandle c) $ Frame 0 $ MethodPayload $ Connection_close --TODO: set these values 0 -- reply_code (ShortString "") -- reply_text 0 -- class_id 0 -- method_id - ))) - ) - (\ (_ :: CE.IOException) -> - --do nothing if connection is already closed - return () ) + (\ (e :: CE.IOException) -> do + -- Make sure the connection is closed. + -- (This pattern match can't fail, since openConnection'' will always fill + -- the variable in, and there's no other way to get a connection.) + Just thrID <- readMVar (connThread c) + killConnection c Abnormal (CE.toException e) thrID + ) -- wait for connection_close_ok by the server; this MVar gets filled in the CE.finally handler in openConnection' readMVar $ connClosedLock c return () +-- | get the server properties sent in connection.start +getServerProperties :: Connection -> IO FieldTable +getServerProperties = return . connServerProperties + -- | @addConnectionClosedHandler conn ifClosed handler@ adds a @handler@ that will be called after the connection is closed (either by calling @closeConnection@ or by an exception). If the @ifClosed@ parameter is True and the connection is already closed, the handler will be called immediately. If @ifClosed == False@ and the connection is already closed, the handler will never be called addConnectionClosedHandler :: Connection -> Bool -> IO () -> IO () addConnectionClosedHandler conn ifClosed handler = do - withMVar (connClosed conn) $ \cc -> - case cc of - -- connection is already closed, so call the handler directly - Just _ | ifClosed == True -> handler + withMVar (connClosed conn) $ \case + -- connection is already closed, so call the handler directly + Just _ | ifClosed -> handler - -- otherwise add it to the list - _ -> modifyMVar_ (connClosedHandlers conn) $ \old -> return $ handler:old + -- otherwise add it to the list + _ -> modifyMVar_ (connClosedHandlers conn) $ \old -> return $ handler:old +-- | @addConnectionBlockedHandler conn blockedHandler unblockedHandler@ adds handlers that will be called +-- when a connection gets blocked/unlocked due to server resource constraints. +-- +-- More information: +addConnectionBlockedHandler :: Connection -> (Text -> IO ()) -> IO () -> IO () +addConnectionBlockedHandler conn blockedHandler unblockedHandler = + modifyMVar_ (connBlockedHandlers conn) $ \old -> return $ (blockedHandler, unblockedHandler):old + readFrame :: Conn.Connection -> IO Frame readFrame handle = do strictDat <- connectionGetExact handle 7 @@ -456,7 +499,7 @@ | y == x = return bs | otherwise = do next <- Conn.connectionGet conn (x - y) - loop (BS.append bs next) (y + (BS.length next)) + loop (BS.append bs next) (y + BS.length next) writeFrame :: Conn.Connection -> Frame -> IO () writeFrame handle f = do @@ -467,25 +510,38 @@ {- | A connection to an AMQP server is made up of separate channels. It is recommended to use a separate channel for each thread in your application that talks to the AMQP server (but you don't have to as channels are thread-safe) -} data Channel = Channel { - connection :: Connection, - inQueue :: Chan FramePayload, --incoming frames (from Connection) - outstandingResponses :: MVar (Seq.Seq (MVar Assembly)), -- for every request an MVar is stored here waiting for the response - channelID :: Word16, - lastConsumerTag :: MVar Int, + connection :: Connection, + inQueue :: Chan FramePayload, -- incoming frames (from Connection) + outstandingResponses :: MVar (Seq.Seq (MVar Assembly)), -- for every request an MVar is stored here waiting for the response + channelID :: Word16, + lastConsumerTag :: MVar Int, - nextPublishSeqNum :: MVar Int, - unconfirmedSet :: TVar IntSet.IntSet, - ackedSet :: TVar IntSet.IntSet, --delivery tags - nackedSet :: TVar IntSet.IntSet, --accumulate here. + nextPublishSeqNum :: MVar Int, + unconfirmedSet :: TVar IntSet.IntSet, + ackedSet :: TVar IntSet.IntSet, -- delivery tags + nackedSet :: TVar IntSet.IntSet, -- accumulate here. - chanActive :: Lock, -- used for flow-control. if lock is closed, no content methods will be sent - chanClosed :: MVar (Maybe String), - consumers :: MVar (M.Map Text ((Message, Envelope) -> IO ())), -- who is consumer of a queue? (consumerTag => callback) - returnListeners :: MVar ([(Message, PublishError) -> IO ()]), - confirmListeners :: MVar ([(Word64, Bool, AckType) -> IO ()]), - chanExceptionHandlers :: MVar [CE.SomeException -> IO ()] - } + chanActive :: Lock, -- used for flow-control. if lock is closed, no content methods will be sent + chanClosed :: MVar (Maybe (CloseType, String)), + consumers :: MVar (M.Map Text ((Message, Envelope) -> IO (), -- who is consumer of a queue? (consumerTag => callback) + ConsumerTag -> IO ())), -- cancellation notification callback + returnListeners :: MVar [(Message, PublishError) -> IO ()], + confirmListeners :: MVar [(Word64, Bool, AckType) -> IO ()], + chanExceptionHandlers :: MVar [CE.SomeException -> IO ()] +} +-- | Thrown in the channel thread when the connection gets closed. +-- When handling exceptions in a subscription callback, make sure to re-throw this so the channel thread can be stopped. +data ChanThreadKilledException = ChanThreadKilledException { cause :: CE.SomeException } + deriving (Show) + +instance CE.Exception ChanThreadKilledException + +-- | If the given exception is an instance of ChanThreadKilledException, this method returns +-- the inner exception. Otherwise the exception is returned unchanged. +unwrapChanThreadKilledException :: CE.SomeException -> CE.SomeException +unwrapChanThreadKilledException e = maybe e cause $ CE.fromException e + msgFromContentHeaderProperties :: ContentHeaderProperties -> BL.ByteString -> Message msgFromContentHeaderProperties (CHBasic content_type content_encoding headers delivery_mode priority correlation_id reply_to expiration message_id timestamp message_type user_id application_id cluster_id) body = let msgId = fromShortString message_id @@ -523,12 +579,13 @@ channelReceiver chan where isResponse :: Assembly -> Bool - isResponse (ContentMethod (Basic_deliver _ _ _ _ _) _ _) = False - isResponse (ContentMethod (Basic_return _ _ _ _) _ _) = False + isResponse (ContentMethod Basic_deliver{} _ _) = False + isResponse (ContentMethod Basic_return{} _ _) = False isResponse (SimpleMethod (Channel_flow _)) = False - isResponse (SimpleMethod (Channel_close _ _ _ _)) = False + isResponse (SimpleMethod Channel_close{}) = False isResponse (SimpleMethod (Basic_ack _ _)) = False - isResponse (SimpleMethod (Basic_nack _ _ _)) = False + isResponse (SimpleMethod Basic_nack{}) = False + isResponse (SimpleMethod (Basic_cancel _ _)) = False isResponse _ = True --Basic.Deliver: forward msg to registered consumer @@ -537,20 +594,27 @@ properties body) = withMVar (consumers chan) (\s -> do case M.lookup consumerTag s of - Just subscriber -> do + Just (subscriber, _) -> do let msg = msgFromContentHeaderProperties properties body let env = Envelope {envDeliveryTag = deliveryTag, envRedelivered = redelivered, envExchangeName = exchange, envRoutingKey = routingKey, envChannel = chan} - CE.catch (subscriber (msg, env)) - (\(e::CE.SomeException) -> hPutStrLn stderr $ "AMQP callback threw exception: " ++ show e) + CE.catches (subscriber (msg, env)) [ + CE.Handler (\(e::ChanThreadKilledException) -> CE.throwIO $ cause e), + CE.Handler (\(e::CE.SomeException) -> hPutStrLn stderr $ "AMQP callback threw exception: " ++ show e) + ] Nothing -> -- got a message, but have no registered subscriber; so drop it return () ) handleAsync (SimpleMethod (Channel_close _ (ShortString errorMsg) _ _)) = do - closeChannel' chan errorMsg - myThreadId >>= flip CE.throwTo (ConnectionClosedException . T.unpack $ errorMsg) + CE.catch (writeAssembly' chan (SimpleMethod Channel_close_ok)) + (\ (_ :: CE.IOException) -> + -- do nothing if connection is already closed + return () + ) + closeChannel' chan Abnormal errorMsg + myThreadId >>= flip CE.throwTo (ChannelClosedException Abnormal . T.unpack $ errorMsg) handleAsync (SimpleMethod (Channel_flow active)) = do if active then openLock $ chanActive chan @@ -558,80 +622,120 @@ -- in theory we should respond with flow_ok but rabbitMQ 1.7 ignores that, so it doesn't matter return () --Basic.return - handleAsync (ContentMethod basicReturn@(Basic_return _ _ _ _) props body) = do + handleAsync (ContentMethod basicReturn@Basic_return{} props body) = do let msg = msgFromContentHeaderProperties props body pubError = basicReturnToPublishError basicReturn withMVar (returnListeners chan) $ \listeners -> forM_ listeners $ \l -> CE.catch (l (msg, pubError)) $ \(ex :: CE.SomeException) -> - hPutStrLn stderr $ "return listener on channel ["++(show $ channelID chan)++"] handling error ["++show pubError++"] threw exception: "++show ex + hPutStrLn stderr $ "return listener on channel ["++show (channelID chan)++"] handling error ["++show pubError++"] threw exception: "++show ex handleAsync (SimpleMethod (Basic_ack deliveryTag multiple)) = handleConfirm deliveryTag multiple BasicAck handleAsync (SimpleMethod (Basic_nack deliveryTag multiple _)) = handleConfirm deliveryTag multiple BasicNack + handleAsync (SimpleMethod (Basic_cancel consumerTag _)) = handleCancel consumerTag handleAsync m = error ("Unknown method: " ++ show m) handleConfirm deliveryTag multiple k = do withMVar (confirmListeners chan) $ \listeners -> forM_ listeners $ \l -> CE.catch (l (deliveryTag, multiple, k)) $ \(ex :: CE.SomeException) -> - hPutStrLn stderr $ "confirm listener on channel ["++(show $ channelID chan)++"] handling method "++(show k)++" threw exception: "++ show ex + hPutStrLn stderr $ "confirm listener on channel ["++show (channelID chan)++"] handling method "++show k++" threw exception: "++show ex let seqNum = fromIntegral deliveryTag let targetSet = case k of - BasicAck -> (ackedSet chan) - BasicNack -> (nackedSet chan) + BasicAck -> ackedSet chan + BasicNack -> nackedSet chan atomically $ do - unconfSet <- readTVar (unconfirmedSet chan) - let (merge, pending) = if multiple - then (IntSet.union confs, pending') - else (IntSet.insert seqNum, IntSet.delete seqNum unconfSet) - where - confs = fst parts - pending' = snd parts - parts = IntSet.partition (\n -> n <= seqNum) unconfSet - modifyTVar' targetSet (\ts -> merge ts) - writeTVar (unconfirmedSet chan) pending + unconfSet <- readTVar (unconfirmedSet chan) + let (merge, pending) = if multiple + then (IntSet.union confs, pending') + else (IntSet.insert seqNum, IntSet.delete seqNum unconfSet) + where + confs = fst parts + pending' = snd parts + parts = IntSet.partition (\n -> n <= seqNum) unconfSet + modifyTVar' targetSet (\ts -> merge ts) + writeTVar (unconfirmedSet chan) pending + handleCancel (ShortString consumerTag) = + withMVar (consumers chan) (\s -> do + case M.lookup consumerTag s of + Just (_, cancelCB) -> + CE.catch (cancelCB consumerTag) $ \(ex :: CE.SomeException) -> + hPutStrLn stderr $ "consumer cancellation listener "++show consumerTag++" on channel ["++show (channelID chan)++"] threw exception: "++ show ex + Nothing -> + -- got a cancellation notification, but have no registered subscriber; so drop it + return () + ) + basicReturnToPublishError (Basic_return code (ShortString errText) (ShortString exchange) (ShortString routingKey)) = let replyError = case code of 312 -> Unroutable errText 313 -> NoConsumers errText 404 -> NotFound errText - num -> error $ "unexpected return error code: " ++ (show num) + num -> error $ "unexpected return error code: " ++show num pubError = PublishError replyError (Just exchange) routingKey in pubError basicReturnToPublishError x = error $ "basicReturnToPublishError fail: "++show x --- | registers a callback function that is called whenever a message is returned from the broker ('basic.return'). +-- | Registers a callback function that is called whenever a message is returned from the broker ('basic.return'). addReturnListener :: Channel -> ((Message, PublishError) -> IO ()) -> IO () addReturnListener chan listener = do modifyMVar_ (returnListeners chan) $ \listeners -> return $ listener:listeners --- | registers a callback function that is called whenever a channel is closed by an exception. +-- | Registers a callback function that is called whenever a channel is closed by an exception. +-- +-- This method will always be called when a channel is closed, whether through normal means +-- ('closeChannel', 'closeConnection') or by some AMQP exception. +-- +-- You can use 'isNormalChannelClose' to figure out if the exception was normal or due to an +-- AMQP exception. addChannelExceptionHandler :: Channel -> (CE.SomeException -> IO ()) -> IO () addChannelExceptionHandler chan handler = do modifyMVar_ (chanExceptionHandlers chan) $ \handlers -> return $ handler:handlers +-- | This can be used with the exception passed to 'addChannelExceptionHandler'. +-- +-- Returns True if the argument is a 'ConnectionClosedException' or 'ChannelClosedException' that happened +-- normally (i.e. by the user calling 'closeChannel' or 'closeConnection') and not due to some +-- AMQP exception. +isNormalChannelClose :: CE.SomeException -> Bool +isNormalChannelClose e = case CE.fromException e :: Maybe AMQPException of + Just (ChannelClosedException Normal _) -> True + Just (ConnectionClosedException Normal _) -> True + _ -> False + -- closes the channel internally; but doesn't tell the server -closeChannel' :: Channel -> Text -> IO () -closeChannel' c reason = do +closeChannel' :: Channel -> CloseType -> Text -> IO () +closeChannel' c closeType reason = do modifyMVar_ (chanClosed c) $ \x -> do if isNothing x then do modifyMVar_ (connChannels $ connection c) $ \old -> do ret <- freeChannel (connChanAllocator $ connection c) $ fromIntegral $ channelID c - when (not ret) $ hPutStrLn stderr "closeChannel error: channel already freed" + unless ret $ hPutStrLn stderr "closeChannel error: channel already freed" return $ IM.delete (fromIntegral $ channelID c) old void $ killLock $ chanActive c killOutstandingResponses $ outstandingResponses c - return $ Just $ maybe (T.unpack reason) id x + return $ Just (closeType, T.unpack reason) else return x where killOutstandingResponses :: MVar (Seq.Seq (MVar a)) -> IO () killOutstandingResponses outResps = do modifyMVar_ outResps $ \val -> do F.mapM_ (\x -> tryPutMVar x $ error "channel closed") val + -- Intentionally put 'undefined' into the MVar, so that reading from it will throw. + -- This MVar will be read from the 'request' method, where we appropriately catch ErrorCall exceptions. return undefined -- | opens a new channel on the connection +-- +-- By default, if a channel is closed by an AMQP exception, this exception will be printed to stderr. You can prevent this behaviour by setting a custom exception handler (using 'addChannelExceptionHandler'). +-- +-- Example of adding an exception-handler: +-- +-- > chan <- openChannel conn +-- > addChannelExceptionHandler chan $ \e -> do +-- > unless (isNormalChannelClose e) $ do +-- > putStrLn $ "channel exception: "++show e openChannel :: Connection -> IO Channel openChannel c = do newInQueue <- newChan @@ -648,71 +752,86 @@ cnfListeners <- newMVar [] handlers <- newMVar [] - --add new channel to connection's channel map + -- add new channel to connection's channel map newChannel <- modifyMVar (connChannels c) $ \mp -> do newChannelID <- allocateChannel (connChanAllocator c) let newChannel = Channel c newInQueue outRes (fromIntegral newChannelID) lastConsTag nxtSeq unconfSet aSet nSet ca closed conss retListeners cnfListeners handlers thrID <- forkFinally' (channelReceiver newChannel) $ \res -> do - closeChannel' newChannel "closed" - case res of - Right _ -> return () - Left ex -> readMVar handlers >>= mapM_ ($ ex) + closeChannel' newChannel Normal "closed" + case res of + Right _ -> return () + Left ex -> do + let unwrappedExc = unwrapChanThreadKilledException ex + handlers' <- readMVar handlers + + case (null handlers', fromAbnormalChannelClose unwrappedExc) of + (True, Just reason) -> hPutStrLn stderr $ "unhandled AMQP channel exception (chanId="++show newChannelID++"): "++reason + _ -> mapM_ ($ unwrappedExc) handlers' when (IM.member newChannelID mp) $ CE.throwIO $ userError "openChannel fail: channel already open" return (IM.insert newChannelID (newChannel, thrID) mp, newChannel) - SimpleMethod (Channel_open_ok _) <- request newChannel (SimpleMethod (Channel_open (ShortString ""))) + SimpleMethod (Channel_open_ok _) <- request newChannel $ SimpleMethod $ Channel_open (ShortString "") return newChannel + where + fromAbnormalChannelClose :: CE.SomeException -> Maybe String + fromAbnormalChannelClose exc = + case CE.fromException exc :: Maybe AMQPException of + Just (ConnectionClosedException _ _) -> Nothing + Just (ChannelClosedException Normal _) -> Nothing + Just (ChannelClosedException Abnormal reason) -> Just reason + Just (AllChannelsAllocatedException _) -> Just "all channels allocated" + Nothing -> Just $ show exc + -- | closes a channel. It is typically not necessary to manually call this as closing a connection will implicitly close all channels. closeChannel :: Channel -> IO () closeChannel c = do SimpleMethod Channel_close_ok <- request c $ SimpleMethod $ Channel_close 0 (ShortString "") 0 0 withMVar (connChannels $ connection c) $ \chans -> do case IM.lookup (fromIntegral $ channelID c) chans of - Just (_, thrID) -> killThread thrID + Just (_, thrID) -> throwTo thrID $ ChannelClosedException Normal "closeChannel was called" Nothing -> return () -- | writes multiple frames to the channel atomically writeFrames :: Channel -> [FramePayload] -> IO () -writeFrames chan payloads = - let conn = connection chan in - withMVar (connChannels conn) $ \chans -> - if IM.member (fromIntegral $ channelID chan) chans - then - CE.catch - -- ensure at most one thread is writing to the socket at any time - (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" - ) - else do - CE.throwIO $ userError "channel not open" +writeFrames chan payloads = do + let conn = connection chan + withMVar (connChannels conn) $ \chans -> + if IM.member (fromIntegral $ channelID chan) chans + then CE.catch + -- ensure at most one thread is writing to the socket at any time + (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" + ) + else do + CE.throwIO $ userError "channel not open" writeAssembly' :: Channel -> Assembly -> IO () writeAssembly' chan (ContentMethod m properties msg) = do -- wait iff the AMQP server instructed us to withhold sending content data (flow control) waitLock $ chanActive chan - let !toWrite = - [(MethodPayload m), - (ContentHeaderPayload + let !toWrite = [ + MethodPayload m, + ContentHeaderPayload (getClassIDOf properties) --classID 0 --weight is deprecated in AMQP 0-9 (fromIntegral $ BL.length msg) --bodySize - properties)] ++ + properties] ++ (if BL.length msg > 0 then do -- split into frames of maxFrameSize -- (need to substract 8 bytes to account for frame header and end-marker) map ContentBodyPayload - (splitLen msg $ (fromIntegral $ connMaxFrameSize $ connection chan) - 8) + (splitLen msg $ fromIntegral (connMaxFrameSize $ connection chan) - 8) else [] ) writeFrames chan toWrite where - splitLen str len | BL.length str > len = (BL.take len str):(splitLen (BL.drop len str) len) + splitLen str len | BL.length str > len = BL.take len str : splitLen (BL.drop len str) len splitLen str _ = [str] writeAssembly' chan (SimpleMethod m) = writeFrames chan [MethodPayload m] @@ -754,17 +873,16 @@ throwMostRelevantAMQPException chan = do cc <- readMVar $ connClosed $ connection chan case cc of - Just r -> CE.throwIO $ ConnectionClosedException r + Just (closeType, r) -> CE.throwIO $ ConnectionClosedException closeType r Nothing -> do chc <- readMVar $ chanClosed chan case chc of - Just r -> CE.throwIO $ ChannelClosedException r - Nothing -> CE.throwIO $ ConnectionClosedException "unknown reason" + Just (ct, r) -> CE.throwIO $ ChannelClosedException ct r + Nothing -> CE.throwIO $ ConnectionClosedException Abnormal "unknown reason" waitForAllConfirms :: Channel -> STM (IntSet.IntSet, IntSet.IntSet) waitForAllConfirms chan = do - pending <- readTVar $ (unconfirmedSet chan) - check (IntSet.null pending) - return =<< (,) - <$> swapTVar (ackedSet chan) IntSet.empty - <*> swapTVar (nackedSet chan) IntSet.empty + pending <- readTVar $ unconfirmedSet chan + check (IntSet.null pending) + (,) <$> swapTVar (ackedSet chan) IntSet.empty + <*> swapTVar (nackedSet chan) IntSet.empty diff --git a/Network/AMQP/Lifted.hs b/Network/AMQP/Lifted.hs --- a/Network/AMQP/Lifted.hs +++ b/Network/AMQP/Lifted.hs @@ -12,10 +12,7 @@ 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'. +-- | Lifted version of 'Network.AMQP.consumeMsgs' (please look there for documentation). -- -- In addition, while the callback function @(('Message', 'Envelope') -> m ())@ -- has access to the captured state, all its side-effects in m are discarded. @@ -29,14 +26,15 @@ liftBaseWith $ \runInIO -> A.consumeMsgs chan queueName ack (void . runInIO . callback) --- | an extended version of @consumeMsgs@ that allows you to include arbitrary arguments. +-- | an extended version of @consumeMsgs@ that allows you to define a consumer cancellation callback and 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 ()) + -> (ConsumerTag -> m ()) -> FieldTable -> m A.ConsumerTag -consumeMsgs' chan queueName ack callback args = +consumeMsgs' chan queueName ack callback cancelled args = liftBaseWith $ \runInIO -> - A.consumeMsgs' chan queueName ack (void . runInIO . callback) args + A.consumeMsgs' chan queueName ack (void . runInIO . callback) (void . runInIO . cancelled) args diff --git a/Network/AMQP/Protocol.hs b/Network/AMQP/Protocol.hs --- a/Network/AMQP/Protocol.hs +++ b/Network/AMQP/Protocol.hs @@ -12,9 +12,9 @@ --True if a content (contentheader and possibly contentbody) will follow the method hasContent :: FramePayload -> Bool -hasContent (MethodPayload (Basic_get_ok _ _ _ _ _)) = True -hasContent (MethodPayload (Basic_deliver _ _ _ _ _)) = True -hasContent (MethodPayload (Basic_return _ _ _ _)) = True +hasContent (MethodPayload Basic_get_ok{}) = True +hasContent (MethodPayload Basic_deliver{}) = True +hasContent (MethodPayload Basic_return{}) = True hasContent _ = False data Frame = Frame ChannelID FramePayload --channel, payload @@ -42,20 +42,20 @@ peekFrameSize = runGet f where f = do - void $ getWord8 -- 1 byte - void $ (get :: Get ChannelID) -- 2 bytes - get :: Get PayloadSize -- 4 bytes + void getWord8 -- 1 byte + void (get :: Get ChannelID) -- 2 bytes + get :: Get PayloadSize -- 4 bytes data FramePayload = - MethodPayload MethodPayload - | ContentHeaderPayload ShortInt ShortInt LongLongInt ContentHeaderProperties --classID, weight, bodySize, propertyFields - | ContentBodyPayload BL.ByteString - | HeartbeatPayload + 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 ContentHeaderPayload{} = 2 frameType (ContentBodyPayload _) = 3 frameType HeartbeatPayload = 8 @@ -76,6 +76,7 @@ -- ignoring the actual payload, but still need to read the bytes from the network buffer _ <- getLazyByteString $ fromIntegral payloadSize return HeartbeatPayload +-- this should never happen: getPayload n _ = error ("Unknown frame payload: " ++ show n) putPayload :: FramePayload -> Put @@ -86,4 +87,4 @@ put bodySize putContentHeaderProperties p putPayload (ContentBodyPayload payload) = putLazyByteString payload -putPayload HeartbeatPayload = putLazyByteString BL.empty +putPayload HeartbeatPayload = putLazyByteString BL.empty diff --git a/Network/AMQP/Types.hs b/Network/AMQP/Types.hs --- a/Network/AMQP/Types.hs +++ b/Network/AMQP/Types.hs @@ -13,12 +13,14 @@ LongLongInt, ShortString(..), LongString(..), + ConsumerTag, Timestamp, FieldTable(..), FieldValue(..), Decimals, DecimalValue(..), ConfirmationResult(..), + CloseType(..) ) where import Control.Applicative @@ -38,11 +40,14 @@ import qualified Data.Map as M import qualified Data.Text.Encoding as T +-- | describes whether a channel was closed by user-request (Normal) or by an AMQP exception (Abnormal) +data CloseType = Normal | Abnormal + deriving (Typeable, Show, Ord, Eq) data AMQPException = - -- | the 'String' contains the reason why the channel was closed - ChannelClosedException String - | ConnectionClosedException String -- ^ String may contain a reason + -- | the 'String' contains the reason why the channel was closed + ChannelClosedException CloseType String + | ConnectionClosedException CloseType String -- ^ String may contain a reason | AllChannelsAllocatedException Int -- ^ the 'Int' contains the channel-max property of the connection (i.e. the highest permitted channel id) deriving (Typeable, Show, Ord, Eq) instance CE.Exception AMQPException @@ -75,13 +80,15 @@ type LongInt = Word32 type LongLongInt = Word64 +type ConsumerTag = Text + newtype ShortString = ShortString Text deriving (Eq, Ord, Read, Show) instance Binary ShortString where get = do - len <- getWord8 - dat <- getByteString (fromIntegral len) - return $ ShortString $ T.decodeUtf8 dat + len <- getWord8 + dat <- getByteString (fromIntegral len) + return $ ShortString $ T.decodeUtf8 dat put (ShortString x) = do let s = T.encodeUtf8 x @@ -95,9 +102,9 @@ deriving (Eq, Ord, Read, Show) instance Binary LongString where get = do - len <- getWord32be - dat <- getByteString (fromIntegral len) - return $ LongString dat + len <- getWord32be + dat <- getByteString (fromIntegral len) + return $ LongString dat put (LongString x) = do putWord32be $ fromIntegral (BS.length x) @@ -118,7 +125,7 @@ fvp <- getLazyByteString (fromIntegral len) let !fields = readMany fvp return $ FieldTable $ M.fromList $ map (\(ShortString a, b) -> (a,b)) fields - else return $ FieldTable $ M.empty + else return $ FieldTable M.empty put (FieldTable fvp) = do let bytes = runPut (putMany $ map (\(a,b) -> (ShortString a, b)) $ M.toList fvp) :: BL.ByteString @@ -135,7 +142,7 @@ | FVFloat Float | FVDouble Double | FVDecimal DecimalValue - | FVString Text + | FVString BS.ByteString | FVFieldArray [FieldValue] | FVTimestamp Timestamp | FVFieldTable FieldTable @@ -157,7 +164,7 @@ 'D' -> FVDecimal <$> get 'S' -> do LongString x <- get :: Get LongString - return $ FVString $ T.decodeUtf8 x + return $ FVString x 'A' -> do len <- get :: Get Int32 if len > 0 @@ -172,6 +179,7 @@ 'x' -> do len <- get :: Get Word32 FVByteArray <$> getByteString (fromIntegral len) + -- this should never happen: c -> error ("Unknown field type: " ++ show c) put (FVBool x) = put 't' >> put x @@ -182,10 +190,10 @@ put (FVFloat x) = put 'f' >> putFloat32be x put (FVDouble x) = put 'd' >> putFloat64be x put (FVDecimal x) = put 'D' >> put x - put (FVString x) = put 'S' >> put (LongString $ T.encodeUtf8 x) + put (FVString x) = put 'S' >> put (LongString x) put (FVFieldArray x) = do put 'A' - if length x == 0 + if null x then put (0 :: Int32) else do let bytes = runPut (putMany x) :: BL.ByteString @@ -193,7 +201,7 @@ putLazyByteString bytes put (FVTimestamp s) = put 'T' >> put s put (FVFieldTable s) = put 'F' >> put s - put (FVVoid) = put 'V' + put FVVoid = put 'V' put (FVByteArray x) = do put 'x' let len = fromIntegral (BS.length x) :: Word32 diff --git a/amqp.cabal b/amqp.cabal --- a/amqp.cabal +++ b/amqp.cabal @@ -1,5 +1,5 @@ Name: amqp -Version: 0.14.1 +Version: 0.24.0 Synopsis: Client library for AMQP servers (currently only RabbitMQ) Description: Client library for AMQP servers (currently only RabbitMQ) License: BSD3 @@ -11,39 +11,37 @@ Build-Type: Simple Homepage: https://github.com/hreinhardt/amqp bug-reports: https://github.com/hreinhardt/amqp/issues -Cabal-Version: >=1.8 +Cabal-Version: >=1.10 Extra-source-files: examples/ExampleConsumer.hs, examples/ExampleProducer.hs, changelog.md -flag network-uri - description: Get Network.URI from the network-uri package - default: True - Library + default-language: Haskell2010 Build-Depends: base >= 4 && < 5, binary >= 0.5, containers>=0.2, bytestring>=0.9, data-binary-ieee754>=0.4.2.1, + data-default-class >= 0.1, text>=0.11.2, split>=0.2, clock >= 0.4.0.1, monad-control >= 0.3, - connection == 0.2.*, + crypton-connection >= 0.4 && <= 0.5, vector, - stm >= 2.4.0 - if flag(network-uri) - build-depends: network-uri >= 2.6, network > 2.6 - else - build-depends: network < 2.6 + stm >= 2.4.0, + network-uri >= 2.6, + network > 2.6 + Exposed-modules: Network.AMQP, Network.AMQP.Types, Network.AMQP.Lifted Other-modules: Network.AMQP.ChannelAllocator, Network.AMQP.Generated, Network.AMQP.Helpers, Network.AMQP.Protocol, Network.AMQP.Internal, Paths_amqp GHC-Options: -Wall Executable amqp-builder + default-language: Haskell2010 Build-Depends: base >= 4 && < 5, xml == 1.3.*, containers >= 0.2 Hs-Source-Dirs: Tools Main-is: Builder.hs @@ -54,6 +52,7 @@ location: https://github.com/hreinhardt/amqp test-suite spec + default-language: Haskell2010 type: exitcode-stdio-1.0 ghc-options: @@ -69,6 +68,7 @@ ConnectionSpec ExchangeDeclareSpec ExchangeDeleteSpec + FromURISpec QueueDeclareSpec QueueDeleteSpec QueuePurgeSpec @@ -78,15 +78,14 @@ , containers>=0.2 , bytestring>=0.9 , data-binary-ieee754>=0.4.2.1 + , data-default-class >= 0.1 , text>=0.11.2 , split>=0.2 , clock >= 0.4.0.1 , hspec >= 1.3 , hspec-expectations >= 0.3.3 - , connection == 0.2.* + , crypton-connection >= 0.4 && <= 0.5 , vector , stm >= 2.4.0 - if flag(network-uri) - build-depends: network-uri >= 2.6, network > 2.6 - else - build-depends: network < 2.6 + , network-uri >= 2.6 + , network > 2.6 diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,76 @@ +### Version 0.24.0 + +* the `fromURI` method now returns `Either String ConnectionOpts` to better handle parsing-errors + +### Version 0.23.0 + +* bump dependency bounds for `crypton-connection` + +### Version 0.22.2 + +* replace `connection` dependency with `crypton-connection` + +### Version 0.22.1 + +* fix a potential deadlock in `closeConnection` + +### Version 0.22.0 + +* `closeChannel` now uses a `ChannelClosedException` to kill the channel-thread instead of `ThreadKilled`. This fixes a bug in 0.21.0 that led to errors being printed on the screen when `closeChannel` was called. + +### Version 0.21.0 + +* added method `isNormalChannelClose` +* Channels that have no explicit exception handlers (added using `addChannelExceptionHandler`) will now print all exceptions to `stderr`, even if those are not of type `ChannelClosedException`. This is an uncommon situation, so it should not have an observable effect for most users. + +### Version 0.20.0 + +* `fromURI` now activates TLS if the URI starts with `ampqs://`. Previously it only changed the port, without activating TLS. +* `fromURI` now supports multi-broker URIs in the form of `amqp://user:pass@host-1:port-2,host-2:port-2/vhost` + +### Version 0.19.1 + +* add `nackMsg` and `nackEnv` methods + +### Version 0.19.0 + +* change `FVString` to be binary instead of UTF-8 encoded + +### Version 0.18.3 + +* respond to `channel.close` messages from the server with `channel.close_ok` + +### Version 0.18.2 + +* support `connection` 0.3 and drop support for `network` < 2.6 + +### Version 0.18.1 + +* new function `addConnectionBlockedHandler` to be notified when the connection is blocked (due to the server being resource-constrained) + +### Version 0.18.0 + +* `ConnectionClosedException` and `ChannelClosedException` now specify whether the close was normal (user-initiated) or abnormal (caused by an AMQP exception) +* channels that are abnormally closed and have no exception handler (set using `addChannelExceptionHandler`) will now print the error to `stderr` +* new function `getServerProperties` to get the RabbitMQ server-properties +* `consumeMsgs'` now allows setting a consumer-cancellation-callback + +### Version 0.17.0 + +* When the server asynchronously closed a channel, this was (erroneously) internally represented as a `ConnectionClosedException`. It is now represented as a `ChannelClosedException`. This could affect you if you explicitly match on `ConnectionClosedException` or `ChannelClosedException` in your code, for example when using `addChannelExceptionHandler`. + +### Version 0.16.0 + +* new `coName` field in `ConnectionOpts` to specify a custom name that will be displayed in the RabbitMQ web interface + +### Version 0.15.1 + +* export the `AckType` data-type and constructors + +### Version 0.15.0 + +* The way channels are closed internally was changed. This may affect you if you have installed an exception handler inside the callback passed to `consumeMsgs`. Specifically, the exceptions used internally to close channels are now wrapped inside `ChanThreadKilledException`. You should make sure to re-throw this exception if you did catch it. + ### Version 0.14.1 * show all exceptions if no host can be connected to diff --git a/examples/ExampleConsumer.hs b/examples/ExampleConsumer.hs --- a/examples/ExampleConsumer.hs +++ b/examples/ExampleConsumer.hs @@ -1,4 +1,6 @@ {-# OPTIONS -XOverloadedStrings #-} +module ExampleConsumer where + import Network.AMQP import qualified Data.ByteString.Lazy.Char8 as BL diff --git a/examples/ExampleProducer.hs b/examples/ExampleProducer.hs --- a/examples/ExampleProducer.hs +++ b/examples/ExampleProducer.hs @@ -1,4 +1,6 @@ {-# OPTIONS -XOverloadedStrings #-} +module ExampleProducer where + import Network.AMQP import qualified Data.ByteString.Lazy.Char8 as BL diff --git a/test/BasicPublishSpec.hs b/test/BasicPublishSpec.hs --- a/test/BasicPublishSpec.hs +++ b/test/BasicPublishSpec.hs @@ -6,8 +6,12 @@ import Network.AMQP import Data.ByteString.Lazy.Char8 as BL +import Data.Map (Map) +import qualified Data.Map as Map +import Data.Word import Control.Concurrent (threadDelay) +import Control.Concurrent.STM main :: IO () main = hspec spec @@ -60,3 +64,66 @@ _ <- deleteQueue ch q closeConnection conn + context "confirmSelect" $ do + it "receives a confirmation message" $ do + let q = "haskell-amqp.queues.publish-over-fanout1" + e = "haskell-amqp.fanout.d.na" + conn <- openConnection "127.0.0.1" "/" "guest" "guest" + ch <- openChannel conn + confirmSelect ch True + (confirmMap, counter) <- atomically $ (,) <$> newTVar Map.empty <*> newTVar 0 + addConfirmationListener ch (handleConfirms counter confirmMap) + _ <- declareExchange ch (newExchange {exchangeName = e, + exchangeType = "fanout", + exchangeDurable = True}) + + (_, _, _) <- declareQueue ch (newQueue {queueName = q, queueDurable = False}) + _ <- purgeQueue ch q + bindQueue ch q e "" + + + _ <- traverse (\n -> do + sn' <- publishMsg ch e "" (newMsg {msgBody = (BL.pack "hello")}) + case sn' of + Just sn -> atomically $ addSequenceNumber confirmMap (fromIntegral sn) n + Nothing -> return () + + ) [1..5] + + + threadDelay (1000 * 100) + + (_, n, _) <- declareQueue ch (newQueue {queueName = q, queueDurable = False}) + n `shouldBe` 5 + + cMap' <- atomically $ readTVar confirmMap + cMap' `shouldBe` Map.empty + + counter' <- atomically $ readTVar counter + counter' `shouldBe` 5 + + _ <- deleteQueue ch q + closeConnection conn + + +addSequenceNumber :: TVar (Map Word64 Int) -> Word64 -> Int -> STM () +addSequenceNumber cMap sn n = modifyTVar' cMap (Map.insert sn n) + +removeSequenceNumber :: TVar (Map Word64 Int) -> Word64 -> STM () +removeSequenceNumber cMap sn = modifyTVar' cMap (Map.delete sn) + +increaseCounter :: TVar Int -> STM () +increaseCounter n = modifyTVar' n (+1) + +handleConfirms :: TVar Int -> TVar (Map Word64 Int) -> (Word64, Bool, AckType) -> IO () +handleConfirms c _ (_, False, BasicNack) = atomically $ increaseCounter c +handleConfirms c _ (_, True, BasicNack) = atomically $ increaseCounter c +handleConfirms c cMap (n, False, BasicAck) = atomically $ removeSequenceNumber cMap n >> increaseCounter c +handleConfirms c cMap (n, True, BasicAck) = atomically $ do + cMap' <- readTVar cMap + let (lt, eq', _) = Map.splitLookup n cMap' + case eq' of + Just _ -> removeSequenceNumber cMap n >> increaseCounter c + Nothing -> return () + _ <- traverse (\i -> removeSequenceNumber cMap i >> increaseCounter c) (Map.keys lt) + return () diff --git a/test/ConnectionSpec.hs b/test/ConnectionSpec.hs --- a/test/ConnectionSpec.hs +++ b/test/ConnectionSpec.hs @@ -26,7 +26,7 @@ context "with custom vhost and valid credentials" $ do it "raises an exception" $ do - let ex = ConnectionClosedException "Handshake failed. Please check the RabbitMQ logs for more information" + let ex = ConnectionClosedException Abnormal "Handshake failed. Please check the RabbitMQ logs for more information" (openConnection "127.0.0.1" "haskell_amqp_testbed" "NxqrbLaNiN3TAenNu:r9Pq]XwABuRs" "RyRxfVDyrKjC8yhJ6htCp}P>FnJxfc") `shouldThrow` (== ex) diff --git a/test/ExchangeDeclareSpec.hs b/test/ExchangeDeclareSpec.hs --- a/test/ExchangeDeclareSpec.hs +++ b/test/ExchangeDeclareSpec.hs @@ -50,8 +50,11 @@ conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () + let x = "haskell-amqp.exchanges.Xiz2mQozyYcFrQgGmN8r" - ex = ChannelClosedException "NOT_FOUND - no exchange 'haskell-amqp.exchanges.Xiz2mQozyYcFrQgGmN8r' in vhost '/'" + ex = ChannelClosedException Abnormal "NOT_FOUND - no exchange 'haskell-amqp.exchanges.Xiz2mQozyYcFrQgGmN8r' in vhost '/'" (declareExchange ch $ newExchange {exchangeName = x, exchangePassive = True}) `shouldThrow` (== ex) closeConnection conn diff --git a/test/ExchangeDeleteSpec.hs b/test/ExchangeDeleteSpec.hs --- a/test/ExchangeDeleteSpec.hs +++ b/test/ExchangeDeleteSpec.hs @@ -17,14 +17,17 @@ conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () + _ <- declareExchange ch (newExchange {exchangeName = eName, exchangeType = "topic", exchangeDurable = False}) _ <- deleteExchange ch eName - let ex = ChannelClosedException "NOT_FOUND - no exchange 'haskell-amqp.exchanges.to-be-deleted' in vhost '/'" + let ex = ChannelClosedException Abnormal "NOT_FOUND - no exchange 'haskell-amqp.exchanges.to-be-deleted' in vhost '/'" (declareExchange ch $ newExchange {exchangeName = eName, exchangePassive = True}) `shouldThrow` (== ex) closeConnection conn @@ -33,9 +36,11 @@ it "throws an exception" $ do conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () let q = "haskell-amqp.exchanges.GmN8rozyXiz2mQYcFrQg" - ex = ChannelClosedException "NOT_FOUND - no exchange 'haskell-amqp.exchanges.GmN8rozyXiz2mQYcFrQg' in vhost '/'" + ex = ChannelClosedException Abnormal "NOT_FOUND - no exchange 'haskell-amqp.exchanges.GmN8rozyXiz2mQYcFrQg' in vhost '/'" (declareExchange ch $ newExchange {exchangeName = q, exchangePassive = True}) `shouldThrow` (== ex) closeConnection conn diff --git a/test/FromURISpec.hs b/test/FromURISpec.hs new file mode 100644 --- /dev/null +++ b/test/FromURISpec.hs @@ -0,0 +1,190 @@ +{-# OPTIONS -XOverloadedStrings #-} + + +module FromURISpec (main, spec) where + + +import Network.AMQP +import Test.Hspec + + +main :: IO () +main = hspec spec + +spec :: Spec +spec = do + describe "fromURI" $ do + it "empty" $ do + let Right o = fromURI "" + coServers o `shouldBe` [("localhost", 5672)] + coVHost o `shouldBe` "/" + -- avoid undefined SASLMechanism Show instance + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULguest\NULguest"] + -- avoid undefined TLSSettings Show instance + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1" + coServers o `shouldBe` [("127.0.0.1", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host-2" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1,127.0.0.2" + coServers o `shouldBe` [("127.0.0.1", 5672),("127.0.0.2", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host port" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1:5672" + coServers o `shouldBe` [("127.0.0.1", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host-2 port" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1,127.0.0.2:5672" + coServers o `shouldBe` [("127.0.0.1", 5672),("127.0.0.2", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host-2 port-2" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1:5672,127.0.0.2:5672" + coServers o `shouldBe` [("127.0.0.1", 5672),("127.0.0.2", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host port vhost" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1:5672/v" + coServers o `shouldBe` [("127.0.0.1", 5672)] + coVHost o `shouldBe` "v" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host-2 port-2 vhost" $ do + let Right o = fromURI "amqp://u:p@127.0.0.1:5673,127.0.0.2:5674/v" + coServers o `shouldBe` [("127.0.0.1", 5673),("127.0.0.2", 5674)] + coVHost o `shouldBe` "v" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp auth host-3 port-3 vhost" $ do + let Right o = fromURI "amqp://a:b@h1:8001,h2:8002,h3:8003/w" + coServers o `shouldBe` [("h1", 8001),("h2", 8002),("h3", 8003)] + coVHost o `shouldBe` "w" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULa\NULb"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp host" $ do + let Right o = fromURI "amqp://127.0.0.1" + -- this appears to break: user host + coServers o `shouldBe` [("localhost", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` + ["\NUL127.0.0.1\NULguest"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqp host-2" $ do + let Right o = fromURI "amqp://127.0.0.1,127.0.0.2" + -- this appears to break: user host + coServers o `shouldBe` [("localhost", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` + ["\NUL127.0.0.1,127.0.0.2\NULguest"] + (isTrusted $ coTLSSettings o) `shouldBe` False + + it "amqps auth host" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1" + coServers o `shouldBe` [("127.0.0.1", 5671)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps auth host-2" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1,127.0.0.2" + coServers o `shouldBe` [("127.0.0.1", 5671),("127.0.0.2", 5671)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps auth host port" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1:5672" + coServers o `shouldBe` [("127.0.0.1", 5672)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps auth host-2 port" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1,127.0.0.1:5673" + coServers o `shouldBe` [("127.0.0.1", 5671),("127.0.0.1", 5673)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps auth host-2 port-2" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1:5672,127.0.0.1:5673" + coServers o `shouldBe` [("127.0.0.1", 5672),("127.0.0.1", 5673)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps auth host port vhost" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1:5672/v" + coServers o `shouldBe` [("127.0.0.1", 5672)] + coVHost o `shouldBe` "v" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps auth host-2 port-2 vhost" $ do + let Right o = fromURI "amqps://u:p@127.0.0.1:5672,127.0.0.2:5673/v" + coServers o `shouldBe` [("127.0.0.1", 5672),("127.0.0.2", 5673)] + coVHost o `shouldBe` "v" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` ["\NULu\NULp"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps host" $ do + let Right o = fromURI "amqps://127.0.0.1" + -- this appears to break: user host + coServers o `shouldBe` [("localhost", 5671)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` + ["\NUL127.0.0.1\NULguest"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + it "amqps host-2" $ do + let Right o = fromURI "amqps://127.0.0.1,127.0.0.2" + -- this appears to break: user host + coServers o `shouldBe` [("localhost", 5671)] + coVHost o `shouldBe` "/" + (saslName <$> coAuth o) `shouldBe` ["PLAIN"] + (saslInitialResponse <$> coAuth o) `shouldBe` + ["\NUL127.0.0.1,127.0.0.2\NULguest"] + (isTrusted $ coTLSSettings o) `shouldBe` True + + +isTrusted :: Maybe TLSSettings -> Bool +isTrusted (Just TLSTrusted) = True +isTrusted _ = False diff --git a/test/QueueDeclareSpec.hs b/test/QueueDeclareSpec.hs --- a/test/QueueDeclareSpec.hs +++ b/test/QueueDeclareSpec.hs @@ -48,7 +48,7 @@ -- consumer count, undelivered message count cn `shouldBe` 0 mn `shouldBe` 0 - + closeConnection conn context "passive declaration when the queue DOES NOT exist" $ do @@ -56,8 +56,11 @@ conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () + let x = "haskell-amqp.queues.mQozyYcFrQgGmN8Xiz2r" - ex = ChannelClosedException "NOT_FOUND - no queue 'haskell-amqp.queues.mQozyYcFrQgGmN8Xiz2r' in vhost '/'" + ex = ChannelClosedException Abnormal "NOT_FOUND - no queue 'haskell-amqp.queues.mQozyYcFrQgGmN8Xiz2r' in vhost '/'" (declareQueue ch $ newQueue {queueName = x, queuePassive = True}) `shouldThrow` (== ex) closeConnection conn diff --git a/test/QueueDeleteSpec.hs b/test/QueueDeleteSpec.hs --- a/test/QueueDeleteSpec.hs +++ b/test/QueueDeleteSpec.hs @@ -15,6 +15,8 @@ it "deletes the queue" $ do conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () (q, _, _) <- declareQueue ch (newQueue {queueName = "haskell-amqp.queues.to-be-deleted", queueExclusive = True}) @@ -23,7 +25,7 @@ n <- deleteQueue ch q n `shouldBe` 0 - let ex = (ChannelClosedException "NOT_FOUND - no queue 'haskell-amqp.queues.to-be-deleted' in vhost '/'") + let ex = (ChannelClosedException Abnormal "NOT_FOUND - no queue 'haskell-amqp.queues.to-be-deleted' in vhost '/'") (declareQueue ch $ newQueue {queueName = q, queuePassive = True}) `shouldThrow` (== ex) closeConnection conn @@ -32,9 +34,11 @@ it "throws an exception" $ do conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () let q = "haskell-amqp.queues.GmN8rozyXiz2mQYcFrQg" - ex = ChannelClosedException "NOT_FOUND - no queue 'haskell-amqp.queues.GmN8rozyXiz2mQYcFrQg' in vhost '/'" + ex = ChannelClosedException Abnormal "NOT_FOUND - no queue 'haskell-amqp.queues.GmN8rozyXiz2mQYcFrQg' in vhost '/'" (declareQueue ch $ newQueue {queueName = q, queuePassive = True}) `shouldThrow` (== ex) closeConnection conn diff --git a/test/QueuePurgeSpec.hs b/test/QueuePurgeSpec.hs --- a/test/QueuePurgeSpec.hs +++ b/test/QueuePurgeSpec.hs @@ -43,8 +43,10 @@ it "empties the queue" $ do conn <- openConnection "127.0.0.1" "/" "guest" "guest" ch <- openChannel conn + -- silence error messages + addChannelExceptionHandler ch $ return . const () - let ex = ChannelClosedException "NOT_FOUND - no queue 'haskell-amqp.queues.avjqmyG{CHrc66MRyzYVA+PwrMVARJ' in vhost '/'" + let ex = ChannelClosedException Abnormal "NOT_FOUND - no queue 'haskell-amqp.queues.avjqmyG{CHrc66MRyzYVA+PwrMVARJ' in vhost '/'" (purgeQueue ch "haskell-amqp.queues.avjqmyG{CHrc66MRyzYVA+PwrMVARJ") `shouldThrow` (== ex) closeConnection conn diff --git a/test/Runner.hs b/test/Runner.hs --- a/test/Runner.hs +++ b/test/Runner.hs @@ -2,6 +2,7 @@ import qualified ConnectionSpec import qualified ChannelSpec +import qualified FromURISpec import qualified QueueDeclareSpec import qualified QueueDeleteSpec import qualified QueuePurgeSpec @@ -21,6 +22,8 @@ describe "ExchangeDeclareSpec" ExchangeDeclareSpec.spec -- exchange.delete describe "ExchangeDeleteSpec" ExchangeDeleteSpec.spec + -- fromuri.* + describe "FromURISpec" FromURISpec.spec -- queue.declare describe "QueueDeclareSpec" QueueDeclareSpec.spec -- queue.delete