amqp 0.17.0 → 0.18.0

raw patch · 12 files changed

+123/−54 lines, 12 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Network/AMQP.hs +8/−8
• Network/AMQP/Internal.hs +68/−31
• Network/AMQP/Lifted.hs +4/−3
• Network/AMQP/Types.hs +9/−2
• amqp.cabal +1/−1
• changelog.md +7/−0
• test/ConnectionSpec.hs +1/−1
• test/ExchangeDeclareSpec.hs +4/−1
• test/ExchangeDeleteSpec.hs +7/−2
• test/QueueDeclareSpec.hs +5/−2
• test/QueueDeleteSpec.hs +6/−2
• test/QueuePurgeSpec.hs +3/−1

@@ -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

@@ -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

@@ -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

@@ -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)

@@ -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

@@ -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`.

@@ -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)

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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