diff --git a/Network/AMQP.hs b/Network/AMQP.hs --- a/Network/AMQP.hs +++ b/Network/AMQP.hs @@ -65,6 +65,7 @@ closeChannel, closeConnection, addConnectionClosedHandler, + getServerProperties, -- * Channel Channel, @@ -128,7 +129,7 @@ addConfirmationListener, ConfirmationResult(..), AckType(..), - + -- * Flow Control flow, @@ -141,6 +142,7 @@ -- * Exceptions AMQPException(..), ChanThreadKilledException, + CloseType(..), -- * URI parsing fromURI @@ -351,8 +353,6 @@ newMsg :: Message newMsg = Message (BL.empty) Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing -type ConsumerTag = Text - -- | 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 deriving (Eq, Ord, Read, Show) @@ -369,16 +369,16 @@ -- 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'. 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) --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 diff --git a/Network/AMQP/Internal.hs b/Network/AMQP/Internal.hs --- a/Network/AMQP/Internal.hs +++ b/Network/AMQP/Internal.hs @@ -145,12 +145,13 @@ connChanAllocator :: ChannelAllocator, connChannels :: MVar (IM.IntMap (Channel, ThreadId)), -- open channels (channelID => (Channel, ChannelThread)) connMaxFrameSize :: Int, --negotiated maximum frame size - connClosed :: MVar (Maybe String), + 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 ()], 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 + connLastSent :: MVar Int64, -- the timestamp from a monotonic clock when the last frame was written + connServerProperties :: FieldTable -- the server properties sent in Connection_start } -- | Represents the parameters to connect to a broker or a cluster of brokers. @@ -194,16 +195,16 @@ 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 payload = hPutStrLn stderr $ "Got unexpected msg on channel zero: " ++ show payload forwardToChannel chanID payload = do @@ -217,7 +218,7 @@ 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 @@ -227,7 +228,7 @@ ]) -- 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 @@ -252,7 +253,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 @@ -264,7 +265,7 @@ cClosedHandlers <- newMVar [] cLastReceived <- getTimestamp >>= newMVar cLastSent <- getTimestamp >>= newMVar - let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cLastReceived cLastSent + let conn = Connection handle cChanAllocator cChannels (fromIntegral maxFrameSize) cClosed ccl writeLock cClosedHandlers cLastReceived cLastSent serverProps --spawn the connectionReceiver connThread <- forkFinally' (connectionReceiver conn) $ \res -> do @@ -272,7 +273,7 @@ CE.catch (Conn.connectionClose handle) (\(_ :: CE.SomeException) -> return ()) -- mark as closed - modifyMVar_ cClosed $ return . Just . maybe "unknown reason" id + 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 @@ -313,7 +314,7 @@ 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)) + 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 @@ -329,8 +330,11 @@ (ShortString "en_US")) )) where clientProperties = FieldTable $ M.fromList $ [ ("platform", FVString "Haskell") , ("version" , FVString . T.pack $ showVersion version) + , ("capabilities", FVFieldTable clientCapabilities) ] ++ maybe [] (\x -> [("connection_name", FVString x)]) (coName connOpts) + clientCapabilities = FieldTable $ M.fromList $ [ ("consumer_cancel_notify", FVBool True) ] + handleSecureUntilTune handle sasl = do tuneOrSecure <- readFrame handle case tuneOrSecure of @@ -351,7 +355,7 @@ 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 @@ -367,10 +371,10 @@ 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) + (killConnection conn Abnormal (CE.toException skippedBeatEx) connThread) checkSendTimeout = doCheck (connLastSent conn) sendTimeout (writeFrame (connHandle conn) (Frame 0 HeartbeatPayload)) @@ -387,9 +391,9 @@ 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) +killConnection :: Connection -> CloseType -> CE.SomeException -> ThreadId -> IO () +killConnection conn closeType ex connThread = do + modifyMVar_ (connClosed conn) $ const $ return $ Just (closeType, show ex) throwTo connThread ex -- | closes a connection @@ -415,6 +419,10 @@ 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 @@ -481,8 +489,9 @@ 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) + 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 ()] @@ -541,6 +550,7 @@ isResponse (SimpleMethod (Channel_close _ _ _ _)) = False isResponse (SimpleMethod (Basic_ack _ _)) = False isResponse (SimpleMethod (Basic_nack _ _ _)) = False + isResponse (SimpleMethod (Basic_cancel _ _)) = False isResponse _ = True --Basic.Deliver: forward msg to registered consumer @@ -549,7 +559,7 @@ 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} @@ -564,8 +574,8 @@ return () ) handleAsync (SimpleMethod (Channel_close _ (ShortString errorMsg) _ _)) = do - closeChannel' chan errorMsg - myThreadId >>= flip CE.throwTo (ChannelClosedException . T.unpack $ errorMsg) + 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 @@ -581,6 +591,7 @@ 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 @@ -604,6 +615,17 @@ 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 @@ -625,8 +647,8 @@ modifyMVar_ (chanExceptionHandlers chan) $ \handlers -> return $ handler:handlers -- 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 @@ -637,7 +659,7 @@ 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 () @@ -647,6 +669,8 @@ 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'). openChannel :: Connection -> IO Channel openChannel c = do newInQueue <- newChan @@ -668,16 +692,29 @@ 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" + closeChannel' newChannel Normal "closed" case res of Right _ -> return () - Left ex -> readMVar handlers >>= mapM_ ($ unwrapChanThreadKilledException ex) + 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 ""))) return newChannel + where + fromAbnormalChannelClose :: CE.SomeException -> Maybe String + fromAbnormalChannelClose exc = + case CE.fromException exc :: Maybe AMQPException of + Just (ChannelClosedException Abnormal reason) -> Just reason + _ -> Nothing + -- | 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 @@ -769,12 +806,12 @@ 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 diff --git a/Network/AMQP/Lifted.hs b/Network/AMQP/Lifted.hs --- a/Network/AMQP/Lifted.hs +++ b/Network/AMQP/Lifted.hs @@ -29,14 +29,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/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 + 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 @@ -74,6 +79,8 @@ type ShortInt = Word16 type LongInt = Word32 type LongLongInt = Word64 + +type ConsumerTag = Text newtype ShortString = ShortString Text deriving (Eq, Ord, Read, Show) diff --git a/amqp.cabal b/amqp.cabal --- a/amqp.cabal +++ b/amqp.cabal @@ -1,5 +1,5 @@ Name: amqp -Version: 0.17.0 +Version: 0.18.0 Synopsis: Client library for AMQP servers (currently only RabbitMQ) Description: Client library for AMQP servers (currently only RabbitMQ) License: BSD3 diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,10 @@ +### 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`. 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/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